Skip to content
Snippets Groups Projects
Select Git revision
  • DBOARD3-1142/token-auth
  • feature/DBOARD3-1137-implement-whois
  • develop default
  • master protected
  • feature/DBOARD3-958
  • release/0.110
  • fix-uuid-validation-error
  • docker-poc
  • 0.143
  • 0.142
  • 0.141
  • 0.140
  • 0.139
  • 0.138
  • 0.137
  • 0.136
  • 0.135
  • 0.134
  • 0.133
  • 0.132
  • 0.131
  • 0.130
  • 0.129
  • 0.128
  • 0.127
  • 0.126
  • 0.125
  • 0.124
28 results
Blame Permalink
test_celery_worker.py 3.10 KiB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105 






"""
just checks that the worker methods call the right functions
and some data ends up in the right place ... otherwise not very detailed
"""
import re
from inventory_provider.tasks import worker
from inventory_provider.tasks.common import _get_redis


def backend_db():
    return _get_redis({
        'redis': {
            'hostname': None,
            'port': None
        },
        'redis-databases': [0, 7]
    }).db


def test_netconf_refresh_config(mocked_worker_module, router):
    del backend_db()['netconf:' + router]
    worker.netconf_refresh_config(router)
    assert backend_db()['netconf:' + router]


def test_snmp_refresh_interfaces(mocked_worker_module, router):

    def _wanted(s):
        return s.startswith(f'snmp-interfaces-single:{router}')

    def _ifc_keys():
        return filter(_wanted, backend_db().keys())

    del backend_db()[f'snmp-interfaces:{router}']
    for k in list(_ifc_keys()):
        del backend_db()[k]

    worker.snmp_refresh_interfaces(router, 'fake-community', [])
    assert backend_db()['snmp-interfaces:' + router]
    assert list(_ifc_keys())


def test_snmp_refresh_peerings(mocked_worker_module, router):

    def _wanted(s):
        return s.startswith('snmp-peerings:')

    def _ifc_keys():
        return filter(_wanted, backend_db().keys())

    for k in list(_ifc_keys()):
        del backend_db()[k]

    worker.snmp_refresh_peerings(router, 'fake-community', [])
    assert list(_ifc_keys())


def test_reload_router_config(mocked_worker_module, router, mocker):
    saved_data = {}
    for key in ('netconf:' + router, 'snmp-interfaces:' + router):
        saved_data[key] = backend_db().pop(key)

    saved_peerings = {}
    for key in backend_db().keys():
        if re.match(r'^snmp-peerings:\d.*', key):
            saved_peerings[key] = backend_db()[key]
    for key in saved_peerings.keys():
        del backend_db()[key]

    assert 'netconf:' + router not in backend_db()
    assert 'snmp-interfaces:' + router not in backend_db()

    def _mocked_netconf_refresh_config_apply(args):
        key = 'netconf:' + args[0]
        backend_db()[key] = saved_data[key]
    mocker.patch(
        'inventory_provider.tasks.worker.netconf_refresh_config.apply',
        _mocked_netconf_refresh_config_apply)

    def _mocked_snmp_refresh_peerings_apply(args):
        assert len(args) == 3
        backend_db().update(saved_peerings)
    mocker.patch(
        'inventory_provider.tasks.worker.snmp_refresh_peerings.apply',
        _mocked_snmp_refresh_peerings_apply)

    def _mocked_snmp_refresh_interfaces_apply(args):
        assert len(args) == 3
        key = 'snmp-interfaces:' + args[0]
        backend_db()[key] = saved_data[key]

    mocker.patch(
        'inventory_provider.tasks.worker.snmp_refresh_interfaces.apply',
        _mocked_snmp_refresh_interfaces_apply)

    def _mocked_update_status(self, **kwargs):
        pass
    mocker.patch(
        'inventory_provider.tasks.worker.InventoryTask.update_state',
        _mocked_update_status)

    worker.reload_router_config(router)
    assert 'netconf:' + router in backend_db()
    assert 'snmp-interfaces:' + router in backend_db()