From 8c3404bf580cfa39e2697d90d3c91b30d581ad39 Mon Sep 17 00:00:00 2001 From: Thomas Grimonet Date: Thu, 8 Sep 2022 07:35:40 +0200 Subject: [PATCH] feat(inventory): Add tag support to filter test execution (#112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(inventory): Add initial support for tags Add support for tags in inventory module as per issue #44 ```yaml anta_inventory: hosts: - host: 192.168.0.10 tags: [] - host: 192.168.0.11 tags: [] networks: - network: 192.168.110.0/24 tags: [] ranges: - start: 10.0.0.9 end: 10.0.0.11 tags: [] ``` * doc: Fix typo in docstring * feat(inventory): Add method to filter inventory per tags - Update inventory module to support tag filtering in get_inventory - Update check-devices.py to support tags Inventory example: ```yaml anta_inventory: hosts: - host: 192.168.0.10 tags: ['test'] - host: 192.168.0.11 tags: ['dc1'] - host: 192.168.0.12 tags: ['dc1'] - host: 192.168.0.13 tags: ['dc2'] - host: 192.168.0.14 - host: 192.168.0.15 - host: 10.73.252.11 tags: ['dc1'] ranges: - start: 10.73.252.12 end: 10.73.252.50 tags: ['clab'] ``` Code execution ```bash python scripts/check-devices.py -i .personal/avd-lab.yml -c .personal/ceos-catalog.yml --table --tags dc1 [08:07:43] INFO Inventory .personal/avd-lab.yml loaded check-devices.py INFO starting running test on inventory ... check-devices.p [08:07:44] INFO testing done ! All tests results ┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓ ┃ Device IP ┃ Test Name ┃ Test Status ┃ Message(s) ┃ ┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩ │ 10.73.252.11 │ verify_eos_version │ failure │ device is running version 4.27.2F-26069621.4272F (engineering build) not in expected versions: ['4.25.4M', '4.26.1F'] │ │ 10.73.252.11 │ verify_field_notice_44_resolution │ skipped │ verify_field_notice_44_resolution test is not supported on cEOSLab. │ │ 10.73.252.11 │ verify_uptime │ success │ │ │ 10.73.252.11 │ verify_zerotouch │ success │ │ │ 10.73.252.11 │ verify_running_config_diffs │ success │ │ │ 10.73.252.11 │ verify_mlag_status │ success │ │ │ 10.73.252.11 │ verify_mlag_interfaces │ success │ │ │ 10.73.252.11 │ verify_mlag_config_sanity │ success │ │ │ 10.73.252.11 │ verify_routing_protocol_model │ success │ │ └──────────────┴───────────────────────────────────┴─────────────┴───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ ``` * doc: Update rwith tag support --- README.md | 7 ++++++ anta/inventory/__init__.py | 48 +++++++++++++++++++++++++++----------- anta/inventory/models.py | 13 +++++++++-- scripts/check-devices.py | 11 +++++++-- 4 files changed, 61 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 3d38c712b..3535bb7c2 100755 --- a/README.md +++ b/README.md @@ -73,15 +73,21 @@ anta_inventory: hosts: - host: 192.168.0.10 - host: 192.168.0.11 + # Optional tag to assign to this device + tags: ['tag01', 'tag02'] - host: 192.168.0.12 - host: 192.168.0.13 - host: 192.168.0.14 - host: 192.168.0.15 networks: - network: '192.168.110.0/24' + # Optional tag to assign to all devices in this subnet + tags: ['tag01', 'tag02'] ranges: - start: 10.0.0.9 end: 10.0.0.11 + # Optional tag to assign to all devices in this range + tags: ['tag01', 'tag02'] - start: 10.0.0.100 end: 10.0.0.101 ``` @@ -164,6 +170,7 @@ optional arguments: eAPI connection timeout --hostip HOSTIP search result for host --test TEST search result for test + --tags TAGS List of device tags to limit scope of testing --list Display internal data --table Result represented in tables --all-results Display all test cases results. Default table view (Only valid with --table) diff --git a/anta/inventory/__init__.py b/anta/inventory/__init__.py index 494f03a24..81404bd21 100644 --- a/anta/inventory/__init__.py +++ b/anta/inventory/__init__.py @@ -20,7 +20,7 @@ from .exceptions import (InventoryIncorrectSchema, InventoryRootKeyErrors, InventoryUnknownFormat) -from .models import AntaInventoryInput, InventoryDevice, InventoryDevices +from .models import AntaInventoryInput, InventoryDevice, InventoryDevices, DEFAULT_TAG # pylint: disable=W1309 @@ -288,7 +288,7 @@ def _build_device_session(self, device: InventoryDevice, timeout: float = 5) -> device.session = connection return device - def _add_device_to_inventory(self, host_ip: str) -> None: + def _add_device_to_inventory(self, host_ip: str, tags: List[str] = None) -> None: """Add a InventoryDevice to final inventory. Create InventoryDevice and append to existing inventory @@ -299,6 +299,9 @@ def _add_device_to_inventory(self, host_ip: str) -> None: assert self._username is not None assert self._password is not None + if tags is None: + tags = [DEFAULT_TAG] + device = InventoryDevice( host=host_ip, username=self._username, @@ -308,7 +311,8 @@ def _add_device_to_inventory(self, host_ip: str) -> None: host=host_ip, username=self._username, password=self._password - ) + ), + tags=tags ) self._inventory.append(device) @@ -319,7 +323,7 @@ def _inventory_read_hosts(self) -> None: """ assert self._read_inventory.hosts is not None for host in self._read_inventory.hosts: - self._add_device_to_inventory(host_ip=str(host.host)) + self._add_device_to_inventory(host_ip=str(host.host), tags=host.tags) def _inventory_read_networks(self) -> None: """Read input data from networks section and create inventory structure. @@ -329,7 +333,7 @@ def _inventory_read_networks(self) -> None: assert self._read_inventory.networks is not None for network in self._read_inventory.networks: for host_ip in IPNetwork(str(network.network)): - self._add_device_to_inventory(host_ip=host_ip) + self._add_device_to_inventory(host_ip=host_ip, tags=network.tags) def _inventory_read_ranges(self) -> None: """Read input data from ranges section and create inventory structure. @@ -341,7 +345,8 @@ def _inventory_read_ranges(self) -> None: range_increment = IPAddress(str(range_def.start)) range_stop = IPAddress(str(range_def.end)) while range_increment <= range_stop: - self._add_device_to_inventory(host_ip=str(range_increment)) + self._add_device_to_inventory( + host_ip=str(range_increment), tags=range_def.tags) range_increment += 1 def _inventory_rebuild(self, list_devices: List[InventoryDevice]) -> InventoryDevices: @@ -360,24 +365,33 @@ def _inventory_rebuild(self, list_devices: List[InventoryDevice]) -> InventoryDe inventory.append(device) return inventory - def _filtered_inventory(self, established_only: bool = False) -> InventoryDevices: + def _filtered_inventory(self, established_only: bool = False, tags: List[str] = None) -> InventoryDevices: """ _filtered_inventory Generate a temporary inventory filtered. Args: established_only (bool, optional): Do we have to include non-established devices. Defaults to False. + tags (List[str], optional): List of tags to use to filter devices. Default is [default]. Returns: InventoryDevices: A inventory with concerned devices """ - inventory = InventoryDevices() - if not established_only: - return self._inventory + if tags is None: + tags = [DEFAULT_TAG] + inventory_filtered_tags = InventoryDevices() for device in self._inventory: + if any(tag in tags for tag in device.tags): + inventory_filtered_tags.append(device) + + if not established_only: + return inventory_filtered_tags + + inventory_final = InventoryDevices() + for device in inventory_filtered_tags: if device.established: - inventory.append(device) - return inventory + inventory_final.append(device) + return inventory_final ########################################################################### # Public methods @@ -387,7 +401,9 @@ def _filtered_inventory(self, established_only: bool = False) -> InventoryDevice # GET methods # TODO refactor this to avoid having a union of return of types .. - def get_inventory(self, format_out: str = 'native', established_only: bool = True) -> Union[List[InventoryDevice], str, InventoryDevices]: + def get_inventory( + self, format_out: str = 'native', established_only: bool = True, tags: List[str] = None + ) -> Union[List[InventoryDevice], str, InventoryDevices]: """get_inventory Expose device inventory. Provides inventory has a list of InventoryDevice objects. If requried, it can be exposed in JSON format. Also, by default expose only active devices. @@ -395,15 +411,19 @@ def get_inventory(self, format_out: str = 'native', established_only: bool = Tru Args: format (str, optional): Format output, can be native, list or JSON. Defaults to 'native'. established_only (bool, optional): Allow to expose also unreachable devices. Defaults to True. + tags (List[str], optional): List of tags to use to filter devices. Default is [default]. Returns: InventoryDevices: List of InventoryDevice """ + if tags is None: + tags = [DEFAULT_TAG] + if format_out not in ['native', 'json', 'list']: raise InventoryUnknownFormat( f'Unsupported inventory format: {format_out}. Only supported format are: {self.INVENTORY_OUTPUT_FORMAT}') - inventory = self._filtered_inventory(established_only) + inventory = self._filtered_inventory(established_only, tags) if format_out == 'list': # pylint: disable=R1721 diff --git a/anta/inventory/models.py b/anta/inventory/models.py index 424528ec7..b6ee6328a 100644 --- a/anta/inventory/models.py +++ b/anta/inventory/models.py @@ -7,6 +7,11 @@ from pydantic import BaseModel, IPvAnyAddress, IPvAnyNetwork +# Default values + +DEFAULT_TAG = 'default' +DEFAULT_HW_MODEL = 'unset' + # Pydantic models for input validation @@ -19,6 +24,7 @@ class AntaInventoryHost(BaseModel): """ host: IPvAnyAddress + tags: List[str] = [DEFAULT_TAG] class AntaInventoryNetwork(BaseModel): @@ -30,6 +36,7 @@ class AntaInventoryNetwork(BaseModel): """ network: IPvAnyNetwork + tags: List[str] = [DEFAULT_TAG] class AntaInventoryRange(BaseModel): @@ -43,6 +50,7 @@ class AntaInventoryRange(BaseModel): start: IPvAnyAddress end: IPvAnyAddress + tags: List[str] = [DEFAULT_TAG] class AntaInventoryInput(BaseModel): @@ -50,7 +58,7 @@ class AntaInventoryInput(BaseModel): User's inventory model. Attributes: - netwrks (List[AntaInventoryNetwork],Optional): List of AntaInventoryNetwork objects for networks. + networks (List[AntaInventoryNetwork],Optional): List of AntaInventoryNetwork objects for networks. hosts (List[AntaInventoryHost],Optional): List of AntaInventoryHost objects for hosts. range (List[AntaInventoryRange],Optional): List of AntaInventoryRange objects for ranges. """ @@ -86,8 +94,9 @@ class InventoryDevice(BaseModel): session: Any established = False is_online = False - hw_model: str = "unset" + hw_model: str = DEFAULT_HW_MODEL url: str + tags: List[str] = [DEFAULT_TAG] def assert_enable_password_is_not_none(self, test_name: Optional[str] = None) -> None: """ diff --git a/scripts/check-devices.py b/scripts/check-devices.py index 1b92b9adb..16eed0866 100644 --- a/scripts/check-devices.py +++ b/scripts/check-devices.py @@ -35,6 +35,7 @@ import anta.loader from anta.inventory import AntaInventory +from anta.inventory.models import DEFAULT_TAG from anta.result_manager import ResultManager from anta.reporter import ReportTable @@ -96,6 +97,9 @@ def cli_manager() -> argparse.Namespace: parser.add_argument('--test', required=False, default=None, help='search result for test') + parser.add_argument('--tags', required=False, + default=DEFAULT_TAG, help='List of device tags to limit scope of testing') + ############################# # Display Options @@ -138,6 +142,9 @@ def cli_manager() -> argparse.Namespace: auto_connect=True ) + scope_tags = cli_options.tags.split(',') if ',' in cli_options.tags else [ + cli_options.tags] + ############################################################################ # Test loader ############################################################################ @@ -152,10 +159,10 @@ def cli_manager() -> argparse.Namespace: # Test Execution ############################################################################ - logger.info('starting running test on by_host ...') + logger.info('starting running test on inventory ...') manager = ResultManager() list_tests = [] - for device, test in itertools.product(inventory_anta.get_inventory(), tests_catalog): + for device, test in itertools.product(inventory_anta.get_inventory(tags=scope_tags), tests_catalog): if ((cli_options.hostip is None or cli_options.hostip == str(device.host)) and (cli_options.test is None or cli_options.test == str(test[0].__name__))): list_tests.append(str(test[0]))