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