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
|
import time
from collections import namedtuple
import pytest
import yaml
from tests.integration_tests.instances import IntegrationInstance
from tests.integration_tests.integration_settings import PLATFORM
from tests.integration_tests.releases import CURRENT_RELEASE, FOCAL
USER_DATA = """\
#cloud-config
updates:
network:
when: ['hotplug']
"""
ip_addr = namedtuple("ip_addr", "interface state ip4 ip6")
def _wait_till_hotplug_complete(client, expected_runs=1):
for _ in range(60):
log = client.read_from_file("/var/log/cloud-init.log")
if log.count("Exiting hotplug handler") == expected_runs:
return log
time.sleep(1)
raise Exception("Waiting for hotplug handler failed")
def _get_ip_addr(client):
ips = []
lines = client.execute("ip --brief addr").split("\n")
for line in lines:
attributes = line.split()
interface, state = attributes[0], attributes[1]
ip4_cidr = attributes[2] if len(attributes) > 2 else None
ip6_cidr = attributes[3] if len(attributes) > 3 else None
ip4 = ip4_cidr.split("/")[0] if ip4_cidr else None
ip6 = ip6_cidr.split("/")[0] if ip6_cidr else None
ip = ip_addr(interface, state, ip4, ip6)
ips.append(ip)
return ips
@pytest.mark.skipif(
PLATFORM != "openstack",
reason=(
f"Test was written for {PLATFORM} but can likely run on "
"other platforms."
),
)
@pytest.mark.skipif(
CURRENT_RELEASE < FOCAL,
reason="Openstack network metadata support was added in focal.",
)
@pytest.mark.user_data(USER_DATA)
def test_hotplug_add_remove(client: IntegrationInstance):
ips_before = _get_ip_addr(client)
log = client.read_from_file("/var/log/cloud-init.log")
assert "Exiting hotplug handler" not in log
assert client.execute(
"test -f /etc/udev/rules.d/10-cloud-init-hook-hotplug.rules"
).ok
# Add new NIC
added_ip = client.instance.add_network_interface()
_wait_till_hotplug_complete(client, expected_runs=1)
ips_after_add = _get_ip_addr(client)
new_addition = [ip for ip in ips_after_add if ip.ip4 == added_ip][0]
assert len(ips_after_add) == len(ips_before) + 1
assert added_ip not in [ip.ip4 for ip in ips_before]
assert added_ip in [ip.ip4 for ip in ips_after_add]
assert new_addition.state == "UP"
netplan_cfg = client.read_from_file("/etc/netplan/50-cloud-init.yaml")
config = yaml.safe_load(netplan_cfg)
assert new_addition.interface in config["network"]["ethernets"]
# Remove new NIC
client.instance.remove_network_interface(added_ip)
_wait_till_hotplug_complete(client, expected_runs=2)
ips_after_remove = _get_ip_addr(client)
assert len(ips_after_remove) == len(ips_before)
assert added_ip not in [ip.ip4 for ip in ips_after_remove]
netplan_cfg = client.read_from_file("/etc/netplan/50-cloud-init.yaml")
config = yaml.safe_load(netplan_cfg)
assert new_addition.interface not in config["network"]["ethernets"]
assert "enabled" == client.execute(
"cloud-init devel hotplug-hook -s net query"
)
@pytest.mark.skipif(
PLATFORM != "openstack",
reason=(
f"Test was written for {PLATFORM} but can likely run on "
"other platforms."
),
)
def test_no_hotplug_in_userdata(client: IntegrationInstance):
ips_before = _get_ip_addr(client)
log = client.read_from_file("/var/log/cloud-init.log")
assert "Exiting hotplug handler" not in log
assert client.execute(
"test -f /etc/udev/rules.d/10-cloud-init-hook-hotplug.rules"
).failed
# Add new NIC
client.instance.add_network_interface()
log = client.read_from_file("/var/log/cloud-init.log")
assert "hotplug-hook" not in log
ips_after_add = _get_ip_addr(client)
if len(ips_after_add) == len(ips_before) + 1:
# We can see the device, but it should not have been brought up
new_ip = [ip for ip in ips_after_add if ip not in ips_before][0]
assert new_ip.state == "DOWN"
else:
assert len(ips_after_add) == len(ips_before)
assert "disabled" == client.execute(
"cloud-init devel hotplug-hook -s net query"
)
|
