test_db_model.py

from random import randint

from resource_management.db import model
from resource_management.db import session_scope
from resource_management.cli import _save_router_info
from resource_management.hardware.router import load_line_cards


def test_model_sanity_check(resources_db, router_name):

    test_node = [randint(1, 50) for _ in range(randint(1, 50))]

    with session_scope() as session:

        node = model.Node(fqdn=router_name)
        session.add(node)

        for position, num_ports in enumerate(test_node):
            line_card = model.LineCard(
                model=f'XYZ-{position}', position=position, node=node)
            session.add(line_card)
            for idx in range(num_ports):
                port = model.Port(
                    name=str(idx), line_card=line_card, speed=1000)
                session.add(port)

    # new session - previous rows should have been committed
    with session_scope() as session:
        node = session.query(model.Node) \
            .filter(model.Node.fqdn == router_name).one()
        assert len(node.line_cards) == len(test_node)
        for expected_num_ports, line_card in zip(test_node, node.line_cards):
            assert expected_num_ports == len(line_card.ports)


def test_save_router_info(resources_db, router_name, mocked_router):

    ssh = {
        'username': 'bogus',
        'private-key': 'no file'
    }

    fpcs = list(load_line_cards(hostname=router_name, ssh_config=ssh))
    _save_router_info(fqdn=router_name, fpcs=fpcs)

    fpc_position_dict = {f['position']: f['ports'] for f in fpcs}

    with session_scope() as session:
        node = session.query(model.Node) \
            .filter(model.Node.fqdn == router_name).one()
        expected_fpc_positions = set(fpc_position_dict.keys())
        fpc_positions = {fpc.position for fpc in node.line_cards}
        assert expected_fpc_positions == fpc_positions

        for fpc in node.line_cards:
            expected_ports = {
                p['name'] for p in fpc_position_dict[fpc.position]}
            port_names = {p.name for p in fpc.ports}
            assert expected_ports == port_names


# TODO: do this in a tmp container ...
# from resource_management.db import mysql_dsn
# from resource_management.migrations import migration_utils
#
#
# def test_migration():
#     dsn = mysql_dsn(
#         username='dummy',
#         password='dummy-pass',
#         hostname='localhost',
#         db_name='resources_test')
#     migration_utils.upgrade(dsn=dsn)