ACLP Logs - stabilize integration tests - fix teardown

Author: sjer-akamai

test/integration/models/monitor/test_monitor_logs.py

@@ -1,3 +1,4 @@
+import logging
 import os
 import urllib.request
 from test.integration.conftest import get_region
@@ -9,7 +10,7 @@
 
 import pytest
 
-from linode_api4 import LinodeClient, LogsStreamType, PaginatedList
+from linode_api4 import LinodeClient, LogsStreamType, PaginatedList, Region
 from linode_api4.objects import (
     ObjectStorageACL,
     ObjectStorageBucket,
@@ -22,18 +23,30 @@
     LogsStreamStatus,
 )
 
+logger = logging.getLogger(__name__)
+
 _RUN_ACLP_LOGS_STREAM_TESTS = "RUN_ACLP_LOGS_STREAM_TESTS"
 _SKIP_STREAM_TESTS = pytest.mark.skipif(
     os.getenv(_RUN_ACLP_LOGS_STREAM_TESTS, "").strip().lower()
     not in {"yes", "true"},
     reason=f"{_RUN_ACLP_LOGS_STREAM_TESTS} environment variable must be set to 'yes' or 'true'",
 )
+_STREAM_FIXTURE_CLEANUP_WAIT = 2700
+_STREAM_FIXTURE_PROVISIONING_WAIT = 3600
 
 
 @pytest.fixture(scope="session")
-def create_object_storage_key(test_linode_client: LinodeClient):
+def region(test_linode_client: LinodeClient):
+    region = get_region(test_linode_client, {"Object Storage"})
+    yield region
+
+
+@pytest.fixture(scope="session")
+def create_object_storage_key(test_linode_client: LinodeClient,
+                              region: Region):
     key = test_linode_client.object_storage.keys_create(
         label=get_test_label(),
+        regions=[region.id]
     )
     yield key
     key.delete()
@@ -43,19 +56,19 @@ def create_object_storage_key(test_linode_client: LinodeClient):
 def test_destination(
     test_linode_client: LinodeClient,
     create_object_storage_key: ObjectStorageKeys,
+    region: Region,
 ):
     dest, bucket = _create_destination_with_bucket(
-        test_linode_client, create_object_storage_key
+        test_linode_client, create_object_storage_key, region
     )
     yield dest
     _delete_destination_with_bucket(test_linode_client, dest, bucket)
 
 
 def _create_destination_with_bucket(
-    client: LinodeClient, key: ObjectStorageKeys
+    client: LinodeClient, key: ObjectStorageKeys, region: Region
 ):
     """Helper that creates an OBJ bucket and a logs destination backed by it."""
-    region = get_region(client, {"Object Storage"})
     bucket = client.object_storage.bucket_create(
         cluster_or_region=region.id,
         label=get_test_label(),
@@ -79,16 +92,38 @@ def _delete_destination_with_bucket(
     client: LinodeClient, dest: LogsDestination, bucket: ObjectStorageBucket
 ):
     """Helper that deletes a logs destination and its backing OBJ bucket."""
-    send_request_when_resource_available(timeout=1800, func=dest.delete)
+    logger.info(
+        f"Deleting logs destination %s (id=%s), waiting up to {_STREAM_FIXTURE_CLEANUP_WAIT}s...",
+        dest.label,
+        dest.id,
+    )
+    # Wait until no stream references this destination before attempting deletion.
+    # create_secondary_destination may be torn down before create_stream due to
+    # pytest's LIFO fixture teardown order, so the stream could still be alive.
+    def no_stream_attached():
+        streams = client.monitor.streams()
+        return all(
+            all(d.id != dest.id for d in s.destinations) for s in streams
+        )
+
+    logger.info(
+        "Waiting for all streams referencing destination %s to be deleted...",
+        dest.id,
+    )
+    wait_for_condition(30, _STREAM_FIXTURE_CLEANUP_WAIT, no_stream_attached)
+    send_request_when_resource_available(timeout=100, func=dest.delete)
+    logger.info("Destination %s deleted, emptying bucket %s...", dest.id, bucket.label)
     _empty_bucket(client, bucket)
-    send_request_when_resource_available(timeout=1800, func=bucket.delete)
+    logger.info("Deleting bucket %s, waiting up to 100s...", bucket.label)
+    send_request_when_resource_available(timeout=100, func=bucket.delete)
+    logger.info("Bucket %s deleted.", bucket.label)
 
 
 def _skip_if_streams_exist(client: LinodeClient):
     """Skip the current test if any streams already exist on the account.
     Only one stream can be present per account at a time."""
     existing_streams = client.monitor.streams()
-    if len(existing_streams) > 0:
+    if len(existing_streams) > 10:
         stream_labels = [s.label for s in existing_streams]
         pytest.skip(
             f"Skipping: existing stream(s) found on this account "
@@ -304,9 +339,10 @@ def test_fails_to_create_stream_invalid_destination(invalid_destination_error):
 def create_secondary_destination(
     test_linode_client: LinodeClient,
     create_object_storage_key: ObjectStorageKeys,
+        region: Region
 ):
     dest, bucket = _create_destination_with_bucket(
-        test_linode_client, create_object_storage_key
+        test_linode_client, create_object_storage_key, region
     )
     yield dest
     _delete_destination_with_bucket(test_linode_client, dest, bucket)
@@ -317,6 +353,7 @@ def create_stream(
     test_linode_client: LinodeClient,
     test_destination: LogsDestination,
     invalid_destination_error,  # This ensures run order to keep negative test case deterministic
+    create_secondary_destination: LogsDestination,
 ):
     _skip_if_streams_exist(test_linode_client)
 
@@ -328,7 +365,59 @@ def create_stream(
     assert stream.id is not None
     assert stream.status == LogsStreamStatus.provisioning
     yield stream
-    send_request_when_resource_available(timeout=1800, func=stream.delete)
+    stream_teardown(test_linode_client, stream)
+
+
+def _wait_for_stream_updatable(client: LinodeClient, stream_id: int):
+    """
+    Blocks until the stream with the given id reaches active or inactive status.
+    Used both by the wait_for_updatable_status fixture (before tests) and by
+    stream_teardown (before deletion) to ensure the stream is in a stable,
+    operable state. Updating destinations or other attributes puts the stream
+    back into a transitional state, and attempting to delete or modify it while
+    transitioning results in [400] errors.
+    """
+
+    def is_stream_updatable():
+        stream = client.load(LogsStream, stream_id)
+        return stream.status in (
+            LogsStreamStatus.active,
+            LogsStreamStatus.inactive,
+        )
+
+    wait_for_condition(30, _STREAM_FIXTURE_PROVISIONING_WAIT, is_stream_updatable)
+
+
+def stream_teardown(test_linode_client: LinodeClient, stream: LogsStream):
+    # Wait for the stream to be in a stable state before deleting.
+    # Updating destinations (or labels/status) puts the stream back into a
+    # transitional state; deleting while transitioning returns [400].
+    logger.info(
+        "Teardown: waiting for stream %s (id=%s) to reach a stable state before deletion...",
+        stream.label,
+        stream.id,
+    )
+    _wait_for_stream_updatable(test_linode_client, stream.id)
+    logger.info(
+        "Teardown: deleting stream %s (id=%s)...",
+        stream.label,
+        stream.id,
+    )
+    send_request_when_resource_available(timeout=100, func=stream.delete)
+
+    # The delete request returns 200 but stream deletion is async on the backend.
+    # Wait until the stream is fully gone before teardown continues, so that
+    # dependent fixtures (e.g. create_secondary_destination) can proceed with teardown.
+    logger.info(
+        "Stream delete request accepted for id=%s, waiting for backend to process request",
+        stream.id,
+    )
+
+    def is_stream_deleted():
+        existing = test_linode_client.monitor.streams()
+        return all(s.id != stream.id for s in existing)
+
+    wait_for_condition(30, _STREAM_FIXTURE_CLEANUP_WAIT, is_stream_deleted)
 
 
 @pytest.fixture(scope="session")
@@ -347,7 +436,7 @@ def is_stream_provisioned():
             LogsStreamStatus.inactive,
         )
 
-    wait_for_condition(60, 3600, is_stream_provisioned)
+    wait_for_condition(60, _STREAM_FIXTURE_PROVISIONING_WAIT, is_stream_provisioned)
 
     yield test_linode_client.load(LogsStream, create_stream.id)
 
@@ -360,15 +449,7 @@ def wait_for_updatable_status(
     Waits for the stream to be in an active or inactive state before a test runs.
     Streams can switch to `provisioning` state between updates. This makes sure the previous update is fully finished.
     """
-
-    def is_stream_updatable():
-        stream = test_linode_client.load(LogsStream, provisioned_stream.id)
-        return stream.status in (
-            LogsStreamStatus.active,
-            LogsStreamStatus.inactive,
-        )
-
-    wait_for_condition(30, 3600, is_stream_updatable)
+    _wait_for_stream_updatable(test_linode_client, provisioned_stream.id)
 
 
 @_SKIP_STREAM_TESTS