Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use set_api_timeout method in yandex-tank
Best Python code snippet using yandex-tank
denonavr.py
Source:denonavr.py 
1#!/usr/bin/env python32# -*- coding: utf-8 -*-3"""4This module implements the interface to Denon AVR receivers.5:copyright: (c) 2016 by Oliver Goetz.6:license: MIT, see LICENSE for more details.7"""8import asyncio9import logging10import time11from typing import Callable, Dict, List, Optional12import attr13import httpx14from .decorators import run_async_synchronously15from .foundation import DenonAVRFoundation, set_api_host, set_api_timeout16from .const import (17 DENON_ATTR_SETATTR, MAIN_ZONE, VALID_ZONES)18from .exceptions import AvrCommandError19from .audyssey import DenonAVRAudyssey, audyssey_factory20from .input import DenonAVRInput, input_factory21from .soundmode import DenonAVRSoundMode, sound_mode_factory22from .tonecontrol import DenonAVRToneControl, tone_control_factory23from .volume import DenonAVRVolume, volume_factory24_LOGGER = logging.getLogger(__name__)25@attr.s(auto_attribs=True, on_setattr=DENON_ATTR_SETATTR)26class DenonAVR(DenonAVRFoundation):27 """28 Representing a Denon AVR Device.29 Initialize MainZone of DenonAVR.30 :param host: IP or HOSTNAME.31 :type host: str32 :param name: Device name, if None FriendlyName of device is used.33 :type name: str or None34 :param show_all_inputs: If True deleted input functions are also shown35 :type show_all_inputs: bool36 :param timeout: Timeout when calling device APIs.37 :type timeout: float38 :param add_zones: Additional Zones for which an instance are created39 :type add_zones: dict [str, str] or None40 """41 _host: str = attr.ib(42 converter=str, on_setattr=[*DENON_ATTR_SETATTR, set_api_host])43 _name: Optional[str] = attr.ib(44 converter=attr.converters.optional(str), default=None)45 _show_all_inputs: bool = attr.ib(converter=bool, default=False)46 _add_zones: Optional[Dict[str, str]] = attr.ib(47 validator=attr.validators.optional(attr.validators.deep_mapping(48 attr.validators.in_(VALID_ZONES),49 attr.validators.optional(attr.validators.instance_of(str)),50 attr.validators.instance_of(dict))),51 default=None)52 _timeout: float = attr.ib(53 converter=float,54 on_setattr=[*DENON_ATTR_SETATTR, set_api_timeout],55 default=2.0)56 _zones: Dict[str, DenonAVRFoundation] = attr.ib(57 validator=attr.validators.deep_mapping(58 attr.validators.in_(VALID_ZONES),59 attr.validators.instance_of(DenonAVRFoundation),60 attr.validators.instance_of(dict)),61 default=attr.Factory(dict),62 init=False)63 _setup_lock: asyncio.Lock = attr.ib(default=attr.Factory(asyncio.Lock))64 audyssey: DenonAVRAudyssey = attr.ib(65 validator=attr.validators.instance_of(DenonAVRAudyssey),66 default=attr.Factory(audyssey_factory, takes_self=True),67 init=False)68 input: DenonAVRInput = attr.ib(69 validator=attr.validators.instance_of(DenonAVRInput),70 default=attr.Factory(input_factory, takes_self=True),71 init=False)72 soundmode: DenonAVRSoundMode = attr.ib(73 validator=attr.validators.instance_of(DenonAVRSoundMode),74 default=attr.Factory(sound_mode_factory, takes_self=True),75 init=False)76 tonecontrol: DenonAVRToneControl = attr.ib(77 validator=attr.validators.instance_of(DenonAVRToneControl),78 default=attr.Factory(tone_control_factory, takes_self=True),79 init=False)80 vol: DenonAVRVolume = attr.ib(81 validator=attr.validators.instance_of(DenonAVRVolume),82 default=attr.Factory(volume_factory, takes_self=True),83 init=False)84 def __attrs_post_init__(self) -> None:85 """Initialize special attributes."""86 # Set host and timeout again to start its custom setattr function87 self._host = self._host88 self._timeout = self._timeout89 # Add own instance to zone dictionary90 self._zones[self._device.zone] = self91 # Create instances of additional zones if requested92 if self._device.zone == MAIN_ZONE and self._add_zones is not None:93 self.create_zones(self._add_zones)94 def create_zones(self, add_zones):95 """Create instances of additional zones for the receiver."""96 for zone, zname in add_zones.items():97 # Name either set explicitly or name of Main Zone with suffix98 zonename = None99 if zname is None and self._name is not None:100 zonename = "{} {}".format(self._name, zone)101 zone_device = attr.evolve(self._device, zone=zone)102 zone_inst = DenonAVR(103 host=self._host,104 device=zone_device,105 name=zonename,106 timeout=self._timeout,107 show_all_inputs=self._show_all_inputs)108 self._zones[zone] = zone_inst109 async def async_setup(self) -> None:110 """Ensure that configuration is loaded from receiver asynchronously."""111 async with self._setup_lock:112 # Device setup113 await self._device.async_setup()114 if self._name is None:115 self._name = self._device.friendly_name116 # Setup other functions117 self.input.setup()118 await self.soundmode.async_setup()119 self.tonecontrol.setup()120 self.vol.setup()121 self.audyssey.setup()122 self._is_setup = True123 @run_async_synchronously(async_func=async_setup)124 def setup(self) -> None:125 """Ensure that configuration is loaded from receiver."""126 async def async_update(self):127 """128 Get the latest status information from device asynchronously.129 Method executes the update method for the current receiver type.130 """131 # Ensure that the device is setup132 if self._is_setup is False:133 await self.async_setup()134 # Create a cache id for this global update135 cache_id = time.time()136 # Verify update method137 await self._device.async_verify_avr_2016_update_method(138 cache_id=cache_id)139 # Update device140 await self._device.async_update(global_update=True, cache_id=cache_id)141 # Update other functions142 await self.input.async_update(global_update=True, cache_id=cache_id)143 await self.soundmode.async_update(144 global_update=True, cache_id=cache_id)145 await self.tonecontrol.async_update(146 global_update=True, cache_id=cache_id)147 await self.vol.async_update(global_update=True, cache_id=cache_id)148 # AppCommand0300.xml interface is very slow, thus it is not included149 # into main update150 # await self.audyssey.async_update(151 # global_update=True, cache_id=cache_id)152 @run_async_synchronously(async_func=async_update)153 def update(self):154 """155 Get the latest status information from device.156 Method executes the update method for the current receiver type.157 """158 async def async_update_tonecontrol(self):159 """Get Tonecontrol settings."""160 await self.tonecontrol.async_update()161 @run_async_synchronously(async_func=async_update_tonecontrol)162 def update_tonecontrol(self):163 """Get Tonecontrol settings."""164 async def async_update_audyssey(self):165 """Get Audyssey settings."""166 await self.audyssey.async_update()167 @run_async_synchronously(async_func=async_update_audyssey)168 def update_audyssey(self):169 """Get Audyssey settings."""170 async def async_get_command(self, request: str) -> str:171 """Send HTTP GET command to Denon AVR receiver asynchronously."""172 return await self._device.api.async_get_command(request)173 @run_async_synchronously(async_func=async_get_command)174 def get_command(self, request: str) -> str:175 """Send HTTP GET command to Denon AVR receiver."""176 @run_async_synchronously(async_func=async_get_command)177 def send_get_command(self, request: str) -> str:178 """Send HTTP GET command to Denon AVR receiver...for compatibility."""179 ##############180 # Properties #181 ##############182 @property183 def zone(self) -> str:184 """Return Zone of this instance."""185 return self._device.zone186 @property187 def zones(self) -> Dict[str, DenonAVRFoundation]:188 """Return all Zone instances of the device."""189 zones = self._zones.copy()190 return zones191 @property192 def name(self) -> Optional[str]:193 """Return the name of the device as string."""194 return self._name195 @property196 def host(self) -> str:197 """Return the host of the device as string."""198 return self._host199 @property200 def manufacturer(self) -> Optional[str]:201 """Return the manufacturer of the device as string."""202 return self._device.manufacturer203 @property204 def model_name(self) -> Optional[str]:205 """Return the model name of the device as string."""206 return self._device.model_name207 @property208 def serial_number(self) -> Optional[str]:209 """Return the serial number of the device as string."""210 return self._device.serial_number211 @property212 def power(self) -> Optional[str]:213 """214 Return the power state of the device.215 Possible values are: "ON", "STANDBY" and "OFF"216 """217 return self._device.power218 @property219 def state(self) -> Optional[str]:220 """221 Return the state of the device.222 Possible values are: "on", "off", "playing", "paused"223 "playing" and "paused" are only available for input functions224 in PLAYING_SOURCES.225 """226 return self.input.state227 @property228 def muted(self) -> bool:229 """230 Boolean if volume is currently muted.231 Return "True" if muted and "False" if not muted.232 """233 return self.vol.muted234 @property235 def volume(self) -> float:236 """237 Return volume of Denon AVR as float.238 Volume is send in a format like -50.0.239 Minimum is -80.0, maximum at 18.0240 """241 return self.vol.volume242 @property243 def input_func(self) -> Optional[str]:244 """Return the current input source as string."""245 return self.input.input_func246 @property247 def input_func_list(self) -> List[str]:248 """Return a list of available input sources as string."""249 return self.input.input_func_list250 @property251 def support_sound_mode(self) -> Optional[bool]:252 """Return True if sound mode supported."""253 return self.soundmode.support_sound_mode254 @property255 def sound_mode(self) -> Optional[str]:256 """Return the matched current sound mode as a string."""257 return self.soundmode.sound_mode258 @property259 def sound_mode_list(self) -> List[str]:260 """Return a list of available sound modes as string."""261 return self.soundmode.sound_mode_list262 @property263 def sound_mode_map(self) -> Dict[str, str]:264 """Return a dict of available sound modes with their mapping values."""265 return self.soundmode.sound_mode_map266 @property267 def sound_mode_map_rev(self) -> Dict[str, str]:268 """Return a dict to map each sound_mode_raw to matching sound_mode."""269 return self.soundmode.sound_mode_map_rev270 @property271 def sound_mode_raw(self) -> Optional[str]:272 """Return the current sound mode as string as received from the AVR."""273 return self.soundmode.sound_mode_raw274 @property275 def image_url(self) -> Optional[str]:276 """Return image URL of current playing media when powered on."""277 return self.input.image_url278 @property279 def title(self) -> Optional[str]:280 """Return title of current playing media as string."""281 return self.input.title282 @property283 def artist(self) -> Optional[str]:284 """Return artist of current playing media as string."""285 return self.input.artist286 @property287 def album(self) -> Optional[str]:288 """Return album name of current playing media as string."""289 return self.input.album290 @property291 def band(self) -> Optional[str]:292 """Return band of current radio station as string."""293 return self.input.band294 @property295 def frequency(self) -> Optional[str]:296 """Return frequency of current radio station as string."""297 return self.input.frequency298 @property299 def station(self) -> Optional[str]:300 """Return current radio station as string."""301 return self.input.station302 @property303 def netaudio_func_list(self) -> List[str]:304 """Return list of network audio devices.305 Those devices should react to play, pause, next and previous306 track commands.307 """308 return self.input.netaudio_func_list309 @property310 def playing_func_list(self) -> List[str]:311 """Return list of playing devices.312 Those devices offer additional information about what they are playing313 (e.g. title, artist, album, band, frequency, station, image_url).314 """315 return self.input.playing_func_list316 @property317 def receiver_port(self) -> int:318 """Return the receiver's port."""319 if self._device.receiver is None:320 return None321 return self._device.receiver.port322 @property323 def receiver_type(self) -> Optional[str]:324 """Return the receiver's type."""325 if self._device.receiver is None:326 return None327 return self._device.receiver.type328 @property329 def show_all_inputs(self) -> Optional[bool]:330 """Indicate if all inputs are shown or just active one."""331 return self._show_all_inputs332 @property333 def bass(self) -> Optional[int]:334 """Return value of bass."""335 return self.tonecontrol.bass336 @property337 def bass_level(self) -> Optional[str]:338 """Return level of bass."""339 return self.tonecontrol.bass_level340 @property341 def treble(self) -> Optional[int]:342 """Return value of treble."""343 return self.tonecontrol.treble344 @property345 def treble_level(self) -> Optional[str]:346 """Return level of treble."""347 return self.tonecontrol.treble_level348 @property349 def dynamic_eq(self) -> Optional[bool]:350 """Return value of Dynamic EQ."""351 return self.audyssey.dynamic_eq352 @property353 def reference_level_offset(self) -> Optional[str]:354 """Return value of Reference Level Offset."""355 return self.audyssey.reference_level_offset356 @property357 def reference_level_offset_setting_list(self) -> List[str]:358 """Return a list of available reference level offset settings."""359 return self.audyssey.reference_level_offset_setting_list360 @property361 def dynamic_volume(self) -> Optional[str]:362 """Return value of Dynamic Volume."""363 return self.audyssey.dynamic_volume364 @property365 def dynamic_volume_setting_list(self) -> List[str]:366 """Return a list of available Dynamic Volume settings."""367 return self.audyssey.dynamic_volume_setting_list368 @property369 def multi_eq(self) -> Optional[str]:370 """Return value of MultiEQ."""371 return self.audyssey.multi_eq372 @property373 def multi_eq_setting_list(self) -> List[str]:374 """Return a list of available MultiEQ settings."""375 return self.audyssey.multi_eq_setting_list376 async def async_dynamic_eq_off(self) -> None:377 """Turn DynamicEQ off."""378 await self.audyssey.async_dynamiceq_off()379 ##########380 # Setter #381 ##########382 def set_async_client_getter(383 self,384 async_client_getter: Callable[[], httpx.AsyncClient]) -> None:385 """386 Set a custom httpx.AsyncClient getter for this instance.387 The function provided must return an instance of httpx.AsyncClient.388 This is a non-blocking method.389 """390 if not callable(async_client_getter):391 raise AvrCommandError("Provided object is not callable")392 self._device.api.async_client_getter = async_client_getter393 @run_async_synchronously(async_func=async_dynamic_eq_off)394 def dynamic_eq_off(self) -> None:395 """Turn DynamicEQ off."""396 async def async_dynamic_eq_on(self) -> None:397 """Turn DynamicEQ on."""398 await self.audyssey.async_dynamiceq_on()399 @run_async_synchronously(async_func=async_dynamic_eq_on)400 def dynamic_eq_on(self) -> None:401 """Turn DynamicEQ on."""402 async def async_toggle_dynamic_eq(self) -> None:403 """Toggle DynamicEQ."""404 await self.audyssey.async_toggle_dynamic_eq()405 @run_async_synchronously(async_func=async_toggle_dynamic_eq)406 def toggle_dynamic_eq(self) -> None:407 """Toggle DynamicEQ."""408 async def async_set_input_func(self, input_func: str) -> None:409 """410 Set input_func of device.411 Valid values depend on the device and should be taken from412 "input_func_list".413 """414 await self.input.async_set_input_func(input_func)415 @run_async_synchronously(async_func=async_set_input_func)416 def set_input_func(self, input_func: str) -> None:417 """418 Set input_func of device.419 Valid values depend on the device and should be taken from420 "input_func_list".421 """422 async def async_set_sound_mode(self, sound_mode: str) -> None:423 """424 Set sound_mode of device.425 Valid values depend on the device and should be taken from426 "sound_mode_list".427 """428 await self.soundmode.async_set_sound_mode(sound_mode)429 @run_async_synchronously(async_func=async_set_sound_mode)430 def set_sound_mode(self, sound_mode: str) -> None:431 """432 Set sound_mode of device.433 Valid values depend on the device and should be taken from434 "sound_mode_list".435 """436 async def async_toggle_play_pause(self) -> None:437 """Toggle play pause media player."""438 await self.input.async_toggle_play_pause()439 @run_async_synchronously(async_func=async_toggle_play_pause)440 def toggle_play_pause(self) -> None:441 """Toggle play pause media player."""442 async def async_play(self) -> None:443 """Send play command to receiver command via HTTP post."""444 await self.input.async_play()445 @run_async_synchronously(async_func=async_play)446 def play(self) -> None:447 """Send play command to receiver command via HTTP post."""448 async def async_pause(self) -> None:449 """Send pause command to receiver command via HTTP post."""450 await self.input.async_pause()451 @run_async_synchronously(async_func=async_pause)452 def pause(self) -> None:453 """Send pause command to receiver command via HTTP post."""454 async def async_previous_track(self) -> None:455 """Send previous track command to receiver command via HTTP post."""456 await self.input.async_previous_track()457 @run_async_synchronously(async_func=async_previous_track)458 def previous_track(self) -> None:459 """Send previous track command to receiver command via HTTP post."""460 async def async_next_track(self) -> None:461 """Send next track command to receiver command via HTTP post."""462 await self.input.async_next_track()463 @run_async_synchronously(async_func=async_next_track)464 def next_track(self) -> None:465 """Send next track command to receiver command via HTTP post."""466 async def async_power_on(self) -> None:467 """Turn on receiver via HTTP get command."""468 await self._device.async_power_on()469 @run_async_synchronously(async_func=async_power_on)470 def power_on(self) -> None:471 """Turn on receiver via HTTP get command."""472 async def async_power_off(self) -> None:473 """Turn off receiver via HTTP get command."""474 await self._device.async_power_off()475 @run_async_synchronously(async_func=async_power_off)476 def power_off(self) -> None:477 """Turn off receiver via HTTP get command."""478 async def async_volume_up(self) -> None:479 """Volume up receiver via HTTP get command."""480 await self.vol.async_volume_up()481 @run_async_synchronously(async_func=async_volume_up)482 def volume_up(self) -> None:483 """Volume up receiver via HTTP get command."""484 async def async_volume_down(self) -> None:485 """Volume down receiver via HTTP get command."""486 await self.vol.async_volume_down()487 @run_async_synchronously(async_func=async_volume_down)488 def volume_down(self) -> None:489 """Volume down receiver via HTTP get command."""490 async def async_set_volume(self, volume: float) -> None:491 """492 Set receiver volume via HTTP get command.493 Volume is send in a format like -50.0.494 Minimum is -80.0, maximum at 18.0495 """496 await self.vol.async_set_volume(volume)497 @run_async_synchronously(async_func=async_set_volume)498 def set_volume(self, volume: float) -> None:499 """500 Set receiver volume via HTTP get command.501 Volume is send in a format like -50.0.502 Minimum is -80.0, maximum at 18.0503 """504 async def async_mute(self, mute: bool) -> None:505 """Mute receiver via HTTP get command."""506 await self.vol.async_mute(mute)507 @run_async_synchronously(async_func=async_mute)508 def mute(self, mute: bool) -> None:509 """Mute receiver via HTTP get command."""510 async def async_enable_tone_control(self) -> None:511 """Enable tone control to change settings like bass or treble."""512 await self.tonecontrol.async_enable_tone_control()513 @run_async_synchronously(async_func=async_enable_tone_control)514 def enable_tone_control(self) -> None:515 """Enable tone control to change settings like bass or treble."""516 async def async_disable_tone_control(self) -> None:517 """Disable tone control to change settings like bass or treble."""518 await self.tonecontrol.async_disable_tone_control()519 @run_async_synchronously(async_func=async_disable_tone_control)520 def disable_tone_control(self) -> None:521 """Disable tone control to change settings like bass or treble."""522 async def async_set_bass(self, value: int) -> None:523 """524 Set receiver bass.525 Minimum is 0, maximum at 12526 Note:527 Doesn't work, if Dynamic Equalizer is active.528 """529 await self.tonecontrol.async_set_bass(value)530 @run_async_synchronously(async_func=async_set_bass)531 def set_bass(self, value: int) -> None:532 """533 Set receiver bass.534 Minimum is 0, maximum at 12535 Note:536 Doesn't work, if Dynamic Equalizer is active.537 """538 async def async_bass_up(self) -> None:539 """540 Increase level of Bass.541 Note:542 Doesn't work, if Dynamic Equalizer is active543 """544 await self.tonecontrol.async_bass_up()545 @run_async_synchronously(async_func=async_bass_up)546 def bass_up(self) -> None:547 """548 Increase level of Bass.549 Note:550 Doesn't work, if Dynamic Equalizer is active551 """552 async def async_bass_down(self) -> None:553 """554 Decrease level of Bass.555 Note:556 Doesn't work, if Dynamic Equalizer is active557 """558 await self.tonecontrol.async_bass_down()559 @run_async_synchronously(async_func=async_bass_down)560 def bass_down(self) -> None:561 """562 Decrease level of Bass.563 Note:564 Doesn't work, if Dynamic Equalizer is active565 """566 async def async_set_treble(self, value: int) -> None:567 """568 Set receiver treble.569 Minimum is 0, maximum at 12570 Note:571 Doesn't work, if Dynamic Equalizer is active.572 """573 await self.tonecontrol.async_set_treble(value)574 @run_async_synchronously(async_func=async_set_treble)575 def set_treble(self, value: int) -> None:576 """577 Set receiver treble.578 Minimum is 0, maximum at 12579 Note:580 Doesn't work, if Dynamic Equalizer is active.581 """582 async def async_treble_up(self) -> None:583 """584 Increase level of Treble.585 Note:586 Doesn't work, if Dynamic Equalizer is active587 """588 await self.tonecontrol.async_treble_up()589 @run_async_synchronously(async_func=async_treble_up)590 def treble_up(self) -> None:591 """592 Increase level of Treble.593 Note:594 Doesn't work, if Dynamic Equalizer is active595 """596 async def async_treble_down(self) -> None:597 """598 Decrease level of Treble.599 Note:600 Doesn't work, if Dynamic Equalizer is active601 """602 await self.tonecontrol.async_treble_down()603 @run_async_synchronously(async_func=async_treble_down)604 def treble_down(self) -> None:605 """606 Decrease level of Treble.607 Note:608 Doesn't work, if Dynamic Equalizer is active...
client.py
Source:client.py
...59 class JobNotCreated(Exception):60 pass61 class NetworkError(Exception):62 pass63 def set_api_timeout(self, timeout):64 self.api_timeout = float(timeout)65 def network_timeouts(self):66 return (self.network_timeout for _ in range(self.network_attempts - 1))67 def api_timeouts(self):68 return (self.api_timeout for _ in range(self.api_attempts - 1))69 def maintenance_timeouts(self):70 return (71 self.maintenance_timeout for _ in range(72 self.maintenance_attempts - 1))73 @staticmethod74 def filter_headers(headers):75 boring = ['X-Content-Security-Policy', 'Content-Security-Policy',76 'Strict-Transport-Security', 'X-WebKit-CSP', 'Set-Cookie',77 'X-DNS-Prefetch-Control', 'X-Frame-Options', 'P3P',...
dma_manager.py
Source:dma_manager.py
...123 self.ready.set()124 self.schedule_backup()125 # After a successful DMA setup126 # let's API timeout for operations to 15 seconds127 self.flow.set_api_timeout(utils.FLOW_API_TIMEOUT)128 def check_flow_connection(self):129 """Performs a test operation on the flow service,130 it raises an exception if it wasn't able to connect.131 """132 if not self.ready.is_set():133 raise Exception("DMA is not configured yet")134 flow_util.check_flow_connection(135 self.flow,136 self.ldap_team_id,137 self.test_cid,138 )139 def get_dma_fingerprint(self):140 if not self.ready.is_set():141 raise Exception("DMA is not configured yet")...
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 yandex-tank 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?
