Skip to content
Snippets Groups Projects
test_db_model.py 2.49 KiB
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_router_ports

#
# 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_router_ports(hostname=router_name, ssh_config=ssh))
    _save_router_info(fqdn=router_name, fpcs=fpcs)

    fpc_slot_dict = {f['slot']: 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_slots = set(fpc_slot_dict.keys())

        fpc_slots = {fpc.slot for fpc in node.line_cards}
        assert expected_fpc_slots == fpc_slots

        for fpc in node.line_cards:
            expected_ports = {
                (p['pic'], p['position']) for p in fpc_slot_dict[fpc.slot]}
            db_ports = {(p.pic, p.position) for p in fpc.ports}
            assert expected_ports == db_ports


# 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)