Best Python code snippet using autotest_python
command_service.py
Source:command_service.py  
1import os2import sys3import platform4import subprocess5import click6import simplejson as json7from .errors import *8from .controller_osx import OsxController9from .controller_raspberry_pi import RaspberryPiController10class CommandService(object):11    ENV_MACOS = 'osx'12    ENV_RASPBERRY_PI = 'raspberry_pi'13    ENV_UNKNOWN = 'unknown'14    _os_controller = None15    _path_services_system = os.path.dirname(os.path.realpath(__file__)) + '/../../../'16    _path_services_skill = os.path.dirname(os.path.realpath(__file__)) + '/../../../../../services/'17    _path_services_droid = os.path.dirname(os.path.realpath(__file__)) + '/../../../../../../droids/'18    _path_current_droid = None19    _storage_dir_path = '/etc/turing'20    _storage_global_config_file = 'config.json'21    _colors = {22        'black':        '\033[0;30m',23        'red':          '\033[0;31m',24        'green':        '\033[0;32m',25        'brown':        '\033[0;33m',26        'blue':         '\033[0;34m',27        'purple':       '\033[0;35m',28        'cyan':         '\033[097;096m',29        'graylight':    '\033[0;37m',30        'graydark':     '\033[1;30m',31        'redlight':     '\033[1;31m',32        'greenlight':    '\033[1;32m',33        'yellow':       '\033[0;93m',34        'bluelight':    '\033[1;34m',35        'purplelight':  '\033[1;35m',36        'cyanlight':    '\033[1;36m',37        'white':        '\033[0;97m',38    }39    # --------------------------------------------------------------------------40    # UTILITY METHODS41    # --------------------------------------------------------------------------42    def get_env(self):43        run_env = self.ENV_UNKNOWN44    def display(self, msg):45        msg = self._colors['white'] + '    ' + str(msg) + self._colors['white']46        for color in self._colors:47            msg = msg.replace('{{'+color.upper()+'}}', self._colors[color])48        print(msg)49    def _cleanup(self):50        print('')51    # --------------------------------------------------------------------------52    # CORE CLASS METHODS53    # --------------------------------------------------------------------------54    def __init__(self):55        plat_name = platform.system().lower()56        plat_machine = platform.machine()57        plat_release = platform.release()58        plat_full_description = os.uname().nodename59        os_name = os.name.lower()60        if plat_name == 'linux':61            if 'raspberry' in plat_full_description.lower():62                self._env = self.ENV_RASPBERRY_PI63            elif plat_machine.startswith('arm'):64                self._env = self.ENV_RASPBERRY_PI65            else:66                self._env = self.ENV_UNKNOWN67        elif plat_name == 'darwin':68            self._env = self.ENV_MACOS69        # Instantiate controller for right platform70        if self._env == self.ENV_MACOS:71            self._os_controller = OsxController()72        elif self._env == self.ENV_RASPBERRY_PI:73            self._os_controller = RaspberryPiController()74    # --------------------------------------------------------------------------75    # STORAGE METHODS76    # --------------------------------------------------------------------------77    def init_storage(self):78        store_exists = False79        store_init_success = True80        if os.path.isdir(self._storage_dir_path):81            store_exists = True82        else:83            try:84                self._os_controller.run_command('sudo mkdir %s' % (self._storage_dir_path))85                store_exists = True86            except Error as e:87                store_exists = False88        if not os.path.isdir(os.path.join(self._storage_dir_path, 'services')):89            self._os_controller.run_command('sudo mkdir %s' % os.path.join(self._storage_dir_path, 'services'))90        if not os.path.isdir(os.path.join(self._storage_dir_path, 'droids')):91            self._os_controller.run_command('sudo mkdir %s' % os.path.join(self._storage_dir_path, 'droids'))92        if store_exists:93            self.display('{{GREEN}}Validating filesystem: {{WHITE}}OS')94            try:95                if not os.path.isdir(self._get_config_value('service-path')):96                    self.display('{{YELLOW}}Validating filesystem: {{WHITE}}"service-path" directory does not actually exist.\n')97                    store_init_success = False98                if not os.path.isdir(self._get_config_value('droid-path')):99                    self.display('{{YELLOW}}Validating filesystem: {{WHITE}}"droid-path" directory does not actually exist.\n')100                    store_init_success = False101            except ConfigKeyNotFoundError:102                pass103        else:104            self.display('{{RED}}Had an issue accessing config storage on disk.{{WHITE}}')105        return store_init_success106    # --------------------------------------------------------------------------107    # CONFIG METHODS108    # --------------------------------------------------------------------------109    def _validate_system_config(self):110        is_config_valid = True111        try:112            cfg = self._get_config()113        except ConfigNotFoundError:114            self.display('{{YELLOW}}Config file not found. Cannot continue.')115            is_config_valid = False116        except ConfigEmptyError:117            self.display('{{YELLOW}}Config file empty. Cannot continue.')118            is_config_valid = False119        except ConfigMalformedError:120            self.display('{{YELLOW}}Config JSON is malformed. Cannot continue.')121            is_config_valid = False122        return is_config_valid123    def _validate_service_config(self, config_obj):124        current_run_file_name = None125        is_config_valid = True126        config_err_msg = None127        if config_obj is None:128            is_config_valid = False129            config_err_msg = 'config object supplied is null'130        else:131            if 'service-name' in config_obj:132                current_name = config_obj['service-name']133            else:134                is_config_valid = False135                config_err_msg = 'service.json missing "service-name" attribute'136        return (is_config_valid, config_err_msg)137    def _get_config_value(self, key):138        r = None139        try:140            cfg = self._get_config()141        except Exception:142            raise ConfigKeyNotFoundError()143        if key in cfg:144            r = cfg[key]145        else:146            raise ConfigKeyNotFoundError()147        return r148    def _set_config_value(self, key, new_val):149        try:150            cfg = self._get_config()151        except Exception:152            raise ConfigKeySetError()153        cfg[key] = new_val154        try:155            self._save_config(cfg)156        except ConfigSaveError as e:157            print(e)158            raise ConfigKeySetError()159    def _save_config(self, config_dict):160        self._elevate_privileges()161        config_file_path = '%s/%s' % (self._storage_dir_path, self._storage_global_config_file)162        try:163            with open(config_file_path, 'w') as f:164                json.dump(config_dict, f, indent=4, sort_keys=True)165        except:166            raise ConfigSaveError()167    def _get_config(self):168        config_file_path = '%s/%s' % (self._storage_dir_path, self._storage_global_config_file)169        config_file_raw = None170        config_file_contents = None171        if not os.path.isfile(config_file_path):172            raise ConfigNotFoundError()173        try:174            with open(config_file_path, 'r') as f:175                config_file_raw = f.read()176            config_file_contents = json.loads(config_file_raw)177        except:178            if len(str(config_file_raw)) <= 1:179                raise ConfigEmptyError()180            else:181                raise ConfigMalformedError()182        return config_file_contents183    # --------------------------------------------------------------------------184    # STATUS METHODS185    # --------------------------------------------------------------------------186    def get_system_status(self):187        self._elevate_privileges()188        self.display('{{WHITE}}OPERATING SYSTEM:')189        self.display('------------------------------------------------')190        self.display('{{GRAYDARK}}Filesystem: {{WHITE}}operational.')191        self._get_service_status_by_dir(self._path_services_system)192        self.display('')193        self.display('{{WHITE}}DROID:')194        self.display('------------------------------------------------')195        try:196            self._get_service_status(197                self._get_config_value('droid-path'),198                self._get_config_value('current-droid')199            )200        except ConfigKeyNotFoundError:201            self.display('{{GRAYDARK}}Droid not configured.')202        self.display('')203        self.display('')204        self.display('{{WHITE}}SKILL SERVICES:')205        self.display('------------------------------------------------')206        try:207            self._get_service_status_by_dir(self._get_config_value('service-path'))208        except ConfigKeyNotFoundError:209            self.display('{{GRAYDARK}}Skills not configured.')210        self.display('')211    def _get_service_status_by_dir(self, dir_services):212        service_dir_names = next(os.walk(dir_services))[1]213        if self._os_controller:214            if len(service_dir_names) == 0:215                msg = '{{YELLOW}}No services installed.{{WHITE}}'216                self.display(msg)217            else:218                for service_name in service_dir_names:219                    self._get_service_status(dir_services, service_name)220        self._cleanup()221    def _get_service_status(self, base_dir_path, service_name):222        path_service_config = os.path.join(223            base_dir_path,224            service_name,225            'service.json'226        )227        is_config_available = True228        is_config_valid = True229        cfg = None230        config_err_msg = None231        try:232            with open(path_service_config) as f:233                cfg = json.loads(f.read())234        except:235            is_config_available = False236        is_config_valid, config_err_msg = self._validate_service_config(cfg)237        if is_config_valid:238            is_always_on = False239            if 'always-on' in cfg:240                is_always_on = True241            if is_always_on:242                msg = '{{GRAYDARK}}Service always on: {{WHITE}}%s' % (service_name)243            else:244                msg = self._os_controller.get_service_status(service_name, cfg)245        else:246            msg = '{{RED}}%s: %s' % (config_err_msg, service_name)247        self.display(msg)248    def get_system_info(self):249        plat_name = platform.system().lower()250        plat_release = platform.release()251        plat_full_description = os.uname().nodename252        plat_machine = platform.machine()253        os_name = os.name.lower()254        self.display('Current running platform: %s, release number %s, (%s), on the %s OS' % (plat_name, plat_release, plat_full_description, plat_machine))255    # --------------------------------------------------------------------------256    # STOP METHODS257    # --------------------------------------------------------------------------258    def stop_all_services(self):259        self._elevate_privileges()260        # Stop core services in the /system/services/ directory261        self.display('{{WHITE}}OPERATING SYSTEM:')262        self.display('------------------------------------------------')263        self.display('{{GRAYDARK}}Filesystem: {{WHITE}}operational.')264        self._stop_services_by_dir(self._path_services_system)265        self.display('')266        self.display('{{WHITE}}DROID:')267        self.display('------------------------------------------------')268        try:269            self._stop_service(self._path_services_droid, self._get_config_value('current-droid'))270            self.display('{{RED}}Droid disabled: {{WHITE}}%s' % (self._get_config_value('current-droid')))271        except ConfigKeyNotFoundError:272            self.display('{{GRAYDARK}}Droid not configured.')273        self.display('')274        self.display('')275        # Stop skill services in the /services directory276        self.display('{{WHITE}}SKILL SERVICES:')277        self.display('------------------------------------------------')278        try:279            self.display(self._get_config_value('service-path'))280            self._stop_services_by_dir(self._get_config_value('service-path'))281        except ConfigKeyNotFoundError:282            self.display('{{GRAYDARK}}Services not configured.')283        self.display('')284    def _stop_service(self, base_dir_path, service_name):285        path_service_config = os.path.join(286            base_dir_path,287            service_name,288            'service.json'289        )290        is_config_available = True291        is_config_valid = True292        cfg = None293        config_err_msg = None294        try:295            with open(path_service_config) as f:296                cfg = json.loads(f.read())297        except Exception:298            is_config_available = False299        is_config_valid, config_err_msg = self._validate_service_config(cfg)300        if is_config_valid:301            is_always_on = False302            if 'always-on' in cfg:303                if cfg['always-on'] == True:304                    is_always_on = True305            if is_always_on:306                msg = '{{GRAYDARK}}Service always on: {{WHITE}}%s' % (service_name)307            else:308                msg = self._os_controller.stop_service(service_name, cfg)309        else:310            msg = '{{PURPLE}}%s:{{WHITE}} %s' % (config_err_msg, service_name)311            msg = base_dir_path + ' ::: ' + service_name312        self.display(msg)313    def _stop_services_by_dir(self, dir_services):314        service_dir_names = next(os.walk(dir_services))[1]315        if self._os_controller:316            if len(service_dir_names) == 0:317                msg = '{{YELLOW}}No services installed.{{WHITE}}'318                self.display(msg)319            else:320                for service_name in service_dir_names:321                    is_config_available = True322                    is_config_valid = True323                    cfg = None324                    config_err_msg = None325                    path_service_config = os.path.join(326                        dir_services,327                        service_name,328                        'service.json'329                    )330                    try:331                        json_data_raw = open(path_service_config).read()332                        cfg = json.loads(json_data_raw)333                    except Exception:334                        is_config_available = False335                    is_config_valid, config_err_msg = self._validate_service_config(cfg)336                    if is_config_valid:337                        is_always_on = False338                        if 'always-on' in cfg:339                            if cfg['always-on'] == True:340                                is_always_on = True341                        if is_always_on:342                            msg = '{{GRAYDARK}}Service always on: {{WHITE}}%s' % (service_name)343                        else:344                            msg = self._os_controller.stop_service(service_name, cfg)345                    else:346                        msg = '{{PURPLE}}%s:{{WHITE}} %s' % (config_err_msg, service_name)347                    self.display(msg)348        self._cleanup()349    # --------------------------------------------------------------------------350    # START METHODS351    # --------------------------------------------------------------------------352    def start_all_services(self):353        self._elevate_privileges()354        # Start core services in the /system/services/ directory355        self.display('{{WHITE}}OPERATING SYSTEM:')356        self.display('------------------------------------------------')357        self.display('{{GRAYDARK}}Filesystem: {{WHITE}}operational.')358        self._start_services_by_dir(self._path_services_system)359        self.display('')360        self.display('{{WHITE}}DROID:')361        self.display('------------------------------------------------')362        self.display('{{GREEN}}Starting droid: {{WHITE}}%s' % (self._get_config_value('current-droid')))363        self._start_service(364            self._get_config_value('droid-path'),365            self._get_config_value('current-droid')366        )367        self.display('')368        self.display('')369        # Start skill services in the /services directory370        self.display('{{WHITE}}SKILL SERVICES:')371        self.display(self._path_services_skill)372        self.display('------------------------------------------------')373        self._start_services_by_dir(self._get_config_value('service-path'))374        self.display('')375    def _start_services_by_dir(self, dir_services):376        service_dir_names = next(os.walk(dir_services))[1]377        if self._os_controller:378            if len(service_dir_names) == 0:379                self.display('{{YELLOW}}No services installed.{{WHITE}}')380            else:381                for service_name in service_dir_names:382                    self._start_service(dir_services, service_name)383        self._cleanup()384    def _start_service(self, base_dir_path, service_name):385        path_service_config = os.path.join(386            base_dir_path,387            service_name,388            'service.json'389        )390        is_config_available = True391        is_config_valid = True392        cfg = None393        config_err_msg = None394        try:395            with open(path_service_config) as f:396                cfg = json.loads(f.read())397        except Exception:398            is_config_available = False399        is_config_valid, config_err_msg = self._validate_service_config(cfg)400        if is_config_valid:401            is_always_on = False402            if 'always-on' in cfg:403                if cfg['always-on'] == True:404                    is_always_on = True405            if is_always_on:406                msg = '{{GRAYDARK}}Service always on: {{WHITE}}%s' % (service_name)407            else:408                msg = self._os_controller.start_service(service_name, cfg)409        else:410            msg = '{{PURPLE}}%s:{{WHITE}} %s' % (config_err_msg, service_name)411        self.display(msg)412    # --------------------------------------------------------------------------413    # INSTALL METHODS414    # --------------------------------------------------------------------------415    def install_service(self, service_name):416        self._elevate_privileges()417        path_requested_service = os.path.join(418            self._storage_dir_path,419            self._get_config_value('service-path'),420            service_name421        )422        is_existing_path_in_storage = os.path.isdir(path_requested_service)423        is_existing_path_in_current_dir = os.path.isdir(service_name)424        if is_existing_path_in_current_dir:425            self.display('{{RED}}This directory already as a \'%s\' sub-directory.' % service_name)426        else:427            if is_existing_path_in_storage:428                self.display('{{RED}}This service is already registered locally with Turing, but is installed in a different location.')429            else:430                self.display('{{WHITE}}Downloading service..')431                # subprocess.check_output('cd $TURING_APP_DIR', shell=True)432                git_path = 'https://github.com/enborra/%s.git' % service_name433                hrm = subprocess.check_output(434                    'pwd',435                    stderr=subprocess.STDOUT,436                    shell=True437                )438                current_dir = hrm.decode('utf-8').rstrip()439                new_repo_dir = os.path.join(440                    current_dir,441                    service_name442                )443                subprocess.check_output('git clone %s --quiet' % git_path, shell=True)444                self.display('{{WHITE}}Registering repo with Turing locally...')445                subprocess.check_output(446                    'sudo ln -s %s %s' % (new_repo_dir, path_requested_service),447                    stderr=subprocess.STDOUT,448                    shell=True449                )450                self.display('{{WHITE}}Service registered successfully!')451                self.display('')452    def uninstall_service(self, service_name):453        self._elevate_privileges()454        path_requested_service = os.path.join(455            self._storage_dir_path,456            self._get_config_value('service-path'),457            service_name458        )459        is_existing_path_in_storage = os.path.isdir(path_requested_service)460        if is_existing_path_in_storage:461            self.display('{{WHITE}}Unregistering %s from Turing locally' % service_name)462            if path_requested_service.startswith('/etc/turing/'):463                subprocess.check_output(464                    'sudo rm -rf %s' % path_requested_service,465                    stderr=subprocess.STDOUT,466                    shell=True467                )468            else:469                self.display('{{RED}}Something went wrong during service uninstall.')470        else:471            self.display('{{WHITE}}%s is not locally registered with Turing')472    def install_all_services(self):473        self._elevate_privileges()474        self.display('{{WHITE}}OPERATING SYSTEM:')475        self.display('------------------------------------------------')476        can_proceed = self.init_storage()477        if not can_proceed:478            self.display('{{YELLOW}}Can\'t proceed. Stopping request.')479        else:480            # Install core services in the /system/services/ directory481            self._install_services_by_dir(self._path_services_system)482            self.display('')483            self.display('{{WHITE}}DROID:')484            self.display('------------------------------------------------')485            self.display('{{GREEN}}Installing droid: {{WHITE}}%s' % (self._get_config_value('current-droid')))486            self._install_service(487                self._get_config_value('droid-path'),488                self._get_config_value('current-droid')489            )490            self.display('')491            self.display('')492            # Install skill services in the /services directory493            self.display('{{WHITE}}SKILL SERVICES:')494            self.display('------------------------------------------------')495            self._install_services_by_dir(self._get_config_value('service-path'))496            self.display('')497    def _install_services_by_dir(self, dir_services):498        service_dir_names = next(os.walk(dir_services))[1]499        if self._os_controller:500            for service_name in service_dir_names:501                self._install_service(dir_services, service_name)502        self._cleanup()503    def _install_service(self, base_dir_path, service_name):504        path_service_config = os.path.join(505            base_dir_path,506            service_name,507            'service.json'508        )509        is_config_available = True510        is_config_valid = True511        cfg = None512        config_err_msg = None513        try:514            json_data_raw = open(path_service_config).read()515            cfg = json.loads(json_data_raw)516        except Exception:517            is_config_available = False518        is_config_valid, config_err_msg = self._validate_service_config(cfg)519        if is_config_valid:520            is_always_on = False521            if 'always-on' in cfg:522                if cfg['always-on'] == True:523                    is_always_on = True524            if is_always_on:525                msg = '{{GRAYDARK}}Service always on: {{WHITE}}%s' % (service_name)526            else:527                msg = self._os_controller.install_service(service_name, base_dir_path, cfg)528        else:529            msg = '{{PURPLE}}%s:{{WHITE}} %s' % (config_err_msg, service_name)530        self.display(msg)531    # --------------------------------------------------------------------------532    # DROID UPDATE METHODS533    # --------------------------------------------------------------------------534    def set_active_droid(self, droid_name):535        self.display('{{YELLOW}}Booting droid: {{WHITE}}%s' % (droid_name))536        self.display('')537        self._set_config_value('current-droid', droid_name)538    # --------------------------------------------------------------------------539    # SOURCE UPDATE METHODS540    # --------------------------------------------------------------------------541    def _services_update_source(self):542        if self._os_controller:543            self._os_controller.update_source()544    def _elevate_privileges(self):545        try:546            self._os_controller.elevate_privileges()547        except PermissionError as e:548            self.display('{{YELLOW}}\n\n    Could not acquire elevated privileges. Can\'t proceed.\n\n{{WHITE}}')...config.py
Source:config.py  
1import configparser2import os3import requests4from simplejson.errors import JSONDecodeError5config = configparser.ConfigParser()6CONFIG_FILE_NAME = "config.ini"7def load_configuration():8    is_config_valid = True9    if os.path.isfile(CONFIG_FILE_NAME):10        read = config.read(CONFIG_FILE_NAME)11        is_config_valid = len(read) > 012    else:13        is_config_valid = False14    if not is_config_valid:15        # Configuration has to be made16        print("===========================")17        print("")18        print(" CONFIGURATION HELP WIZARD")19        print("")20        print("===========================")21        print("")22        print("Please, input your server's hostname.")23        print("This must be only the hostname or IP address, examples:")24        print("- example.com")25        print("- 81.126.59.185")26        hostname = None27        while not hostname:28            host = input("> ")29            if len(host) <= 0:30                print("Please, enter a valid hostname.")31            else:32                hostname = host33        print("")34        print("Please, input your server's port.")35        print("This must be a number, for example, the default http port is 80.")36        port = -137        while port < 0:38            port_str = input("> ")39            if port_str.isdigit():40                port = int(port_str)41            else:42                print("Please, enter a valid port name.")43        print("")44        print("Please, input your server's path.")45        print("This is your EmonCMS's instance path, for example, you may access EmonCMS")46        print(" through http://example.com/emoncms/, then you should introduce emoncms/.")47        print("If your instance is at the root of the hostname, you can simply leave the")48        print(" field blank.")49        path = input("> ")50        if len(path) > 0 and not path.endswith("/"):51            path = path + "/"52        print("")53        print("Do you want to use https?")54        print("Type yes or no")55        protocol = None56        while protocol is None:57            https_str = input("> ")58            if https_str == "yes":59                protocol = 'https'60            elif https_str == "no":61                protocol = 'http'62            else:63                print("Please, type \"yes\" or \"no\".")64        print("")65        print("Please, enter your EmonCMS write-enabled API key. You can get it at:")66        print(f"{protocol}://{hostname}:{port}/input/api")67        api_key = None68        while not api_key:69            key = input("> ")70            if len(key) != 32:71                print("The API key is 32 characters long. Please check the input.")72            else:73                api_key = key74        print("")75        print("Checking key validity...", end="")76        addr = f"{protocol}://{hostname}:{port}/input/list/?apikey={api_key}"77        valid_request = requests.get(addr)78        valid_data = True79        if valid_request.status_code == 200:80            try:81                valid_request.json()82                print("ok")83            except JSONDecodeError:84                valid_data = False85        else:86            valid_data = False87        if not valid_data:88            print("error!")89            print("The introduced API key is not valid. This can be caused because you")90            print(" are not connected to the Internet, or simply because the API key is")91            print(" not valid. Type \"yes\" to continue.")92            shall_continue = input("> ")93            if shall_continue != "yes":94                exit()95        print("")96        print("Please, enter your desired node name. This can be almost anything, it's")97        print(" just an identifier for your device at EmonCMS.")98        node_name = None99        while not node_name:100            name = input("> ")101            if len(name) <= 0:102                print("Please, enter a valid name.")103            else:104                node_name = name105        print("")106        print("Configuration process complete, storing...")107        config['server'] = {}108        config['emoncms'] = {}109        config['server']['hostname'] = hostname110        config['server']['port'] = str(port)111        config['server']['path'] = path112        config['server']['protocol'] = protocol113        config['emoncms']['apikey'] = api_key114        config['emoncms']['node_name'] = node_name115        with open(CONFIG_FILE_NAME, 'w') as configfile:116            config.write(configfile)117        return load_configuration()118    else:119        data = {120            "hostname": config['server']['hostname'],121            "port": int(config['server']['port']),122            "path": config['server']['path'],123            "protocol": config['server']['protocol'],124            "api_key": config['emoncms']['apikey'],125            "node_name": config['emoncms']['node_name'],126        }127        if "mqtt" in config:128            data["mqtt_address"] = config["mqtt"]["address"]129            data["mqtt_port"] = config["mqtt"]["port"] if "port" in config["mqtt"] else 1883130            data["mqtt_keepalive"] = config["mqtt"]["keepalive"] if "keepalive" in config["mqtt"] else 60131            data["mqtt_topic"] = config["mqtt"]["basetopic"] if "basetopic" in config["mqtt"] else 60...config_parser.py
Source:config_parser.py  
1from helpers.open_config_file import OpenFile2from pykwalify.core import Core3from icecream import ic4class ConfigParser(OpenFile):5    ic.disable()6    # ic.enable()7    def __init__(self, config_dir) -> None:8        self.config_dir = config_dir9        ic(self.config_dir)10        self.file_path = f'{self.config_dir}/config.yaml'11        self.schema_path = './config/schema/config_schema.yaml'12        super().__init__(self.file_path)13        self.config_file = self.open_yaml_file()14    def __call__(self) -> tuple:15        is_config_valid = self.config_validation()16        # ic(is_config_valid)17        if is_config_valid:18            parsed_config = self.config_parse()19        else:20            raise Exception21        return parsed_config22    def config_validation(self) -> bool:23        c = Core(source_file=self.file_path, schema_files=[self.schema_path])24        c.validate(raise_exception=True)25        return True26    def config_parse(self) -> tuple:27        browsers = self.parse_browsers()28        aerokube = self.parse_aerokube()29        hosts = self.parse_hosts()30        teams = self.parse_teams_quota()31        # ic(browsers)32        # ic(aerokube)33        # ic(hosts)34        # ic(teams)35        return browsers, aerokube, hosts, teams36    def parse_browsers(self) -> dict:37        browsers = self.config_file['browsers']38        return browsers39    def parse_aerokube(self) -> dict:40        aerokube = self.config_file['aerokube']41        return aerokube42    def parse_hosts(self) -> dict:43        hosts = self.config_file['hosts']44        return hosts45    def parse_teams_quota(self) -> dict:46        try:47            teams = self.config_file['teams-quota']48        except KeyError:49            teams = {}50        return teams51    def __del__(self) -> None:...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
