codekingpro/portable-devtools
114k
1# testing/config.py
2# Copyright (C) 2005-2024 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 i - the given argument is a string that is part of the id only, don't
128 pass it as an argument
129
130 n - the given argument should be passed and it should be added to the
131 id by calling the .__name__ attribute
132
133 r - the given argument should be passed and it should be added to the
134 id by calling repr()
135
136 s - the given argument should be passed and it should be added to the
137 id by calling str()
138
139 a - (argument) the given argument should be passed and it should not
140 be used to generated the id
141
142 e.g.::
143
144 @testing.combinations(
145 (operator.eq, "eq"),
146 (operator.ne, "ne"),
147 (operator.gt, "gt"),
148 (operator.lt, "lt"),
149 id_="na"
150 )
151 def test_operator(self, opfunc, name):
152 pass
153
154 The above combination will call ``.__name__`` on the first member of
155 each tuple and use that as the "id" to pytest.param().
156
157
158 """
159 return _fixture_functions.combinations(
160 *comb, id_=id_, argnames=argnames, **kw
161 )
162
163
164def combinations_list(arg_iterable: Iterable[Tuple[Any, ...]], **kw):
165 "As combination, but takes a single iterable"
166 return combinations(*arg_iterable, **kw)
167
168
169class Variation:
170 __slots__ = ("_name", "_argname")
171
172 def __init__(self, case, argname, case_names):
173 self._name = case
174 self._argname = argname
175 for casename in case_names:
176 setattr(self, casename, casename == case)
177
178 if typing.TYPE_CHECKING:
179
180 def __getattr__(self, key: str) -> bool: ...
181
182 @property
183 def name(self):
184 return self._name
185
186 def __bool__(self):
187 return self._name == self._argname
188
189 def __nonzero__(self):
190 return not self.__bool__()
191
192 def __str__(self):
193 return f"{self._argname}={self._name!r}"
194
195 def __repr__(self):
196 return str(self)
197
198 def fail(self) -> NoReturn:
199 fail(f"Unknown {self}")
200
201 @classmethod
202 def idfn(cls, variation):
203 return variation.name
204
205 @classmethod
206 def generate_cases(cls, argname, cases):
207 case_names = [
208 argname if c is True else "not_" + argname if c is False else c
209 for c in cases
210 ]
211
212 typ = type(
213 argname,
214 (Variation,),
215 {
216 "__slots__": tuple(case_names),
217 },
218 )
219
220 return [typ(casename, argname, case_names) for casename in case_names]
221
222
223def variation(argname_or_fn, cases=None):
224 """a helper around testing.combinations that provides a single namespace
225 that can be used as a switch.
226
227 e.g.::
228
229 @testing.variation("querytyp", ["select", "subquery", "legacy_query"])
230 @testing.variation("lazy", ["select", "raise", "raise_on_sql"])
231 def test_thing(
232 self,
233 querytyp,
234 lazy,
235 decl_base
236 ):
237 class Thing(decl_base):
238 __tablename__ = 'thing'
239
240 # use name directly
241 rel = relationship("Rel", lazy=lazy.name)
242
243 # use as a switch
244 if querytyp.select:
245 stmt = select(Thing)
246 elif querytyp.subquery:
247 stmt = select(Thing).subquery()
248 elif querytyp.legacy_query:
249 stmt = Session.query(Thing)
250 else:
251 querytyp.fail()
252
253
254 The variable provided is a slots object of boolean variables, as well
255 as the name of the case itself under the attribute ".name"
256
257 """
258
259 if inspect.isfunction(argname_or_fn):
260 argname = argname_or_fn.__name__
261 cases = argname_or_fn(None)
262
263 @variation_fixture(argname, cases)
264 def go(self, request):
265 yield request.param
266
267 return go
268 else:
269 argname = argname_or_fn
270 cases_plus_limitations = [
271 (
272 entry
273 if (isinstance(entry, tuple) and len(entry) == 2)
274 else (entry, None)
275 )
276 for entry in cases
277 ]
278
279 variations = Variation.generate_cases(
280 argname, [c for c, l in cases_plus_limitations]
281 )
282 return combinations(
283 *[
284 (
285 (variation._name, variation, limitation)
286 if limitation is not None
287 else (variation._name, variation)
288 )
289 for variation, (case, limitation) in zip(
290 variations, cases_plus_limitations
291 )
292 ],
293 id_="ia",
294 argnames=argname,
295 )
296
297
298def variation_fixture(argname, cases, scope="function"):
299 return fixture(
300 params=Variation.generate_cases(argname, cases),
301 ids=Variation.idfn,
302 scope=scope,
303 )
304
305
306def fixture(*arg: Any, **kw: Any) -> Any:
307 return _fixture_functions.fixture(*arg, **kw)
308
309
310def get_current_test_name() -> str:
311 return _fixture_functions.get_current_test_name()
312
313
314def mark_base_test_class() -> Any:
315 return _fixture_functions.mark_base_test_class()
316
317
318class _AddToMarker:
319 def __getattr__(self, attr: str) -> Any:
320 return getattr(_fixture_functions.add_to_marker, attr)
321
322
323add_to_marker = _AddToMarker()
324
325
326class Config:
327 def __init__(self, db, db_opts, options, file_config):
328 self._set_name(db)
329 self.db = db
330 self.db_opts = db_opts
331 self.options = options
332 self.file_config = file_config
333 self.test_schema = "test_schema"
334 self.test_schema_2 = "test_schema_2"
335
336 self.is_async = db.dialect.is_async and not util.asbool(
337 db.url.query.get("async_fallback", False)
338 )
339
340 _stack = collections.deque()
341 _configs = set()
342
343 def _set_name(self, db):
344 suffix = "_async" if db.dialect.is_async else ""
345 if db.dialect.server_version_info:
346 svi = ".".join(str(tok) for tok in db.dialect.server_version_info)
347 self.name = "%s+%s%s_[%s]" % (db.name, db.driver, suffix, svi)
348 else:
349 self.name = "%s+%s%s" % (db.name, db.driver, suffix)
350
351 @classmethod
352 def register(cls, db, db_opts, options, file_config):
353 """add a config as one of the global configs.
354
355 If there are no configs set up yet, this config also
356 gets set as the "_current".
357 """
358 global any_async
359
360 cfg = Config(db, db_opts, options, file_config)
361
362 # if any backends include an async driver, then ensure
363 # all setup/teardown and tests are wrapped in the maybe_async()
364 # decorator that will set up a greenlet context for async drivers.
365 any_async = any_async or cfg.is_async
366
367 cls._configs.add(cfg)
368 return cfg
369
370 @classmethod
371 def set_as_current(cls, config, namespace):
372 global db, _current, db_url, test_schema, test_schema_2, db_opts
373 _current = config
374 db_url = config.db.url
375 db_opts = config.db_opts
376 test_schema = config.test_schema
377 test_schema_2 = config.test_schema_2
378 namespace.db = db = config.db
379
380 @classmethod
381 def push_engine(cls, db, namespace):
382 assert _current, "Can't push without a default Config set up"
383 cls.push(
384 Config(
385 db, _current.db_opts, _current.options, _current.file_config
386 ),
387 namespace,
388 )
389
390 @classmethod
391 def push(cls, config, namespace):
392 cls._stack.append(_current)
393 cls.set_as_current(config, namespace)
394
395 @classmethod
396 def pop(cls, namespace):
397 if cls._stack:
398 # a failed test w/ -x option can call reset() ahead of time
399 _current = cls._stack[-1]
400 del cls._stack[-1]
401 cls.set_as_current(_current, namespace)
402
403 @classmethod
404 def reset(cls, namespace):
405 if cls._stack:
406 cls.set_as_current(cls._stack[0], namespace)
407 cls._stack.clear()
408
409 @classmethod
410 def all_configs(cls):
411 return cls._configs
412
413 @classmethod
414 def all_dbs(cls):
415 for cfg in cls.all_configs():
416 yield cfg.db
417
418 def skip_test(self, msg):
419 skip_test(msg)
420
421
422def skip_test(msg):
423 raise _fixture_functions.skip_test_exception(msg)
424
425
426def async_test(fn):
427 return _fixture_functions.async_test(fn)
428