test: improve test initialization error path (#5920)

- eliminated vestigial setup_image() fixture use
- make setup_image() a function not a fixture
- move one-time setup into pytest_sessionstart()
- use a global variable to store the session for cleanup
- any failure in pytest_sessionstart() triggers pytest.exit()
- make failure warnings more descriptive

tests/integration_tests/bugs/test_gh671.py

@@ -21,7 +21,7 @@ def _check_password(instance, unhashed_password):
 
 
 @pytest.mark.skipif(PLATFORM != "azure", reason="Test is Azure specific")
-def test_update_default_password(setup_image, session_cloud: IntegrationCloud):
+def test_update_default_password(session_cloud: IntegrationCloud):
     os_profile = {
         "os_profile": {
             "admin_password": "",

tests/integration_tests/bugs/test_lp1835584.py

@@ -92,8 +92,5 @@ def test_azure_kernel_upgrade_case_insensitive_uuid(
             )
         }
     ) as instance:
-        # We can't use setup_image fixture here because we want to avoid
-        # taking a snapshot or cleaning the booted machine after cloud-init
-        # upgrade.
         instance.install_new_cloud_init(source, clean=False)
         _check_iid_insensitive_across_kernel_upgrade(instance)

tests/integration_tests/bugs/test_lp1901011.py

@@ -20,7 +20,7 @@
     ],
 )
 def test_ephemeral(
-    instance_type, is_ephemeral, session_cloud: IntegrationCloud, setup_image
+    instance_type, is_ephemeral, session_cloud: IntegrationCloud
 ):
     if is_ephemeral:
         expected_log = (

tests/integration_tests/bugs/test_lp1910835.py

@@ -29,7 +29,7 @@
 
 
 @pytest.mark.skipif(PLATFORM != "azure", reason="Test is Azure specific")
-def test_crlf_in_azure_metadata_ssh_keys(session_cloud, setup_image):
+def test_crlf_in_azure_metadata_ssh_keys(session_cloud):
     authorized_keys_path = "/home/{}/.ssh/authorized_keys".format(
         session_cloud.cloud_instance.username
     )

tests/integration_tests/clouds.py

@@ -7,7 +7,7 @@
 import string
 from abc import ABC, abstractmethod
 from copy import deepcopy
-from typing import Type
+from typing import Optional, Type
 from uuid import UUID
 
 from pycloudlib import (
@@ -65,7 +65,7 @@ def __init__(
         self.settings = settings
         self.cloud_instance = self._get_cloud_instance()
         self.initial_image_id = self._get_initial_image()
-        self.snapshot_id = None
+        self.snapshot_id: Optional[str] = None
 
     @property
     def image_id(self):
@@ -182,7 +182,7 @@ def snapshot(self, instance):
 
     def delete_snapshot(self):
         if self.snapshot_id:
-            if self.settings.KEEP_IMAGE:  # type: ignore
+            if self.settings.KEEP_IMAGE:
                 log.info(
                     "NOT deleting snapshot image created for this testrun "
                     "because KEEP_IMAGE is True: %s",

tests/integration_tests/cmd/test_status.py

@@ -34,7 +34,7 @@ def retry_read_from_file(client: IntegrationInstance, path: str):
     PLATFORM != "lxd_container",
     reason="Test is LXD specific",
 )
-def test_wait_when_no_datasource(session_cloud: IntegrationCloud, setup_image):
+def test_wait_when_no_datasource(session_cloud: IntegrationCloud):
     """Ensure that when no datasource is found, we get status: disabled
 
     LP: #1966085

tests/integration_tests/conftest.py

@@ -74,8 +74,21 @@ def disable_subp_usage(request):
     pass
 
 
+_SESSION_CLOUD: IntegrationCloud
+
+
 @pytest.fixture(scope="session")
 def session_cloud() -> Generator[IntegrationCloud, None, None]:
+    """a shared session is created in pytest_sessionstart()
+
+    yield this shared session
+    """
+    global _SESSION_CLOUD
+    yield _SESSION_CLOUD
+
+
+def get_session_cloud() -> IntegrationCloud:
+    """get_session_cloud() creates a session from configuration"""
     if integration_settings.PLATFORM not in platforms.keys():
         raise ValueError(
             f"{integration_settings.PLATFORM} is an invalid PLATFORM "
@@ -91,8 +104,7 @@ def session_cloud() -> Generator[IntegrationCloud, None, None]:
         )
     cloud = platforms[integration_settings.PLATFORM](image_type=image_type)
     cloud.emit_settings_to_log()
-    yield cloud
-    cloud.destroy()
+    return cloud
 
 
 def get_validated_source(
@@ -118,12 +130,8 @@ def get_validated_source(
     raise ValueError(f"Invalid value for CLOUD_INIT_SOURCE setting: {source}")
 
 
-@pytest.fixture(scope="session")
-def setup_image(session_cloud: IntegrationCloud, request):
-    """Setup the target environment with the correct version of cloud-init.
-
-    So we can launch instances / run tests with the correct image
-    """
+def setup_image(session_cloud: IntegrationCloud) -> None:
+    """create image with correct version of cloud-init, then make a snapshot"""
     source = get_validated_source(session_cloud)
     if not (
         source.installs_new_version()
@@ -141,7 +149,8 @@ def setup_image(session_cloud: IntegrationCloud, request):
         and integration_settings.INCLUDE_COVERAGE
     ):
         log.error(
-            "Invalid configuration, cannot enable both profile and coverage."
+            "Invalid configuration, cannot enable both profile and "
+            "coverage."
         )
         raise ValueError()
     if integration_settings.INCLUDE_COVERAGE:
@@ -158,11 +167,6 @@ def setup_image(session_cloud: IntegrationCloud, request):
     client.destroy()
     log.info("Done with environment setup")
 
-    # For some reason a yield here raises a
-    # ValueError: setup_image did not yield a value
-    # during setup so use a finalizer instead.
-    request.addfinalizer(session_cloud.delete_snapshot)
-
 
 def _collect_logs(instance: IntegrationInstance, log_dir: Path):
     instance.execute(
@@ -311,32 +315,37 @@ def _client(
         local_launch_kwargs["lxd_setup"] = lxd_setup
 
     with session_cloud.launch(
-        user_data=user_data, launch_kwargs=launch_kwargs, **local_launch_kwargs
+        user_data=user_data,
+        launch_kwargs=launch_kwargs,
+        **local_launch_kwargs,
     ) as instance:
         if lxd_use_exec is not None and isinstance(
             instance.instance, LXDInstance
         ):
             # Existing instances are not affected by the launch kwargs, so
-            # ensure it here; we still need the launch kwarg so waiting works
+            # ensure it here; we still need the launch kwarg so waiting
+            # works
             instance.instance.execute_via_ssh = False
         previous_failures = request.session.testsfailed
         yield instance
         test_failed = request.session.testsfailed - previous_failures > 0
         _collect_artifacts(instance, request.node.nodeid, test_failed)
     # conflicting requirements:
     # - pytest thinks that it can cleanup loggers after tests run
-    # - pycloudlib thinks that at garbage collection is a good place to tear
-    #   down sftp connections
-    # After the final test runs, pytest might clean up loggers which will cause
-    # paramiko to barf when it logs that the connection is being closed.
+    # - pycloudlib thinks that at garbage collection is a good place to
+    # tear down sftp connections
+    #
+    # After the final test runs, pytest might clean up loggers which will
+    # cause paramiko to barf when it logs that the connection is being
+    # closed.
     #
     # Manually run __del__() to prevent this teardown mess.
     instance.instance.__del__()
 
 
 @pytest.fixture
 def client(  # pylint: disable=W0135
-    request, fixture_utils, session_cloud, setup_image
+    request, fixture_utils, session_cloud
 ) -> Iterator[IntegrationInstance]:
     """Provide a client that runs for every test."""
     with _client(request, fixture_utils, session_cloud) as client:
@@ -345,7 +354,7 @@ def client(  # pylint: disable=W0135
 
 @pytest.fixture(scope="module")
 def module_client(  # pylint: disable=W0135
-    request, fixture_utils, session_cloud, setup_image
+    request, fixture_utils, session_cloud
 ) -> Iterator[IntegrationInstance]:
     """Provide a client that runs once per module."""
     with _client(request, fixture_utils, session_cloud) as client:
@@ -354,7 +363,7 @@ def module_client(  # pylint: disable=W0135
 
 @pytest.fixture(scope="class")
 def class_client(  # pylint: disable=W0135
-    request, fixture_utils, session_cloud, setup_image
+    request, fixture_utils, session_cloud
 ) -> Iterator[IntegrationInstance]:
     """Provide a client that runs once per class."""
     with _client(request, fixture_utils, session_cloud) as client:
@@ -459,8 +468,45 @@ def _generate_profile_report() -> None:
     log.info(command, "final.stats")
 
 
+# https://docs.pytest.org/en/stable/reference/reference.html#pytest.hookspec.pytest_sessionstart
+def pytest_sessionstart(session) -> None:
+    """do session setup"""
+    global _SESSION_CLOUD
+    try:
+        _SESSION_CLOUD = get_session_cloud()
+        setup_image(_SESSION_CLOUD)
+    except Exception as e:
+        if _SESSION_CLOUD:
+            # if a _SESSION_CLOUD was allocated, clean it up
+            if _SESSION_CLOUD.snapshot_id:
+                # if a snapshot id was set, then snapshot succeeded, teardown
+                _SESSION_CLOUD.delete_snapshot()
+            _SESSION_CLOUD.destroy()
+        pytest.exit(
+            f"{type(e).__name__} in session setup: {str(e)}", returncode=2
+        )
+
+
 def pytest_sessionfinish(session, exitstatus) -> None:
-    if integration_settings.INCLUDE_COVERAGE:
-        _generate_coverage_report()
-    elif integration_settings.INCLUDE_PROFILE:
-        _generate_profile_report()
+    """do session teardown"""
+    try:
+        if integration_settings.INCLUDE_COVERAGE:
+            _generate_coverage_report()
+        elif integration_settings.INCLUDE_PROFILE:
+            _generate_profile_report()
+    except Exception as e:
+        log.warning("Could not generate report during teardown: %s", e)
+    try:
+        _SESSION_CLOUD.delete_snapshot()
+    except Exception as e:
+        log.warning(
+            "Could not delete snapshot. Leaked snapshot id %s: %s",
+            _SESSION_CLOUD.snapshot_id,
+            e,
+        )
+    try:
+        _SESSION_CLOUD.destroy()
+    except Exception as e:
+        log.warning(
+            "Could not destroy session cloud: %s(%s)", type(e).__name__, e
+        )

tests/integration_tests/datasources/test_azure.py

@@ -78,9 +78,7 @@ def parse_resolvectl_dns(output: str) -> dict:
 @pytest.mark.skipif(
     CURRENT_RELEASE < BIONIC, reason="Easier to test on Bionic+"
 )
-def test_azure_multi_nic_setup(
-    setup_image, session_cloud: IntegrationCloud
-) -> None:
+def test_azure_multi_nic_setup(session_cloud: IntegrationCloud) -> None:
     """Integration test for https://warthogs.atlassian.net/browse/CPC-3999.
 
     Azure should have the primary NIC only route to DNS.

tests/integration_tests/decorators.py

@@ -27,7 +27,7 @@ def wrapper(*args, **kwargs):
                     time.sleep(delay)
             else:
                 if last_error:
-                    raise last_error
+                    raise TimeoutError from last_error
             return retval
 
         return wrapper

tests/integration_tests/modules/test_apt_functionality.py

@@ -478,7 +478,7 @@ def test_apt_proxy(client: IntegrationInstance):
 
 
 @pytest.mark.skipif(not IS_UBUNTU, reason="Apt usage")
-def test_install_missing_deps(setup_image, session_cloud: IntegrationCloud):
+def test_install_missing_deps(session_cloud: IntegrationCloud):
     # Two stage install: First stage:  remove gpg noninteractively from image
     instance1 = session_cloud.launch(user_data=REMOVE_GPG_USERDATA)
     snapshot_id = instance1.snapshot()

tests/integration_tests/modules/test_hotplug.py

@@ -309,7 +309,7 @@ def test_multi_nic_hotplug(client: IntegrationInstance):
 
 @pytest.mark.skipif(PLATFORM != "ec2", reason="test is ec2 specific")
 @pytest.mark.skip(reason="IMDS race, see GH-5373. Unskip when fixed.")
-def test_multi_nic_hotplug_vpc(setup_image, session_cloud: IntegrationCloud):
+def test_multi_nic_hotplug_vpc(session_cloud: IntegrationCloud):
     """Tests that additional secondary NICs are routable from local
     networks after the hotplug hook is executed when network updates
     are configured on the HOTPLUG event."""

tests/integration_tests/modules/test_lxd.py

@@ -247,7 +247,7 @@ def test_storage_btrfs(client):
 @pytest.mark.skipif(
     CURRENT_RELEASE < FOCAL, reason="tested on Focal and later"
 )
-def test_storage_preseed_btrfs(setup_image, session_cloud: IntegrationCloud):
+def test_storage_preseed_btrfs(session_cloud: IntegrationCloud):
     # TODO: If test is marked as not bionic, why is there a bionic section?
     if CURRENT_RELEASE.series in ("bionic",):
         nictype = "nictype: bridged"
@@ -311,7 +311,7 @@ def test_storage_zfs(client):
 @pytest.mark.skipif(
     CURRENT_RELEASE < FOCAL, reason="Tested on focal and later"
 )
-def test_storage_preseed_zfs(setup_image, session_cloud: IntegrationCloud):
+def test_storage_preseed_zfs(session_cloud: IntegrationCloud):
     # TODO: If test is marked as not bionic, why is there a bionic section?
     if CURRENT_RELEASE.series in ("bionic",):
         nictype = "nictype: bridged"

tests/integration_tests/test_networking.py

@@ -182,9 +182,7 @@ def test_applied(self, client: IntegrationInstance):
         pytest.param(NET_V2_MATCH_CONFIG, id="v2"),
     ),
 )
-def test_netplan_rendering(
-    net_config, session_cloud: IntegrationCloud, setup_image
-):
+def test_netplan_rendering(net_config, session_cloud: IntegrationCloud):
     mac_addr = random_mac_address()
     launch_kwargs = {
         "config_dict": {
@@ -222,9 +220,7 @@ def test_netplan_rendering(
     reason="Test requires custom networking provided by LXD",
 )
 @pytest.mark.parametrize("net_config", (NET_V1_NAME_TOO_LONG,))
-def test_schema_warnings(
-    net_config, session_cloud: IntegrationCloud, setup_image
-):
+def test_schema_warnings(net_config, session_cloud: IntegrationCloud):
     # TODO: This test takes a lot more time than it needs to.
     # The default launch wait will wait until cloud-init done, but the
     # init network stage will wait 2 minutes for network timeout.
@@ -266,9 +262,7 @@ def test_schema_warnings(
     PLATFORM not in ("lxd_vm", "lxd_container"),
     reason="Test requires lxc exec feature due to broken network config",
 )
-def test_invalid_network_v2_netplan(
-    session_cloud: IntegrationCloud, setup_image
-):
+def test_invalid_network_v2_netplan(session_cloud: IntegrationCloud):
     mac_addr = random_mac_address()
 
     if PLATFORM == "lxd_vm":
@@ -316,7 +310,7 @@ def test_invalid_network_v2_netplan(
 
 
 @pytest.mark.skipif(PLATFORM != "ec2", reason="test is ec2 specific")
-def test_ec2_multi_nic_reboot(setup_image, session_cloud: IntegrationCloud):
+def test_ec2_multi_nic_reboot(session_cloud: IntegrationCloud):
     """Tests that additional secondary NICs and secondary IPs on them are
     routable from non-local networks after a reboot event when network updates
     are configured on every boot."""
@@ -347,7 +341,7 @@ def test_ec2_multi_nic_reboot(setup_image, session_cloud: IntegrationCloud):
 
 @pytest.mark.adhoc  # costly instance not available in all regions / azs
 @pytest.mark.skipif(PLATFORM != "ec2", reason="test is ec2 specific")
-def test_ec2_multi_network_cards(setup_image, session_cloud: IntegrationCloud):
+def test_ec2_multi_network_cards(session_cloud: IntegrationCloud):
     """
     Tests that with an interface type with multiple network cards (non unique
     device indexes).