provision.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603
  1. # testing/provision.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. from __future__ import annotations
  9. import collections
  10. import contextlib
  11. import logging
  12. from . import config
  13. from . import engines
  14. from . import util
  15. from .. import exc
  16. from .. import inspect
  17. from ..engine import Connection
  18. from ..engine import Engine
  19. from ..engine import url as sa_url
  20. from ..schema import sort_tables_and_constraints
  21. from ..sql import ddl
  22. from ..sql import schema
  23. from ..util import decorator
  24. log = logging.getLogger(__name__)
  25. FOLLOWER_IDENT = None
  26. class register:
  27. def __init__(self, decorator=None):
  28. self.fns = {}
  29. self.decorator = decorator
  30. @classmethod
  31. def init(cls, fn):
  32. return register().for_db("*")(fn)
  33. @classmethod
  34. def init_decorator(cls, decorator):
  35. return register(decorator).for_db("*")
  36. def for_db(self, *dbnames):
  37. def decorate(fn):
  38. if self.decorator:
  39. fn = self.decorator(fn)
  40. for dbname in dbnames:
  41. self.fns[dbname] = fn
  42. return self
  43. return decorate
  44. def call_original(self, cfg, *arg, **kw):
  45. return self.fns["*"](cfg, *arg, **kw)
  46. def __call__(self, cfg, *arg, **kw):
  47. if isinstance(cfg, str):
  48. url = sa_url.make_url(cfg)
  49. elif isinstance(cfg, sa_url.URL):
  50. url = cfg
  51. elif isinstance(cfg, (Engine, Connection)):
  52. url = cfg.engine.url
  53. else:
  54. url = cfg.db.url
  55. backend = url.get_backend_name()
  56. if backend in self.fns:
  57. return self.fns[backend](cfg, *arg, **kw)
  58. else:
  59. return self.fns["*"](cfg, *arg, **kw)
  60. def create_follower_db(follower_ident):
  61. for cfg in _configs_for_db_operation():
  62. log.info("CREATE database %s, URI %r", follower_ident, cfg.db.url)
  63. create_db(cfg, cfg.db, follower_ident)
  64. def setup_config(db_url, options, file_config, follower_ident):
  65. # load the dialect, which should also have it set up its provision
  66. # hooks
  67. dialect = sa_url.make_url(db_url).get_dialect()
  68. dialect.load_provisioning()
  69. if follower_ident:
  70. db_url = follower_url_from_main(db_url, follower_ident)
  71. db_opts = {}
  72. update_db_opts(db_url, db_opts, options)
  73. db_opts["scope"] = "global"
  74. eng = engines.testing_engine(db_url, db_opts)
  75. post_configure_engine(db_url, eng, follower_ident)
  76. eng.connect().close()
  77. cfg = config.Config.register(eng, db_opts, options, file_config)
  78. # a symbolic name that tests can use if they need to disambiguate
  79. # names across databases
  80. if follower_ident:
  81. config.ident = follower_ident
  82. if follower_ident:
  83. configure_follower(cfg, follower_ident)
  84. return cfg
  85. def drop_follower_db(follower_ident):
  86. for cfg in _configs_for_db_operation():
  87. log.info("DROP database %s, URI %r", follower_ident, cfg.db.url)
  88. drop_db(cfg, cfg.db, follower_ident)
  89. def generate_db_urls(db_urls, extra_drivers):
  90. """Generate a set of URLs to test given configured URLs plus additional
  91. driver names.
  92. Given:
  93. .. sourcecode:: text
  94. --dburi postgresql://db1 \
  95. --dburi postgresql://db2 \
  96. --dburi postgresql://db2 \
  97. --dbdriver=psycopg2 --dbdriver=asyncpg?async_fallback=true
  98. Noting that the default postgresql driver is psycopg2, the output
  99. would be:
  100. .. sourcecode:: text
  101. postgresql+psycopg2://db1
  102. postgresql+asyncpg://db1
  103. postgresql+psycopg2://db2
  104. postgresql+psycopg2://db3
  105. That is, for the driver in a --dburi, we want to keep that and use that
  106. driver for each URL it's part of . For a driver that is only
  107. in --dbdrivers, we want to use it just once for one of the URLs.
  108. for a driver that is both coming from --dburi as well as --dbdrivers,
  109. we want to keep it in that dburi.
  110. Driver specific query options can be specified by added them to the
  111. driver name. For example, to enable the async fallback option for
  112. asyncpg::
  113. .. sourcecode:: text
  114. --dburi postgresql://db1 \
  115. --dbdriver=asyncpg?async_fallback=true
  116. """
  117. urls = set()
  118. backend_to_driver_we_already_have = collections.defaultdict(set)
  119. urls_plus_dialects = [
  120. (url_obj, url_obj.get_dialect())
  121. for url_obj in [sa_url.make_url(db_url) for db_url in db_urls]
  122. ]
  123. for url_obj, dialect in urls_plus_dialects:
  124. # use get_driver_name instead of dialect.driver to account for
  125. # "_async" virtual drivers like oracledb and psycopg
  126. driver_name = url_obj.get_driver_name()
  127. backend_to_driver_we_already_have[dialect.name].add(driver_name)
  128. backend_to_driver_we_need = {}
  129. for url_obj, dialect in urls_plus_dialects:
  130. backend = dialect.name
  131. dialect.load_provisioning()
  132. if backend not in backend_to_driver_we_need:
  133. backend_to_driver_we_need[backend] = extra_per_backend = set(
  134. extra_drivers
  135. ).difference(backend_to_driver_we_already_have[backend])
  136. else:
  137. extra_per_backend = backend_to_driver_we_need[backend]
  138. for driver_url in _generate_driver_urls(url_obj, extra_per_backend):
  139. if driver_url in urls:
  140. continue
  141. urls.add(driver_url)
  142. yield driver_url
  143. def _generate_driver_urls(url, extra_drivers):
  144. main_driver = url.get_driver_name()
  145. extra_drivers.discard(main_driver)
  146. url = generate_driver_url(url, main_driver, "")
  147. yield url
  148. for drv in list(extra_drivers):
  149. if "?" in drv:
  150. driver_only, query_str = drv.split("?", 1)
  151. else:
  152. driver_only = drv
  153. query_str = None
  154. new_url = generate_driver_url(url, driver_only, query_str)
  155. if new_url:
  156. extra_drivers.remove(drv)
  157. yield new_url
  158. @register.init
  159. def is_preferred_driver(cfg, engine):
  160. """Return True if the engine's URL is on the "default" driver, or
  161. more generally the "preferred" driver to use for tests.
  162. Backends can override this to make a different driver the "prefeferred"
  163. driver that's not the default.
  164. """
  165. return (
  166. engine.url._get_entrypoint()
  167. is engine.url.set(
  168. drivername=engine.url.get_backend_name()
  169. )._get_entrypoint()
  170. )
  171. @register.init
  172. def generate_driver_url(url, driver, query_str):
  173. backend = url.get_backend_name()
  174. new_url = url.set(
  175. drivername="%s+%s" % (backend, driver),
  176. )
  177. if query_str:
  178. new_url = new_url.update_query_string(query_str)
  179. try:
  180. new_url.get_dialect()
  181. except exc.NoSuchModuleError:
  182. return None
  183. else:
  184. return new_url
  185. def _configs_for_db_operation():
  186. hosts = set()
  187. for cfg in config.Config.all_configs():
  188. cfg.db.dispose()
  189. for cfg in config.Config.all_configs():
  190. url = cfg.db.url
  191. backend = url.get_backend_name()
  192. host_conf = (backend, url.username, url.host, url.database)
  193. if host_conf not in hosts:
  194. yield cfg
  195. hosts.add(host_conf)
  196. for cfg in config.Config.all_configs():
  197. cfg.db.dispose()
  198. @register.init
  199. def drop_all_schema_objects_pre_tables(cfg, eng):
  200. pass
  201. @register.init
  202. def drop_all_schema_objects_post_tables(cfg, eng):
  203. pass
  204. def drop_all_schema_objects(cfg, eng):
  205. drop_all_schema_objects_pre_tables(cfg, eng)
  206. drop_views(cfg, eng)
  207. if config.requirements.materialized_views.enabled:
  208. drop_materialized_views(cfg, eng)
  209. inspector = inspect(eng)
  210. consider_schemas = (None,)
  211. if config.requirements.schemas.enabled_for_config(cfg):
  212. consider_schemas += (cfg.test_schema, cfg.test_schema_2)
  213. util.drop_all_tables(eng, inspector, consider_schemas=consider_schemas)
  214. drop_all_schema_objects_post_tables(cfg, eng)
  215. if config.requirements.sequences.enabled_for_config(cfg):
  216. with eng.begin() as conn:
  217. for seq in inspector.get_sequence_names():
  218. conn.execute(ddl.DropSequence(schema.Sequence(seq)))
  219. if config.requirements.schemas.enabled_for_config(cfg):
  220. for schema_name in [cfg.test_schema, cfg.test_schema_2]:
  221. for seq in inspector.get_sequence_names(
  222. schema=schema_name
  223. ):
  224. conn.execute(
  225. ddl.DropSequence(
  226. schema.Sequence(seq, schema=schema_name)
  227. )
  228. )
  229. def drop_views(cfg, eng):
  230. inspector = inspect(eng)
  231. try:
  232. view_names = inspector.get_view_names()
  233. except NotImplementedError:
  234. pass
  235. else:
  236. with eng.begin() as conn:
  237. for vname in view_names:
  238. conn.execute(
  239. ddl._DropView(schema.Table(vname, schema.MetaData()))
  240. )
  241. if config.requirements.schemas.enabled_for_config(cfg):
  242. try:
  243. view_names = inspector.get_view_names(schema=cfg.test_schema)
  244. except NotImplementedError:
  245. pass
  246. else:
  247. with eng.begin() as conn:
  248. for vname in view_names:
  249. conn.execute(
  250. ddl._DropView(
  251. schema.Table(
  252. vname,
  253. schema.MetaData(),
  254. schema=cfg.test_schema,
  255. )
  256. )
  257. )
  258. def drop_materialized_views(cfg, eng):
  259. inspector = inspect(eng)
  260. mview_names = inspector.get_materialized_view_names()
  261. with eng.begin() as conn:
  262. for vname in mview_names:
  263. conn.exec_driver_sql(f"DROP MATERIALIZED VIEW {vname}")
  264. if config.requirements.schemas.enabled_for_config(cfg):
  265. mview_names = inspector.get_materialized_view_names(
  266. schema=cfg.test_schema
  267. )
  268. with eng.begin() as conn:
  269. for vname in mview_names:
  270. conn.exec_driver_sql(
  271. f"DROP MATERIALIZED VIEW {cfg.test_schema}.{vname}"
  272. )
  273. @register.init
  274. def create_db(cfg, eng, ident):
  275. """Dynamically create a database for testing.
  276. Used when a test run will employ multiple processes, e.g., when run
  277. via `tox` or `pytest -n4`.
  278. """
  279. raise NotImplementedError(
  280. "no DB creation routine for cfg: %s" % (eng.url,)
  281. )
  282. @register.init
  283. def drop_db(cfg, eng, ident):
  284. """Drop a database that we dynamically created for testing."""
  285. raise NotImplementedError("no DB drop routine for cfg: %s" % (eng.url,))
  286. def _adapt_update_db_opts(fn):
  287. insp = util.inspect_getfullargspec(fn)
  288. if len(insp.args) == 3:
  289. return fn
  290. else:
  291. return lambda db_url, db_opts, _options: fn(db_url, db_opts)
  292. @register.init_decorator(_adapt_update_db_opts)
  293. def update_db_opts(db_url, db_opts, options):
  294. """Set database options (db_opts) for a test database that we created."""
  295. @register.init
  296. def post_configure_engine(url, engine, follower_ident):
  297. """Perform extra steps after configuring the main engine for testing.
  298. (For the internal dialects, currently only used by sqlite, oracle, mssql)
  299. """
  300. @register.init
  301. def post_configure_testing_engine(url, engine, options, scope):
  302. """perform extra steps after configuring any engine within the
  303. testing_engine() function.
  304. this includes the main engine as well as most ad-hoc testing engines.
  305. steps here should not get in the way of test cases that are looking
  306. for events, etc.
  307. """
  308. @register.init
  309. def follower_url_from_main(url, ident):
  310. """Create a connection URL for a dynamically-created test database.
  311. :param url: the connection URL specified when the test run was invoked
  312. :param ident: the pytest-xdist "worker identifier" to be used as the
  313. database name
  314. """
  315. url = sa_url.make_url(url)
  316. return url.set(database=ident)
  317. @register.init
  318. def configure_follower(cfg, ident):
  319. """Create dialect-specific config settings for a follower database."""
  320. pass
  321. @register.init
  322. def run_reap_dbs(url, ident):
  323. """Remove databases that were created during the test process, after the
  324. process has ended.
  325. This is an optional step that is invoked for certain backends that do not
  326. reliably release locks on the database as long as a process is still in
  327. use. For the internal dialects, this is currently only necessary for
  328. mssql and oracle.
  329. """
  330. def reap_dbs(idents_file):
  331. log.info("Reaping databases...")
  332. urls = collections.defaultdict(set)
  333. idents = collections.defaultdict(set)
  334. dialects = {}
  335. with open(idents_file) as file_:
  336. for line in file_:
  337. line = line.strip()
  338. db_name, db_url = line.split(" ")
  339. url_obj = sa_url.make_url(db_url)
  340. if db_name not in dialects:
  341. dialects[db_name] = url_obj.get_dialect()
  342. dialects[db_name].load_provisioning()
  343. url_key = (url_obj.get_backend_name(), url_obj.host)
  344. urls[url_key].add(db_url)
  345. idents[url_key].add(db_name)
  346. for url_key in urls:
  347. url = list(urls[url_key])[0]
  348. ident = idents[url_key]
  349. run_reap_dbs(url, ident)
  350. @register.init
  351. def temp_table_keyword_args(cfg, eng):
  352. """Specify keyword arguments for creating a temporary Table.
  353. Dialect-specific implementations of this method will return the
  354. kwargs that are passed to the Table method when creating a temporary
  355. table for testing, e.g., in the define_temp_tables method of the
  356. ComponentReflectionTest class in suite/test_reflection.py
  357. """
  358. raise NotImplementedError(
  359. "no temp table keyword args routine for cfg: %s" % (eng.url,)
  360. )
  361. @register.init
  362. def prepare_for_drop_tables(config, connection):
  363. pass
  364. @register.init
  365. def stop_test_class_outside_fixtures(config, db, testcls):
  366. pass
  367. @register.init
  368. def get_temp_table_name(cfg, eng, base_name):
  369. """Specify table name for creating a temporary Table.
  370. Dialect-specific implementations of this method will return the
  371. name to use when creating a temporary table for testing,
  372. e.g., in the define_temp_tables method of the
  373. ComponentReflectionTest class in suite/test_reflection.py
  374. Default to just the base name since that's what most dialects will
  375. use. The mssql dialect's implementation will need a "#" prepended.
  376. """
  377. return base_name
  378. @register.init
  379. def set_default_schema_on_connection(cfg, dbapi_connection, schema_name):
  380. raise NotImplementedError(
  381. "backend does not implement a schema name set function: %s"
  382. % (cfg.db.url,)
  383. )
  384. @register.init
  385. def upsert(
  386. cfg,
  387. table,
  388. returning,
  389. *,
  390. set_lambda=None,
  391. sort_by_parameter_order=False,
  392. index_elements=None,
  393. ):
  394. """return the backends insert..on conflict / on dupe etc. construct.
  395. while we should add a backend-neutral upsert construct as well, such as
  396. insert().upsert(), it's important that we continue to test the
  397. backend-specific insert() constructs since if we do implement
  398. insert().upsert(), that would be using a different codepath for the things
  399. we need to test like insertmanyvalues, etc.
  400. """
  401. raise NotImplementedError(
  402. f"backend does not include an upsert implementation: {cfg.db.url}"
  403. )
  404. @register.init
  405. def normalize_sequence(cfg, sequence):
  406. """Normalize sequence parameters for dialect that don't start with 1
  407. by default.
  408. The default implementation does nothing
  409. """
  410. return sequence
  411. @register.init
  412. def allow_stale_update_impl(cfg):
  413. return contextlib.nullcontext()
  414. @decorator
  415. def allow_stale_updates(fn, *arg, **kw):
  416. """decorator around a test function that indicates the test will
  417. be UPDATING rows that have been read and are now stale.
  418. This normally doesn't require intervention except for mariadb 12
  419. which now raises its own error for that, and we want to turn off
  420. that setting just within the scope of the test that needs it
  421. to be turned off (i.e. ORM stale version tests)
  422. """
  423. with allow_stale_update_impl(config._current):
  424. return fn(*arg, **kw)
  425. @register.init
  426. def delete_from_all_tables(connection, cfg, metadata):
  427. """an absolutely foolproof delete from all tables routine.
  428. dialects should override this to add special instructions like
  429. disable constraints etc.
  430. """
  431. savepoints = getattr(cfg.requirements, "savepoints", False)
  432. if savepoints:
  433. savepoints = savepoints.enabled
  434. inspector = inspect(connection)
  435. for table in reversed(
  436. [
  437. t
  438. for (t, fks) in sort_tables_and_constraints(
  439. metadata.tables.values()
  440. )
  441. if t is not None
  442. # remember that inspector.get_table_names() is cached,
  443. # so this emits SQL once per unique schema name
  444. and t.name in inspector.get_table_names(schema=t.schema)
  445. ]
  446. ):
  447. if savepoints:
  448. with connection.begin_nested():
  449. connection.execute(table.delete())
  450. else:
  451. connection.execute(table.delete())