diff options
Diffstat (limited to 'anta/cli/exec/utils.py')
| -rw-r--r-- | anta/cli/exec/utils.py | 65 |
1 files changed, 35 insertions, 30 deletions
diff --git a/anta/cli/exec/utils.py b/anta/cli/exec/utils.py index 758072c..a5f7da2 100644 --- a/anta/cli/exec/utils.py +++ b/anta/cli/exec/utils.py @@ -14,12 +14,13 @@ import re from pathlib import Path from typing import TYPE_CHECKING, Literal -from aioeapi import EapiCommandError from click.exceptions import UsageError from httpx import ConnectError, HTTPError +from anta.custom_types import REGEXP_PATH_MARKERS from anta.device import AntaDevice, AsyncEOSDevice from anta.models import AntaCommand +from asynceapi import EapiCommandError if TYPE_CHECKING: from anta.inventory import AntaInventory @@ -29,7 +30,7 @@ INVALID_CHAR = "`~!@#$/" logger = logging.getLogger(__name__) -async def clear_counters_utils(anta_inventory: AntaInventory, tags: set[str] | None = None) -> None: +async def clear_counters(anta_inventory: AntaInventory, tags: set[str] | None = None) -> None: """Clear counters.""" async def clear(dev: AntaDevice) -> None: @@ -60,7 +61,7 @@ async def collect_commands( async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "text"]) -> None: outdir = Path() / root_dir / dev.name / outformat outdir.mkdir(parents=True, exist_ok=True) - safe_command = re.sub(r"(/|\|$)", "_", command) + safe_command = re.sub(rf"{REGEXP_PATH_MARKERS}", "_", command) c = AntaCommand(command=command, ofmt=outformat) await dev.collect(c) if not c.collected: @@ -72,6 +73,9 @@ async def collect_commands( elif c.ofmt == "text": outfile = outdir / f"{safe_command}.log" content = c.text_output + else: + logger.error("Command outformat is not in ['json', 'text'] for command '%s'", command) + return with outfile.open(mode="w", encoding="UTF-8") as f: f.write(content) logger.info("Collected command '%s' from device %s (%s)", command, dev.name, dev.hw_model) @@ -91,7 +95,7 @@ async def collect_commands( logger.error("Error when collecting commands: %s", str(r)) -async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, *, configure: bool, tags: set[str] | None = None, latest: int | None = None) -> None: +async def collect_show_tech(inv: AntaInventory, root_dir: Path, *, configure: bool, tags: set[str] | None = None, latest: int | None = None) -> None: """Collect scheduled show-tech on devices.""" async def collect(device: AntaDevice) -> None: @@ -103,12 +107,12 @@ async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, *, con cmd += f" | head -{latest}" command = AntaCommand(command=cmd, ofmt="text") await device.collect(command=command) - if command.collected and command.text_output: - filenames = [Path(f"{EOS_SCHEDULED_TECH_SUPPORT}/{f}") for f in command.text_output.splitlines()] - else: + if not (command.collected and command.text_output): logger.error("Unable to get tech-support filenames on %s: verify that %s is not empty", device.name, EOS_SCHEDULED_TECH_SUPPORT) return + filenames = [Path(f"{EOS_SCHEDULED_TECH_SUPPORT}/{f}") for f in command.text_output.splitlines()] + # Create directories outdir = Path() / root_dir / f"{device.name.lower()}" outdir.mkdir(parents=True, exist_ok=True) @@ -119,31 +123,32 @@ async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, *, con if command.collected and not command.text_output: logger.debug("'aaa authorization exec default local' is not configured on device %s", device.name) - if configure: - commands = [] - # TODO: @mtache - add `config` field to `AntaCommand` object to handle this use case. - # Otherwise mypy complains about enable as it is only implemented for AsyncEOSDevice - # TODO: Should enable be also included in AntaDevice? - if not isinstance(device, AsyncEOSDevice): - msg = "anta exec collect-tech-support is only supported with AsyncEOSDevice for now." - raise UsageError(msg) - if device.enable and device._enable_password is not None: # pylint: disable=protected-access - commands.append({"cmd": "enable", "input": device._enable_password}) # pylint: disable=protected-access - elif device.enable: - commands.append({"cmd": "enable"}) - commands.extend( - [ - {"cmd": "configure terminal"}, - {"cmd": "aaa authorization exec default local"}, - ], - ) - logger.warning("Configuring 'aaa authorization exec default local' on device %s", device.name) - command = AntaCommand(command="show running-config | include aaa authorization exec default local", ofmt="text") - await device._session.cli(commands=commands) # pylint: disable=protected-access - logger.info("Configured 'aaa authorization exec default local' on device %s", device.name) - else: + if not configure: logger.error("Unable to collect tech-support on %s: configuration 'aaa authorization exec default local' is not present", device.name) return + + commands = [] + # TODO: @mtache - add `config` field to `AntaCommand` object to handle this use case. + # Otherwise mypy complains about enable as it is only implemented for AsyncEOSDevice + # TODO: Should enable be also included in AntaDevice? + if not isinstance(device, AsyncEOSDevice): + msg = "anta exec collect-tech-support is only supported with AsyncEOSDevice for now." + raise UsageError(msg) + if device.enable and device._enable_password is not None: # pylint: disable=protected-access + commands.append({"cmd": "enable", "input": device._enable_password}) # pylint: disable=protected-access + elif device.enable: + commands.append({"cmd": "enable"}) + commands.extend( + [ + {"cmd": "configure terminal"}, + {"cmd": "aaa authorization exec default local"}, + ], + ) + logger.warning("Configuring 'aaa authorization exec default local' on device %s", device.name) + command = AntaCommand(command="show running-config | include aaa authorization exec default local", ofmt="text") + await device._session.cli(commands=commands) # pylint: disable=protected-access + logger.info("Configured 'aaa authorization exec default local' on device %s", device.name) + logger.debug("'aaa authorization exec default local' is already configured on device %s", device.name) await device.copy(sources=filenames, destination=outdir, direction="from") |