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
|
import json
import pytest
from cloudinit import safeyaml
from cloudinit.subp import subp
from tests.integration_tests.clouds import ImageSpecification
from tests.integration_tests.decorators import retry
from tests.integration_tests.instances import IntegrationInstance
from tests.integration_tests.util import lxd_has_nocloud
USER_DATA = """\
#cloud-config
updates:
network:
when: ["hotplug"]
"""
UPDATED_NETWORK_CONFIG = """\
version: 2
ethernets:
eth0:
dhcp4: true
eth2:
dhcp4: true
"""
@retry()
def ensure_hotplug_exited(client):
assert "cloud-init" not in client.execute("ps -A")
def get_parent_network(instance_name: str):
lxd_network = json.loads(
subp("lxc network list --format json".split()).stdout
)
for net in lxd_network:
if net["type"] == "bridge" and net["managed"]:
if f"/1.0/instances/{instance_name}" in net.get("used_by", []):
return net["name"]
return "lxdbr0"
def _prefer_lxd_datasource_over_nocloud(client: IntegrationInstance):
"""For hotplug support we need LXD datasource detected instead of NoCloud
Bionic and Focal still deliver nocloud-net seed files so override it
with /etc/cloud/cloud.cfg.d/99-detect-lxd-first.cfg
"""
client.write_to_file(
"/etc/cloud/cloud.cfg.d/99-detect-lxd-first.cfg",
"datasource_list: [LXD, NoCloud]\n",
)
client.execute("cloud-init clean --logs")
client.restart()
# TODO: Once LXD adds MACs to the devices endpoint, support LXD VMs here
# Currently the names are too unpredictable to be worth testing on VMs.
@pytest.mark.lxd_container
@pytest.mark.user_data(USER_DATA)
class TestLxdHotplug:
@pytest.fixture(autouse=True, scope="class")
def class_teardown(self, class_client: IntegrationInstance):
# We need a teardown here because on IntegrationInstance teardown,
# if KEEP_INSTANCE=True, we grab the instance IP for logging, but
# we're currently running into
# https://github.com/canonical/pycloudlib/issues/220 .
# Once that issue is fixed, we can remove this teardown
yield
name = class_client.instance.name
subp(f"lxc config device remove {name} eth1".split())
subp(f"lxc config device remove {name} eth2".split())
subp("lxc network delete ci-test-br-eth1".split())
subp("lxc network delete ci-test-br-eth2".split())
def test_no_network_change_default(
self, class_client: IntegrationInstance
):
client = class_client
if lxd_has_nocloud(client):
_prefer_lxd_datasource_over_nocloud(client)
assert "eth1" not in client.execute("ip address")
pre_netplan = client.read_from_file("/etc/netplan/50-cloud-init.yaml")
networks = subp("lxc network list".split())
if "ci-test-br-eth1" not in networks.stdout:
subp(
"lxc network create ci-test-br-eth1 --type=bridge "
"ipv4.address=10.10.41.1/24 ipv4.nat=true".split()
)
subp(
f"lxc config device add {client.instance.name} eth1 nic name=eth1 "
f"nictype=bridged parent=ci-test-br-eth1".split()
)
ensure_hotplug_exited(client)
post_netplan = client.read_from_file("/etc/netplan/50-cloud-init.yaml")
assert pre_netplan == post_netplan
ip_info = json.loads(client.execute("ip --json address"))
eth1s = [i for i in ip_info if i["ifname"] == "eth1"]
assert len(eth1s) == 1
assert eth1s[0]["operstate"] == "DOWN"
def test_network_config_applied(self, class_client: IntegrationInstance):
client = class_client
if lxd_has_nocloud(client):
_prefer_lxd_datasource_over_nocloud(client)
assert "eth2" not in client.execute("ip address")
pre_netplan = client.read_from_file("/etc/netplan/50-cloud-init.yaml")
assert "eth2" not in pre_netplan
if ImageSpecification.from_os_image().release in [
"bionic",
"focal",
]: # pyright: ignore
top_key = "user"
else:
top_key = "cloud-init"
assert subp(
[
"lxc",
"config",
"set",
client.instance.name,
f"{top_key}.network-config={UPDATED_NETWORK_CONFIG}",
]
)
assert (
client.read_from_file("/etc/netplan/50-cloud-init.yaml")
== pre_netplan
)
networks = subp("lxc network list".split())
if "ci-test-br-eth2" not in networks.stdout:
assert subp(
"lxc network create ci-test-br-eth2 --type=bridge"
" ipv4.address=10.10.42.1/24 ipv4.nat=true".split()
)
assert subp(
f"lxc config device add {client.instance.name} eth2 nic name=eth2 "
f"nictype=bridged parent=ci-test-br-eth2".split()
)
ensure_hotplug_exited(client)
post_netplan = safeyaml.load(
client.read_from_file("/etc/netplan/50-cloud-init.yaml")
)
expected_netplan = safeyaml.load(UPDATED_NETWORK_CONFIG)
expected_netplan = {"network": expected_netplan}
assert post_netplan == expected_netplan, client.read_from_file(
"/var/log/cloud-init.log"
)
ip_info = json.loads(client.execute("ip --json address"))
eth2s = [i for i in ip_info if i["ifname"] == "eth2"]
assert len(eth2s) == 1
assert eth2s[0]["operstate"] == "UP"
|
