KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run avocado automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
