test.py

#!/usr/bin/python
# coding: utf-8 -*-

import sys
import logging
import pprint
import random
from pydantic import BaseModel, ValidationError, validator
from anta import inventory
from anta.inventory import AntaInventory
from anta.inventory.models import InventoryDevices
# from anta.result_manager import ResultManager
# from anta.result_manager.models import TestResult

logging.basicConfig(level=logging.DEBUG)


class Test(BaseModel):
    test: str
    test2: str

    @validator('test2')
    def name_must_contain_space(cls, v):
        if 'success' not in v:
            raise ValueError('must contain success')
        return v.title()




class TestInputError(Exception):
    """Error raised when inventory root key is not found."""

def verify_eos_version(device, enable_password: str = None, versions: str = None):
    result = TestResult(host=str(device.host), test='verify_eos_version', )
    if not versions:
        raise TestInputError('Version not set in your test')

    try:
        response = device.session.runCmds(1, ['show version'], 'json')
    except Exception:
        result.result = 'failure'
        result.message = 'error collecting data from device'
    else:
        try:
            if response[0]['version'] in versions:
                result.result = 'success'
            else:
                result.result = 'failure'
                result.message = f'device is running version {response[0]["version"]} and test expect {versions}'
        except KeyError:
            result.result = 'failure'
            result.message = f'version not found in device response: {response}'
    return result


def verify_uptime(device, enable_password: str = None, minimum: int = None):
    result = TestResult(host=str(device.host), test='verify_uptime', )
    if not minimum:
        raise TestInputError('Version not set in your test')

    try:
        response = device.session.runCmds(1, ['show uptime'], 'json')
    except Exception:
        result.result = 'failure'
        result.message = 'error collecting data from device'
    else:
        try:
            if response[0]['upTime'] > minimum:
                result.result = 'success'
            else:
                result.result = 'failure'
                result.message = f'device is running version {response[0]["upTime"]} and test expect {minimum}'
        except KeyError:
            result.result = 'failure'
            result.message = f'version not found in device response: {response}'
    return result


if __name__ == '__main__':

    logging.disable(sys.maxsize)

    print('Load inventory file')

    inventory_anta = AntaInventory(
        inventory_file='examples/inventory.yml',
        username='ansible',
        password='ansible',
        timeout=0.5,
        auto_connect=True
    )

    print('Run testing')

    print(len(inventory_anta.get_inventory()))

    # manager = ResultManager()
    # for device in inventory_anta.get_inventory():
    #     result = verify_eos_version(
    #         device=device, versions=['4.28.0F']
    #     )
    #     manager.add_test_result(result)

    #     manager.add_test_result(
    #         verify_eos_version(
    #             device=device, versions=[
    #                 '4.27.3F-26379303.4273F (engineering build)']
    #         )
    #     )
    #     manager.add_test_result(
    #         verify_uptime(
    #             device=device, minimum=1
    #         )
    #     )

    # print('List of test results:')
    # print(manager.get_results())
    # print('------------------')
    # print(manager.get_result_by_host(host_ip='192.168.0.10'))
    # print('------------------')
    # print('Display result\n')

    # print(
    #     manager.table_report(
    #         sort_by='host',
    #         reverse=False
    #     )
    # )

    # test = TestResult(host='1.1.1.1', test='validator', result='success')

    # print(test.dict())

    # f = open("report.txt", "w")
    # f.write(
    #     manager.table_report(
    #         sort_by='host',
    #         reverse=False,
    #         colors=False
    #     )
    # )
    # f.close()
    sys.exit(0)