tests: Unify IPv6 disabling

Add a decorator that disables IPv6 and requires running
under a VM, rather than open-coding that many times.

Signed-off-by: Johannes Berg <johannes.berg@intel.com>

tests/hwsim/test_ap_ciphers.py

@@ -663,24 +663,9 @@ def run_ap_cipher_replay_protection_sta(dev, apdev, cipher, keytype=KT_PTK):
         if replays < 1:
             raise Exception("Replays not reported")
 
+@disable_ipv6
 def test_ap_wpa2_delayed_m3_retransmission(dev, apdev):
     """Delayed M3 retransmission"""
-    require_under_vm()
-    try:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        run_ap_wpa2_delayed_m3_retransmission(dev, apdev)
-    finally:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-
-def run_ap_wpa2_delayed_m3_retransmission(dev, apdev):
     params = hostapd.wpa2_params(ssid="test-wpa2-psk", passphrase="12345678")
     hapd = hostapd.add_ap(apdev[0], params)
 
@@ -726,39 +711,15 @@ def run_ap_wpa2_delayed_m3_retransmission(dev, apdev):
         if a < b:
             raise Exception("GTK RX counter decreased: idx=%d before=%d after=%d" % (i, b, a))
 
+@disable_ipv6
 def test_ap_wpa2_delayed_m1_m3_retransmission(dev, apdev):
     """Delayed M1+M3 retransmission"""
-    require_under_vm()
+    run_ap_wpa2_delayed_m1_m3_retransmission(dev, apdev, False)
-    try:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        run_ap_wpa2_delayed_m1_m3_retransmission(dev, apdev)
-    finally:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
 
+@disable_ipv6
 def test_ap_wpa2_delayed_m1_m3_retransmission2(dev, apdev):
     """Delayed M1+M3 retransmission (change M1 ANonce)"""
-    require_under_vm()
-    try:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
     run_ap_wpa2_delayed_m1_m3_retransmission(dev, apdev, True)
-    finally:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
 
 def run_ap_wpa2_delayed_m1_m3_retransmission(dev, apdev,
                                              change_m1_anonce=False):
@@ -812,24 +773,9 @@ def run_ap_wpa2_delayed_m1_m3_retransmission(dev, apdev,
         if a < b:
             raise Exception("GTK RX counter decreased: idx=%d before=%d after=%d" % (i, b, a))
 
+@disable_ipv6
 def test_ap_wpa2_delayed_group_m1_retransmission(dev, apdev):
     """Delayed group M1 retransmission"""
-    require_under_vm()
-    try:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        run_ap_wpa2_delayed_group_m1_retransmission(dev, apdev)
-    finally:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-
-def run_ap_wpa2_delayed_group_m1_retransmission(dev, apdev):
     params = hostapd.wpa2_params(ssid="test-wpa2-psk", passphrase="12345678")
     hapd = hostapd.add_ap(apdev[0], params)
 
@@ -867,24 +813,9 @@ def run_ap_wpa2_delayed_group_m1_retransmission(dev, apdev):
         if a < b:
             raise Exception("RX counter decreased: idx=%d before=%d after=%d" % (i, b, a))
 
+@disable_ipv6
 def test_ap_wpa2_delayed_group_m1_retransmission_igtk(dev, apdev):
     """Delayed group M1 retransmission (check IGTK protection)"""
-    require_under_vm()
-    try:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=1'],
-                        stdout=open('/dev/null', 'w'))
-        run_ap_wpa2_delayed_group_m1_retransmission_igtk(dev, apdev)
-    finally:
-        subprocess.call(['sysctl', '-w', 'net.ipv6.conf.all.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-        subprocess.call(['sysctl', '-w',
-                         'net.ipv6.conf.default.disable_ipv6=0'],
-                        stdout=open('/dev/null', 'w'))
-
-def run_ap_wpa2_delayed_group_m1_retransmission_igtk(dev, apdev):
     params = hostapd.wpa2_params(ssid="test-wpa2-psk", passphrase="12345678",
                                  ieee80211w="2")
     hapd = hostapd.add_ap(apdev[0], params)

tests/hwsim/utils.py

@@ -254,3 +254,27 @@ def set_world_reg(apdev0=None, apdev1=None, dev0=None):
     if dev0:
         dev0.cmd_execute(['iw', 'reg', 'set', '00'])
     time.sleep(0.1)
+
+def sysctl_write(val):
+    subprocess.call(['sysctl', '-w', val], stdout=open('/dev/null', 'w'))
+
+def disable_ipv6(fn):
+    def wrapper(dev, apdev, params):
+        require_under_vm()
+        try:
+            sysctl_write('net.ipv6.conf.all.disable_ipv6=1')
+            sysctl_write('net.ipv6.conf.default.disable_ipv6=1')
+            if fn.__code__.co_argcount > 2:
+                return fn(dev, apdev, params)
+            elif fn.__code__.co_argcount > 1:
+                return fn(dev, apdev)
+            return fn(dev)
+        finally:
+            sysctl_write('net.ipv6.conf.all.disable_ipv6=0')
+            sysctl_write('net.ipv6.conf.default.disable_ipv6=0')
+    # we need the name set right for selecting / printing etc.
+    wrapper.__name__ = fn.__name__
+    wrapper.__doc__ = fn.__doc__
+    # reparent to the right module for module filtering
+    wrapper.__module__ = fn.__module__
+    return wrapper