Unblock CI: stop asserting server prose, fix task-def leak and retry gaps

tests/integration/client/orkes/test_orkes_clients.py

@@ -53,6 +53,51 @@ def _retry_on_404(func, *args, retries=5, **kwargs):
             raise
 
 
+def _assert_not_found(fetch, *identifiers):
+    # Assert the 404 status, not the server's prose. The wording is not part of
+    # the API contract: a server release changed "Workflow with id: X not found."
+    # to "No execution found for id: X" and turned CI red with no SDK change.
+    # Also fail loudly when the resource is still readable -- the bare
+    # try/except this replaces passed silently in exactly that case.
+    try:
+        fetch()
+    except ApiException as e:
+        assert e.code == 404, f"expected a 404, got {e.code}: {e.message}"
+        for identifier in identifiers:
+            assert str(identifier) in str(e.message), \
+                f"expected {identifier!r} in the 404 message, got: {e.message}"
+        return
+    raise AssertionError("expected a 404, but the resource is still readable")
+
+
+def _clear_tags(get_tags, delete_tag, target):
+    # Best-effort: leave the target with no tags so a scenario that asserts on
+    # exact tag counts can be re-run. Failures are ignored -- a tag that is
+    # already gone is nothing to report, and cleanup must not fail the test.
+    try:
+        existing = get_tags(target) or []
+    except Exception:
+        return
+    for tag in existing:
+        try:
+            delete_tag(MetadataTag(tag.key, tag.value), target)
+        except Exception:
+            pass
+
+
+def _await_value(fetch, expected, timeout=15, interval=1):
+    # Queue size is eventually consistent -- the server enqueues and indexes
+    # asynchronously, so reading straight after starting or draining work
+    # intermittently returns a stale count. Poll until it settles, then let the
+    # caller assert on the last value seen so failures still show the mismatch.
+    deadline = time.time() + timeout
+    value = fetch()
+    while value != expected and time.time() < deadline:
+        time.sleep(interval)
+        value = fetch()
+    return value
+
+
 class TestOrkesClients:
     def __init__(self, configuration: Configuration):
         self.api_client = ApiClient(configuration)
@@ -129,11 +174,7 @@ def test_task_lifecycle(self):
         self.__test_task_execution_lifecycle()
 
         self.metadata_client.unregister_task_def(TASK_TYPE)
-        try:
-            self.metadata_client.get_task_def(TASK_TYPE)
-        except ApiException as e:
-            assert e.code == 404
-            assert e.message == "Task {0} not found".format(TASK_TYPE)
+        _assert_not_found(lambda: self.metadata_client.get_task_def(TASK_TYPE), TASK_TYPE)
 
     def test_secret_lifecycle(self):
         self.secret_client.put_secret(SECRET_NAME, "secret_value")
@@ -164,11 +205,7 @@ def test_secret_lifecycle(self):
         assert self.secret_client.secret_exists(SECRET_NAME) == False
 
         self.secret_client.delete_secret(SECRET_NAME + "_2")
-
-        try:
-            self.secret_client.get_secret(SECRET_NAME + "_2")
-        except ApiException as e:
-            assert e.code == 404
+        _assert_not_found(lambda: self.secret_client.get_secret(SECRET_NAME + "_2"))
 
     def test_scheduler_lifecycle(self, workflowDef):
         startWorkflowRequest = StartWorkflowRequest(
@@ -212,12 +249,9 @@ def test_scheduler_lifecycle(self, workflowDef):
         assert len(fetched_tags) == 0
 
         self.scheduler_client.delete_schedule(SCHEDULE_NAME)
-
-        try:
-            schedule = self.scheduler_client.get_schedule(SCHEDULE_NAME)
-        except ApiException as e:
-            assert e.code == 404
-            assert e.message == "Schedule '{0}' not found".format(SCHEDULE_NAME)
+        _assert_not_found(
+            lambda: self.scheduler_client.get_schedule(SCHEDULE_NAME), SCHEDULE_NAME
+        )
 
     def test_application_lifecycle(self):
         req = CreateOrUpdateApplicationRequest(APPLICATION_NAME)
@@ -263,11 +297,10 @@ def test_application_lifecycle(self):
         self.authorization_client.delete_access_key(created_app.id, created_access_key.id)
 
         self.authorization_client.delete_application(created_app.id)
-        try:
-            application = self.authorization_client.get_application(created_app.id)
-        except ApiException as e:
-            assert e.code == 404
-            assert e.message == "Application '{0}' not found".format(created_app.id)
+        _assert_not_found(
+            lambda: self.authorization_client.get_application(created_app.id),
+            created_app.id
+        )
 
     def test_user_group_permissions_lifecycle(self, workflowDef):
         req = UpsertUserRequest("Integration User", ["USER"])
@@ -340,18 +373,10 @@ def test_user_group_permissions_lifecycle(self, workflowDef):
         self.authorization_client.remove_user_from_group(GROUP_ID, USER_ID)
 
         self.authorization_client.delete_user(USER_ID)
-        try:
-            self.authorization_client.get_user(USER_ID)
-        except ApiException as e:
-            assert e.code == 404
-            assert e.message == "User '{0}' not found".format(USER_ID)
+        _assert_not_found(lambda: self.authorization_client.get_user(USER_ID), USER_ID)
 
         self.authorization_client.delete_group(GROUP_ID)
-        try:
-            self.authorization_client.get_group(GROUP_ID)
-        except ApiException as e:
-            assert e.code == 404
-            assert e.message == "Group '{0}' not found".format(GROUP_ID)
+        _assert_not_found(lambda: self.authorization_client.get_group(GROUP_ID), GROUP_ID)
 
     def __test_register_workflow_definition(self, workflowDef: WorkflowDef):
         self.__create_workflow_definition(workflowDef)
@@ -482,12 +507,9 @@ def __poll_workflow_until_complete(self, workflow_id, timeout_seconds=60,
 
     def __test_unregister_workflow_definition(self):
         self.metadata_client.unregister_workflow_def(WORKFLOW_NAME, 1)
-
-        try:
-            self.metadata_client.get_workflow_def(WORKFLOW_NAME, 1)
-        except ApiException as e:
-            assert e.code == 404
-            assert e.message == 'No such workflow found by name: {0}, version: 1'.format(WORKFLOW_NAME)
+        _assert_not_found(
+            lambda: self.metadata_client.get_workflow_def(WORKFLOW_NAME, 1), WORKFLOW_NAME
+        )
 
     def __test_task_tags(self):
         tags = [
@@ -496,6 +518,14 @@ def __test_task_tags(self):
             MetadataTag("tag3", "val3")
         ]
 
+        # retry_scenario re-runs this whole scenario from the top on a transient
+        # blip, but tags survive the failed attempt -- so the "add one tag, now
+        # there is exactly one" assertion below saw the leftovers and failed with
+        # `assert 2 == 1`. Start from a known-empty set so the scenario is
+        # re-runnable.
+        _clear_tags(self.metadata_client.getTaskTags,
+                    self.metadata_client.deleteTaskTag, TASK_TYPE)
+
         self.metadata_client.addTaskTag(tags[0], TASK_TYPE)
         fetchedTags = self.metadata_client.getTaskTags(TASK_TYPE)
         assert len(fetchedTags) == 1
@@ -512,6 +542,10 @@ def __test_task_tags(self):
     def __test_workflow_tags(self):
         singleTag = MetadataTag("wftag", "val")
 
+        # Same re-runnability problem as __test_task_tags.
+        _clear_tags(self.metadata_client.get_workflow_tags,
+                    self.metadata_client.delete_workflow_tag, WORKFLOW_NAME)
+
         self.metadata_client.add_workflow_tag(singleTag, WORKFLOW_NAME)
         fetchedTags = self.metadata_client.get_workflow_tags(WORKFLOW_NAME)
         assert len(fetchedTags) == 1
@@ -575,11 +609,9 @@ def __test_workflow_execution_lifecycle(self):
         assert workflow.status == "RUNNING"
 
         self.workflow_client.delete_workflow(workflow_uuid)
-        try:
-            workflow = self.workflow_client.get_workflow(workflow_uuid, False)
-        except ApiException as e:
-            assert e.code == 404
-            assert str(e.message).lower() == "workflow with id: {} not found.".format(workflow_uuid)
+        _assert_not_found(
+            lambda: self.workflow_client.get_workflow(workflow_uuid, False), workflow_uuid
+        )
 
     def __test_task_execution_lifecycle(self):
 
@@ -606,7 +638,9 @@ def __test_task_execution_lifecycle(self):
         workflow_uuid_2 = self.workflow_client.start_workflow(startWorkflowRequest)
 
         # First task of each workflow is in the queue
-        assert self.task_client.get_queue_size_for_task(TASK_TYPE) == 2
+        assert _await_value(
+            lambda: self.task_client.get_queue_size_for_task(TASK_TYPE), 2
+        ) == 2
 
         polledTask = self.task_client.poll_task(TASK_TYPE)
         assert polledTask.status == TaskResultStatus.IN_PROGRESS
@@ -617,7 +651,9 @@ def __test_task_execution_lifecycle(self):
         assert taskExecLogs[0].log == "Polled task..."
 
         # First task of second workflow is still in the queue
-        assert self.task_client.get_queue_size_for_task(TASK_TYPE) == 1
+        assert _await_value(
+            lambda: self.task_client.get_queue_size_for_task(TASK_TYPE), 1
+        ) == 1
 
         taskResult = TaskResult(
             workflow_instance_id=polledTask.workflow_instance_id,
@@ -666,7 +702,9 @@ def __test_task_execution_lifecycle(self):
             )
             completed += 1
 
-        queue_size = self.task_client.get_queue_size_for_task(TASK_TYPE)
+        queue_size = _await_value(
+            lambda: self.task_client.get_queue_size_for_task(TASK_TYPE), 0
+        )
         print(f'queue size for {TASK_TYPE} is {queue_size}')
         assert queue_size == 0
 

tests/integration/client/orkes/test_orkes_service_registry_client.py

@@ -12,6 +12,10 @@
 from conductor.client.http.models.proto_registry_entry import ProtoRegistryEntry
 from conductor.client.orkes.orkes_service_registry_client import OrkesServiceRegistryClient
 from conductor.client.http.rest import ApiException
+from tests.integration.retry_helpers import (
+    DEFAULT_OVERALL_DEADLINE_SECONDS,
+    retry_scenario,
+)
 
 SUFFIX = str(uuid())
 HTTP_SERVICE_NAME = 'IntegrationTestServiceRegistryHttp_' + SUFFIX
@@ -38,11 +42,20 @@ def __init__(self, configuration: Configuration):
         self.client = OrkesServiceRegistryClient(configuration)
         logger.info(f'Setting up TestOrkesServiceRegistryClient with config {configuration}')
 
-    def run(self) -> None:
-        """Run all service registry tests"""
-        self.test_http_service_registry()
-        self.test_grpc_service()
-        self.test_proto_operations()
+    def run(self, deadline=None) -> None:
+        """Run all service registry tests.
+
+        Each is a scenario: on a transient blip against the shared dev server
+        (gateway 5xx, 423 contention, status-0 transport hiccup) it retries from
+        the top until the shared deadline passes. Real failures raise at once.
+        This suite used to call them bare, so a single 502 from the proxy failed
+        the whole run.
+        """
+        retry_scenario('test_http_service_registry', self.test_http_service_registry,
+                       deadline=deadline)
+        retry_scenario('test_grpc_service', self.test_grpc_service, deadline=deadline)
+        retry_scenario('test_proto_operations', self.test_proto_operations,
+                       deadline=deadline)
 
     def setUp(self):
         """Clean up services before each test"""
@@ -276,8 +289,12 @@ def test_all(self):
         logger.info('START: service registry integration tests')
         configuration = self.config
 
+        # One shared wall-clock budget for the whole suite, as in
+        # test_workflow_client_intg.test_all.
+        deadline = time.monotonic() + DEFAULT_OVERALL_DEADLINE_SECONDS
+
         # Run service registry tests
-        TestOrkesServiceRegistryClient(configuration=configuration).run()
+        TestOrkesServiceRegistryClient(configuration=configuration).run(deadline=deadline)
 
         logger.info('END: service registry integration tests')
 

tests/integration/conftest.py

@@ -52,6 +52,36 @@ def _check_server_connectivity():
     return _server_available, _skip_reason
 
 
+_reclaimed = False
+
+
+def _reclaim_once():
+    """Reclaim task-def quota leaked by earlier runs, before any test registers.
+
+    Without this a filled-up account stays filled: registration answers 402 for
+    every branch, so no run gets far enough to clean up after itself.
+
+    Called from both entry points below — pytest_sessionstart covers the CI
+    invocations (tests/integration/conftest.py is an initial conftest for every
+    bucket), skip_if_server_unavailable covers a session that reaches the
+    unittest suites some other way. The flag makes the second call a no-op.
+    """
+    global _reclaimed
+    if _reclaimed:
+        return
+    _reclaimed = True
+
+    available, _ = _check_server_connectivity()
+    if not available:
+        return
+    from tests.integration.leaked_task_defs import reclaim_task_def_quota
+    reclaim_task_def_quota(Configuration())
+
+
+def pytest_sessionstart(session):
+    _reclaim_once()
+
+
 def skip_if_server_unavailable():
     """
     Call from unittest.TestCase.setUpClass to skip the entire test class
@@ -70,6 +100,61 @@ def setUpClass(cls):
         raise unittest.SkipTest(reason)
 
 
+# ---------------------------------------------------------------------------
+# Metadata cleanup
+# ---------------------------------------------------------------------------
+
+def cleanup_metadata(config, task_defs=(), workflow_defs=()):
+    """Best-effort delete of metadata a test class registered.
+
+    Suites that register per-run task defs (``worker_task(...,
+    register_task_def=True)`` with a RUN_ID-suffixed name) used to leave every
+    one behind, so each CI run added a handful to the shared server until it
+    hit its Task Definitions cap and answered 402 to all further
+    registrations. Call this from tearDownClass.
+
+    Failures are logged and swallowed: cleanup must never turn a passing suite
+    red, and a def that was never registered is nothing to report.
+
+    ``workflow_defs`` items are names (version 1 assumed) or (name, version).
+    """
+    from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient
+
+    client = OrkesMetadataClient(config)
+
+    def _log(what, e):
+        # "no such definition" is the normal case for anything a deselected or
+        # skipped test never registered, so it logs at debug — only a cleanup
+        # that failed for some other reason is worth a warning.
+        level = logging.DEBUG if _is_missing(e) else logging.WARNING
+        logger.log(level, "cleanup: could not unregister %s: %s", what, e)
+
+    for name in task_defs:
+        try:
+            client.unregister_task_def(name)
+        except Exception as e:
+            _log(f"task def {name}", e)
+
+    for entry in workflow_defs:
+        name, version = entry if isinstance(entry, tuple) else (entry, 1)
+        try:
+            client.unregister_workflow_def(name, version)
+        except Exception as e:
+            _log(f"workflow def {name} v{version}", e)
+
+
+def _is_missing(exc):
+    """True when a delete failed only because the definition was not there.
+
+    The server reports this inconsistently — 404, or a 500 whose body reads
+    "No such task definition" / "No such workflow definition" — so match on
+    both the status and the text.
+    """
+    if getattr(exc, "code", None) == 404:
+        return True
+    return "no such" in str(getattr(exc, "body", "") or exc).lower()
+
+
 # ---------------------------------------------------------------------------
 # Pytest session-scoped fixtures
 # ---------------------------------------------------------------------------