/
test_routing.py
178 lines (140 loc) · 5.46 KB
/
test_routing.py
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
"""Main module for functional testing."""
import json
import os
import cfg_opts
import pytest
pytestmark = pytest.mark.asyncio
SERIES = [
'focal',
'bionic',
]
CHARM_BUILD_DIR = os.getenv('CHARM_BUILD_DIR', '/tmp/charm-builds')
BUILT_CHARM = os.path.join(CHARM_BUILD_DIR, "advanced-routing")
############
# FIXTURES #
############
@pytest.fixture(scope='module', params=SERIES)
async def deploy_app(request, model):
"""Deploy the advanced-routing charm as a subordinate of ubuntu."""
release = request.param
await model.deploy(
'ubuntu',
application_name='ubuntu-{}'.format(release),
series=release,
channel='stable'
)
advanced_routing = await model.deploy(
BUILT_CHARM,
application_name='advanced-routing-{}'.format(release),
series=release,
num_units=0,
)
await model.add_relation(
'ubuntu-{}'.format(release),
'advanced-routing-{}'.format(release),
)
yield advanced_routing
@pytest.fixture(scope='module')
async def unit(deploy_app):
"""Return the advanced-routing unit we've deployed."""
return deploy_app.units.pop()
#########
# TESTS #
#########
async def test_deploy(deploy_app, model):
"""Test the deployment."""
status, message = "blocked", "Advanced routing is disabled"
await model.block_until(
lambda: (deploy_app.status == status
and all(unit.workload_status_message == message
for unit in deploy_app.units)),
timeout=300,
)
assert True
@pytest.mark.parametrize(
"cfg",
[pytest.param(cfg, id="cfg-{}".format(i))
for i, cfg in enumerate(cfg_opts.JSON_CONFIGS)],
)
async def test_juju_routing(cfg, file_contents, file_exists, deploy_app, model):
"""Test juju routing file contents with config."""
json_config = cfg["input"]
await deploy_app.set_config({
'advanced-routing-config': json.dumps(json_config),
'enable-advanced-routing': 'true',
'action-managed-update': 'false',
})
status, agent_status, message = "active", "idle", "Unit is ready"
await model.block_until(
lambda: (deploy_app.status == status
and all(unit.agent_status == agent_status
and unit.workload_status_message == message
for unit in deploy_app.units)),
timeout=300,
)
common_path = "/usr/local/lib/juju-charm-advanced-routing"
up_path = "{}/if-up/95-juju_routing".format(common_path)
cleanup_path = "{}/cleanup/95-juju_routing".format(common_path)
unit = deploy_app.units.pop()
if_up_content = await file_contents(path=up_path, target=unit)
if_down_content = await file_contents(path=cleanup_path, target=unit)
if_up_expected_content = cfg["expected_ifup"]
if_down_expected_content = cfg["expected_ifdown"]
assert if_up_expected_content == if_up_content
assert if_down_expected_content == if_down_content
series = deploy_app.name.split("-")[-1]
if series >= "xenial" or series < "bionic":
ifup_path = "/etc/network/if-up.d"
else:
ifup_path = "/etc/networkd-dispatcher/routable.d"
ifup_expected_file_path = "{}/95-juju_routing".format(ifup_path)
ifup_exists = await file_exists(path=ifup_expected_file_path, target=unit)
assert ifup_exists == "1\n"
async def test_juju_routing_disable(file_exists, unit, deploy_app, model):
"""Test juju routing file non-existance when conf disabled."""
status, message = "blocked", "Advanced routing is disabled"
await deploy_app.set_config({'enable-advanced-routing': 'false'})
await model.block_until(
lambda: (deploy_app.status == status
and all(unit.workload_status_message == message
for unit in deploy_app.units)),
timeout=300,
)
series = deploy_app.name.split("-")[-1]
if series >= "xenial" or series < "bionic":
ifup_path = "/etc/network/if-up.d"
ifdown_path = "/etc/network/if-down.d"
else:
ifup_path = "/etc/networkd-dispatcher/routable.d"
ifdown_path = "/etc/networkd-dispatcher/off.d"
for if_path in [ifup_path, ifdown_path]:
filename = "{}/95-juju_routing".format(if_path)
if_exists = await file_exists(path=filename, target=unit)
assert if_exists == "0\n"
async def test_apply_changes_disabled(file_exists, deploy_app, unit):
"""Tests that apply-changes action complete."""
await deploy_app.set_config({
'enable-advanced-routing': 'false',
'action-managed-update': 'true',
})
action = await unit.run_action('apply-changes')
action = await action.wait()
assert action.status == 'failed'
async def test_apply_changes(file_exists, deploy_app, unit):
"""Tests that apply-changes action complete."""
ifup_path = "/etc/networkd-dispatcher/routable.d"
ifup_filename = "{}/95-juju_routing".format(ifup_path)
ifup_exists = await file_exists(path=ifup_filename, target=unit)
# ifup file doesn't exist before running the action
assert ifup_exists == "0\n"
await deploy_app.set_config({
'enable-advanced-routing': 'true',
'action-managed-update': 'true',
})
unit = deploy_app.units[0]
action = await unit.run_action('apply-changes')
action = await action.wait()
assert action.status == 'completed'
ifup_exists = await file_exists(path=ifup_filename, target=unit)
# ifup file exists after running the action
assert ifup_exists == "1\n"