KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use disable_device method in lisa

Best Python code snippet using lisa_python

devices.py

Source:devices.py

...358    def del_user(self, _id: int, disable_device=True):359        if not self._connected:360            raise Exception("you need to connect to machine first")361        if disable_device:362            self.disable_device()363        try:364            self.send_cmd("del_user", payload=struct.pack("<H", _id))365        except Exception as ex:366            raise ex367        finally:368            if disable_device:369                self.enable_device()370    @property371    def database_structure(self):372        """373        return database structure (tables)374        :return:375        """376        if not self._connected:377            raise Exception("you need to connect to machine first")378        self.send_cmd("get_table_struct", res_cmd_name=["recv_buff_header"])379        self.receive()380        self.verify_response(["recv_buff_content"])381        data = self._response382        self.receive()383        self.verify_response()384        # footer = self._response385        return data.payload.decode('latin-1')386    def get_fp_data(self, user_id: int, finger_id: int, disable_device=True):387        """388        get fingerprint data389        :param user_id: user serial number390        :param finger_id: finger id391        :param disable_device: disable device during the command execution392        :return: fingerprint data in bytes393        """394        if not self._connected:395            raise Exception("you need to connect to machine first")396        if disable_device:397            self.disable_device()398        try:399            self.send_cmd("get_fp", struct.pack("<HB", user_id, finger_id), ["recv_buff_header", "nak", "no_data"])400            if self._response.cmd == device_cmd["nak"]:401                return b''402            elif self._response.cmd == device_cmd["no_data"]:403                return b''404            else:405                self.receive()406                self.verify_response(["recv_buff_content"])407                data = self._response408                self.receive()409                self.verify_response()410                # footer = self._response411                return data.payload[:-6]412        except Exception as ex:413            raise ex414        finally:415            if disable_device:416                self.enable_device()417    def send_file(self, file_name: str, file_data: bytes, disable_device=True):418        """419        send file to the device420        :param file_name: name of the file421        :param file_data: file content in bytes422        :param disable_device: disable device during the command execution423        :return:424        """425        if not self._connected:426            raise Exception("you need to connect to machine first")427        if disable_device:428            self.disable_device()429        try:430            self._upload_data(file_data)431            file_name = file_name.encode(self._encoding)432            file_name = file_name + struct.pack("<{}x".format(39 - len(file_name)))433            payload = struct.pack("<I", 0x6A4) + file_name434            self.send_cmd("send_file", payload, ["ack", "nak"])435            ret = self._response.cmd == device_cmd["ack"]436            self.send_cmd("end_buff_stream")437            if not ret:438                raise Exception("operation failed")439        finally:440            if disable_device:441                self.enable_device()442        self.send_cmd("save_data")443    def set_user_pic(self, person_id: str, pic_data: bytes, disable_device=True):444        """445        set picture of the user446        :param person_id: string that represent person id447        :param pic_data: picture data in bytes448        :param disable_device: disable device during the command execution449        :return:450        """451        if not self._connected:452            raise Exception("you need to connect to machine first")453        if disable_device:454            self.disable_device()455        try:456            self._upload_data(pic_data)457            file_name = person_id.encode(self._encoding) + b".jpg"458            file_name = file_name + struct.pack("<{}x".format(39 - len(file_name)))459            payload = file_name + bytes([0] * 4)460            self.send_cmd("set_user_pic", payload)461            self.send_cmd("end_buff_stream")462        finally:463            if disable_device:464                self.enable_device()465        self.send_cmd("save_data")466    def del_user_pic(self, person_id: str, disable_device=True):467        """468        delete user picture469        :param person_id: person id470        :param disable_device: disable device during the command execution471        :return:472        """473        if not self._connected:474            raise Exception("you need to connect to machine first")475        if disable_device:476            self.disable_device()477        try:478            self.send_cmd("del_user_pic", person_id.encode(self._encoding) + b".jpg\x00", ["ack", "no_pic"])479            if self._response.cmd == device_cmd["no_pic"]:480                return False481            return True482        finally:483            if disable_device:484                self.enable_device()485            self.send_cmd("save_data")486    def get_user_pic(self, person_id: str):487        """488        return bytes of user picture489        :param person_id: person id490        :return: picture data in bytes491        """492        if not self._connected:493            raise Exception("you need to connect to machine first")494        self.send_cmd("get_user_pic", person_id.encode(self._encoding) + b".jpg\x00",495                      ["recv_buff_header", "nak", "no_pic"])496        if self._response.cmd != device_cmd["recv_buff_header"]:497            return b''498        data_size, _ = struct.unpack("<II", self._response.payload[:8])499        self.receive()500        self.verify_response(["recv_buff_content"])501        data = self._response502        self.receive()503        self.verify_response()504        _ = self._response505        if len(data.payload) != data_size:506            raise Exception("data integrity is incorrect")507        return data.payload508    def del_op_logs(self):509        """510        delete all op logs from the device511        :return:512        """513        self.send_cmd("clear_op_log")514        self.send_cmd("save_data")515    def del_users(self):516        """517        delete all users from the device518        :return:519        """520        self.clear_data(0x05)521    def del_fps(self):522        """523        delete all fingerprints from the device524        :return:525        """526        self.clear_data(0x02)527    def del_admins(self):528        """529        clear all device admins530        :return:531        """532        if not self._connected:533            raise Exception("you need to connect to machine first")534        self.send_cmd("cls_admins")535        self.send_cmd("save_data")536    def get_photo_count(self):537        """538        return number of photo store in the device539        :return: number of photo540        """541        if not self._connected:542            raise Exception("you need to connect to machine first")543        self.send_cmd("get_photo_count")544        if len(self._response.payload) == 1:545            return struct.unpack("<B", self._response.payload)[0]546        elif len(self._response.payload) == 2:547            return struct.unpack("<H", self._response.payload)[0]548        elif len(self._response.payload) == 4:549            return struct.unpack("<I", self._response.payload)[0]550        else:551            raise Exception("unknown response format for photo count")552    def get_state(self, disable_device=True):553        """554        get the machine state and capacity555        :param disable_device: disable device during the command execution556        :return: structure describe machine state and capacity557        """558        if not self._connected:559            raise Exception("you need to connect to machine first")560        if disable_device:561            self.disable_device()562        try:563            self.send_cmd("logs_count")564            if len(self._response.payload or b'') < (23 * 4):565                raise Exception("error in logs_count command")566            return models.MachineState.from_bytes(self._response.payload)567        except Exception as ex:568            raise ex569        finally:570            if disable_device:571                self.enable_device()572    def del_att_logs(self, disable_device=True):573        """574        clear all attendance logs575        :param disable_device:576        :return:577        """578        if not self._connected:579            raise Exception("you need to connect to machine first")580        if disable_device:581            self.disable_device()582        try:583            self.send_cmd("del_logs")584        except Exception as ex:585            raise ex586        finally:587            if disable_device:588                self.enable_device()589        self.send_cmd("save_data")590    def disable_device(self, time_out_in_sec=0):591        """592        disable the device for time period593        :param time_out_in_sec: number of second to disable the device 0 mean forever until you call enable_device594        :return:595        """596        if not self._connected:597            raise Exception("you need to connect to machine first")598        self.send_cmd("disable", struct.pack("<I", time_out_in_sec))599    def enable_device(self):600        """601        enable the device602        :return:603        """604        if not self._connected:605            raise Exception("you need to connect to machine first")606        self.send_cmd("enable")607    def check_hash(self):608        """609        use as follows after send data from client to machine using header-content-footer610        use this function optionally to return the hash of the buffer received by device and611        compare it with hash you make from buffer before send612        """613        if not self._connected:614            raise Exception("you need to connect to machine first")615        self.send_cmd("check_hash")616        return struct.unpack("<I", self._response.payload)617    def get_fps(self, disable_device=True):618        """619        return all fingerprints that stored in the device620        :param disable_device:621        :return: list of FPInfo struct622        """623        ret = self._get_data_buffer(0x02000701, disable_device)624        if ret is not None:625            ret.remove_leading()626            return ret.fps627    def del_user_face(self, person_id: str, face_id=50, disable_device=True):628        """629        delete user face630        :param person_id: person id631        :param face_id: face id default = 50632        :param disable_device: disable device during the command execution633        :return:634        """635        if not self._connected:636            raise Exception("you need to connect to machine first")637        if disable_device:638            self.disable_device()639        try:640            payload = [0] * 28641            temp = person_id.encode(self._encoding)642            length = len(temp)643            if length > 24:644                length = 24645            payload[:length] = temp[:length]646            payload[25] = face_id647            self.send_cmd("del_face", bytes(payload))648        finally:649            if disable_device:650                self.enable_device()651        self.send_cmd("save_data")652    def get_user_face(self, person_id: str, face_id=50, disable_device=True):653        """654        return face data in bytes655        :param person_id: person id656        :param face_id: face id default = 50657        :param disable_device: disable device during the command execution658        :return: face data in bytes659        """660        if not self._connected:661            raise Exception("you need to connect to machine first")662        if disable_device:663            self.disable_device()664        try:665            # payload = [0] * 28666            person_id = person_id.encode(self._encoding)667            payload = person_id + struct.pack("<{}xxB2x".format(24 - len(person_id)), face_id)668            self.send_cmd("get_face", payload, ["recv_buff_header", "nak", "no_data"])669            if self._response.cmd == device_cmd["nak"]:670                return b''671            elif self._response.cmd == device_cmd["no_data"]:672                return b''673            else:674                self.receive()675                self.verify_response(["recv_buff_content"])676                data = self._response677                self.receive()678                self.verify_response()679                _ = self._response680                return data.payload681        finally:682            if disable_device:683                self.enable_device()684    def set_user_face(self, person_id: str, face_data: bytes, face_index=50, disable_device=True):685        """686        set user face687        :param person_id: person id688        :param face_data: face data in bytes689        :param face_index: face index default 50690        :param disable_device: disable device during the command execution691        :return:692        """693        if not self._connected:694            raise Exception("you need to connect to machine first")695        if disable_device:696            self.disable_device()697        try:698            self._upload_data(face_data)699            person_id = person_id.encode(self._encoding)700            payload = person_id + struct.pack("<{}xxBH".format(24 - len(person_id)), face_index, len(face_data))701            self.send_cmd("set_face", payload)702            self.send_cmd("end_buff_stream")703        finally:704            if disable_device:705                self.enable_device()706        self.send_cmd("save_data")707    # old function708    # def delFP(self, user_id, finger_id, disable_device=True):709    #     if not self._connected:710    #         raise Exception("you need to connect to machine first")711    #     if disable_device:712    #         self.disableDevice()713    #     try:714    #         self.sendCmd("del_fp", struct.pack("<HB", user_id, finger_id), ["ack", "nak", "no_fp"])715    #         ret = self._response.cmd == device_cmd["ack"]716    #         self.sendCmd("save_data")717    #         return ret718    #     finally:719    #         if disable_device:720    #             self.enableDevice()721    def set_user(self, user_info: models.UserInfo):722        """723        set user info at specific serial number724        :param user_info: contain user information to be set. serial field is the determinant of set position725        :return:726        """727        if not self._connected:728            raise Exception("you need to connect to machine first")729        self.send_cmd("set_user", bytes(user_info))730    def del_fp(self, person_id: str, finger_id: int, disable_device=True):731        """732        delete fingerprint of a user733        :param person_id: person id734        :param finger_id: finger id735        :param disable_device: disable device during the command execution736        :return:737        """738        if not self._connected:739            raise Exception("you need to connect to machine first")740        if disable_device:741            self.disable_device()742        try:743            payload = [0] * 25744            temp = person_id.encode(self._encoding)745            length = len(temp)746            if length > 24:747                length = 24748            payload[:length] = temp[:length]749            payload[24] = finger_id750            # if len(temp) >= 24:751            #     payload = temp[:24] + struct.pack("<B", finger_id)752            # else:753            #     remain = 24 - len(temp)754            #     fmt = "<{}xB".format(remain)755            #     payload = temp + struct.pack(fmt, finger_id)756            self.send_cmd("del_fp_ex", bytes(payload), ["ack", "nak", "no_fp"])757            ret = self._response.cmd == device_cmd["ack"]758        finally:759            if disable_device:760                self.enable_device()761        if ret:762            self.send_cmd("save_data")763        return ret764    def _upload_data(self, data: bytes):765        """766        upload data to the device it is intermediate function767        :param data: data to be uploaded in bytes768        :return:769        """770        self.send_cmd("recv_buff_header", struct.pack("<I", len(data)))771        self.send_cmd("recv_buff_content", data)772        buffer = DataBuffer(self._encoding, data)773        hash_code = hash(buffer)774        self.send_cmd("check_hash")775        if hash_code != struct.unpack("<I", self._response.payload)[0]:776            raise Exception("problem in data sending")777    def set_fp(self, fp_info: models.FPInfo, disable_device=True):778        """779        set fingerprint for a user780        :param fp_info: fingerprint information special user serial, finger id and fingerprint data in bytes781        :param disable_device:782        :return:783        """784        if not self._connected:785            raise Exception("you need to connect to machine first")786        if disable_device:787            self.disable_device()788        try:789            self._upload_data(fp_info.data)790            # self.sendCmd("recv_buff_header", struct.pack("<I", len(fp_info.data)))791            # self.sendCmd("recv_buff_content", fp_info.data)792            # buffer = DataBuffer(fp_info.data)793            # hash_code = hash(buffer)794            # self.sendCmd("check_hash")795            # if hash_code != struct.unpack("<I", self._response.payload)[0]:796            #     raise Exception("problem in data sending")797            payload = struct.pack("<HBBH", fp_info.user_id, fp_info.finger_id, fp_info.enabled, len(fp_info.data))798            self.send_cmd("set_fp_ex", payload)799            self.send_cmd("end_buff_stream")800        finally:801            if disable_device:802                self.enable_device()803        self.send_cmd("save_data")804    def set_fps(self, fp_infos: list, disable_device=True):805        """806        bulk set of fingerprints in the device807        :param fp_infos: list of all fingerprints808        :param disable_device: disable device during the command execution809        :return:810        """811        if not self._connected:812            raise Exception("you need to connect to machine first")813        if disable_device:814            self.disable_device()815        try:816            for fp_info in fp_infos:817                self.send_cmd("recv_buff_header", struct.pack("<I", len(fp_info.data)))818                self.send_cmd("recv_buff_content", fp_info.data)819                buffer = DataBuffer(self._encoding, fp_info.data)820                hash_code = hash(buffer)821                self.send_cmd("check_hash")822                if hash_code != struct.unpack("<I", self._response.payload)[0]:823                    raise Exception("problem in data sending")824                payload = struct.pack("<HBBH", fp_info.user_id, fp_info.finger_id, fp_info.enabled, len(fp_info.data))825                self.send_cmd("set_fp_ex", payload)826                self.send_cmd("end_buff_stream")827        finally:828            if disable_device:829                self.enable_device()830        self.send_cmd("save_data")831    def get_fp(self, user_id: int, finger_id: int, disable_device=True):832        """833        return fingerprint of user834        :param user_id: user serial835        :param finger_id: finger index836        :param disable_device: disable device during the command execution837        :return: fingerprint data in bytes838        """839        if not self._connected:840            raise Exception("you need to connect to machine first")841        if disable_device:842            self.disable_device()843        try:844            self.send_cmd("get_fp_ex", struct.pack("<HB", user_id, finger_id), ["recv_buff_header", "nak", "no_data"])845            if self._response.cmd == device_cmd["nak"]:846                return847            elif self._response.cmd == device_cmd["no_data"]:848                return849            else:850                self.receive()851                self.verify_response(["recv_buff_content"])852                data = self._response853                self.receive()854                self.verify_response()855                # footer = self._response856                return models.FPInfo(user_id=user_id, finger_id=finger_id, enabled=data.payload[-1],857                                     data=data.payload[:-7])858        except Exception as ex:859            raise ex860        finally:861            if disable_device:862                self.enable_device()863    def get_users(self, disable_device=True):864        """865        get all users866        :param disable_device: disable device during the command execution867        :return: list of UserInfo868        """869        ret = self._get_data_buffer(0x05000901, disable_device)870        if ret is not None:871            ret.remove_leading()872            return ret.users873    def get_op_logs(self, disable_device=True):874        """875        get all op logs876        :param disable_device: disable device during the command execution877        :return: list of OpLog struct878        """879        ret = self._get_data_buffer(0x2201, disable_device)880        if ret is not None:881            ret.remove_leading()882            return ret.op_logs883    def get_att_logs(self, disable_device=True):884        """885        get all attendance logs886        :param disable_device: disable device during the command execution887        :return: list of AttLog888        """889        ret = self._get_data_buffer(0x0D01, disable_device)890        if ret is not None:891            ret.remove_leading()892            return ret.att_logs893    def _get_data_buffer(self, data_id: int, disable_device=True):894        """895        read data buffer from the stream896        :param data_id: data type to be read897        :param disable_device: disable device during the command execution898        :return: DataBuffer object899        """900        if not self._connected:901            raise Exception("you need to connect to machine first")902        if disable_device:903            self.disable_device()904        try:905            data_buffer = DataBuffer(self._encoding)906            payload = struct.pack("<I7x", data_id)907            self.send_cmd("start_buff_stream", payload, ["ack", "nak", "recv_buff_content", "no_record"])908            if self._response.cmd == device_cmd["recv_buff_content"]:909                data_buffer.append(self._response.payload)910                return data_buffer911            elif self._response.cmd in [device_cmd["nak"], device_cmd["no_record"]]:912                return data_buffer913            else:914                data_size = struct.unpack("<I", self._response.payload[1:5])[0]915                packet_size = 0xFFC0916                packets = data_size // packet_size917                rem_size = data_size...

systemdatahandler.py

Source:systemdatahandler.py

1"""2Various constants used for the system data handler library.3The constant values are not fully documented here, see the4`constants source code <https://github.com/yombo/yombo-gateway/blob/master/yombo/lib/systemdatahandler/constants.py>`_5for a full list.6.. moduleauthor:: Mitch Schwenk <mitch-gw@yombo.net>7:copyright: Copyright 2018-2020 by Yombo.8:license: LICENSE for details.9:view-source: `View Source Code <https://github.com/yombo/yombo-gateway/blob/master/yombo/lib/systemdatahandler/constants.py>`_10"""11CONFIG_ITEM_MAP = {12    "devices": "gateway_devices"13}14CONFIG_ITEMS = {15    "categories": {16        "dbclass": "Category",17        "schemaclass": "CategorySchema",18        "table": "categories",19        "library": "categories",20        "functions": {21            # "process": "enable_command",22            # "enabled": "enable_device",23            # "disabled": "disable_device",24            # "deleted": "delete_device",25        },26        "purgeable": False,27        "map": {  # api name : database field name28            "id": "id",29            "category_parent_id": "category_parent_id",30            "category_type": "category_type",31            "machine_label": "machine_label",32            "label": "label",33            "description": "description",34            "status": "status",35            "created_at": "created_at",36            "updated_at": "updated_at",37            # "": "",38        }39    },40    "commands": {41        "dbclass": "Command",42        "schemaclass": "CommandSchema",43        "table": "commands",44        "library": "commands",45        "functions": {46            # "process": "enable_command",47            # "enabled": "enable_device",48            # "disabled": "disable_device",49            # "deleted": "delete_device",50        },51        "purgeable": False,52        "map": {  # api name : database field name53            "id": "id",54            "user_id": "user_id",55            "original_user_id": "original_user_id",56            "machine_label": "machine_label",57            "label": "label",58            "description": "description",59            "public": "public",60            "status": "status",61            "created_at": "created_at",62            "updated_at": "updated_at",63        }64    },65    "devices": {66        "dbclass": "Device",67        "schemaclass": "DeviceSchema",68        "table": "devices",69        "library": "devices",70        "functions": {71            # "enabled": "enable_device",72            # "disabled": "disable_device",73            # "deleted": "delete_device",74        },75        "purgeable": True,76        "map": {  # api name : database field name77            "id": "id",78            "gateway_id": "gateway_id",79            "user_id": "user_id",80            "device_type_id": "device_type_id",81            "machine_label": "machine_label",82            "label": "label",83            "description": "description",84            "location_id": "location_id",85            "area_id": "area_id",86            "notes": "notes",87            "attributes": "attributes",88            "intent_allow": "intent_allow",89            "intent_text": "intent_text",90            "pin_code": "pin_code",91            "pin_required": "pin_required",92            "pin_timeout": "pin_timeout",93            "statistic_label": "statistic_label",94            "statistic_lifetime": "statistic_lifetime",95            "statistic_type": "statistic_type",96            "statistic_bucket_size": "statistic_bucket_size",97            "energy_type": "energy_type",98            "energy_tracker_source_type": "energy_tracker_source_type",99            "energy_tracker_source_id": "energy_tracker_source_id",100            "energy_map": "energy_map",101            "scene_controllable": "scene_controllable",102            "allow_direct_control": "allow_direct_control",103            "status": "status",104            "created_at": "created_at",105            "updated_at": "updated_at",106        }107    },108    "device_command_inputs": {109        "dbclass": "DeviceCommandInput",110        "schemaclass": "DeviceCommandInputSchema",111        "table": "device_command_inputs",112        "library": None,113        "functions": {114            # "enabled": "enable_device",115            # "disabled": "disable_device",116            # "deleted": "delete_device",117        },118        "purgeable": False,119        "map": {  # api name : database field name120            "id": "id",121            "device_type_id": "device_type_id",122            "command_id": "command_id",123            "input_type_id": "input_type_id",124            "machine_label": "machine_label",125            "label": "label",126            "live_update": "live_update",127            "value_required": "value_required",128            "value_max": "value_max",129            "value_min": "value_min",130            "value_casing": "value_casing",131            "encryption": "encryption",132            "notes": "notes",133            "created_at": "created_at",134            "updated_at": "updated_at",135        }136    },137    "device_types": {138        "dbclass": "DeviceType",139        "schemaclass": "DeviceTypeSchema",140        "table": "device_types",141        "library": "devicestypes",142        "functions": {143            # "enabled": "enable_device",144            # "disabled": "disable_device",145            # "deleted": "delete_device",146        },147        "purgeable": False,148        "map": {  # api name : database field name149            "id": "id",150            "user_id": "user_id",151            "original_user_id": "original_user_id",152            "category_id": "category_id",153            "machine_label": "machine_label",154            "label": "label",155            "description": "description",156            "public": "public",157            "status": "status",158            "created_at": "created_at",159            "updated_at": "updated_at",160        }161    },162    "device_type_commands": {163        "dbclass": "DeviceTypeCommand",164        "schemaclass": "DeviceTypeCommandSchema",165        "table": "device_type_commands",166        "library": None,167        "functions": {168            # "enabled": "enable_device",169            # "disabled": "disable_device",170            # "deleted": "delete_device",171        },172        "purgeable": False,173        "map": {  # api name : database field name174            "id": "id",175            "device_type_id": "device_type_id",176            "command_id": "command_id",177            "created_at": "created_at",178        }179    },180    "gateways": {181        "dbclass": "Gateway",182        "schemaclass": "GatewaySchema",183        "table": "gateways",184        "library": "gateways",185        "functions": {186            # "process": "enable_command",187            # "enabled": "enable_device",188            # "disabled": "disable_device",189            # "deleted": "delete_device",190        },191        "purgeable": False,192        "map": {  # api name : database field name193            "id": "id",194            "machine_label": "machine_label",195            "label": "label",196            "description": "description",197            "user_id": "user_id",198            "mqtt_auth": "mqtt_auth",199            "mqtt_auth_next": "mqtt_auth_next",200            "mqtt_auth_last_rotate_at": "mqtt_auth_last_rotate_at",201            "internal_ipv4": "internal_ipv4",202            "external_ipv4": "external_ipv4",203            "internal_ipv6": "internal_ipv6",204            "external_ipv6": "external_ipv6",205            "internal_http_port": "internal_http_port",206            "external_http_port": "external_http_port",207            "internal_http_secure_port": "internal_http_secure_port",208            "external_http_secure_port": "external_http_secure_port",209            "internal_mqtt": "internal_mqtt",210            "internal_mqtt_le": "internal_mqtt_le",211            "internal_mqtt_ss": "internal_mqtt_ss",212            "internal_mqtt_ws": "internal_mqtt_ws",213            "internal_mqtt_ws_le": "internal_mqtt_ws_le",214            "internal_mqtt_ws_ss": "internal_mqtt_ws_ss",215            "externalmqtt": "externalmqtt",216            "externalmqtt_le": "externalmqtt_le",217            "externalmqtt_ss": "externalmqtt_ss",218            "externalmqtt_ws": "externalmqtt_ws",219            "externalmqtt_ws_le": "externalmqtt_ws_le",220            "externalmqtt_ws_ss": "externalmqtt_ws_ss",221            "is_master": "is_master",222            "master_gateway_id": "master_gateway_id",223            "dns_name": "dns_name",224            "status": "status",225            "created_at": "created_at",226            "updated_at": "updated_at",227        }228    },229    # "gateway_dns_name": {230    #     "dbclass": "none",231    #     "table": "none",232    #     "library": None,233    #     "functions": {234    #     },235    #     "purgeable": False,236    #     "map": {  # api name : database field name237    #     }238    # },239    "input_types": {240        "dbclass": "InputType",241        "schemaclass": "InputTypeSchema",242        "table": "input_types",243        "library": "inputtypes",244        "functions": {245            # "enabled": "enable_device",246            # "disabled": "disable_device",247            # "deleted": "delete_device",248        },249        "purgeable": False,250        "map": {  # api name : database field name251            "id": "id",252            "user_id": "user_id",253            "original_user_id": "original_user_id",254            "category_id": "category_id",255            "machine_label": "machine_label",256            "label": "label",257            "description": "description",258            "public": "public",259            "status": "status",260            "created_at": "created_at",261            "updated_at": "updated_at",262        }263    },264    "locations": {265        "dbclass": "Location",266        "schemaclass": "LocationSchema",267        "table": "locations",268        "library": None,269        "functions": {270            # "enabled": "enable_device",271            # "disabled": "disable_device",272            # "deleted": "delete_device",273        },274        "purgeable": False,275        "map": {  # api name : database field name276            "id": "id",277            "user_id": "user_id",278            "location_type": "location_type",279            "machine_label": "machine_label",280            "label": "label",281            "description": "description",282            "created_at": "created_at",283            "updated_at": "updated_at",284        }285    },286    "modules": {287        "dbclass": "Modules",288        "schemaclass": "ModuleSchema",289        "table": "modules",290        "library": "modules",291        "functions": {292            # "enabled": "enable_command",293            # "disabled": "enable_command",294            # "deleted": "enable_command",295        },296        "purgeable": True,297        "map": {  # api name : database field name298            "id": "id",299            "user_id": "user_id",300            "original_user_id": "original_user_id",301            "module_type": "module_type",302            "machine_label": "machine_label",303            "label": "label",304            "short_description": "short_description",305            "medium_description": "medium_description",306            "description": "description",307            "medium_description_html": "medium_description_html",308            "description_html": "description_html",309            "see_also": "see_also",310            "repository_link": "repository_link",311            "issue_tracker_link": "issue_tracker_link",312            "install_count": "install_count",313            "doc_link": "doc_link",314            "git_link": "git_link",315            "git_auto_approve": "git_auto_approve",316            "install_branch": "install_branch",317            "require_approved": "require_approved",318            "public": "public",319            "status": "status",320            "created_at": "created_at",321            "updated_at": "updated_at",322        }323    },324    "module_commits": {325        "dbclass": "ModuleCommits",326        "schemaclass": "ModuleCommitSchema",327        "table": "module_commits",328        "library": "modules",329        "functions": {330            # "enabled": "enable_command",331            # "disabled": "enable_command",332            # "deleted": "enable_command",333        },334        "purgeable": True,335        "map": {  # api name : database field name336            "id": "id",337            "module_id": "module_id",338            "branch": "branch",339            "commit": "commit",340            "committed_at": "committed_at",341            "approved": "approved",342            "approved_at": "approved_at",343            "created_at": "created_at",344        }345    },346    "module_device_types": {347        "dbclass": "ModuleDeviceTypes",348        "schemaclass": "ModuleDeviceTypeSchema",349        "table": "module_device_types",350        "library": None,351        "functions": {352            # "enabled": "enable_device",353            # "disabled": "disable_device",354            # "deleted": "delete_device",355        },356        "purgeable": False,357        "map": {  # api name : database field name358            "id": "id",359            "module_id": "module_id",360            "device_type_id": "device_type_id",361            "created_at": "created_at",362        }363    },364    "nodes": {365        "dbclass": "Node",366        "schemaclass": "NodeSchema",367        "table": "nodes",368        "library": None,369        "functions": {370            # "enabled": "enable_device",371            # "disabled": "disable_device",372            # "deleted": "delete_device",373        },374        "purgeable": True,375        "map": {  # api name : database field name376            "id": "id",377            "node_parent_id": "node_parent_id",378            "node_id": "node_id",379            "gateway_id": "gateway_id",380            "node_type": "node_type",381            "weight": "weight",382            "label": "label",383            "machine_label": "machine_label",384            "always_load": "always_load",385            "destination": "destination",386            "data": "data",387            "data_content_type": "data_content_type",388            "status": "status",389            "created_at": "created_at",390            "updated_at": "updated_at",391        }392    },393    "gateway_users": {394        "dbclass": "Users",395        "schemaclass": "UserSchema",396        "table": "users",397        "library": None,398        "functions": {399        },400        "purgeable": True,401        "map": {  # api name : database field name402            "user_id": "id",403            "gateway_id": "gateway_id",404            "email": "email",405            "name": "name",406            "access_code_string": "access_code_string",407            "created_at": "created_at",408            "updated_at": "updated_at",409        }410    },411    "variable_groups": {412        "dbclass": "VariableGroups",413        "schemaclass": "VariableGroupSchema",414        "table": "variable_groups",415        "library": "configuration",416        "functions": {417        },418        "purgeable": False,419        "map": {  # api name : database field name420            "id": "id",421            "relation_id": "group_relation_id",422            "relation_type": "group_relation_type",423            "group_machine_label": "group_machine_label",424            "group_label": "group_label",425            "group_description": "group_description",426            "group_weight": "group_weight",427            "status": "status",428            "created_at": "created_at",429            "updated_at": "updated_at",430        }431    },432    "variable_fields": {433        "dbclass": "VariableFields",434        "schemaclass": "VariableFieldSchema",435        "table": "variable_fields",436        "library": "configuration",437        "functions": {438        },439        "purgeable": False,440        "map": {  # api name : database field name441            "id": "id",442            "user_id": "user_id",443            "variable_group_id": "variable_group_id",444            "field_machine_label": "field_machine_label",445            "field_label": "field_label",446            "field_description": "field_description",447            "field_weight": "field_weight",448            "value_required": "value_required",449            "value_max": "value_max",450            "value_min": "value_min",451            "value_casing": "value_casing",452            "encryption": "encryption",453            "input_type_id": "input_type_id",454            "default_value": "default_value",455            "field_help_text": "field_help_text",456            "multiple": "multiple",457            "created_at": "created_at",458            "updated_at": "updated_at",459        }460    },461    "variables": {462        "dbclass": "VariableData",463        "schemaclass": "VariableDataSchema",464        "table": "variable_data",465        "library": "configuration",466        "functions": {467        },468        "purgeable": True,469        "map": {  # api name : database field name470            "id": "id",471            "user_id": "user_id",472            "gateway_id": "gateway_id",473            "variable_field_id": "variable_field_id",474            "variable_relation_id": "variable_relation_id",475            "variable_relation_type": "variable_relation_type",476            "data": "data",477            "data_content_type": "data_content_type",478            "data_weight": "data_weight",479            "created_at": "created_at",480            "updated_at": "updated_at",481        }482    },...

traps.py

Source:traps.py

1def create_trap_details_table(curs):2	sql = '\n'.join([3		"CREATE TABLE trap_details (",4		"  trap_details_id INTEGER PRIMARY KEY,",5		"  section_id INTEGER NOT NULL,",6		"  cr TEXT,",7		"  trap_type TEXT,",8		"  perception TEXT,",9		"  disable_device TEXT,",10		"  duration TEXT,",11		"  effect TEXT,",12		"  trigger TEXT,",13		"  reset TEXT",14		")"])15	curs.execute(sql)16def create_trap_details_index(curs):17	sql = '\n'.join([18		"CREATE INDEX trap_details_section_id",19		" ON trap_details (section_id)"])20	curs.execute(sql)21def insert_trap_detail(curs, section_id, cr=None, trap_type=None, perception=None, disable_device=None, duration=None, effect=None, trigger=None, reset=None, **kwargs):22	values = [section_id, cr, trap_type, perception, disable_device, duration, effect, trigger, reset]23	sql = '\n'.join([24		"INSERT INTO trap_details",25		" (section_id, cr, trap_type, perception, disable_device, duration, effect, trigger, reset)",26		" VALUES",27		" (?, ?, ?, ?, ?, ?, ?, ?, ?)"])28	curs.execute(sql, values)29def delete_trap_detail(curs, section_id):30	values = [section_id]31	sql = '\n'.join([32		"DELETE FROM trap_details",33		" WHERE section_id = ?"])34	curs.execute(sql, values)35def fetch_trap_detail(curs, section_id):36	values = [section_id]37	sql = '\n'.join([38		"SELECT *",39		" FROM trap_details",40		" WHERE section_id = ?"])...

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.