Arista Network Test Automation (ANTA) Framework

import os
from anta.inventory import AntaInventory
from anta.models import AntaTest
from anta.tests.connectivity import VerifyReachability # Requires anta-lib-tests
from anta.result_manager import ResultManager

# 1. Define your inventory programmatically or load from a file
# This example uses a simple in-memory inventory for a single device.
# Replace '127.0.0.1' with your actual device's IP/hostname.
inventory_data = {
    "lab": {
        "hosts": {
            "veos-01": {
                "host": "127.0.0.1",
                "port": 8443,
                "username": os.environ.get('ANTA_USERNAME', 'admin'),
                "password": os.environ.get('ANTA_PASSWORD', 'arista'),
                "protocol": "https_vfy"
            }
        }
    }
}
inventory = AntaInventory(inventory_data=inventory_data)

# 2. Define your tests using AntaTest objects
# This uses a built-in test from anta-lib-tests. Adjust 'hosts' for your network.
# Note: '1.1.1.1' and '2.2.2.2' are example unreachable hosts for a quick fail result.
# If you have real devices, replace with actual reachable/unreachable IPs.
tests_to_run = [
    AntaTest(test=VerifyReachability, args={"hosts": ["1.1.1.1", "2.2.2.2"]})
]

# 3. Instantiate a ResultManager to collect test outcomes
result_manager = ResultManager()

# 4. Iterate over inventory devices and execute tests
print("--- Starting ANTA tests ---")
for device_name, device in inventory.get_devices().items():
    print(f"\nRunning tests on device: {device_name} ({device.host})")
    try:
        device.connect() # Establish connection to the device
        for anta_test in tests_to_run:
            # Execute each test and record its result
            print(f"  Executing test: {anta_test.test.__name__}...")
            anta_test.test(device=device, result_manager=result_manager) # Pass device and result_manager
        device.close() # Close connection after tests
    except Exception as e:
        # Handle connection errors or general exceptions during test execution
        print(f"  ERROR during tests on {device_name}: {e}")
        # Add a skipped/failed result for the first test if connection failed
        if tests_to_run:
            result_manager.add_failure(device=device, test=tests_to_run[0].test, result="SKIPPED", messages=[f"Could not connect or run tests: {e}"])
        else:
            print(f"  No tests defined for {device_name}, but connection failed: {e}")

# 5. Print the aggregated test results
print("\n--- ANTA Test Results Summary ---")
for result in result_manager.get_results():
    print(f"Device: {result.name}, Test: {result.test}, Result: {result.result}")
    if result.messages:
        for message in result.messages:
            print(f"  - {message}")