Make HSTS test infrastructure more robust

Enhance wait_for to return the callback value and use it to retry
file server header retrieval until the server is ready. Change
host_with_hsts fixture to function scope for test isolation.

Signed-off-by: Gaëtan Lehmann <gaetan.lehmann@vates.tech>

Author: glehmann

conftest.py

@@ -426,7 +426,7 @@ def host_less_than_8_3(host: Host) -> None:
     if not host.xcp_version < version.parse(version_str):
         pytest.skip(f"This test requires an XCP-ng < {version_str} host")
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope='function')
 def host_with_hsts(host: Host) -> Generator[Host, None, None]:
     host.enable_hsts_header()
     yield host

lib/common.py

@@ -192,17 +192,17 @@ def callable_marker(value: T | Callable[..., T], request: pytest.FixtureRequest)
     else:
         return value
 
-def wait_for(fn: Callable[[], object], msg: str | None = None, timeout_secs: int = 2 * 60, retry_delay_secs: int = 2,
-             invert: bool = False) -> None:
+def wait_for(fn: Callable[[], T], msg: str | None = None, timeout_secs: int = 2 * 60, retry_delay_secs: int = 2,
+             invert: bool = False) -> T:
     if msg is not None:
         logging.info(msg)
     start_time = time.perf_counter()
     while True:
         ret = fn()
         if not invert and ret:
-            return
+            return ret
         if invert and not ret:
-            return
+            return ret
         if time.perf_counter() - start_time >= timeout_secs:
             expected = 'True' if not invert else 'False'
             raise TimeoutError(

tests/misc/test_file_server.py

@@ -3,6 +3,7 @@
 import re
 
 import lib.commands as commands
+from lib.common import wait_for
 from lib.host import Host
 from lib.netutil import wrap_ip
 
@@ -31,10 +32,15 @@ class TestHSTS:
 
     @staticmethod
     def __get_header(host: Host) -> list[str]:
-        res = commands.local_cmd(
-            ["curl", "-s", "-XGET", "-k", "-I", "https://" + wrap_ip(host.hostname_or_ip)]
-        )
-        return res.stdout.splitlines()
+        def get_or_none():
+            res = commands.local_cmd(["curl", "-s", "-XGET", "-k", "-I", "https://"
+                                      + wrap_ip(host.hostname_or_ip)], check=False)
+            if res.returncode != 0:
+                return None
+            return res.stdout.splitlines()
+        headers = wait_for(get_or_none)
+        assert headers is not None
+        return headers
 
     def test_fileserver_hsts_default(self, host: Host) -> None:
         # By default HSTS header should not be set