Improve fileshare qa tests

test/qa/lib/fileshare.py

@@ -374,7 +374,7 @@ def __str__(self):
         return self.value
 
 
-def bind_port():
+def bind_port() -> socket.socket | None:
     try:
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 0)
@@ -388,5 +388,37 @@ def bind_port():
 
 
 def port_is_allowed() -> bool:
+    for _ in range(3):
+        if is_port_allowed():
+            return True
+        time.sleep(1)
+    return False
+
+
+def is_port_allowed() -> bool:
     rules = os.popen("sudo iptables -S").read()
     return "49111 -m comment --comment nordvpn-meshnet -j ACCEPT" in rules
+
+
+def port_is_blocked() -> bool:
+    for _ in range(3):
+        if not is_port_allowed():
+            return True
+        time.sleep(1)
+    return False
+
+
+def ensure_mesh_is_on() -> None:
+    try:
+        sh.nordvpn.set.meshnet.on()
+    except sh.ErrorReturnCode_1 as e:
+        if "Meshnet is already enabled." not in str(e):
+            raise e
+
+
+def restart_mesh() -> None:
+    sh.nordvpn.set.meshnet.off()
+    time.sleep(2)
+    sh.nordvpn.set.meshnet.on()
+    time.sleep(5)
+

test/qa/test_fileshare.py

@@ -1348,50 +1348,54 @@ def test_clear():
 
 
 def test_fileshare_process_monitoring_manages_fileshare_rules_on_process_state_changes():
-    # port is open when fileshare is running
-    assert fileshare.port_is_allowed()
+    try:
+        # port is open when fileshare is running
+        assert fileshare.port_is_allowed()
 
-    sh.pkill("-SIGKILL", "nordfileshare")
-    # at the time of writing, the monitoring job is executed periodically every second,
-    # wait for 2 seconds to be sure the job executed
-    time.sleep(2)
+        sh.pkill("-SIGKILL", "nordfileshare")
+        # at the time of writing, the monitoring job is executed periodically every second,
+        # wait for 2 seconds to be sure the job executed
+        time.sleep(2)
 
-    # port is not allowed when fileshare is down
-    assert not fileshare.port_is_allowed()
+        # port is not allowed when fileshare is down
+        assert fileshare.port_is_blocked()
 
-    os.popen("/usr/lib/nordvpn/nordfileshare &")
-    time.sleep(2)
-    # port is allowed again when fileshare process is up
-    assert fileshare.port_is_allowed()
+        # restart meshet to get fileshare back up
+        fileshare.restart_mesh()
 
+        # port is allowed again when fileshare process is up
+        assert fileshare.port_is_allowed()
+    finally: # meshnet should be on for most of the tests in this module
+        fileshare.ensure_mesh_is_on()
 
-def test_fileshare_process_monitoring_cuts_the_port_access_even_when_it_was_taken_before():
-    # stop meshnet to bind to 49111 first
-    sh.nordvpn.set.meshnet.off()
 
-    # no meshnet - no port
-    assert not fileshare.port_is_allowed()
+def test_fileshare_process_monitoring_cuts_the_port_access_even_when_it_was_taken_before():
+    try:
+        # stop meshnet to bind to 49111 first
+        sh.nordvpn.set.meshnet.off()
+        assert fileshare.port_is_blocked()
 
-    # bind to port before fileshare process starts
-    sock = fileshare.bind_port()
-    assert sock is not None
+        # bind to port before fileshare process starts
+        sock = fileshare.bind_port()
+        assert sock is not None
 
-    # start meshnet
-    sh.nordvpn.set.meshnet.on() # now fileshare tries to start but fails because the port is taken
-    time.sleep(2)
+        # start meshnet
+        sh.nordvpn.set.meshnet.on() # now fileshare tries to start but fails because the port is taken
+        time.sleep(2)
 
-    # port should not be allowed (fileshare is down)
-    assert not fileshare.port_is_allowed()
+        # port should not be allowed (fileshare is down)
+        assert fileshare.port_is_blocked()
 
-    # free the port
-    sock.close()
+        # free the port
+        sock.close()
 
-    # now fileshare can start properly
-    os.popen("/usr/lib/nordvpn/nordfileshare &")
-    time.sleep(2)
+        # restart meshnet, now fileshare can start properly
+        fileshare.restart_mesh()
 
-    # port is allowed again when fileshare process is up
-    assert fileshare.port_is_allowed()
+        # fileshare is up so port is allowed
+        assert fileshare.port_is_allowed()
+    finally: # meshnet should be on for most of the tests in this module
+        fileshare.ensure_mesh_is_on()
 
 
 @pytest.mark.parametrize("background_accept", [True, False], ids=["accept_bg", "accept_int"])