test_db_model.py 2.48 KiB
from resource_management.db import model
from resource_management.db import session_scope
from resource_management.router_interfaces import _save_router_info
from resource_management.hardware.juniper 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)