Coverage for tests/test_client.py: 100%

1import warnings 

2from typing import TYPE_CHECKING, Any, Mapping 

3from pathlib import Path 

4from datetime import timedelta 

5 

6import httpx 

7import pytest 

8from mock import AsyncMock 

9from pytest_mock import MockerFixture 

10 

11from prisma import ENGINE_TYPE, SCHEMA_PATH, Prisma, errors, get_client 

12from prisma.types import HttpConfig 

13from prisma.testing import reset_client 

14from prisma.cli.prisma import run 

15from prisma.engine.http import HTTPEngine 

16from prisma.engine.errors import AlreadyConnectedError 

17from prisma.http_abstract import DEFAULT_CONFIG 

18 

19from .utils import Testdir, patch_method 

20 

21if TYPE_CHECKING: 

22 from _pytest.monkeypatch import MonkeyPatch 

23 

24 

25@pytest.mark.asyncio 

26async def test_catches_not_connected() -> None: 

27 """Trying to make a query before connecting raises an error""" 

28 client = Prisma() 

29 with pytest.raises(errors.ClientNotConnectedError) as exc: 

30 await client.post.delete_many() 

31 

32 assert 'connect()' in str(exc) 

33 

34 

35@pytest.mark.asyncio 

36async def test_create_many_skip_duplicates_invalid_provider(client: Prisma) -> None: 

37 """Trying to call skip_duplicates fails as SQLite does not support it""" 

38 with pytest.raises(errors.UnsupportedDatabaseError) as exc: 

39 await client.user.create_many([{'name': 'Robert'}], skip_duplicates=True) 

40 

41 assert exc.match(r'skip_duplicates is not supported by sqlite') 

42 

43 

44@pytest.mark.asyncio 

45async def test_datasource_overwriting(testdir: Testdir, client: Prisma) -> None: 

46 """Ensure the client can connect and query to a custom datasource""" 

47 # we have to do this messing with the schema so that we can run db push on the new database 

48 schema = Path(__file__).parent / 'data' / 'schema.prisma' 

49 testdir.path.joinpath('schema.prisma').write_text(schema.read_text().replace('"file:dev.db"', 'env("_PY_DB")')) 

50 run(['db', 'push', '--skip-generate'], env={'_PY_DB': 'file:./tmp.db'}) 

51 

52 other = Prisma( 

53 datasource={'url': 'file:./tmp.db'}, 

54 ) 

55 await other.connect(timeout=timedelta(seconds=1)) 

56 

57 user = await other.user.create({'name': 'Robert'}) 

58 assert user.name == 'Robert' 

59 

60 assert await client.user.count() == 0 

61 

62 

63@pytest.mark.asyncio 

64async def test_context_manager() -> None: 

65 """Client can be used as a context manager to connect and disconnect from the database""" 

66 client = Prisma() 

67 assert not client.is_connected() 

68 

69 async with client: 

70 assert client.is_connected() 

71 

72 assert not client.is_connected() 

73 

74 # ensure exceptions are propagated 

75 with pytest.raises(AlreadyConnectedError): 

76 async with client: 

77 assert client.is_connected() 

78 await client.connect() 

79 

80 

81def test_auto_register() -> None: 

82 """Client(auto_register=True) correctly registers the client instance""" 

83 with reset_client(): 

84 with pytest.raises(errors.ClientNotRegisteredError): 

85 get_client() 

86 

87 client = Prisma(auto_register=True) 

88 assert get_client() == client 

89 

90 

91def test_engine_type() -> None: 

92 """The exported ENGINE_TYPE enum matches the actual engine type""" 

93 assert ENGINE_TYPE.value == 'binary' 

94 

95 

96@pytest.mark.asyncio 

97async def test_connect_timeout(mocker: MockerFixture) -> None: 

98 """Setting the timeout on a client and a per-call basis works""" 

99 client = Prisma(connect_timeout=timedelta(seconds=7)) 

100 mocked = mocker.patch.object( 

101 client._engine_class, 

102 'connect', 

103 new=AsyncMock(), 

104 ) 

105 

106 await client.connect() 

107 mocked.assert_called_once_with( 

108 timeout=timedelta(seconds=7), 

109 datasources=[client._make_sqlite_datasource()], 

110 ) 

111 mocked.reset_mock() 

112 

113 await client.connect(timeout=timedelta(seconds=5)) 

114 mocked.assert_called_once_with( 

115 timeout=timedelta(seconds=5), 

116 datasources=[client._make_sqlite_datasource()], 

117 ) 

118 

119 

120@pytest.mark.asyncio 

121async def test_custom_http_options(monkeypatch: 'MonkeyPatch') -> None: 

122 """Custom http options are passed to the HTTPX Client""" 

123 

124 def mock___init__(real__init__: Any, *args: Any, **kwargs: Any) -> None: 

125 # pass to real __init__ method to ensure types passed will actually work at runtime 

126 real__init__(*args, **kwargs) 

127 

128 getter = patch_method(monkeypatch, httpx.AsyncClient, '__init__', mock___init__) 

129 

130 def mock_app(args: Mapping[str, object], data: object) -> object: ... 

131 

132 async def _test(config: HttpConfig) -> None: 

133 client = Prisma( 

134 http=config, 

135 ) 

136 engine = client._create_engine() # pylint: disable=protected-access 

137 assert isinstance(engine, HTTPEngine) 

138 engine.session.open() 

139 await engine.session.close() 

140 

141 captured = getter() 

142 assert captured is not None 

143 assert captured == ((), {**DEFAULT_CONFIG, **config}) 

144 

145 await _test({'timeout': 1}) 

146 await _test({'timeout': httpx.Timeout(5, connect=10, read=30)}) 

147 await _test({'max_redirects': 1, 'trust_env': True}) 

148 await _test({'http1': True, 'http2': False}) 

149 await _test( 

150 config={ 

151 'timeout': 200, 

152 'http1': True, 

153 'http2': False, 

154 'limits': httpx.Limits(max_connections=10), 

155 'max_redirects': 2, 

156 'trust_env': False, 

157 }, 

158 ) 

159 

160 with warnings.catch_warnings(): 

161 warnings.simplefilter('ignore') 

162 await _test({'app': mock_app}) 

163 

164 

165def test_old_client_alias() -> None: 

166 """Ensure that Prisma can be imported from the root package under the Client alias""" 

167 from prisma import Client, Prisma 

168 

169 assert Client == Prisma 

170 

171 

172def test_sqlite_url(client: Prisma) -> None: 

173 """Ensure that the default overriden SQLite URL uses the correct relative path 

174 

175 https://github.com/RobertCraigie/prisma-client-py/issues/409 

176 """ 

177 rootdir = Path(__file__).parent 

178 

179 url = client._make_sqlite_url('file:dev.db') 

180 assert url == f'file:{SCHEMA_PATH.parent.joinpath("dev.db")}' 

181 

182 url = client._make_sqlite_url('file:dev.db', relative_to=rootdir) 

183 assert url == f'file:{rootdir.joinpath("dev.db")}' 

184 

185 url = client._make_sqlite_url('sqlite:../dev.db', relative_to=rootdir) 

186 assert url == f'file:{rootdir.parent.joinpath("dev.db")}' 

187 

188 # already absolute paths are not updated 

189 url = client._make_sqlite_url( 

190 f'sqlite:{rootdir.parent.joinpath("foo.db").absolute()}', 

191 relative_to=rootdir, 

192 ) 

193 assert url == f'sqlite:{rootdir.parent.joinpath("foo.db").absolute()}' 

194 

195 # unknown prefixes are not updated 

196 url = client._make_sqlite_url('unknown:dev.db', relative_to=rootdir) 

197 assert url == 'unknown:dev.db' 

198 

199 # prefixes being dropped without affecting file name 

200 url = client._make_sqlite_url('file:file.db') 

201 assert url == f'file:{SCHEMA_PATH.parent.joinpath("file.db")}' 

202 

203 url = client._make_sqlite_url('sqlite:sqlite.db') 

204 assert url == f'file:{SCHEMA_PATH.parent.joinpath("sqlite.db")}' 

205 

206 

207@pytest.mark.asyncio 

208async def test_copy() -> None: 

209 """The Prisma._copy() method forwards all relevant properties""" 

210 client1 = Prisma( 

211 log_queries=True, 

212 datasource={ 

213 'url': 'file:foo.db', 

214 }, 

215 connect_timeout=timedelta(seconds=15), 

216 http={ 

217 'trust_env': False, 

218 }, 

219 ) 

220 client2 = client1._copy() 

221 assert not client2.is_connected() is None 

222 assert client2._log_queries is True 

223 assert client2._datasource == {'url': 'file:foo.db'} 

224 assert client2._connect_timeout == timedelta(seconds=15) 

225 assert client2._http_config == {'trust_env': False} 

226 

227 await client1.connect() 

228 assert client1.is_connected() 

229 client3 = client1._copy() 

230 assert client3.is_connected() 

231 

232 

233@pytest.mark.asyncio 

234async def test_copied_client_does_not_close_engine(client: Prisma) -> None: 

235 """Deleting a Prisma._copy()'d client does not cause the engine to be stopped""" 

236 copied = client._copy() 

237 assert copied.is_connected() 

238 assert client.is_connected() 

239 

240 del copied 

241 

242 assert client.is_connected() 

243 await client.user.count() # ensure queries can still be executed 

244 

245 

246def test_is_registered(client: Prisma) -> None: 

247 """The Prisma.is_registered() method can be used both when the client is registered 

248 and when there is no client registered at all. 

249 """ 

250 assert client.is_registered() 

251 

252 other_client = Prisma() 

253 assert not other_client.is_registered() 

254 

255 with reset_client(): 

256 assert not client.is_registered() 

257 assert not other_client.is_registered()