codekingpro/portable-devtools
115k
1# testing/config.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9
10from __future__ import annotations
11
12from argparse import Namespace
13import collections
14import inspect
15import typing
16from typing import Any
17from typing import Callable
18from typing import Iterable
19from typing import NoReturn
20from typing import Optional
21from typing import Tuple
22from typing import TypeVar
23from typing import Union
24
25from . import mock
26from . import requirements as _requirements
27from .util import fail
28from .. import util
29
30# default requirements; this is replaced by plugin_base when pytest
31# is run
32requirements = _requirements.SuiteRequirements()
33
34db = None
35db_url = None
36db_opts = None
37file_config = None
38test_schema = None
39test_schema_2 = None
40any_async = False
41_current = None
42ident = "main"
43options: Namespace = None # type: ignore
44
45if typing.TYPE_CHECKING:
46 from .plugin.plugin_base import FixtureFunctions
47
48 _fixture_functions: FixtureFunctions
49else:
50
51 class _NullFixtureFunctions:
52 def _null_decorator(self):
53 def go(fn):
54 return fn
55
56 return go
57
58 def skip_test_exception(self, *arg, **kw):
59 return Exception()
60
61 @property
62 def add_to_marker(self):
63 return mock.Mock()
64
65 def mark_base_test_class(self):
66 return self._null_decorator()
67
68 def combinations(self, *arg_sets, **kw):
69 return self._null_decorator()
70
71 def param_ident(self, *parameters):
72 return self._null_decorator()
73
74 def fixture(self, *arg, **kw):
75 return self._null_decorator()
76
77 def get_current_test_name(self):
78 return None
79
80 def async_test(self, fn):
81 return fn
82
83 # default fixture functions; these are replaced by plugin_base when
84 # pytest runs
85 _fixture_functions = _NullFixtureFunctions()
86
87
88_FN = TypeVar("_FN", bound=Callable[..., Any])
89
90
91def combinations(
92 *comb: Union[Any, Tuple[Any, ...]],
93 argnames: Optional[str] = None,
94 id_: Optional[str] = None,
95 **kw: str,
96) -> Callable[[_FN], _FN]:
97 r"""Deliver multiple versions of a test based on positional combinations.
98
99 This is a facade over pytest.mark.parametrize.
100
101
102 :param \*comb: argument combinations. These are tuples that will be passed
103 positionally to the decorated function.
104
105 :param argnames: optional list of argument names. These are the names
106 of the arguments in the test function that correspond to the entries
107 in each argument tuple. pytest.mark.parametrize requires this, however
108 the combinations function will derive it automatically if not present
109 by using ``inspect.getfullargspec(fn).args[1:]``. Note this assumes the
110 first argument is "self" which is discarded.
111
112 :param id\_: optional id template. This is a string template that
113 describes how the "id" for each parameter set should be defined, if any.
114 The number of characters in the template should match the number of
115 entries in each argument tuple. Each character describes how the
116 corresponding entry in the argument tuple should be handled, as far as
117 whether or not it is included in the arguments passed to the function, as
118 well as if it is included in the tokens used to create the id of the
119 parameter set.
120
121 If omitted, the argument combinations are passed to parametrize as is. If
122 passed, each argument combination is turned into a pytest.param() object,
123 mapping the elements of the argument tuple to produce an id based on a
124 character value in the same position within the string template using the
125 following scheme:
126
127 .. sourcecode:: text
128
129 i - the given argument is a string that is part of the id only, don't
130 pass it as an argument
131
132 n - the given argument should be passed and it should be added to the
133 id by calling the .__name__ attribute
134
135 r - the given argument should be passed and it should be added to the
136 id by calling repr()
137
138 s - the given argument should be passed and it should be added to the
139 id by calling str()
140
141 a - (argument) the given argument should be passed and it should not
142 be used to generated the id
143
144 e.g.::
145
146 @testing.combinations(
147 (operator.eq, "eq"),
148 (operator.ne, "ne"),
149 (operator.gt, "gt"),
150 (operator.lt, "lt"),
151 id_="na",
152 )
153 def test_operator(self, opfunc, name):
154 pass
155
156 The above combination will call ``.__name__`` on the first member of
157 each tuple and use that as the "id" to pytest.param().
158
159
160 """
161 return _fixture_functions.combinations(
162 *comb, id_=id_, argnames=argnames, **kw
163 )
164
165
166def combinations_list(arg_iterable: Iterable[Tuple[Any, ...]], **kw):
167 "As combination, but takes a single iterable"
168 return combinations(*arg_iterable, **kw)
169
170
171class Variation:
172 __slots__ = ("_name", "_argname")
173
174 def __init__(self, case, argname, case_names):
175 self._name = case
176 self._argname = argname
177 for casename in case_names:
178 setattr(self, casename, casename == case)
179
180 if typing.TYPE_CHECKING:
181
182 def __getattr__(self, key: str) -> bool: ...
183
184 @property
185 def name(self):
186 return self._name
187
188 def __bool__(self):
189 return self._name == self._argname
190
191 def __nonzero__(self):
192 return not self.__bool__()
193
194 def __str__(self):
195 return f"{self._argname}={self._name!r}"
196
197 def __repr__(self):
198 return str(self)
199
200 def fail(self) -> NoReturn:
201 fail(f"Unknown {self}")
202
203 @classmethod
204 def idfn(cls, variation):
205 return variation.name
206
207 @classmethod
208 def generate_cases(cls, argname, cases):
209 case_names = [
210 argname if c is True else "not_" + argname if c is False else c
211 for c in cases
212 ]
213
214 typ = type(
215 argname,
216 (Variation,),
217 {
218 "__slots__": tuple(case_names),
219 },
220 )
221
222 return [typ(casename, argname, case_names) for casename in case_names]
223
224
225def variation(argname_or_fn, cases=None):
226 """a helper around testing.combinations that provides a single namespace
227 that can be used as a switch.
228
229 e.g.::
230
231 @testing.variation("querytyp", ["select", "subquery", "legacy_query"])
232 @testing.variation("lazy", ["select", "raise", "raise_on_sql"])
233 def test_thing(self, querytyp, lazy, decl_base):
234 class Thing(decl_base):
235 __tablename__ = "thing"
236
237 # use name directly
238 rel = relationship("Rel", lazy=lazy.name)
239
240 # use as a switch
241 if querytyp.select:
242 stmt = select(Thing)
243 elif querytyp.subquery:
244 stmt = select(Thing).subquery()
245 elif querytyp.legacy_query:
246 stmt = Session.query(Thing)
247 else:
248 querytyp.fail()
249
250 The variable provided is a slots object of boolean variables, as well
251 as the name of the case itself under the attribute ".name"
252
253 """
254
255 if inspect.isfunction(argname_or_fn):
256 argname = argname_or_fn.__name__
257 cases = argname_or_fn(None)
258
259 @variation_fixture(argname, cases)
260 def go(self, request):
261 yield request.param
262
263 return go
264 else:
265 argname = argname_or_fn
266 cases_plus_limitations = [
267 (
268 entry
269 if (isinstance(entry, tuple) and len(entry) == 2)
270 else (entry, None)
271 )
272 for entry in cases
273 ]
274
275 variations = Variation.generate_cases(
276 argname, [c for c, l in cases_plus_limitations]
277 )
278 return combinations(
279 *[
280 (
281 (variation._name, variation, limitation)
282 if limitation is not None
283 else (variation._name, variation)
284 )
285 for variation, (case, limitation) in zip(
286 variations, cases_plus_limitations
287 )
288 ],
289 id_="ia",
290 argnames=argname,
291 )
292
293
294def variation_fixture(argname, cases, scope="function"):
295 return fixture(
296 params=Variation.generate_cases(argname, cases),
297 ids=Variation.idfn,
298 scope=scope,
299 )
300
301
302def fixture(*arg: Any, **kw: Any) -> Any:
303 return _fixture_functions.fixture(*arg, **kw)
304
305
306def get_current_test_name() -> str:
307 return _fixture_functions.get_current_test_name()
308
309
310def mark_base_test_class() -> Any:
311 return _fixture_functions.mark_base_test_class()
312
313
314class _AddToMarker:
315 def __getattr__(self, attr: str) -> Any:
316 return getattr(_fixture_functions.add_to_marker, attr)
317
318
319add_to_marker = _AddToMarker()
320
321
322class Config:
323 def __init__(self, db, db_opts, options, file_config):
324 self._set_name(db)
325 self.db = db
326 self.db_opts = db_opts
327 self.options = options
328 self.file_config = file_config
329 self.test_schema = "test_schema"
330 self.test_schema_2 = "test_schema_2"
331
332 self.is_async = db.dialect.is_async and not util.asbool(
333 db.url.query.get("async_fallback", False)
334 )
335
336 from . import provision
337
338 self.is_default_dialect = provision.is_preferred_driver(self, db)
339
340 _stack = collections.deque()
341 _configs = set()
342
343 def __repr__(self):
344 return (
345 f"sqlalchemy.testing.config.Config"
346 f"({self.db.name}+{self.db.driver}, "
347 f"{self.db.dialect.server_version_info})"
348 )
349
350 def _set_name(self, db):
351 suffix = "_async" if db.dialect.is_async else ""
352 if db.dialect.server_version_info:
353 svi = ".".join(str(tok) for tok in db.dialect.server_version_info)
354 self.name = "%s+%s%s_[%s]" % (db.name, db.driver, suffix, svi)
355 else:
356 self.name = "%s+%s%s" % (db.name, db.driver, suffix)
357
358 @classmethod
359 def register(cls, db, db_opts, options, file_config):
360 """add a config as one of the global configs.
361
362 If there are no configs set up yet, this config also
363 gets set as the "_current".
364 """
365 global any_async
366
367 cfg = Config(db, db_opts, options, file_config)
368
369 # if any backends include an async driver, then ensure
370 # all setup/teardown and tests are wrapped in the maybe_async()
371 # decorator that will set up a greenlet context for async drivers.
372 any_async = any_async or cfg.is_async
373
374 cls._configs.add(cfg)
375 return cfg
376
377 @classmethod
378 def set_as_current(cls, config, namespace):
379 global db, _current, db_url, test_schema, test_schema_2, db_opts
380 _current = config
381 db_url = config.db.url
382 db_opts = config.db_opts
383 test_schema = config.test_schema
384 test_schema_2 = config.test_schema_2
385 namespace.db = db = config.db
386
387 @classmethod
388 def push_engine(cls, db, namespace):
389 assert _current, "Can't push without a default Config set up"
390 cls.push(
391 Config(
392 db, _current.db_opts, _current.options, _current.file_config
393 ),
394 namespace,
395 )
396
397 @classmethod
398 def push(cls, config, namespace):
399 cls._stack.append(_current)
400 cls.set_as_current(config, namespace)
401
402 @classmethod
403 def pop(cls, namespace):
404 if cls._stack:
405 # a failed test w/ -x option can call reset() ahead of time
406 _current = cls._stack[-1]
407 del cls._stack[-1]
408 cls.set_as_current(_current, namespace)
409
410 @classmethod
411 def reset(cls, namespace):
412 if cls._stack:
413 cls.set_as_current(cls._stack[0], namespace)
414 cls._stack.clear()
415
416 @classmethod
417 def all_configs(cls):
418 return cls._configs
419
420 @classmethod
421 def all_dbs(cls):
422 for cfg in cls.all_configs():
423 yield cfg.db
424
425 def skip_test(self, msg):
426 skip_test(msg)
427
428
429def skip_test(msg):
430 raise _fixture_functions.skip_test_exception(msg)
431
432
433def async_test(fn):
434 return _fixture_functions.async_test(fn)
435