test_ims.py

from requests import HTTPError

import inventory_provider


class MockResponse:
    def __init__(self, json_data, status_code):
        self.json_data = json_data
        self.status_code = status_code
        self.text = '' if json_data else 'No records found for Entity:XXXXX'

    def json(self):
        return self.json_data

    def raise_for_status(self):
        pass


def test_ims_class_login(mocker):
    mock_post = mocker.patch('inventory_provider.db.ims.requests.post')
    mock_get = mocker.patch('inventory_provider.db.ims.requests.get')
    mock_post.return_value = MockResponse("my_bearer_token", 200)
    mock_post.return_value.text = "my_bearer_token"

    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password')
    ds.get_entity_by_id('Node', 1234)
    mock_post.assert_called_once_with(
        'dummy_base/login', auth=('dummy_username', 'dummy_password'))
    mock_get.assert_called_once_with(
        'dummy_base/ims/Node/1234',
        headers={'Authorization': 'Bearer my_bearer_token'},
        params=None)


def test_ims_class_entity_by_id(mocker):
    mock_get = mocker.patch('inventory_provider.db.ims.requests.get')
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password', 'dummy_bt')

    ds.get_entity_by_id('Node', 1234)
    mock_get.assert_called_once_with(
        'dummy_base/ims/Node/1234',
        headers={'Authorization': 'Bearer dummy_bt'},
        params=None)


def test_ims_class_entity_by_name(mocker):
    mock_get = mocker.patch('inventory_provider.db.ims.requests.get')
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password', 'dummy_bt')

    ds.get_entity_by_name('Node', 'dummy_name')
    mock_get.assert_called_once_with(
        'dummy_base/ims/Node/byname/dummy_name',
        headers={'Authorization': 'Bearer dummy_bt'},
        params=None)


def test_ims_class_filtered_entities(mocker):
    mock_get = mocker.patch('inventory_provider.db.ims.requests.get')
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password', 'dummy_bt')

    list(ds.get_filtered_entities(
        'Node', 'dummy_param=dummy value', step_count=50))
    mock_get.assert_called_once_with(
        'dummy_base/ims/Node/filtered/dummy_param=dummy value',
        headers={'Authorization': 'Bearer dummy_bt'},
        params={
            'paginatorStartElement': 0,
            'paginatorNumberOfElements': 50
        })

    def side_effect(*args, **kargs):
        if kargs['params']['paginatorStartElement'] == 0:
            return MockResponse([1, 2], 200)
        return MockResponse([3], 200)

    mock_multi_get = mocker.patch(
        'inventory_provider.db.ims.requests.get', side_effect=side_effect)

    res = list(ds.get_filtered_entities(
        'Node', 'dummy_param=dummy value', step_count=2))
    mock_multi_get.assert_any_call(
        'dummy_base/ims/Node/filtered/dummy_param=dummy value',
        headers={'Authorization': 'Bearer dummy_bt'},
        params={
            'paginatorStartElement': 0,
            'paginatorNumberOfElements': 2
        })
    mock_multi_get.assert_any_call(
        'dummy_base/ims/Node/filtered/dummy_param=dummy value',
        headers={'Authorization': 'Bearer dummy_bt'},
        params={
            'paginatorStartElement': 2,
            'paginatorNumberOfElements': 2
        })
    assert mock_multi_get.call_count == 2
    assert res == [1, 2, 3]

    def side_effect_no_recs(*args, **kargs):
        if kargs['params']['paginatorStartElement'] == 0:
            return MockResponse([1, 2], 200)
        e = HTTPError()
        e.response = MockResponse('', 404)
        raise e

    mocker.patch('inventory_provider.db.ims.requests.get',
                 side_effect=side_effect_no_recs)

    res = list(ds.get_filtered_entities(
        'Node', 'dummy_param=dummy value', step_count=2))
    assert res == [1, 2]


def test_ims_class_get_all_entities(mocker):
    mock_get = mocker.patch('inventory_provider.db.ims.requests.get')
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password', 'dummy_bt')

    list(ds.get_all_entities('Node', step_count=10))
    mock_get.assert_called_once_with(
        'dummy_base/ims/Node/filtered/Id <> 0',
        headers={'Authorization': 'Bearer dummy_bt'},
        params={
            'paginatorStartElement': 0,
            'paginatorNumberOfElements': 10
        })


def test_ims_class_navigation_properties(mocker):
    mock_get = mocker.patch('inventory_provider.db.ims.requests.get')
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password', 'dummy_bt')

    ds.get_entity_by_id('Node', 1234, navigation_properties=[1, 2, 3])
    mock_get.assert_called_with(
        'dummy_base/ims/Node/1234',
        headers={'Authorization': 'Bearer dummy_bt'},
        params={'navigationproperty': 6})


def test_ims_class_cache(mocker):
    mock_get = mocker.patch('inventory_provider.db.ims.requests.get')
    ds = inventory_provider.db.ims.IMS(
        'dummy_base', 'dummy_username', 'dummy_password', 'dummy_bt')

    ds.get_entity_by_id(
        'Node', 4567, navigation_properties=[1, 2, 3, 4], use_cache=True)
    assert mock_get.call_count == 1
    ds.get_entity_by_id(
        'Node', 4567, navigation_properties=[1, 2, 3, 4], use_cache=True)
    assert mock_get.call_count == 1
    ds.get_entity_by_id(
        'Node', 4567, navigation_properties=[1, 2, 3, 4], use_cache=False)
    assert mock_get.call_count == 2
    ds.get_entity_by_id(
        'Node', 4567, navigation_properties=[1, 2, 3, 4])
    assert mock_get.call_count == 3
    ds.get_entity_by_id(
        'Node', 4567, navigation_properties=[1, 2, 3, 4, 5], use_cache=True)
    assert mock_get.call_count == 4