Device - Arista Network Test Automation (2024)

888990

def __hash__(self) -> int: """Implement hashing for AntaDevice objects.""" return hash(self._keys)

139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168

async def collect(self, command: AntaCommand, *, collection_id: str | None = None) -> None: """Collect the output for a specified command. When caching is activated on both the device and the command, this method prioritizes retrieving the output from the cache. In cases where the output isn't cached yet, it will be freshly collected and then stored in the cache for future access. The method employs asynchronous locks based on the command's UID to guarantee exclusive access to the cache. When caching is NOT enabled, either at the device or command level, the method directly collects the output via the private `_collect` method without interacting with the cache. Args: ---- command: The command to collect. collection_id: An identifier used to build the eAPI request ID. """ # Need to ignore pylint no-member as Cache is a proxy class and pylint is not smart enough # https://github.com/pylint-dev/pylint/issues/7258 if self.cache is not None and self.cache_locks is not None and command.use_cache: async with self.cache_locks[command.uid]: cached_output = await self.cache.get(command.uid) # pylint: disable=no-member if cached_output is not None: logger.debug("Cache hit for %s on %s", command.command, self.name) command.output = cached_output else: await self._collect(command=command, collection_id=collection_id) await self.cache.set(command.uid, command.output) # pylint: disable=no-member else: await self._collect(command=command, collection_id=collection_id)

170171172173174175176177178

async def collect_commands(self, commands: list[AntaCommand], *, collection_id: str | None = None) -> None: """Collect multiple commands. Args: ---- commands: The commands to collect. collection_id: An identifier used to build the eAPI request ID. """ await asyncio.gather(*(self.collect(command=command, collection_id=collection_id) for command in commands))

190191192193194195196197198199200201202203204

async def copy(self, sources: list[Path], destination: Path, direction: Literal["to", "from"] = "from") -> None: """Copy files to and from the device, usually through SCP. It is not mandatory to implement this for a valid AntaDevice subclass. Args: ---- sources: List of files to copy to or from the device. destination: Local or remote destination when copying the files. Can be a folder. direction: Defines if this coroutine copies files to or from the device. """ _ = (sources, destination, direction) msg = f"copy() method has not been implemented in {self.__class__.__name__} definition" raise NotImplementedError(msg)

180181182183184185186187188

@abstractmethodasync def refresh(self) -> None: """Update attributes of an AntaDevice instance. This coroutine must update the following attributes of AntaDevice: - `is_online`: When the device IP is reachable and a port can be open - `established`: When a command execution succeeds - `hw_model`: The hardware model of the device """

405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443

async def copy(self, sources: list[Path], destination: Path, direction: Literal["to", "from"] = "from") -> None: """Copy files to and from the device using asyncssh.scp(). Args: ---- sources: List of files to copy to or from the device. destination: Local or remote destination when copying the files. Can be a folder. direction: Defines if this coroutine copies files to or from the device. """ async with asyncssh.connect( host=self._ssh_opts.host, port=self._ssh_opts.port, tunnel=self._ssh_opts.tunnel, family=self._ssh_opts.family, local_addr=self._ssh_opts.local_addr, options=self._ssh_opts, ) as conn: src: list[tuple[SSHClientConnection, Path]] | list[Path] dst: tuple[SSHClientConnection, Path] | Path if direction == "from": src = [(conn, file) for file in sources] dst = destination for file in sources: message = f"Copying '{file}' from device {self.name} to '{destination}' locally" logger.info(message) elif direction == "to": src = sources dst = conn, destination for file in src: message = f"Copying '{file}' to device {self.name} to '{destination}' remotely" logger.info(message) else: logger.critical("'direction' argument to copy() function is invalid: %s", direction) return await asyncssh.scp(src, dst)

381382383384385386387388389390391392393394395396397398399400401402403

async def refresh(self) -> None: """Update attributes of an AsyncEOSDevice instance. This coroutine must update the following attributes of AsyncEOSDevice: - is_online: When a device IP is reachable and a port can be open - established: When a command execution succeeds - hw_model: The hardware model of the device """ logger.debug("Refreshing device %s", self.name) self.is_online = await self._session.check_connection() if self.is_online: show_version = AntaCommand(command="show version") await self._collect(show_version) if not show_version.collected: logger.warning("Cannot get hardware information from device %s", self.name) else: self.hw_model = show_version.json_output.get("modelName", None) if self.hw_model is None: logger.critical("Cannot parse 'show version' returned by device %s", self.name) else: logger.warning("Could not connect to device %s: cannot open eAPI port", self.name) self.established = bool(self.is_online and self.hw_model)