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 = get_files_from_templates(BASE_PACKAGE_DIR / 'generator' / 'templates') 

81THIS_DIR = Path(__file__).parent 

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

83 

84 

85def path_replacer( 

86 schema_path: Path, 

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

88 def pathlib_matcher(data: object, path: object) -> Optional[object]: 

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

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

91 

92 data = data.replace( 

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

94 "Path('<absolute-schema-path>')", 

95 ) 

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

97 return data 

98 

99 return pathlib_matcher 

100 

101 

102# TODO: support running snapshot tests on windows 

103 

104 

105@skipif_windows 

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

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

108 """Ensure synchronous client files match""" 

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

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

111 ) 

112 

113 

114@skipif_windows 

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

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

117 """Ensure asynchronous client files match""" 

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

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

120 ) 

121 

122 

123def test_sync_client_can_be_imported() -> None: 

124 """Synchronous client can be imported""" 

125 proc = subprocess.run( 

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

127 cwd=str(SYNC_ROOTDIR.parent), 

128 check=True, 

129 stdout=subprocess.PIPE, 

130 ) 

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

132 

133 

134def test_async_client_can_be_imported() -> None: 

135 """Asynchronous client can be imported""" 

136 proc = subprocess.run( 

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

138 cwd=str(ASYNC_ROOTDIR.parent), 

139 check=True, 

140 stdout=subprocess.PIPE, 

141 ) 

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