Coverage for tests/test_engine.py: 100%

1import signal 

2import asyncio 

3import contextlib 

4from typing import Iterator, Optional 

5from pathlib import Path 

6 

7import pytest 

8from pytest_subprocess import FakeProcess 

9from _pytest.monkeypatch import MonkeyPatch 

10 

11from prisma import BINARY_PATHS, Prisma, config 

12from prisma.utils import temp_env_update 

13from prisma.engine import utils, errors 

14from prisma._compat import get_running_loop 

15from prisma.binaries import platform 

16from prisma.engine.query import QueryEngine 

17 

18from .utils import Testdir, skipif_windows 

19 

20 

21@contextlib.contextmanager 

22def no_event_loop() -> Iterator[None]: 

23 try: 

24 current: Optional[asyncio.AbstractEventLoop] = get_running_loop() 

25 except RuntimeError: 

26 current = None 

27 

28 # if there is no running loop then we don't touch the event loop 

29 # as this can cause weird issues breaking other tests 

30 if not current: # pragma: no cover 

31 yield 

32 else: # pragma: no cover 

33 try: 

34 asyncio.set_event_loop(None) 

35 yield 

36 finally: 

37 asyncio.set_event_loop(current) 

38 

39 

40@pytest.mark.asyncio 

41async def test_engine_connects() -> None: 

42 """Can connect to engine""" 

43 db = Prisma() 

44 await db.connect() 

45 

46 with pytest.raises(errors.AlreadyConnectedError): 

47 await db.connect() 

48 

49 await db.disconnect() 

50 

51 

52@pytest.mark.asyncio 

53@skipif_windows 

54async def test_engine_process_sigint_mask() -> None: 

55 """Block SIGINT in current process""" 

56 signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGINT]) 

57 db = Prisma() 

58 await db.connect() 

59 

60 with pytest.raises(errors.AlreadyConnectedError): 

61 await db.connect() 

62 

63 await asyncio.wait_for(db.disconnect(), timeout=5) 

64 signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGINT]) 

65 

66 

67@pytest.mark.asyncio 

68@skipif_windows 

69async def test_engine_process_sigterm_mask() -> None: 

70 """Block SIGTERM in current process""" 

71 signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGTERM]) 

72 db = Prisma() 

73 await db.connect() 

74 

75 with pytest.raises(errors.AlreadyConnectedError): 

76 await db.connect() 

77 

78 await asyncio.wait_for(db.disconnect(), timeout=5) 

79 signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGTERM]) 

80 

81 

82def test_stopping_engine_on_closed_loop() -> None: 

83 """Stopping the engine with no event loop available does not raise an error""" 

84 with no_event_loop(): 

85 engine = QueryEngine(dml_path=Path.cwd()) 

86 engine.stop() 

87 

88 

89def test_engine_binary_does_not_exist(monkeypatch: MonkeyPatch) -> None: 

90 """No query engine binary found raises an error""" 

91 

92 def mock_exists(path: Path) -> bool: 

93 return False 

94 

95 monkeypatch.setattr(Path, 'exists', mock_exists, raising=True) 

96 

97 with pytest.raises(errors.BinaryNotFoundError) as exc: 

98 utils.ensure(BINARY_PATHS.query_engine) 

99 

100 assert exc.match( 

101 r'Expected .*, .* or .* to exist but none were found or could not be executed\.\nTry running prisma py fetch' 

102 ) 

103 

104 

105def test_engine_binary_does_not_exist_no_binary_paths( 

106 monkeypatch: MonkeyPatch, 

107) -> None: 

108 """No query engine binary found raises an error""" 

109 

110 def mock_exists(path: Path) -> bool: 

111 return False 

112 

113 monkeypatch.setattr(Path, 'exists', mock_exists, raising=True) 

114 

115 with pytest.raises(errors.BinaryNotFoundError) as exc: 

116 utils.ensure({}) 

117 

118 assert exc.match( 

119 r'Expected .* or .* to exist but neither were found or could not be executed\.\nTry running prisma py fetch' 

120 ) 

121 

122 

123def test_mismatched_version_error(fake_process: FakeProcess) -> None: 

124 """Mismatched query engine versions raises an error""" 

125 

126 fake_process.register_subprocess( 

127 [ 

128 str(utils._resolve_from_binary_paths(BINARY_PATHS.query_engine)), 

129 '--version', 

130 ], 

131 stdout='query-engine unexpected-hash', 

132 ) 

133 

134 with pytest.raises(errors.MismatchedVersionsError) as exc: 

135 utils.ensure(BINARY_PATHS.query_engine) 

136 

137 assert exc.match(f'Expected query engine version `{config.expected_engine_version}` but got `unexpected-hash`') 

138 

139 

140def test_ensure_local_path(testdir: Testdir, fake_process: FakeProcess) -> None: 

141 """Query engine in current directory required to be the expected version""" 

142 

143 fake_engine = testdir.path / platform.check_for_extension(f'prisma-query-engine-{platform.binary_platform()}') 

144 fake_engine.touch() 

145 

146 fake_process.register_subprocess( 

147 [str(fake_engine), '--version'], 

148 stdout='query-engine a-different-hash', 

149 ) 

150 with pytest.raises(errors.MismatchedVersionsError): 

151 path = utils.ensure(BINARY_PATHS.query_engine) 

152 

153 fake_process.register_subprocess( 

154 [str(fake_engine), '--version'], 

155 stdout=f'query-engine {config.expected_engine_version}', 

156 ) 

157 path = utils.ensure(BINARY_PATHS.query_engine) 

158 assert path == fake_engine 

159 

160 

161def test_ensure_env_override(testdir: Testdir, fake_process: FakeProcess) -> None: 

162 """Query engine path in environment variable can be any version""" 

163 fake_engine = testdir.path / 'my-query-engine' 

164 fake_engine.touch() 

165 

166 fake_process.register_subprocess( 

167 [str(fake_engine), '--version'], 

168 stdout='query-engine a-different-hash', 

169 ) 

170 

171 with temp_env_update({'PRISMA_QUERY_ENGINE_BINARY': str(fake_engine)}): 

172 path = utils.ensure(BINARY_PATHS.query_engine) 

173 

174 assert path == fake_engine 

175 

176 

177def test_ensure_env_override_does_not_exist() -> None: 

178 """Query engine path in environment variable not found raises an error""" 

179 with temp_env_update({'PRISMA_QUERY_ENGINE_BINARY': 'foo'}): 

180 with pytest.raises(errors.BinaryNotFoundError) as exc: 

181 utils.ensure(BINARY_PATHS.query_engine) 

182 

183 assert exc.match(r'PRISMA_QUERY_ENGINE_BINARY was provided, but no query engine was found at foo')