test_ims_data.py

import json

import inventory_provider
from inventory_provider.db.ims import InventoryStatus
from inventory_provider.db.ims_data import lookup_lg_routers, \
    otrs_get_customer_company_rows, \
    otrs_get_customer_users_rows, get_node_locations, IMS_OPSDB_STATUS_MAP


def test_lookup_lg_routers(mocker):

    ims = mocker.patch('inventory_provider.db.ims.IMS')
    with open('test/data/ims_lg_data.json') as data:
        ims.return_value.get_filtered_entities.return_value = json.load(data)
    ims.return_value.get_entity_by_id.return_value = {
        'name': 'pop name',
        'longitude': 'long',
        'latitude': 'lat',
        'city': {
            'name': 'city name',
            'country': {
                'name': 'country name',
                'abbreviation': 'country code'
            }
        },
        'sitealiases': [
            {
                'aliasname': 'abbr'
            }
        ]
    }

    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password')

    res = list(lookup_lg_routers(ds))

    ds.get_filtered_entities.assert_called_once_with(
        'EquipmentDefinition',
        'Name like mx',
        inventory_provider.db.ims.EQUIP_DEF_PROPERTIES['Nodes'])

    assert ds.get_entity_by_id.call_count == 35
    assert len(res) == 35

    pop = {
        'name': 'pop name',
        'city': 'city name',
        'country': 'country name',
        'country code': 'country code',
        'abbreviation': 'abbr',
        'longitude': 'long',
        'latitude': 'lat'
    }
    assert res[0] == {
        'equipment name': 'MX3.LAB.OFFICE.GEANT.NET',
        'type': 'CORE',
        'pop': pop
    }


def test_get_node_location(mocker):
    ims = mocker.patch('inventory_provider.db.ims.IMS')
    with open('test/data/ims_nodes_data.json') as data:
        resp_data = json.load(data)
    ims.return_value.get_all_entities.return_value = resp_data

    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password')
    res = list(get_node_locations(ds))
    assert len(res) == 35
    assert res[0] == ('LON3_CX_01', {
        'equipment-name': 'LON3_CX_01',
        'status': IMS_OPSDB_STATUS_MAP[InventoryStatus.IN_SERVICE],
        'pop': {
            'name': 'LONDON 3 POWERGATE',
            'city': 'LONDON',
            'country': 'UNITED KINGDOM',
            'abbreviation': 'LON3',
            'longitude': -0.257712,
            'latitude': 51.5308142,
        }
    })


def test_otrs_get_customer_company_rows(mocker):
    ims = mocker.patch('inventory_provider.db.ims.IMS')
    with open('test/data/ims_otrs_customers.json') as data:
        resp_data = json.load(data)

    def se(*args, **kargs):
        return resp_data[args[0]]
    mocked_get_all_entities = ims.return_value.get_all_entities
    mocked_get_all_entities.side_effect = se
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password')
    cus_comp_rows = list(otrs_get_customer_company_rows(ds))
    assert len(cus_comp_rows) == 6
    mocked_get_all_entities.assert_any_call('Customer')
    mocked_get_all_entities.assert_any_call('Vendor')
    assert cus_comp_rows[0] == ['customer_id', 'name', 'street', 'zip',
                                'city', 'country', 'url', 'comments']
    ids = []
    names = []
    for row in cus_comp_rows[1:]:
        assert len(row) == 8
        ids.append(row[0])
        names.append(row[1])
    assert ids == ['DUMMY1', 'DUMMY2', 'DUMMY3', 'DUMMY4', 'DUMMY5']
    assert names == ['DUMMY 1', 'DUMMY 2', 'DUMMY 3', 'DUMMY 4', 'DUMMY 5']


def test_otrs_get_customer_users(mocker):
    ims = mocker.patch('inventory_provider.db.ims.IMS')
    resp_data = {}
    with open('test/data/ims_otrs_customers_extra.json') as data:
        resp_data['Customer'] = json.load(data)
    with open('test/data/ims_otrs_vendor_contact_extra.json') as data:
        resp_data['VendorRelatedContact'] = json.load(data)

    def se(*args, **kargs):
        return resp_data.get(args[0], [])

    mocked_get_all_entities = ims.return_value.get_all_entities
    mocked_get_all_entities.side_effect = se

    mocked_get_all_entities = ims.return_value.get_all_entities
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password')

    customer_users = list(otrs_get_customer_users_rows(ds))
    mocked_get_all_entities.assert_any_call('Customer', [32768, 262144])
    mocked_get_all_entities.assert_any_call('VendorRelatedContact', [8, 16])

    assert customer_users[0] == [
        'email',
        'username',
        'customer_id',
        'customer_id_2',
        'title',
        'firstname',
        'lastname',
        'phone',
        'fax',
        'mobile',
        'street',
        'zip',
        'city',
        'country',
        'comments'
    ]

    assert len(customer_users) == 13
    assert customer_users[1] == [
        'BANOC@DUMMY2.COM', 'BANOC@DUMMY2.COM', 'DUMMY2', '', '',
        'DUMMY 2 NOC', '-', '', '', '',
        '', '', '', '', ''
    ]
    assert customer_users[6] == [
        'HNOC@DUMMY8.COM', 'HNOC@DUMMY8.COM', 'DUMMY8', 'OTRS-GEANT-NREN', '',
        'H D_FIRST', 'H D_INFIX H D_LAST', '', '', '',
        '', '', '', '', ''
    ]
    assert customer_users[10] == [
        'K@DUMMY10.FR', 'K@DUMMY10.FR', 'DUMMY10', '', '',
        'K D_FIRST', 'K D_INFIX K D_LAST', '', '', '',
        '', '', '', '', ''
    ]
    assert customer_users[12][6] == 'M LAST'