Skip to content
Snippets Groups Projects
Select Git revision
  • a5f8e95f76dca54017b0acb0e38c504f2ddbb949
  • develop default protected
  • main protected
3 results

conf.py

Blame
  • test_db_model.py 2.46 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_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)