Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _get_json_result method in avocado

Best Python code snippet using avocado_python

daos_utils.py

Source:daos_utils.py

...29            dict: JSON output30        Raises:31            CommandFailure: if the daos pool query command fails.32        """33        return self._get_json_result(34            ("pool", "query"), pool=pool, sys_name=sys_name, sys=sys)35    def pool_autotest(self, pool):36        """Runs autotest for pool37        Args:38            pool (str): pool UUID39        Returns:40            CmdResult: Object that contains exit status, stdout, and other41                information.42        Raises:43            CommandFailure: if the daos pool autotest command fails.44        """45        return self._get_result(46            ("pool", "autotest"), pool=pool)47    def container_create(self, pool, sys_name=None, cont=None,48                         path=None, cont_type=None, oclass=None,49                         chunk_size=None, properties=None, acl_file=None):50        # pylint: disable=too-many-arguments51        """Create a container.52        Args:53            pool (str): UUID of the pool in which to create the container54            sys_name (str, optional):  DAOS system name context for servers.55                Defaults to None.56            cont (str, optional): container UUID. Defaults to None.57            path (str, optional): container namespace path. Defaults to None.58            cont_type (str, optional): the type of container to create. Defaults59                to None.60            oclass (str, optional): object class. Defaults to None.61            chunk_size (str, optional): chunk size of files created. Supports62                suffixes: K (KB), M (MB), G (GB), T (TB), P (PB), E (EB).63                Defaults to None.64            properties (str, optional): String of comma-separated <name>:<value>65                pairs defining the container properties. Defaults to None66            acl_file (str, optional): ACL file. Defaults to None.67        Returns:68            dict: the daos json command output converted to a python dictionary69        Raises:70            CommandFailure: if the daos container create command fails.71        """72        return self._get_json_result(73            ("container", "create"), pool=pool, sys_name=sys_name,74            cont=cont, path=path, type=cont_type, oclass=oclass,75            chunk_size=chunk_size, properties=properties, acl_file=acl_file)76    def container_clone(self, src, dst):77        """Clone a container to a new container.78        Args:79            src (str): the source, formatted as daos://<pool>/<cont>80            dst (str): the destination, formatted as daos://<pool>/<cont>81        Returns:82            CmdResult: Object that contains exit status, stdout, and other83                information.84        Raises:85            CommandFailure: if the daos container clone command fails.86        """87        return self._get_result(88            ("container", "clone"), src=src, dst=dst)89    def container_destroy(self, pool, cont, force=None, sys_name=None):90        """Destroy a container.91        Args:92            pool (str): UUID of the pool in which to create the container93            cont (str): container UUID.94            force (bool, optional): Force the container destroy. Defaults to95                None.96            sys_name (str, optional):  DAOS system name context for servers.97                Defaults to None.98        Returns:99            CmdResult: Object that contains exit status, stdout, and other100                information.101        Raises:102            CommandFailure: if the daos container destroy command fails.103        """104        return self._get_result(105            ("container", "destroy"), pool=pool, sys_name=sys_name,106            cont=cont, force=force)107    def container_check(self, pool, cont, sys_name=None, path=None):108        """Check the integrity of container objects.109        Args:110            pool (str): UUID of the pool in which to create the container111            cont (str): container UUID.112            sys_name (str, optional):  DAOS system name context for servers.113                Defaults to None.114            path (str): Container namespace path. Defaults to None115        Returns:116            CmdResult: Object that contains exit status, stdout, and other117                information.118        Raises:119            CommandFailure: if the daos container check command fails.120        """121        return self._get_result(122            ("container", "check"), pool=pool, cont=cont,123            sys_name=sys_name, path=path)124    def container_get_acl(self, pool, cont,125                          verbose=False, outfile=None):126        """Get the ACL for a given container.127        Args:128            pool (str): Pool UUID129            cont (str): Container for which to get the ACL.130            verbose (bool, optional): Verbose mode.131            outfile (str, optional): Write ACL to file.132        Returns:133            CmdResult: Object that contains exit status, stdout, and other134                information.135        Raises:136            CommandFailure: if the daos container get-acl command fails.137        """138        return self._get_result(139            ("container", "get-acl"), pool=pool, cont=cont,140            verbose=verbose, outfile=outfile)141    def container_delete_acl(self, pool, cont, principal):142        """Delete an entry for a given principal in an existing container ACL.143        Args:144            pool (str): Pool UUID145            cont (str): Container for which to get the ACL.146            principal (str): principal portion of the ACL.147        Returns:148            CmdResult: Object that contains exit status, stdout, and other149                information.150        Raises:151            CommandFailure: if the daos container delete-acl command fails.152        """153        return self._get_result(154            ("container", "delete-acl"), pool=pool, cont=cont,155            principal=principal)156    def container_overwrite_acl(self, pool, cont, acl_file):157        """Overwrite the ACL for a given container.158        Args:159            pool (str): Pool UUID160            cont (str): Container for which to get the ACL.161            acl_file (str): input file containing ACL162        Returns:163            CmdResult: Object that contains exit status, stdout, and other164                information.165        Raises:166            CommandFailure: if the daos container overwrite-acl command fails.167        """168        return self._get_result(169            ("container", "overwrite-acl"), pool=pool, cont=cont,170            acl_file=acl_file)171    def container_update_acl(self, pool, cont, entry=None, acl_file=None):172        """Add or update the ACL entries for a given container.173        Args:174            pool (str): Pool UUID175            cont (str): Container for which to get the ACL.176            entry (bool, optional): Add or modify a single ACL entry177            acl_file (str, optional): Input file containing ACL178        Returns:179            CmdResult: Object that contains exit status, stdout, and other180                information.181        Raises:182            CommandFailure: if the daos container get-acl command fails.183        """184        return self._get_result(185            ("container", "update-acl"), pool=pool, cont=cont,186            entry=entry, acl_file=acl_file)187    def container_list(self, pool, sys_name=None):188        """List containers in the given pool.189        Args:190            pool (str): Pool label or UUID191            sys_name (str, optional): System name. Defaults to None.192        Returns:193            dict: JSON output194        Raises:195            CommandFailure: if the daos container list command fails.196        """197        # Sample output.198        # {199        #   "response": [200        #     {201        #       "UUID": "bad80a98-aabd-498c-b001-6547cd061c8c",202        #       "Label": "container_label_not_set"203        #     },204        #     {205        #       "UUID": "dd9fc365-5729-4736-9d34-e46504a4a92d",206        #       "Label": "mkc1"207        #     }208        #   ],209        #   "error": null,210        #   "status": 0211        # }212        return self._get_json_result(213            ("container", "list"), pool=pool, sys_name=sys_name)214    def pool_set_attr(self, pool, attr, value, sys_name=None):215        """Set pool attribute.216        Args:217            pool (str): Pool UUID.218            attr (str): Attribute name.219            value (str): Attribute value.220            sys_name (str): DAOS system name. Defaults to None.221        Returns:222            CmdResult: Object that contains exit status, stdout, and other223                information.224        Raises:225            CommandFailure: if the daos pool set-attr command fails.226        """227        return self._get_result(228            ("pool", "set-attr"), pool=pool, attr=attr, value=value,229            sys_name=sys_name)230    def pool_get_attr(self, pool, attr, sys_name=None):231        """Set pool attribute.232        Args:233            pool (str): Pool UUID.234            attr (str): Pool UUID.235            sys_name (str): DAOS system name. Defaults to None.236        Returns:237            dict: JSON output238        Raises:239            CommandFailure: if the daos pool query command fails.240        """241        return self._get_json_result(242            ("pool", "get-attr"), pool=pool, attr=attr, sys_name=sys_name)243    def pool_list_attrs(self, pool, sys_name=None, verbose=False):244        """List pool attributes.245        Args:246            pool (str): Pool UUID.247            sys_name (str): DAOS system name. Defaults to None.248            verbose (bool): False - name only. True - name and value. Defaults249                to False.250        Returns:251            dict: JSON output252        Raises:253            CommandFailure: if the daos pool list-attrs command fails.254        """255        return self._get_json_result(256            ("pool", "list-attrs"), pool=pool, sys_name=sys_name,257            verbose=verbose)258    def container_query(self, pool, cont, sys_name=None):259        """Query a container.260        Args:261            pool (str): Pool UUID.262            cont (str): Container UUID.263            sys_name (str, optional): DAOS system name context for servers.264                Defaults to None.265        Returns:266            dict: JSON output267        Raises:268            CommandFailure: if the daos container query command fails.269        """270        return self._get_json_result(271            ("container", "query"), pool=pool, cont=cont, sys_name=sys_name)272    def container_set_prop(self, pool, cont, prop, value):273        """Call daos container set-prop.274        Args:275            pool (str): Pool UUID.276            cont (str): Container UUID.277            prop (str): Container property-name.278            value (str): Container property-name value to set.279        Returns:280            CmdResult: Object that contains exit status, stdout, and other281                information.282        Raises:283            CommandFailure: if the daos container set-prop command fails.284        """285        prop_value = ":".join([prop, value])286        return self._get_result(287            ("container", "set-prop"),288            pool=pool, cont=cont, prop=prop_value)289    def container_get_prop(self, pool, cont):290        """Call daos container get-prop.291        Args:292            pool (str): Pool UUID.293            cont (str): Container UUID.294        Returns:295            CmdResult: Object that contains exit status, stdout, and other296                information.297        Raises:298            CommandFailure: if the daos container get-prop command fails.299        """300        return self._get_result(301            ("container", "get-prop"), pool=pool, cont=cont)302    def container_set_owner(self, pool, cont, user, group):303        """Call daos container set-owner.304        Args:305            pool (str): Pool UUID.306            cont (str): Container UUID.307            user (str): New-user who will own the container.308            group (str): New-group who will own the container.309        Returns:310            CmdResult: Object that contains exit status, stdout, and other311                information.312        Raises:313            CommandFailure: if the daos container set-owner command fails.314        """315        return self._get_result(316            ("container", "set-owner"),317            pool=pool, cont=cont, user=user, group=group)318    def container_set_attr(319            self, pool, cont, attr, val, sys_name=None):320        """Call daos container set-attr.321        Args:322            pool (str): Pool UUID.323            cont (str): Container UUID.324            attr (str): Attribute name.325            val (str): Attribute value.326            sys_name (str, optional): DAOS system name context for servers.327                Defaults to None.328        Returns:329            CmdResult: Object that contains exit status, stdout, and other330                information.331        Raises:332            CommandFailure: if the daos container set-attr command fails.333        """334        return self._get_result(335            ("container", "set-attr"), pool=pool, cont=cont,336            sys_name=sys_name, attr=attr, value=val)337    def container_get_attr(self, pool, cont, attr, sys_name=None):338        """Call daos container get-attr.339        Args:340            pool (str): Pool UUID.341            cont (str): Container UUID.342            attr (str): Attribute name.343            sys_name (str, optional): DAOS system name context for servers.344                Defaults to None.345        Returns:346            dict: the daos json command output converted to a python dictionary347        Raises:348            CommandFailure: if the daos get-attr command fails.349        """350        return self._get_json_result(351            ("container", "get-attr"), pool=pool, cont=cont, attr=attr, sys_name=sys_name)352    def container_list_attrs(self, pool, cont, sys_name=None, verbose=False):353        """Call daos container list-attrs.354        Args:355            pool (str): Pool UUID.356            cont (str): Container UUID.357            sys_name (str, optional): DAOS system name context for servers.358                Defaults to None.359            verbose (bool, optional): True - fetch values of all attributes.360        Returns:361            dict: the daos json command output converted to a python dictionary362        Raises:363            CommandFailure: if the daos container list-attrs command fails.364        """365        return self._get_json_result(366            ("container", "list-attrs"), pool=pool, cont=cont, sys_name=sys_name,367            verbose=verbose)368    def container_create_snap(self, pool, cont, snap_name=None, epoch=None,369                              sys_name=None):370        """Call daos container create-snap.371        Args:372            pool (str): Pool UUID.373            cont (str): Container UUID.374            snap_name (str, optional): Snapshot name. Defaults to None.375            epoch (str, optional): Epoch number. Defaults to None.376            sys_name (str, optional): DAOS system name context for servers.377                Defaults to None.378        Returns:379            dict: Dictionary that stores the created epoch in the key "epoch"....

asuswrt.py

Source:asuswrt.py

...64_LOGGER = logging.getLogger(__name__)65def _nvram_cmd(info_type):66    """Return the cmd to get nvram data."""67    return f"{CMD_NVRAM}({info_type})"68def _get_json_result(result: str, json_key: str | None = None):69    """Return the json result from a text result."""70    try:71        json_res = json.loads(result)72    except json.JSONDecodeError as exc:73        raise AsusWrtValueError(str(exc)) from exc74    if not json_key:75        return json_res76    if (json_val := json_res.get(json_key)) is None:77        raise AsusWrtValueError("No value available")78    return json_val79class AsusWrtError(Exception):80    """Base class for all errors raised by this library."""81    def __init__(self, *args: Any, message: Optional[str] = None, **_kwargs: Any) -> None:82        """Initialize base UpnpError."""83        super().__init__(*args, message)84class AsusWrtConnectionError(AsusWrtError, aiohttp.ClientConnectionError):85    """Error connecting with the router."""86class AsusWrtConnectionTimeoutError(AsusWrtError, aiohttp.ServerTimeoutError, asyncio.TimeoutError):87    """Timeout while communicating with the device."""88class AsusWrtLoginError(AsusWrtError):89    """Login error / invalid credential."""90class AsusWrtResponseError(AsusWrtError, aiohttp.ClientResponseError):91    """Error communicating with the router."""92class AsusWrtValueError(AsusWrtError, ValueError):93    """Error invalid value received."""94class AsusWrtHttp:95    """Class for AsusWrt router HTTP/HTTPS connection."""96    def __init__(97        self,98        hostname: str,99        username: str,100        password: str,101        *,102        use_https: bool = False,103        port: int | None = None,104        timeout: int = DEFAULT_TIMEOUT,105        session: aiohttp.ClientSession | None = None,106    ):107        """108        Create the router object109        Parameters:110            hostname: HostName or IP Address of the router111            username: Router username112            password: Password required to login113            use_https: if True use https instead of http (default False)114            port: the tcp port to use (leave None or 0 for protocol default)115            timeout: the tcp timeout (default = 5 sec.)116            session: the AioHttp session to use (if None a new session is created)117        """118        self._hostname = hostname119        self._username = username120        self._password = password121        self._protocol = "https" if use_https else "http"122        if port and port > 0:123            self._port = port124        else:125            self._port = DEFAULT_HTTPS_PORT if use_https else DEFAULT_HTTP_PORT126        self._timeout = timeout if timeout > 0 else DEFAULT_TIMEOUT127        self._auth_headers = None128        if session:129            self._session = session130            self._managed_session = False131        else:132            self._session = None133            self._managed_session = True134        self._mac = None135        self._latest_transfer_data = None136        self._latest_transfer_rate = {"rx_rate": 0.0, "tx_rate": 0.0}137        self._latest_transfer_check = None138    def __url(self, path):139        """Return the url to a specific path."""140        return f"{self._protocol}://{self._hostname}:{self._port}/{path}"141    async def __http_post(self, url, headers, payload, *, get_json=False):142        """Perform aiohttp POST request."""143        try:144            async with self._session.post(145                url=url,146                headers=headers,147                data=payload,148                timeout=self._timeout,149                raise_for_status=True,150                ssl=False,151            ) as resp:152                if get_json:153                    result = await resp.json()154                else:155                    result = await resp.text()156        except (asyncio.TimeoutError, aiohttp.ServerTimeoutError) as err:157            raise AsusWrtConnectionTimeoutError(str(err)) from err158        except aiohttp.ClientConnectorError as err:159            raise AsusWrtConnectionError(str(err)) from err160        except aiohttp.ClientConnectionError as err:161            self._auth_headers = None162            raise AsusWrtConnectionError(str(err)) from err163        except aiohttp.ClientResponseError as err:164            raise AsusWrtResponseError(165                request_info=err.request_info,166                history=err.history,167                status=err.status,168                message=err.message,169                headers=err.headers,170            ) from err171        except aiohttp.ClientError as err:172            self._auth_headers = None173            raise AsusWrtError(str(err)) from err174        return result175    async def __post(self, command, path=ASUSWRT_GET_PATH, *, retry=True):176        """177        Private POST method to execute a hook on the router and return the result178        :param command: Command to send to the return179        :returns: string result from the router180        """181        payload = f"hook={command}"182        try:183            await self.async_connect()184            result = await self.__http_post(self.__url(path), self._auth_headers, payload)185        except (AsusWrtConnectionError, AsusWrtResponseError):186            if retry:187                return await self.__post(command, path, retry=False)188            raise189        if result.find(ASUSWRT_ERROR_KEY, 0, len(ASUSWRT_ERROR_KEY) + 5) >= 0:190            self._auth_headers = None191            if retry:192                return await self.__post(command, path, retry=False)193            raise AsusWrtConnectionError("Not connected to the router")194        return result195    async def __send_cmd(self, commands: dict[str, str], action_mode: str = "apply"):196        """Command device to run a service or set parameter"""197        request: dict = {198            "action_mode": action_mode,199            **commands,200        }201        result = await self.__post(str(request), ASUSWRT_CMD_PATH)202        return result203    @property204    def hostname(self) -> str:205        """Return the device hostname."""206        return self._hostname207    @property208    def mac(self) -> str | None:209        """Return the device mac address."""210        return self._mac211    @property212    def is_connected(self) -> bool:213        """Return if connection is active."""214        return self._auth_headers is not None215    async def async_disconnect(self):216        """Close the managed session on exit."""217        if self._managed_session and self._session is not None:218            await self._session.close()219            self._session = None220        self._auth_headers = None221    async def async_connect(self):222        """Authenticate with the router."""223        if self.is_connected:224            return225        if self._managed_session and self._session is None:226            self._session = aiohttp.ClientSession()227        auth = f"{self._username}:{self._password}".encode("ascii")228        login_token = base64.b64encode(auth).decode("ascii")229        payload = f"login_authorization={login_token}"230        headers = {"user-agent": ASUSWRT_USR_AGENT}231        result = await self.__http_post(self.__url(ASUSWRT_LOGIN_PATH), headers, payload, get_json=True)232        if ASUSWRT_TOKEN_KEY not in result:233            raise AsusWrtLoginError("Login Failed")234        token = result[ASUSWRT_TOKEN_KEY]235        self._auth_headers = {236            "user-agent": ASUSWRT_USR_AGENT,237            "cookie": f"{ASUSWRT_TOKEN_KEY}={token}",238        }239        # try to get the main properties after connect240        await self._load_props()241    async def _load_props(self) -> str | None:242        """Load device properties from NVRam."""243        if self._mac is not None:244            return245        try:246            result = await self.async_get_settings(PROP_MAC_ADDR)247        except AsusWrtError:248            return249        self._mac = result.get(PROP_MAC_ADDR)250    async def async_get_uptime(self):251        """252        Return uptime of the router253        Format: {'since': 'Thu, 22 Jul 2021 14:32:38 +0200', 'uptime': '375001'}254        :returns: JSON with last boot time and uptime in seconds255        """256        r = await self.__post(f"{CMD_UPTIME}()")257        since = r.partition(":")[2].partition("(")[0]258        up = int(r.partition("(")[2].partition(" ")[0])259        return {"since": since, "uptime": up}260    async def async_get_memory_usage(self):261        """262        Return memory usage of the router263        Format: {'mem_total': 262144, 'mem_free': 107320, 'mem_used': 154824}264        :returns: JSON with memory variables265        """266        s = await self.__post(f"{CMD_MEMORY_USAGE}({PARAM_APPOBJ})")267        result = _get_json_result(s, CMD_MEMORY_USAGE)268        return {k: int(v) for k, v in result.items()}269    async def async_get_cpu_usage(self):270        """271        Return CPUs usage of the router272        Format: {'cpu1_total': 38106047, 'cpu1_usage': 3395512,273                 'cpu2_total': 38106008, 'cpu2_usage': 2384694, ...}274        :returns: JSON with CPUs load statistics275        """276        s = await self.__post(f"{CMD_CPU_USAGE}({PARAM_APPOBJ})")277        result = _get_json_result(s, CMD_CPU_USAGE)278        return {k: int(v) for k, v in result.items()}279    async def async_get_wan_info(self):280        """281        Get the status of the WAN connection282        Format: {"status": "1", "statusstr": "'Connected'", "type": "'dhcp'", "ipaddr": "'192.168.1.2'",283                 "netmask": "'255.255.255.0'", "gateway": "'192.168.1.1'", "dns": "1.1.1.1'",284                 "lease": "86400", "expires": "81967", "xtype": "''", "xipaddr": "'0.0.0.0'",285                 "xnetmask": "'0.0.0.0'", "xgateway": "'0.0.0.0'", "xdns": "''", "xlease": "0",286                 "xexpires": "0"}287        :returns: JSON with status information on the WAN connection288        """289        r = await self.__post(f"{CMD_WAN_INFO}()")290        status = {}291        for f in r.split("\n"):292            if "return" in f:293                if f"{CMD_WAN_INFO}_" in f:294                    key = f.partition("(")[0].partition("_")[2]295                    value = (f.rpartition(" ")[-1][:-2]).replace("'", "")296                    status[key] = value297        return status298    async def async_is_wan_online(self):299        """300        Returns if the WAN connection in online301        :returns: True if WAN is connected302        """303        r = await self.async_get_wan_info()304        return r["status"] == "1"305    async def async_get_dhcp_leases(self):306        """307        Obtain a list of DHCP leases308        Format: [["00:00:00:00:00:00", "name"], ...]309        :returns: JSON with a list of DHCP leases310        """311        s = await self.__post(f"{CMD_DHCP_LEASE}()")312        return _get_json_result(s, CMD_DHCP_LEASE)313    async def async_get_traffic_bytes(self):314        """315        Get total amount of traffic since last restart (bytes format)316        Format: {'rx': 15901, 'tx': 10926}317        :returns: JSON with sent and received bytes since last boot318        """319        s = await self.__post(f"{CMD_NET_TRAFFIC}({PARAM_APPOBJ})")320        meas = _get_json_result(s, CMD_NET_TRAFFIC)321        traffic = None322        if "INTERNET_rx" in meas:323            traffic = "INTERNET"324        elif "BRIDGE_rx" in meas:325            traffic = "BRIDGE"326        if traffic:327            rx = int(meas[f"{traffic}_rx"], base=16)328            tx = int(meas[f"{traffic}_tx"], base=16)329        else:330            rx = tx = 0331        return {"rx": rx, "tx": tx}332    async def async_get_traffic_rates(self):333        """334        Get total and current amount of traffic since last restart (bytes format)335        Note that at least 2 calls with an interval of min 10 seconds is required to have valid data336        Format: {"rx_rate": 0.13004302978515625, "tx_rate": 4.189826965332031}337        :returns: JSON with current up and down stream in byte/s338        """339        now = datetime.utcnow()340        meas_1 = None341        if self._latest_transfer_data:342            meas_1 = self._latest_transfer_data.copy()343        meas_2 = await self.async_get_traffic_bytes()344        prev_check = self._latest_transfer_check345        self._latest_transfer_data = meas_2.copy()346        self._latest_transfer_check = now347        if meas_1 is None:348            return self._latest_transfer_rate349        meas_delta = (now - prev_check).total_seconds()350        if meas_delta < 10:351            return self._latest_transfer_rate352        rates = {}353        for key in ["rx", "tx"]:354            if meas_2[key] < meas_1[key]:355                rates[key] = meas_2[key]356            else:357                rates[key] = meas_2[key] - meas_1[key]358        self._latest_transfer_rate = {359            "rx_rate": math.ceil(rates["rx"] / meas_delta),360            "tx_rate": math.ceil(rates["tx"] / meas_delta),361        }362        return self._latest_transfer_rate363    async def async_get_settings(self, setting: str = None):364        """365        Get settings from the router NVRam366        Format:{'time_zone': 'MEZ-1DST', 'time_zone_dst': '1', 'time_zone_x': 'MEZ-1DST,M3.2.0/2,M10.2.0/2',367               'time_zone_dstoff': 'M3.2.0/2,M10.2.0/2', 'ntp_server0': 'pool.ntp.org', 'acs_dfs': '1',368               'productid': 'RT-AC68U', 'apps_sq': '', 'lan_hwaddr': '04:D4:C4:C4:AD:D0',369               'lan_ipaddr': '192.168.2.1', 'lan_proto': 'static', 'x_Setting': '1',370               'label_mac': '04:D4:C4:C4:AD:D0', 'lan_netmask': '255.255.255.0', 'lan_gateway': '0.0.0.0',371               'http_enable': '2', 'https_lanport': '8443', 'wl0_country_code': 'EU', 'wl1_country_code': 'EU'}372        :param setting: the setting name to query (leave empty to get all main settings)373        :returns: JSON with main Router settings or specific one374        """375        setting_list = [setting] if setting else NVRAM_INFO376        result = {}377        for s in setting_list:378            resp = await self.__post(_nvram_cmd(s))379            if resp:380                result[s] = _get_json_result(resp, s)381        return result382    async def async_get_clients_fullinfo(self) -> list[dict[str, any]]:383        """384        Obtain a list of all clients385        Format: [386                    "AC:84:C6:6C:A7:C0":{"type": "2", "defaultType": "0", "name": "Archer_C1200",387                                         "nickName": "Router Forlindon", "ip": "192.168.2.175",388                                         "mac": "AC:84:C6:6C:A7:C0", "from": "networkmapd",389                                         "macRepeat": "1", "isGateway": "0", "isWebServer": "0",390                                         "isPrinter": "0", "isITunes": "0", "dpiType": "",391                                         "dpiDevice": "", "vendor": "TP-LINK", "isWL": "0",392                                         "isOnline": "1", "ssid": "", "isLogin": "0", "opMode": "0",393                                         "rssi": "0", "curTx": "", "curRx": "", "totalTx": "",394                                         "totalRx": "", "wlConnectTime": "", "ipMethod": "Manual",395                                         "ROG": "0", "group": "", "callback": "", "keeparp": "",396                                         "qosLevel": "", "wtfast": "0", "internetMode": "allow",397                                         "internetState": "1", "amesh_isReClient": "1",398                                         "amesh_papMac": "04:D4:C4:C4:AD:D0"},399                     "maclist": ["AC:84:C6:6C:A7:C0"],400                     "ClientAPILevel": "2" }401                ]402        :returns: JSON with list of mac address and all client related info403        """404        s = await self.__post(f"{CMD_CLIENT_LIST}()")405        result = _get_json_result(s)406        return [result.get(CMD_CLIENT_LIST, {})]407    async def async_get_connected_mac(self):408        """409        Obtain a list of MAC-addresses from online clients410        Format: ["00:00:00:00:00:00", ...]411        :returns: JSON list with MAC adresses412        """413        clnts = await self.async_get_clients_fullinfo()414        lst = [mac for mac, info in clnts[0].items() if len(mac) == 17 and info.get("isOnline", "0") == "1"]415        return lst416    async def async_get_connected_devices(self):417        """418        Obtain info on all clients419        Format: {"AC:84:C6:6C:A7:C0": {mac: "AC:84:C6:6C:A7:C0", ip: "x.x.x.x" name: "Archer_C1200"}, ...}...

json_printer.py

Source:json_printer.py

...8    def print_results(self, context: click.Context, results: List[DocumentDetections]):9        detections = [detection for document_detections in results for detection in document_detections.detections]10        detections_schema = DetectionSchema(many=True)11        detections_dict = detections_schema.dump(detections)12        json_result = self._get_json_result(context, detections_dict)13        click.secho(json_result)14    def _get_json_result(self, context, detections):15        scan_id = context.obj.get('scan_id')16        result = {17            'scan_id': str(scan_id),18            'detections': detections19        }...

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.