codekingpro/portable-devtools

1#############################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##############################################################9 10import sys11import traceback12from abc import ABCMeta, abstractmethod13from importlib import import_module14 15from werkzeug.utils import find_modules16from pgadmin.utils import server_utils17from pgadmin.utils.constants import PSYCOPG318from .. import socketio19 20import unittest21import config22 23 24class TestsGeneratorRegistry(ABCMeta):25    """26    class TestsGeneratorRegistry()27        Every module will be registered automatically by its module name.28 29    Class-level Methods:30    ----------- -------31    * __init__(...)32      - This is used to register test modules. You don't need to33      call this function explicitly. This will be automatically executed,34      whenever we create a class and inherit from BaseTestGenerator -35      it will register it as an available module in TestsGeneratorRegistry.36      By setting the __metaclass__ for BaseTestGenerator to37      TestsGeneratorRegistry it will create new instance of this38      TestsGeneratorRegistry per class.39 40    * load_generators():41      - This function will load all the modules from __init__()42      present in registry.43    """44 45    registry = dict()46 47    def __init__(self, name, bases, d):48 49        # Register this type of module, based on the module name50        # Avoid registering the BaseDriver itself51 52        if name != 'BaseTestGenerator' and name != 'BaseFeatureTest':53            # Store/append test classes in 'registry' if test modules has54            # multiple classes55            if d['__module__'] in TestsGeneratorRegistry.registry:56                TestsGeneratorRegistry.registry[d['__module__']].append(self)57            else:58                TestsGeneratorRegistry.registry[d['__module__']] = [self]59 60        ABCMeta.__init__(self, name, bases, d)61 62    @classmethod63    def load_generators(cls, pkg_args, pkg_root, exclude_pkgs, for_modules=[],64                        is_resql_only=False):65 66        cls.registry = dict()67 68        all_modules = []69 70        try:71            for module_name in find_modules(pkg_root, False, True):72                all_modules.append(module_name)73        except Exception:74            pass75 76        if 'resql' not in exclude_pkgs:77            # Append reverse engineered test case module78            all_modules.append('regression.re_sql.tests.test_resql')79 80        if (pkg_args is None or pkg_args == "all") and \81                'feature_tests' not in exclude_pkgs:82            # Append feature tests module83            all_modules += find_modules(84                'regression.feature_tests', False, True)85 86        # If specific modules are to be tested, exclude others87        # for modules are handled differently for resql88        if not is_resql_only and len(for_modules) > 0:89            all_modules = [module_name90                           for module_name in all_modules91                           for fmod in for_modules92                           if module_name.endswith(fmod)]93 94        # Set the module list and exclude packages in the BaseTestGenerator95        # for Reverse Engineer SQL test cases.96        BaseTestGenerator.setReSQLModuleList(all_modules)97        BaseTestGenerator.setExcludePkgs(exclude_pkgs)98 99        # Check if only reverse engineered sql test cases to run100        # if yes then import only that module101        if is_resql_only:102            BaseTestGenerator.setForModules(for_modules)103            # In case of RESQL only clear the registry of modules, as104            # RESQL test cases should be run.105            cls.registry = dict()106            try:107                import_module('regression.re_sql.tests.test_resql')108            except ImportError:109                traceback.print_exc(file=sys.stderr)110        else:111            # Check for SERVER mode112            TestsGeneratorRegistry._exclude_packages(all_modules,113                                                     exclude_pkgs)114 115    @staticmethod116    def _exclude_packages(all_modules, exclude_pkgs):117        """118        This function check for server mode test cases.119        :param all_modules: all modules.120        :param exclude_pkgs: exclude package list.121        """122        for module_name in all_modules:123            try:124                if "tests." in str(module_name) and not any(125                    str(module_name).startswith(126                        'pgadmin.' + str(exclude_pkg)127                    ) for exclude_pkg in exclude_pkgs128                ):129                    import_module(module_name)130            except ImportError:131                traceback.print_exc(file=sys.stderr)132 133 134class BaseTestGenerator(unittest.TestCase, metaclass=TestsGeneratorRegistry):135    # Defining abstract method which will override by individual testcase.136 137    def setUp(self):138        super().setUp()139        self.server_id = self.server_information["server_id"]140        server_con = server_utils.connect_server(self, self.server_id)141        if hasattr(self, 'skip_on_database') and \142            'data' in server_con and 'type' in server_con['data'] and \143                server_con['data']['type'] in self.skip_on_database:144            self.skipTest('cannot run in: %s' % server_con['data']['type'])145        if hasattr(self, 'mock_data') and 'function_name' in self.mock_data:146            self.mock_data['function_name'] =\147                self.mock_data['function_name'].replace(148                    PSYCOPG3, config.PG_DEFAULT_DRIVER)149 150    def setTestServer(self, server):151        self.server = server152 153    @abstractmethod154    def runTest(self):155        pass156 157    # Initializing app.158    def setApp(self, app):159        self.app = app160 161    # Initializing test_client.162    @classmethod163    def setTestClient(cls, test_client):164        cls.tester = test_client165 166    def setDriver(self, driver):167        self.driver = driver168 169    def setParallelUI_tests(self, parallel_ui_tests):170        self.parallel_ui_tests = parallel_ui_tests171 172    def setServerInformation(self, server_information):173        self.server_information = server_information174 175    def setTestDatabaseName(self, database_name):176        self.test_db = database_name177 178    @classmethod179    def setReSQLModuleList(cls, module_list):180        cls.re_sql_module_list = module_list181 182    @classmethod183    def setExcludePkgs(cls, exclude_pkgs):184        cls.exclude_pkgs = exclude_pkgs185 186    @classmethod187    def setForModules(cls, for_modules):188        cls.for_modules = for_modules189 190 191class BaseSocketTestGenerator(BaseTestGenerator):192    SOCKET_NAMESPACE = ""193 194    def setUp(self):195        super().setUp()196        self.tester.get("/")197        self.socket_client = socketio.test_client(198            self.app, namespace=self.SOCKET_NAMESPACE,199            flask_test_client=self.tester)200        self.assertTrue(self.socket_client.is_connected(self.SOCKET_NAMESPACE))201 202    def runTest(self):203        super().runTest()204 205    def tearDown(self):206        super().tearDown()207        self.socket_client.disconnect(namespace=self.SOCKET_NAMESPACE)208        self.assertFalse(209            self.socket_client.is_connected(self.SOCKET_NAMESPACE))210