Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _validate_keys method in gabbi

Best Python code snippet using gabbi_python

packagejson.py

Source:packagejson.py

...168                if self.json['identity']['domogik_min_version'] > DMG_VERSION:169                    raise PackageException("Domogik version check failed! min_version={0} current version={1}".format(self.json['identity']['domogik_min_version'], DMG_VERSION));170        else:171            return False172    def _validate_keys(self, expected, name, lst, optional=[]):173        for exp in expected:174            if exp not in lst:175                raise PackageException("key '{0}' not found in {1}".format(exp, name))176        explst = expected + optional177        for item in lst:178            if item not in explst:179                raise PackageException("unknown key '{0}' found in {1}".format(item, name))180    def _validate_dataType(self, msg, dataType):181        if dataType not in self._datatypes:182            raise PackageException(msg)183    def _validate_02(self):184        fieldTypes = ["boolean", "string", "choice", "date", "time", "datetime", "float", "integer", "email", "ipv4", "ipv6", "url", "password"]185        try:186            #check that all main keys are in the file187            expected = ["configuration", "xpl_commands", "xpl_stats", "commands", "sensors", "device_types", "identity", "json_version"]188            self._validate_keys(expected, "file", self.json.keys(), ["products", "external"])189            # validate identity190            expected = ["author", "author_email", "description", "domogik_min_version", "name", "type", "version"]191            optional = ["tags", "dependencies", "package_id", "icon_file", "xpl_clients_only", "compliant_xpl_clients"]192            if type(self.json["identity"]) != dict:193                raise PackageException("Identity part is NOT a dictionary!")194            self._validate_keys(expected, "an identity param", self.json["identity"].keys(), optional)195            # validate configuration196            expected = ["default", "description", "key", "name", "required", "type"]197            optional = ["sort", "max_value", "min_value", "choices", "mask", "multiline"]198            if type(self.json["configuration"]) != list:199                raise PackageException("Configuration part is NOT a list!")200            for conf in self.json["configuration"]:201                self._validate_keys(expected, "a configuration item param", conf.keys(), optional)202                if conf['type'] not in fieldTypes:203                    raise PackageException("Type ({0}) in a config item is not in the allowed list: {1}".format(conf['type'], fieldTypes))204            # validate products205            if 'products' in self.json.keys():206                expected = ["name", "id", "documentation", "type"]207                for prod in self.json['products']:208                    self._validate_keys(expected, "a product", prod.keys())209            #validate the device_type210            if type(self.json["device_types"]) != dict:211                raise PackageException("Device_types part is NOT a dictionary!")212            for devtype in self.json["device_types"]:213                devt = self.json["device_types"][devtype]214                expected = ['id', 'name', 'description', 'commands', 'sensors', 'parameters']215                self._validate_keys(expected, "device_type {0}".format(devtype), devt.keys())216                #check that all commands exists inisde each device_type217                if type(devt["commands"]) != list:218                    raise PackageException("Commands list for device_type {0} is NOT a list!".format(devtype))219                for cmd in devt["commands"]:220                    if cmd not in self.json["commands"].keys():    221                        raise PackageException("cmd {0} defined in device_type {1} is not found".format(cmd, devtype))222                #check that all sensors exists inside each device type223                if type(devt["sensors"]) != list:224                    raise PackageException("Sensors list for device_type {0} is NOT a list!".format(devtype))225                for sens in devt["sensors"]:226                    if sens not in self.json["sensors"].keys():    227                        raise PackageException("sensor {0} defined in device_type {1} is not found".format(sens, devtype))228                #see that each xplparam inside device_type has the following keys: key, description, type229                expected = ["key", "type", "description", "xpl"]230                optional = ["max_value", "min_value", "choices", "mask", "multiline", "default"]231                if type(devt["parameters"]) != list:232                    raise PackageException("Parameters list for device_type {0} is NOT a list!".format(devtype))233                for par in devt["parameters"]:234                    self._validate_keys(expected, "a param for device_type {0}".format(devtype), par.keys(), optional)235                    if par['type'] not in fieldTypes:236                        raise PackageException("Type ({0}) in a config item is not in the allowed list: {1}".format(par['type'], fieldTypes))237            #validate the commands238            if type(self.json["commands"]) != dict:239                raise PackageException("Commands part is NOT a dictionary!")240            for cmdid in self.json["commands"]:241                cmd = self.json["commands"][cmdid]242                expected = ['name', 'return_confirmation', 'parameters']243                optional = ['xpl_command']244                self._validate_keys(expected, "command {0}".format(cmdid), cmd.keys(), optional)245                # validate the params246                expected = ['key', 'data_type', 'conversion']247                if type(cmd['parameters']) != list:248                    raise PackageException("Parameters for command {0} is not a list".format(cmdid))249                for par in cmd['parameters']:250                    self._validate_keys(expected, "a param for command {0}".format(cmdid), par.keys())251                    self._validate_dataType("DataType in command {0} is not valid".format(cmdid), par['data_type'])252                # see that the xpl_command is defined253                if "xpl_command" in cmd and cmd["xpl_command"] not in self.json["xpl_commands"].keys():254                    raise PackageException("xpl_command {0} defined in command {1} is not found".format(cmd["xpl_command"], cmdid))255            #validate the sensors256            if type(self.json["sensors"]) != dict:257                raise PackageException("Sensor part is NOT a dictionary!")258            for senid in self.json["sensors"]:259                sens = self.json["sensors"][senid]260                expected = ['name', 'data_type', 'conversion', 'history', 'incremental', 'timeout']261                hexpected = ['store', 'max', 'expire', 'round_value', 'duplicate']262                self._validate_keys(expected, "sensor {0}".format(senid), list(sens.keys()))263                self._validate_keys(hexpected, "sensor {0} history".format(senid), list(sens['history'].keys()))264                self._validate_dataType("DataType in sensor {0} is not valid".format(senid), sens['data_type'])265            #validate the xpl command266            if type(self.json["xpl_commands"]) != dict:267                raise PackageException("Xpl_commands part is NOT a dictionary!")268            for xcmdid in self.json["xpl_commands"]:269                xcmd = self.json["xpl_commands"][xcmdid]270                expected = ["name", "schema", "xplstat_name", "parameters"]271                self._validate_keys(expected, "xpl_command {0}".format(xcmdid), xcmd.keys())272                # parameters273                expected = ["static", "device"]274                self._validate_keys(expected, "parameters for xpl_command {0}".format(xcmdid), xcmd['parameters'].keys())275                # static parameter276                expected = ["key", "value"]277                if type(xcmd['parameters']['static']) != list:278                    raise PackageException("Static parameters for xpl_command {0} is not a list".format(xcmdid))279                for stat in xcmd['parameters']['static']:280                    self._validate_keys(expected, "a static parameter for xpl_command {0}".format(xcmdid), stat.keys())281                # device parameter282                expected = ["key", "description", "type"]283                optional = ["default"]284                if type(xcmd['parameters']['device']) != list:285                    raise PackageException("Device parameters for xpl_command {0} is not a list".format(xcmdid))286                for stat in xcmd['parameters']['device']:287                    self._validate_keys(expected, "a device parameter for xpl_command {0}".format(xcmdid), stat.keys(), optional)288                    if stat['type'] not in fieldTypes:289                        raise PackageException("Type ({0}) in a config item is not in the allowed list: {1}".format(stat['type'], fieldTypes))290                # see that the xpl_stat is defined291                if xcmd["xplstat_name"] not in self.json["xpl_stats"].keys():292                    raise PackageException("xplstat_name {0} defined in xpl_command {1} is not found".format(xcmd["xplstat_name"], xcmdid))293            #validate the xpl stats294            if type(self.json["xpl_stats"]) != dict:295                raise PackageException("Xpl_stats part is NOT a dictionary!")296            for xstatid in self.json["xpl_stats"]:297                xstat = self.json["xpl_stats"][xstatid]298                expected = ["name", "schema", "parameters"]299                self._validate_keys(expected, "xpl_command {0}".format(xstatid), xstat.keys())300                # parameters301                expected = ["static", "device", "dynamic"]302                self._validate_keys(expected, "parameters for xpl_stat {0}".format(xstatid), xstat['parameters'].keys())303                # static parameter304                expected = ["key", "value"]305                if type(xstat['parameters']['static']) != list:306                    raise PackageException("Static parameters for xpl_stat {0} is not a list".format(xstatid))307                for stat in xstat['parameters']['static']:308                    self._validate_keys(expected, "a static parameter for xpl_stat {0}".format(xstatid), stat.keys())309                # device parameter310                expected = ["key", "description", "type"]311                optional = ["default", "multiple"]312                if type(xstat['parameters']['device']) != list:313                    raise PackageException("Device parameters for xpl_stat {0} is not a list".format(xstatid))314                for stat in xstat['parameters']['device']:315                    self._validate_keys(expected, "a device parameter for xpl_stat {0}".format(xstatid), stat.keys(), optional)316                    if stat['type'] not in fieldTypes:317                        raise PackageException("Type ({0}) in a config item is not in the allowed list: {1}".format(stat['type'], fieldTypes))318                # dynamic parameter319                expected = ["key", "sensor"]320                opt = ["ignore_values"]321                if type(xstat['parameters']['dynamic']) != list:322                    raise PackageException("Dynamic parameters for xpl_stat {0} is not a list".format(xstatid))323                for stat in xstat['parameters']['dynamic']:324                    self._validate_keys(expected, "a dynamic parameter for xpl_stat {0}".format(xstatid), stat.keys(), opt)325                    # check that the sensor exists326                    if stat['sensor'] not in self.json["sensors"].keys():    327                        raise PackageException("sensor {0} defined in xpl_stat {1} is not found".format(stat['sensor'], xstatid))328        except PackageException as exp:329            raise PackageException("Error validating the json: {0}".format(exp.value))330    def set_generated(self, path):331        """ Add generation date info in json data332            @param path : path to json file333        """334        my_json = json.load(open(path))335        my_json["identity"]["generated"] = str(datetime.datetime.now())336        my_file = open(path, "w")337        my_file.write(json.dumps(my_json))338        my_file.close()...

parser.py

Source:parser.py

...30META_DATA_KEYS = set(DEFAULT_KKT_CONFIG["meta_data"].keys())31META_DATA_KEYS.add("slug")32META_DATA_KEYS.add("code_file")33META_DATA_KEYS.add("competition")34def _validate_keys(obj: Dict[str, Any], mandatory_keys: List, path: str = "") -> None:35    """36    Raise MandatoryKeyNotFound if given obj doesn't contain all elements of mandatory_keys.37    >>> _validate_keys({"meta_data": {"slug": "slug", "code_file": "code_file"}}, MANDATORY_KEYS)38    # Comment out because pytest integration joesn't work39    #>>> _validate_keys({"meta_data": {"slug": "slug"}}, MANDATORY_KEYS)40    #MandatoryKeyNotFound('.meta_data.slug.code_file',)41    """42    for cur, children in mandatory_keys:43        path = "{}.{}".format(path, cur)44        if cur not in obj:45            raise MandatoryKeyNotFound(path)46        _validate_keys(obj[cur], children, path=path)47def _get_meta_data_keys(key: str) -> List[str]:48    """49    Return a list of keys to fetch partial meta_data from meta_data root.50    >>> _get_meta_data_keys(".")51    ['meta_data']52    >>> _get_meta_data_keys(".a.b.c")53    ['meta_data', 'a', 'b', 'c']54    """55    if re.match(r"^(\.|(\.[^.]+)+)$", key) is None:56        raise InvalidTarget(key)57    # key == "." means the root of meta_data58    return ["meta_data"] if key == "." else f"meta_data{key}".split(".")59def _trim_redundant_fields(meta_data: Dict[str, Any]) -> Dict[str, Any]:60    """61    Trim field whose key are not in META_DATA_KEYS.62    >>> _trim_redundant_fields({"slug": "slug", "redundant": "redundant"})63    {'slug': 'slug'}64    """65    return {k: v for k, v in meta_data.items() if k in META_DATA_KEYS}66def _compose_meta_data(kkt: Dict[str, Any], key: str) -> Dict[str, Any]:67    """68    Compose a meta_data dict by recurcive sampling and merging them.69    >>> kkt = {                                \70        "meta_data": {                         \71            "slug": "slug",                    \72            "nest1": {                         \73                "code_file": "script.py",      \74                "nest2": {                     \75                    "kernel_type" : "script"   \76                }                              \77            }                                  \78        }                                      \79    }80    >>> _compose_meta_data(kkt, ".nest1.nest2")81    {'slug': 'slug', 'code_file': 'script.py', 'kernel_type': 'script'}82    """83    def _aux(meta_data: Dict[str, Any], keys: List[str]) -> Dict[str, Any]:84        if len(keys) == 0:85            return meta_data86        try:87            meta_data = meta_data[keys[0]]88        except KeyError:89            raise MetaDataNotFound(key)90        return {**meta_data, **_aux(meta_data, keys[1:])}91    keys = _get_meta_data_keys(key)92    return _trim_redundant_fields(_aux(kkt, keys))93class PyprojectParser(TOMLFile):94    def __init__(self, path: Union[str, Path]) -> None:95        super().__init__(str(path))96        self._path = Path(path)97    @property98    def path(self) -> Path:99        return self._path100class KktParser(PyprojectParser):101    def __init__(self, path: Union[str, Path]) -> None:102        super().__init__(path)103    def read_all(self) -> Dict[str, Any]:104        return super().read()105    def read(self, key: str = ".") -> Dict[str, Any]:106        pyproj = super().read()107        try:108            kkt = pyproj["tool"]["kkt"]109        except (NonExistentKey, KeyError):110            raise KktSectionNotFound()111        # following code didn't work.112        # kkt["meta_data"] =  _compose_meta_data(kkt, key)113        tmp = _compose_meta_data(kkt, key)114        del kkt["meta_data"]115        kkt.add("meta_data", tmp)116        kkt = merge(DEFAULT_KKT_CONFIG, kkt)117        _validate_keys(kkt, MANDATORY_KEYS)118        return kkt119    def write(self, kkt_config: Dict[str, Any]) -> None:120        pyproj = super().read()121        pyproj.setdefault("tool", {})["kkt"] = kkt_config...

fields.py

Source:fields.py

...20            if not len(int_list) == 2: return False21        except:22            return False23        return True24    def _validate_keys(self, value):25        keys = value.keys() or []26        if not all(self._split_check_string(key) for key in keys):27            message = "key should be in format of 'row:colomn'"28            raise exceptions.ValidationError(message, code='invalid')29    def validate(self, value, model_instance):30        super().validate(value, model_instance)31        self._validate_schema(value)32        self._validate_keys(value)33    def pre_save(self, model_instance, add):34        value = super().pre_save(model_instance, add)35        if value and not self.null:36            self._validate_schema(value)37            self._validate_keys(value)...

Automation Testing Tutorials

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.