test_ims_worker.py

import os
import re

from inventory_provider.tasks.ims_worker import export_data_for_otrs, OTRSFiles


def test_otrs_exports(data_config_filename, data_config, mocker):
    os.environ['INVENTORY_PROVIDER_CONFIG_FILENAME'] = data_config_filename
    otrs_config = data_config["otrs-export"]
    mocked_writer = \
        mocker.patch('inventory_provider.tasks.ims_worker.csv.writer')
    mocked_row_writer = mocked_writer.return_value.writerows
    mocked_run = mocker.patch(
        'inventory_provider.tasks.ims_worker.subprocess.run')
    mocker.patch(
        'inventory_provider.tasks.ims_worker.ims_data.otrs_get_customer_company_rows',  # noqa
        return_value=[[1, 2, 3, 4], [5, 6, 7, 8]]
    )
    mocker.patch(
        'inventory_provider.tasks.ims_worker.ims_data.otrs_get_customer_users_rows',  # noqa
        return_value=[[9, 10, 11, 12], [13, 14, 15, 16]]
    )

    export_data_for_otrs(OTRSFiles.CUSTOMER_COMPANIES)
    mocked_row_writer.assert_called_with([[1, 2, 3, 4], [5, 6, 7, 8]])

    export_data_for_otrs(OTRSFiles.CUSTOMER_USERS)
    mocked_row_writer.assert_called_with([[9, 10, 11, 12], [13, 14, 15, 16]])

    export_data_for_otrs()
    assert mocked_row_writer.call_count == 4

    args, kwargs = mocked_run.call_args
    called_with = args[0]

    t = r'^rsync -aP --rsh="ssh -l {user} -p 22 -i {key_file} -o \'UserKnownHostsFile {known_hosts}\'" /\w+/\w+/\* {destination}$'  # noqa

    p = t.format(
        user=otrs_config['username'],
        key_file=otrs_config['private-key'],
        destination=otrs_config['destination'],
        known_hosts=otrs_config['known-hosts']
    )
    assert bool(re.match(p, called_with))