Coverage for tests/test_generation/exhaustive/test_exhaustive.py: 100%

1from __future__ import annotations 

2 

3import re 

4import sys 

5import subprocess 

6from typing import Any, List, Callable, Iterator, Optional 

7from pathlib import Path 

8from typing_extensions import override 

9 

10import pytest 

11from syrupy.assertion import SnapshotAssertion 

12from syrupy.extensions.single_file import SingleFileSnapshotExtension 

13from syrupy.extensions.amber.serializer import AmberDataSerializer 

14 

15from prisma.generator import BASE_PACKAGE_DIR 

16from prisma.generator.utils import remove_suffix 

17 

18from .utils import ROOTDIR 

19from ...utils import skipif_windows 

20 

21 

22class OSAgnosticSingleFileExtension(SingleFileSnapshotExtension): 

23 # syrupy's types are only written to target mypy, as such 

24 # pyright does not understand them and reports them as unknown. 

25 # As this method is only called internally it is safe to type as Any 

26 @override 

27 def serialize( 

28 self, 

29 data: Any, 

30 *, 

31 include: Optional[Any] = None, 

32 exclude: Optional[Any] = None, 

33 matcher: Optional[Any] = None, 

34 ) -> bytes: 

35 serialized = AmberDataSerializer.serialize(data, exclude=exclude, include=include, matcher=matcher) 

36 return bytes(serialized, 'utf-8') 

37 

38 # we disable diffs as we don't really care what the diff is 

39 # we just care that there is a diff and it can take a very 

40 # long time for syrupy to calculate the diff 

41 # https://github.com/tophat/syrupy/issues/581 

42 @override 

43 def diff_snapshots(self, serialized_data: Any, snapshot_data: Any) -> str: 

44 return 'diff-is-disabled' # pragma: no cover 

45 

46 @override 

47 def diff_lines(self, serialized_data: Any, snapshot_data: Any) -> Iterator[str]: 

48 yield 'diff-is-disabled' # pragma: no cover 

49 

50 

51@pytest.fixture 

52def snapshot(snapshot: SnapshotAssertion) -> SnapshotAssertion: 

53 return snapshot.use_extension(OSAgnosticSingleFileExtension) 

54 

55 

56def _clean_line(proc: 'subprocess.CompletedProcess[bytes]') -> str: 

57 return proc.stdout.decode('utf-8').rstrip('\n').rstrip('\r') 

58 

59 

60def get_files_from_templates(directory: Path) -> List[str]: 

61 """Return a list of all auto-generated python modules""" 

62 files: List[str] = [] 

63 

64 for template in directory.iterdir(): 

65 if template.is_dir(): 

66 files.extend(get_files_from_templates(template)) 

67 elif template.name.endswith('.py.jinja') and not template.name.startswith('_'): 

68 if directory.name == 'templates': 

69 name = template.name 

70 else: 

71 name = str(template.relative_to(template.parent.parent)) 

72 

73 files.append(remove_suffix(name, '.jinja').replace('\\', '/')) 

74 

75 return files 

76 

77 

78SYNC_ROOTDIR = ROOTDIR / '__prisma_sync_output__' / 'prisma' 

79ASYNC_ROOTDIR = ROOTDIR / '__prisma_async_output__' / 'prisma' 

80FILES = [ 

81 *get_files_from_templates(BASE_PACKAGE_DIR / 'generator' / 'templates'), 

82 'schema.prisma', 

83] 

84THIS_DIR = Path(__file__).parent 

85BINARY_PATH_RE = re.compile(r'BINARY_PATHS = (.*)') 

86 

87 

88def path_replacer( 

89 schema_path: Path, 

90) -> Callable[[object, object], Optional[object]]: 

91 def path_str_matcher(data: object, path: object) -> Optional[object]: 

92 if not isinstance(data, str): # pragma: no cover 

93 raise RuntimeError(f'schema_path_matcher expected data to be a `str` but received {type(data)} instead.') 

94 

95 data = data.replace( 

96 f"'{schema_path.absolute().as_posix()}'", 

97 "'<absolute-schema-path>'", 

98 ) 

99 data = BINARY_PATH_RE.sub("BINARY_PATHS = '<binary-paths-removed>'", data) 

100 return data 

101 

102 return path_str_matcher 

103 

104 

105# TODO: support running snapshot tests on windows 

106 

107 

108@skipif_windows 

109@pytest.mark.parametrize('file', FILES) 

110def test_sync(snapshot: SnapshotAssertion, file: str) -> None: 

111 """Ensure synchronous client files match""" 

112 assert SYNC_ROOTDIR.joinpath(file).absolute().read_text() == snapshot( 

113 matcher=path_replacer(THIS_DIR / 'sync.schema.prisma') # type: ignore 

114 ) 

115 

116 

117@skipif_windows 

118@pytest.mark.parametrize('file', FILES) 

119def test_async(snapshot: SnapshotAssertion, file: str) -> None: 

120 """Ensure asynchronous client files match""" 

121 assert ASYNC_ROOTDIR.joinpath(file).absolute().read_text() == snapshot( 

122 matcher=path_replacer(THIS_DIR / 'async.schema.prisma') # type: ignore 

123 ) 

124 

125 

126def test_sync_client_can_be_imported() -> None: 

127 """Synchronous client can be imported""" 

128 proc = subprocess.run( 

129 [sys.executable, '-c', 'import prisma; print(prisma.__file__)'], 

130 cwd=str(SYNC_ROOTDIR.parent), 

131 check=True, 

132 stdout=subprocess.PIPE, 

133 ) 

134 assert _clean_line(proc) == str(SYNC_ROOTDIR / '__init__.py') 

135 

136 

137def test_async_client_can_be_imported() -> None: 

138 """Asynchronous client can be imported""" 

139 proc = subprocess.run( 

140 [sys.executable, '-c', 'import prisma; print(prisma.__file__)'], 

141 cwd=str(ASYNC_ROOTDIR.parent), 

142 check=True, 

143 stdout=subprocess.PIPE, 

144 ) 

145 assert _clean_line(proc) == str(ASYNC_ROOTDIR / '__init__.py')