ANTA Class Diagram

classDiagram
  direction LR
  namespace Inventory {
    class AntaInventory {
      devices : list[AntaDevice]
      max_potential_connections : int | None
      INVENTORY_ROOT_KEY : str$
      INVENTORY_OUTPUT_FORMAT : list[str]$
      parse(filename: str | Path, username: str, password: str, enable_password: str | None, timeout: float | None, file_format: Literal['yaml', 'json'], enable: bool, insecure: bool, disable_cache: bool, use_session_auth: bool | None) AntaInventory$
      add_device(device: AntaDevice) None
      connect_inventory() None
      disconnect_inventory() None
      dump() AntaInventoryInput
      get_inventory(established_only: bool, tags: set[str] | None, devices: set[str] | None) AntaInventory
      is_base_class(device: AntaDevice) TypeIs[AntaDevice]
    }
    class AntaInventoryHost:::pydantic {
      name : str | None
      host : Hostname | IPvAnyAddress
      port : Port | None
      tags : set[str] | None
      disable_cache : bool
      use_session_auth : bool
    }
    class AntaInventoryInput:::pydantic {
      hosts : list[AntaInventoryHost] | None
      networks : list[AntaInventoryNetwork] | None
      ranges : list[AntaInventoryRange] | None
      yaml() str
      to_json() str
    }
    class AntaInventoryNetwork:::pydantic {
      network : IPvAnyNetwork
      tags : set[str] | None
      disable_cache : bool
      use_session_auth : bool
    }
    class AntaInventoryRange:::pydantic {
      start : IPvAnyAddress
      end : IPvAnyAddress
      tags : set[str] | None
      disable_cache : bool
      use_session_auth : bool
    }
  }

  namespace Device {
    class AntaDevice {
      <<Abstract>>
      capabilities : AntaDeviceCapabilities$
      name : str
      tags : set[str]
      hw_model : str | None
      established : bool
      is_online : bool
      cache : AntaCache | None
      cache_locks : defaultdict[str, asyncio.Lock] | None
      max_connections : int | None
      cache_statistics : dict[str, Any] | None
      collect(command: AntaCommand, collection_id: str | None) None
      collect_commands(commands: list[AntaCommand], collection_id: str | None) None
      copy(sources: list[Path], destination: Path, direction: Literal['to', 'from']) None
      disconnect() None
      refresh()* None
      _collect(command: AntaCommand, collection_id: str | None)* None
    }
    class AsyncEOSDevice {
      capabilities : AntaDeviceCapabilities$
      enable : bool
      max_connections : int | None
      use_session_auth : bool
      _client : asynceapi.Device
      _eapi_opts : EAPIClientConnectionOptions
      _ssh_opts : SSHClientConnectionOptions
      _command_semaphore : asyncio.Semaphore
      copy(sources: list[Path], destination: Path, direction: Literal['to', 'from']) None
      disconnect() None
      refresh() None
      _collect(command: AntaCommand, collection_id: str | None) None
    }
    class AntaDeviceCapabilities:::dataclass {
      <<Frozen Dataclass>>
      supports_session_auth : bool
    }
    class EAPIClientConnectionOptions:::dataclass {
      <<Frozen Dataclass>>
      host : str
      username : str
      password : str
      port : int | None
      proto : Literal['http', 'https']
      timeout : float | None
      use_session_auth : bool
    }
  }

  namespace Catalog {
    class AntaCatalog {
      tests : list[AntaTestDefinition]
      filename : Path | None
      indexes_built : bool
      tag_to_tests : defaultdict[str | None, set[AntaTestDefinition]]
      parse(filename: str | Path, file_format: Literal['yaml', 'json']) AntaCatalog$
      from_dict(data: RawCatalogInput, filename: str | Path | None) AntaCatalog$
      from_list(data: ListAntaTestTuples) AntaCatalog$
      merge_catalogs(catalogs: list[AntaCatalog]) AntaCatalog$
      build_indexes(filtered_tests: set[str] | None) None
      clear_indexes() None
      dump() AntaCatalogFile
      get_tests_by_tags(tags: set[str], strict: bool) set[AntaTestDefinition]
      merge(catalog: AntaCatalog) AntaCatalog
    }
    class AntaCatalogFile:::pydantic {
      root : dict[ImportString[Any], list[AntaTestDefinition]]
      flatten_modules(data: dict[str, Any], package: str | None) dict[ModuleType, list[Any]]$
      yaml() str
      to_json() str
    }
    class AntaTestDefinition:::pydantic {
      inputs : Input
      test : type[AntaTest]
      check_inputs() Self
      instantiate_inputs(data: AntaTest.Input | dict[str, Any] | None, info: ValidationInfo) AntaTest.Input
      serialize_model() dict[str, AntaTest.Input]
    }
  }

  namespace Test {
    class AntaTest {
      <<Abstract>>
      name : str$
      description : str$
      categories : list[str]$
      commands : list[AntaTemplate | AntaCommand]$
      progress : Progress | None$
      nrfu_task : TaskID | None$
      device : AntaDevice
      inputs : Input
      result : TestResult
      instance_commands : list[AntaCommand]
      failed_commands : list[AntaCommand]
      collected : bool
      blocked : bool
      module : str
      logger : Logger
      anta_test(function: Callable[..., Any]) Callable[..., Coroutine[Any, Any, TestResult]]$
      collect() None
      render(template: AntaTemplate) list[AntaCommand]
      save_commands_data(eos_data: list[dict[str, Any] | str]) None
      update_progress() None$
      test() Coroutine[Any, Any, TestResult]*
    }
    class Input:::pydantic {
      filters : Filters | None
      result_overwrite : ResultOverwrite | None
    }
    class Filters:::pydantic {
      tags : set[str] | None
    }
    class ResultOverwrite:::pydantic {
      categories : list[str] | None
      custom_field : str | None
      description : str | None
    }
    class AntaCommand:::pydantic {
      command : str
      version : Literal[1, 'latest']
      revision : Revision | None
      ofmt : Literal['json', 'text']
      output : dict[str, Any] | str | None
      template : AntaTemplate | None
      errors : list[str]
      params : AntaParamsBaseModel
      use_cache : bool
      uid : str
      json_output : dict[str, Any]
      text_output : str
      error : bool
      collected : bool
      requires_privileges : bool
      returned_known_eos_error : bool
      supported : bool
    }
    class AntaTemplate {
      template : str
      version : Literal[1, 'latest']
      revision : Revision | None
      ofmt : Literal['json', 'text']
      use_cache : bool
      params_schema : type[AntaParamsBaseModel]
      render(params: str | int | bool) AntaCommand
    }
  }

  namespace Result_Manager {
    class ResultManager {
      results : list[TestResult]
      status : AntaTestStatus
      error_status : bool
      dump : list[dict[str, Any]]
      json : str
      results_by_status : dict[AntaTestStatus, list[TestResult]]
      results_by_category : list[TestResult]
      device_stats : dict[str, DeviceStats]
      category_stats : dict[str, CategoryStats]
      sorted_category_stats : dict[str, CategoryStats]
      test_stats : dict[str, TestStats]
      add(result: TestResult) None
      filter(hide: set[AntaTestStatus]) ResultManager
      filter_by_devices(devices: set[str]) ResultManager
      filter_by_tests(tests: set[str]) ResultManager
      get_devices() set[str]
      get_results(status: set[AntaTestStatus] | None, sort_by: list[str] | None) list[TestResult]
      get_status(ignore_error: bool) str
      get_tests() set[str]
      get_total_results(status: set[AntaTestStatus] | None) int
      merge_results(results_managers: list[ResultManager]) ResultManager$
      reset() None
      sort(sort_by: list[str]) ResultManager
    }
    class TestResult:::pydantic {
      name : str
      test : str
      categories : list[str]
      description : str
      result : AntaTestStatus
      messages : list[str]
      atomic_results : list[AtomicTestResult]
      custom_field : str | None
      add(description: str, status: AntaTestStatus, messages: list[str] | None) AtomicTestResult
      is_error(message: str | None) None
      is_failure(message: str | None) None
      is_skipped(message: str | None) None
      is_success(message: str | None) None
    }
    class AtomicTestResult:::pydantic {
      description : str
      result : AntaTestStatus
      messages : list[str]
      parent : TestResult
      is_error(message: str | None) None
      is_failure(message: str | None) None
      is_skipped(message: str | None) None
      is_success(message: str | None) None
    }
    class AntaTestStatus {
      <<Enumeration>>
      UNSET
      SUCCESS
      FAILURE
      ERROR
      SKIPPED
    }
    class DeviceStats:::dataclass {
      <<Dataclass>>
      tests_success_count : int
      tests_skipped_count : int
      tests_failure_count : int
      tests_error_count : int
      tests_unset_count : int
      tests_failure : set[str]
      categories_failed : set[str]
      categories_skipped : set[str]
    }
    class CategoryStats:::dataclass {
      <<Dataclass>>
      tests_success_count : int
      tests_skipped_count : int
      tests_failure_count : int
      tests_error_count : int
      tests_unset_count : int
    }
    class TestStats:::dataclass {
      <<Dataclass>>
      devices_success_count : int
      devices_skipped_count : int
      devices_failure_count : int
      devices_error_count : int
      devices_unset_count : int
      devices_failure : set[str]
    }
  }

  AntaInventory o-- AntaDevice : devices
  AntaInventory ..> AntaInventoryInput : parse()/dump()
  AntaInventoryInput ..> AntaInventoryHost
  AntaInventoryInput ..> AntaInventoryNetwork
  AntaInventoryInput ..> AntaInventoryRange
  AntaCatalog o-- AntaTestDefinition : tests
  AntaCatalog ..> AntaCatalogFile : dump()
  AntaTestDefinition *-- Input : inputs
  AntaTestDefinition ..> AntaTest : test
  AsyncEOSDevice --|> AntaDevice
  AntaDevice *-- AntaDeviceCapabilities : capabilities
  AsyncEOSDevice *-- AntaDeviceCapabilities : capabilities
  AsyncEOSDevice *-- EAPIClientConnectionOptions : _eapi_opts
  AntaDevice --o AntaTest : device
  AntaTest *-- Input : inputs
  Input *-- Filters : filters
  Input *-- ResultOverwrite : result_overwrite
  AntaTest o-- AntaCommand : commands
  AntaTest o-- AntaCommand : instance_commands
  AntaTest o-- AntaTemplate : commands
  AntaTemplate ..> AntaCommand : render()
  AntaTest *-- TestResult : result
  ResultManager o-- TestResult : results
  ResultManager *-- AntaTestStatus : status
  ResultManager *-- DeviceStats : device_stats
  ResultManager *-- CategoryStats : category_stats
  ResultManager *-- TestStats : test_stats
  TestResult o-- AtomicTestResult : atomic_results
  TestResult *-- AntaTestStatus : result
  AtomicTestResult *-- AntaTestStatus : result
  classDef pydantic fill:#D63965
  classDef dataclass fill:#2F9E44