1# testing/config.py 2# Copyright (C) 2005-2019 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: http://www.opensource.org/licenses/mit-license.php 7 8import collections 9from unittest import SkipTest as _skip_test_exception 10 11requirements = None 12db = None 13db_url = None 14db_opts = None 15file_config = None 16test_schema = None 17test_schema_2 = None 18_current = None 19 20 21class Config(object): 22 def __init__(self, db, db_opts, options, file_config): 23 self._set_name(db) 24 self.db = db 25 self.db_opts = db_opts 26 self.options = options 27 self.file_config = file_config 28 self.test_schema = "test_schema" 29 self.test_schema_2 = "test_schema_2" 30 31 _stack = collections.deque() 32 _configs = set() 33 34 def _set_name(self, db): 35 if db.dialect.server_version_info: 36 svi = ".".join(str(tok) for tok in db.dialect.server_version_info) 37 self.name = "%s+%s_[%s]" % (db.name, db.driver, svi) 38 else: 39 self.name = "%s+%s" % (db.name, db.driver) 40 41 @classmethod 42 def register(cls, db, db_opts, options, file_config): 43 """add a config as one of the global configs. 44 45 If there are no configs set up yet, this config also 46 gets set as the "_current". 47 """ 48 cfg = Config(db, db_opts, options, file_config) 49 cls._configs.add(cfg) 50 return cfg 51 52 @classmethod 53 def set_as_current(cls, config, namespace): 54 global db, _current, db_url, test_schema, test_schema_2, db_opts 55 _current = config 56 db_url = config.db.url 57 db_opts = config.db_opts 58 test_schema = config.test_schema 59 test_schema_2 = config.test_schema_2 60 namespace.db = db = config.db 61 62 @classmethod 63 def push_engine(cls, db, namespace): 64 assert _current, "Can't push without a default Config set up" 65 cls.push( 66 Config( 67 db, _current.db_opts, _current.options, _current.file_config 68 ), 69 namespace, 70 ) 71 72 @classmethod 73 def push(cls, config, namespace): 74 cls._stack.append(_current) 75 cls.set_as_current(config, namespace) 76 77 @classmethod 78 def reset(cls, namespace): 79 if cls._stack: 80 cls.set_as_current(cls._stack[0], namespace) 81 cls._stack.clear() 82 83 @classmethod 84 def all_configs(cls): 85 return cls._configs 86 87 @classmethod 88 def all_dbs(cls): 89 for cfg in cls.all_configs(): 90 yield cfg.db 91 92 def skip_test(self, msg): 93 skip_test(msg) 94 95 96def skip_test(msg): 97 raise _skip_test_exception(msg) 98