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