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
179
|
#!/usr/bin/python3 -cimport os, sys; os.execv(os.path.dirname(sys.argv[1]) + "/../common/pywrap", sys.argv)
# This file is part of Cockpit.
#
# Copyright (C) 2013 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <https://www.gnu.org/licenses/>.
import netlib
import testlib
@testlib.skipImage("No network checkpoint support", "ubuntu-*")
@testlib.nondestructive
class TestNetworkingCheckpoints(netlib.NetworkCase):
def testCheckpoint(self):
b = self.browser
m = self.machine
iface = self.get_iface("52:54:00:12:34:56")
if "debian" in m.image:
self.sed_file("s/managed=false/managed=true/", "/etc/NetworkManager/NetworkManager.conf",
"systemctl restart NetworkManager")
if m.image == "arch":
self.sed_file("s/unmanaged-devices=interface-name:eth0//", "/etc/NetworkManager/conf.d/noauto.conf",
"systemctl restart NetworkManager")
self.login_and_go("/network")
self.nm_checkpoints_enable()
self.wait_for_iface(iface, prefix="172.")
self.select_iface(iface)
b.wait_visible("#network-interface")
# Disconnect
self.wait_onoff("label[for='interface-switch']", val=True)
self.toggle_onoff("label[for='interface-switch']")
# Wait for dialog to appear and dismiss it
with b.wait_timeout(60):
b.click("#confirm-breaking-change-popup button:contains('Keep connection')")
b.wait_not_present("#confirm-breaking-change-popup")
# Change IP
self.configure_iface_setting('IPv4')
b.wait_visible("#network-ip-settings-dialog")
b.select_from_dropdown("#network-ip-settings-select-method", "manual")
b.set_input_text('#network-ip-settings-address-0', "1.2.3.4")
b.set_input_text('#network-ip-settings-netmask-0', "24")
b.click("#network-ip-settings-save")
with b.wait_timeout(60):
b.click("#confirm-breaking-change-popup button:contains('Keep connection')")
b.wait_not_present("#confirm-breaking-change-popup")
@testlib.skipImage("Main interface settings are read-only", "debian-*")
@testlib.skipImage("no dhclient on OS image", "arch", "centos-10*", "rhel-10*")
@testlib.skipOstree("not using dhclient")
def testCheckpointSlowRollback(self):
b = self.browser
m = self.machine
# A slow rollback would normally cause the global health check
# to fail during rollback and prevent showing the
# #confirm-breaking-change-popup. We expect the global health
# check failure to be ignored.
#
# We test slow rollbacks by slowing down DHCP requests.
#
# We need at least 60 seconds of network disconnection to let
# the health check fail reliably. A rollback is started 7
# seconds after disconnection, so we need to delay the DHCP
# request by at least 53 seconds. However, NetworkManager has
# a default DHCP timeout of 45 seconds, so we need to increase
# that.
#
# Sometimes, NetworkManager extends the DHCP timeout by an
# additional 480 seconds grace period. This doesn't always
# happen, so we don't rely on it.
dhcp_delay = 60
dhcp_timeout = 120
# There are a couple of considerations for ordering the
# following actions:
#
# - Changing the main.dhcp config value requires a restart of NM.
#
# - A restart of NM might run dhclient, and we don't want it
# to be slowed down already at that point.
#
# - After restarting NM, we need to wait for it to settle
# again before messing with dhclient.
#
# - Simply setting ipv4.dhcp_timeout is not enough if it
# should be used immediately for a checkpoint rollback, see
# https://bugzilla.redhat.com/show_bug.cgi?id=1690389. A
# additional restart is enough.
#
# So we set ipv4.dhcp_timeout and main.dhcp, do a restart, log
# into the UI and wait for the expected interface to appear
# and be active, and then slow down dhclient.
iface = self.get_iface("52:54:00:12:34:56")
con_id = self.iface_con_id(iface)
m.execute(f'nmcli con mod "{con_id}" ipv4.dhcp-timeout {dhcp_timeout}')
self.ensure_nm_uses_dhclient()
self.login_and_go("/network")
self.nm_checkpoints_enable()
self.wait_for_iface(iface, prefix="172.")
self.select_iface(iface)
b.wait_visible("#network-interface")
# checkpoints are realtime sensitive, avoid long NM operations
self.settle_cpu()
self.slow_down_dhclient(dhcp_delay)
# Disconnect and trigger a slow rollback
self.wait_onoff("label[for='interface-switch']", val=True)
self.toggle_onoff("label[for='interface-switch']")
with b.wait_timeout(120):
b.click("#confirm-breaking-change-popup button:contains('Keep connection')")
b.wait_not_present("#confirm-breaking-change-popup")
def testNoRollback(self):
b = self.browser
m = self.machine
self.login_and_go("/network")
self.nm_checkpoints_enable()
b.wait_visible("#networking")
iface = 'cockpit1'
self.add_veth(iface, dhcp_cidr="10.111.113.2/20")
self.nm_activate_eth(iface)
self.wait_for_iface(iface)
self.select_iface(iface)
b.wait_visible("#network-interface")
# Disconnect
self.wait_onoff("label[for='interface-switch']", val=True)
self.toggle_onoff("label[for='interface-switch']")
# The checkpoint should be destroyed before it is being rolled
# back.
def checkpoint_was_destroyed():
lines = m.execute("journalctl -u NetworkManager | grep op=").split("\n")
last_checkpoint = None
for line in lines:
match = netlib.re.search('op="checkpoint-create" arg="([^"]*)"', line)
if match:
last_checkpoint = match[1]
if last_checkpoint:
for line in lines:
if f'op="checkpoint-destroy" arg="{last_checkpoint}"' in line:
return True
return False
testlib.wait(checkpoint_was_destroyed)
if __name__ == '__main__':
testlib.test_main()
|
