17
17
18
18
import re
19
19
import subprocess
20
+ import unittest
21
+ from collections .abc import Sequence , Set
20
22
21
- from testlib import Error , MachineCase , wait
23
+ from machine import testvm
24
+ from machine .machine_core .machine_virtual import VirtMachine
25
+ from testlib import Browser , Error , MachineCase , wait
22
26
23
27
24
- class NetworkHelpers :
28
+ class NetworkHelpers ( unittest . TestCase ) :
25
29
"""Mix-in class for tests that require network setup"""
30
+ browser : Browser
31
+ machine : testvm .Machine
26
32
27
- def add_veth (self , name , dhcp_cidr = None , dhcp_range = None ):
33
+ def add_veth (self , name : str , dhcp_cidr : str | None = None , dhcp_range : Sequence [ str ] | None = None ) -> None :
28
34
"""Add a veth device that is manageable with NetworkManager
29
35
30
36
This is safe for @nondestructive tests, the interface gets cleaned up automatically.
@@ -53,7 +59,7 @@ def add_veth(self, name, dhcp_cidr=None, dhcp_range=None):
53
59
self .addCleanup (self .machine .execute , f"kill { server } ; rm -rf /run/dnsmasq" )
54
60
self .machine .execute ("if firewall-cmd --state >/dev/null 2>&1; then firewall-cmd --add-service=dhcp; fi" )
55
61
56
- def nm_activate_eth (self , iface ) :
62
+ def nm_activate_eth (self , iface : str ) -> None :
57
63
"""Create an NM connection for a given interface"""
58
64
59
65
m = self .machine
@@ -62,26 +68,26 @@ def nm_activate_eth(self, iface):
62
68
m .execute (f"nmcli con up { iface } ifname { iface } " )
63
69
self .addCleanup (m .execute , f"nmcli con delete { iface } " )
64
70
65
- def nm_checkpoints_disable (self ):
71
+ def nm_checkpoints_disable (self ) -> None :
66
72
self .browser .eval_js ("window.cockpit_tests_disable_checkpoints = true;" )
67
73
68
- def nm_checkpoints_enable (self , settle_time = 3.0 ):
74
+ def nm_checkpoints_enable (self , settle_time : float = 3.0 ) -> None :
69
75
self .browser .eval_js ("window.cockpit_tests_disable_checkpoints = false;" )
70
76
self .browser .eval_js (f"window.cockpit_tests_checkpoint_settle_time = { settle_time } ;" )
71
77
72
78
73
79
class NetworkCase (MachineCase , NetworkHelpers ):
74
- def setUp (self ):
80
+ def setUp (self ) -> None :
75
81
super ().setUp ()
76
82
77
83
m = self .machine
78
84
79
85
# clean up after nondestructive tests
80
86
if self .is_nondestructive ():
81
- def devs ():
87
+ def devs () -> Set [ str ] :
82
88
return set (self .machine .execute ("ls /sys/class/net/ | grep -v bonding_masters" ).strip ().split ())
83
89
84
- def cleanupDevs ():
90
+ def cleanupDevs () -> None :
85
91
new = devs () - self .orig_devs
86
92
self .machine .execute (f"for d in { ' ' .join (new )} ; do nmcli dev del $d; done" )
87
93
@@ -140,8 +146,10 @@ def getit():
140
146
print (f"{ mac } -> { iface } " )
141
147
return iface
142
148
143
- def add_iface (self , activate = True ):
149
+ def add_iface (self , activate : bool = True ) -> str :
144
150
m = self .machine
151
+ assert isinstance (m , VirtMachine )
152
+ assert self .network is not None
145
153
mac = m .add_netiface (networking = self .network .interface ())
146
154
# Wait for the interface to show up
147
155
self .get_iface (m , mac )
@@ -152,7 +160,7 @@ def add_iface(self, activate=True):
152
160
self .nm_activate_eth (iface )
153
161
return iface
154
162
155
- def wait_for_iface (self , iface , active = True , state = None , prefix = "10.111." ):
163
+ def wait_for_iface (self , iface : str , active : bool = True , state : str | None = None , prefix : str = "10.111." ) -> None :
156
164
sel = f"#networking-interfaces tr[data-interface='{ iface } ']"
157
165
158
166
if state :
@@ -170,31 +178,31 @@ def wait_for_iface(self, iface, active=True, state=None, prefix="10.111."):
170
178
print (self .machine .execute (f"grep . /sys/class/net/*/address; nmcli con; nmcli dev; nmcli dev show { iface } || true" ))
171
179
raise e
172
180
173
- def select_iface (self , iface ) :
181
+ def select_iface (self , iface : str ) -> None :
174
182
b = self .browser
175
183
b .click (f"#networking-interfaces tr[data-interface='{ iface } '] button" )
176
184
177
- def iface_con_id (self , iface ) :
185
+ def iface_con_id (self , iface : str ) -> str | None :
178
186
con_id = self .machine .execute (f"nmcli -m tabular -t -f GENERAL.CONNECTION device show { iface } " ).strip ()
179
187
if con_id == "" or con_id == "--" :
180
188
return None
181
189
else :
182
190
return con_id
183
191
184
- def wait_for_iface_setting (self , setting_title , setting_value ) :
192
+ def wait_for_iface_setting (self , setting_title : str , setting_value : str ) -> None :
185
193
b = self .browser
186
194
b .wait_in_text (f"dt:contains('{ setting_title } ') + dd" , setting_value )
187
195
188
- def configure_iface_setting (self , setting_title ) :
196
+ def configure_iface_setting (self , setting_title : str ) -> None :
189
197
b = self .browser
190
198
b .click (f"dt:contains('{ setting_title } ') + dd button" )
191
199
192
- def ensure_nm_uses_dhclient (self ):
200
+ def ensure_nm_uses_dhclient (self ) -> None :
193
201
m = self .machine
194
202
m .write ("/etc/NetworkManager/conf.d/99-dhcp.conf" , "[main]\n dhcp=dhclient\n " )
195
203
m .execute ("systemctl restart NetworkManager" )
196
204
197
- def slow_down_dhclient (self , delay ) :
205
+ def slow_down_dhclient (self , delay : int ) -> None :
198
206
self .machine .execute (f"""
199
207
mkdir -p { self .vm_tmpdir }
200
208
cp -a /usr/sbin/dhclient { self .vm_tmpdir } /dhclient.real
@@ -205,12 +213,25 @@ def slow_down_dhclient(self, delay):
205
213
""" )
206
214
self .addCleanup (self .machine .execute , "umount /usr/sbin/dhclient" )
207
215
208
- def wait_onoff (self , sel , val ) :
216
+ def wait_onoff (self , sel : str , * , val : bool ) -> None :
209
217
self .browser .wait_visible (sel + " input[type=checkbox]" + (":checked" if val else ":not(:checked)" ))
210
218
211
- def toggle_onoff (self , sel ) :
219
+ def toggle_onoff (self , sel : str ) -> None :
212
220
self .browser .click (sel + " input[type=checkbox]" )
213
221
214
- def login_and_go (self , * args , ** kwargs ):
215
- super ().login_and_go (* args , ** kwargs )
222
+ def login_and_go (
223
+ self ,
224
+ path : str | None = None ,
225
+ * ,
226
+ user : str | None = None ,
227
+ password : str | None = None ,
228
+ host : str | None = None ,
229
+ superuser : bool = True ,
230
+ urlroot : str | None = None ,
231
+ tls : bool = False ,
232
+ enable_root_login : bool = False
233
+ ) -> None :
234
+ super ().login_and_go (path = path , user = user , password = password ,
235
+ host = host , superuser = superuser , urlroot = urlroot ,
236
+ tls = tls , enable_root_login = enable_root_login )
216
237
self .nm_checkpoints_disable ()
0 commit comments