UN-3636 [MISC] enable the unit-workers rig group and de-flake its suite

tests/groups.yaml

@@ -17,8 +17,7 @@ defaults:
 groups:
   # ── Unit tier: pure pytest, no external services ───────────────────────────
   unit-rig:
-    # Self-tests for the rig's pure logic (dep-graph, evaluate, junit parse).
-    # Run from the repo root so the `tests.rig` package import resolves.
+    # Rig self-tests. Run from repo root so the `tests.rig` import resolves.
     tier: unit
     workdir: .
     paths: [tests/rig/tests]
@@ -44,27 +43,27 @@ groups:
     tier: unit
     workdir: workers
     paths: [tests, shared/tests]
-    markers: "unit"
+    # Negative filter like the other unit groups: worker tests aren't tagged
+    # `unit`. Tests tagged `integration`/`slow` are excluded here but not yet
+    # routed to any group — add one before relying on those markers.
+    markers: "not integration and not slow"
     uv_sync_group: test
     coverage_source: shared
 
   unit-backend:
     tier: unit
     workdir: backend
-    # Pure backend tests — no DB. Collect the whole tree and let markers, not a
-    # hand-kept file list, decide membership: tests needing live infra carry
-    # `@pytest.mark.integration` (see integration-backend) and are excluded here.
+    # Marker-selected, not a hand-kept file list: live-infra tests carry
+    # `integration` and run in integration-backend.
     paths: ["."]
     markers: "not integration"
     uv_sync_group: test
-    # Anchored: integration-backend reuses the identical Django settings env so
-    # the two halves of the backend suite can't drift apart.
+    # Shared with integration-backend so both halves use identical settings.
     env: &backend_test_env
       DJANGO_SETTINGS_MODULE: backend.settings.test
       # Tenancy is row-level, not schema-per-tenant; tests run in public.
       DB_SCHEMA: public
-      # base.py resolves these at import time with no default; supply test-safe
-      # values here.
+      # Resolved at import with no default; supply test-safe values.
       DJANGO_SECRET_KEY: test-secret-key-not-for-production
       # All-zero Fernet key: valid format, obviously a placeholder.
       ENCRYPTION_KEY: AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=
@@ -79,8 +78,7 @@ groups:
       SYSTEM_ADMIN_USERNAME: admin
       SYSTEM_ADMIN_PASSWORD: admin
       SYSTEM_ADMIN_EMAIL: admin@example.com
-      # ExecutionFileHandler builds execution-dir paths from this at init; only
-      # constructed, never written to in these tests.
+      # Only used to build paths at init; never written in these tests.
       WORKFLOW_EXECUTION_DIR_PREFIX: /tmp/unstract-workflow-exec
     coverage_source: .
 
@@ -106,13 +104,9 @@ groups:
   integration-backend:
     tier: integration
     workdir: backend
-    # Backend ORM tests — need live infra. The rig provisions the declared
-    # requires_services via testcontainers and injects their connection env into
-    # pytest (tests/rig/cli.py:_inject_infra_env). Not optional: these gate the
-    # integration tier.
-    # Marker-only selection — exact complement of unit-backend: conftest.py
-    # auto-marks DB-bound tests `integration`, so new DB tests join this
-    # group with no edit here. Tests needing external creds skip without them.
+    # Complement of unit-backend: conftest auto-marks DB-bound tests
+    # `integration`, so new ones join here with no edit. The rig provisions
+    # requires_services and injects their env; cred-gated tests skip without.
     paths: ["."]
     markers: "integration"
     uv_sync_group: test
@@ -125,10 +119,8 @@ groups:
     workdir: unstract/connectors
     paths: [tests]
     markers: "integration"
-    # Most connector integration tests need real third-party credentials
-    # (Snowflake, GDrive, Box, Dropbox, …) and skip when those are absent. The
-    # MinIO test actually runs: the rig provisions MinIO via testcontainers and
-    # injects MINIO_* creds (tests/rig/cli.py).
+    # Most need real third-party creds and skip when absent; MinIO runs
+    # because the rig provisions it and injects the creds.
     requires_services: [minio]
     uv_sync_group: test
     coverage_source: src
@@ -141,12 +133,8 @@ groups:
     optional: true   # placeholder until tests are authored
 
   # ── E2E tier: real HTTP against running platform ───────────────────────────
-  # Every group below has `requires_platform: true` and must (transitively)
-  # depend on `e2e-smoke` — the rig's `_validate_platform_groups_depend_on_gate`
-  # enforces this structural invariant so the manifest can't ship a platform
-  # test that bypasses the smoke gate, and so smoke runs first in topo order.
-  # (Runtime skip-on-gate-failure is a future enhancement; see that function's
-  # docstring and the TODO in cmd_run.)
+  # Platform groups must depend (transitively) on e2e-smoke; the rig enforces
+  # this so smoke gates and runs first. Nothing ships bypassing it.
   e2e-smoke:
     tier: e2e
     paths: [tests/e2e/smoke]
@@ -155,8 +143,7 @@ groups:
     timeout_seconds: 1200
     depends_on: [unit-sdk1, unit-workers]
 
-  # Not optional: login is the first e2e critical path we gate on. A red login
-  # test fails the build and (on main) surfaces as an in-scope critical gap.
+  # Not optional: login is a gated critical path; red fails the build.
   e2e-login:
     tier: e2e
     paths: [tests/e2e/auth]

tests/rig/cli.py

@@ -484,6 +484,20 @@ def cmd_run(args: argparse.Namespace) -> int:
                 and overall_exit == 0
             ):
                 overall_exit = exit_code
+            # A non-optional pytest group that collected zero tests (exit 5) is
+            # almost always a broken marker/path, not intent — fail rather than
+            # fold in green. Skip hurl (its exit 5 means "no files", a
+            # placeholder) and dev runs with a marker/paths override that may
+            # legitimately match nothing.
+            if (
+                exit_code == 5
+                and not group.optional
+                and group.runner == "pytest"
+                and not args.marker
+                and not args.paths
+                and overall_exit == 0
+            ):
+                overall_exit = 1
             # Belt-and-braces: if the junit attests to errors/failures the exit
             # code didn't (truncated junit → errors=1 with exit 0), the report
             # shows ❌ but exit would otherwise stay 0. Keep them in sync.

tests/rig/tests/test_cli.py

@@ -216,6 +216,86 @@ def fake_execute_group(group, **kwargs):
     )
 
 
+def _run_empty_group_scenario(
+    tmp_path: Path, monkeypatch, *, extra_args: list[str]
+) -> int:
+    """Drive cmd_run with one non-optional pytest group that collects nothing
+    (exit 5). Returns the overall exit code."""
+    from tests.rig.reporting import GroupResult
+
+    test_dir = Path(__file__).parent
+    (tmp_path / "groups.yaml").write_text(
+        "version: 1\n"
+        "groups:\n"
+        "  unit-empty:\n"
+        "    tier: unit\n"
+        f"    workdir: {test_dir}\n"
+        "    paths: [.]\n"
+    )
+    (tmp_path / "critical_paths.yaml").write_text("version: 1\npaths: []\n")
+
+    import tests.rig.cli as cli_mod
+    import tests.rig.critical_paths as cp_mod
+    import tests.rig.groups as groups_mod
+
+    monkeypatch.setattr(groups_mod, "DEFAULT_MANIFEST", tmp_path / "groups.yaml")
+    monkeypatch.setattr(cp_mod, "DEFAULT_REGISTRY", tmp_path / "critical_paths.yaml")
+
+    def fake_execute_group(group, **kwargs):
+        result = GroupResult(
+            name=group.name,
+            tier=group.tier,
+            exit_code=5,  # pytest "no tests collected"
+            passed=0,
+            failed=0,
+            errors=0,
+            skipped=0,
+            duration_seconds=0.01,
+        )
+        return result, 5
+
+    monkeypatch.setattr(cli_mod, "_execute_group", fake_execute_group)
+
+    args = cli_mod._build_parser().parse_args(
+        [
+            "run",
+            "unit-empty",
+            "--no-coverage",
+            "--no-parallel",
+            "--reports-dir",
+            str(tmp_path / "reports"),
+            "--baseline",
+            str(tmp_path / "reports" / "previous-summary.json"),
+            *extra_args,
+        ]
+    )
+    return cli_mod.cmd_run(args)
+
+
+def test_empty_nonoptional_group_fails_build(tmp_path: Path, monkeypatch) -> None:
+    """A non-optional pytest group that collects zero tests is a broken
+    marker/path, not a pass — it must gate the overall exit."""
+    exit_code = _run_empty_group_scenario(tmp_path, monkeypatch, extra_args=[])
+    assert exit_code != 0, (
+        "a non-optional group that collected nothing must fail the build; "
+        f"got exit_code={exit_code}"
+    )
+
+
+def test_empty_group_with_marker_override_does_not_fail(
+    tmp_path: Path, monkeypatch
+) -> None:
+    """A ``--marker`` override may legitimately match nothing; the empty-collect
+    gate must not fire in that case."""
+    exit_code = _run_empty_group_scenario(
+        tmp_path, monkeypatch, extra_args=["--marker", "some_marker"]
+    )
+    assert exit_code == 0, (
+        "an empty result under a marker override must not gate the build; "
+        f"got exit_code={exit_code}"
+    )
+
+
 def _run_gap_scenario(
     tmp_path: Path, monkeypatch, *, covered_by: str, fail_on_gap: bool
 ) -> int:

workers/conftest.py

@@ -12,6 +12,8 @@
 os.environ.setdefault("CELERY_BROKER_BASE_URL", "amqp://localhost:5672//")
 os.environ.setdefault("CELERY_BROKER_USER", "guest")
 os.environ.setdefault("CELERY_BROKER_PASS", "guest")
+# tests/conftest.py strips these back out for workers/tests so Celery result
+# backends stay in-memory; they remain in effect for shared/tests.
 os.environ.setdefault("DB_HOST", "localhost")
 os.environ.setdefault("DB_USER", "test")
 os.environ.setdefault("DB_PASSWORD", "test")

workers/tests/conftest.py

@@ -6,9 +6,47 @@
 time if INTERNAL_API_BASE_URL is not set.
 """
 
+import os
 from pathlib import Path
 
+import pytest
 from dotenv import load_dotenv
 
 _env_test = Path(__file__).resolve().parent.parent / ".env.test"
 load_dotenv(_env_test)
+
+# Worker Celery apps build a Postgres result backend from DB_*/CELERY_BACKEND_DB_*.
+# Strip these before any app is imported so tests don't reach (or leak) a live DB
+# the unit tier has no server for; eager results then stay in-memory. This undoes
+# the DB_* defaults set in ../conftest.py, which stay in effect for shared/tests
+# (not covered by this file).
+for _var in (
+    "DB_HOST",
+    "DB_PORT",
+    "DB_NAME",
+    "DB_USER",
+    "DB_PASSWORD",
+    "CELERY_BACKEND_DB_HOST",
+    "CELERY_BACKEND_DB_PORT",
+    "CELERY_BACKEND_DB_NAME",
+    "CELERY_BACKEND_DB_USER",
+    "CELERY_BACKEND_DB_PASSWORD",
+):
+    os.environ.pop(_var, None)
+
+
+@pytest.fixture(autouse=True)
+def _restore_current_celery_app():
+    """Pin celery's default app as current_app around each test. Worker modules
+    build their own apps at import and set them current, so `@worker_task` proxies
+    otherwise fail to resolve (`NotRegistered`) against a drifted current_app.
+    Finalize so every shared task is registered on it.
+    """
+    from celery._state import default_app
+
+    default_app.finalize()
+    default_app.set_current()
+    try:
+        yield
+    finally:
+        default_app.set_current()

workers/tests/test_sanity_phase6h.py → workers/tests/test_agentic_operations.py

@@ -1,26 +1,18 @@
-"""Phase 6H Sanity — AgenticPromptStudioExecutor + agentic operations.
-
-Verifies:
-1. All 8 agentic Operation enums exist
-2. AGENTIC_EXTRACTION removed from Operation enum
-3. Mock AgenticPromptStudioExecutor — registration and all 8 operations
-4. Queue routing: executor_name="agentic" → celery_executor_agentic
-5. LegacyExecutor does NOT handle any agentic operations
-6. Dispatch sends to correct queue
-7. Structure tool routes to agentic executor (not legacy)
+"""Agentic operations: the agentic executor registers and routes all 8
+operations to its queue, LegacyExecutor rejects them, and the structure tool
+dispatches agentic extract to it (not legacy). Also guards that the old
+AGENTIC_EXTRACTION enum stays removed.
 """
 
 from unittest.mock import MagicMock, patch
 
 import pytest
-
 from unstract.sdk1.execution.context import ExecutionContext, Operation
 from unstract.sdk1.execution.dispatcher import ExecutionDispatcher
 from unstract.sdk1.execution.executor import BaseExecutor
 from unstract.sdk1.execution.registry import ExecutorRegistry
 from unstract.sdk1.execution.result import ExecutionResult
 
-
 AGENTIC_OPERATIONS = [
     "agentic_extract",
     "agentic_summarize",
@@ -34,15 +26,11 @@
 
 
 # ---------------------------------------------------------------------------
-# 1. Operation enums
+# Operation enum
 # ---------------------------------------------------------------------------
 
-class TestAgenticOperations:
-    @pytest.mark.parametrize("op", AGENTIC_OPERATIONS)
-    def test_agentic_operation_enum_exists(self, op):
-        values = {o.value for o in Operation}
-        assert op in values
 
+class TestAgenticOperations:
     def test_agentic_extraction_removed(self):
         """Old AGENTIC_EXTRACTION enum no longer exists."""
         assert not hasattr(Operation, "AGENTIC_EXTRACTION")
@@ -54,9 +42,11 @@ def test_agentic_extraction_removed(self):
 # 2. Mock AgenticPromptStudioExecutor — registration and all operations
 # ---------------------------------------------------------------------------
 
+
 class TestAgenticExecutorRegistration:
     def test_mock_agentic_executor_registers_and_routes_all_ops(self):
         """Simulate cloud executor discovery and execution of all 8 ops."""
+
         @ExecutorRegistry.register
         class MockAgenticExecutor(BaseExecutor):
             _OPERATION_MAP = {op: f"_handle_{op}" for op in AGENTIC_OPERATIONS}
@@ -115,6 +105,7 @@ def execute(self, context):
 # 3. Queue routing
 # ---------------------------------------------------------------------------
 
+
 class TestAgenticQueueRouting:
     def test_agentic_routes_to_correct_queue(self):
         queue = ExecutionDispatcher._get_queue("agentic")
@@ -148,10 +139,12 @@ def test_dispatch_sends_to_agentic_queue(self, op):
 # 4. LegacyExecutor does NOT handle agentic operations
 # ---------------------------------------------------------------------------
 
+
 class TestLegacyExcludesAgentic:
     @pytest.mark.parametrize("op", AGENTIC_OPERATIONS)
     def test_agentic_op_not_in_legacy_operation_map(self, op):
         from executor.executors.legacy_executor import LegacyExecutor
+
         assert op not in LegacyExecutor._OPERATION_MAP
 
     def test_legacy_returns_failure_for_agentic_extract(self):
@@ -197,11 +190,11 @@ def test_legacy_returns_failure_for_agentic_summarize(self):
 # 5. Structure tool routes to agentic executor
 # ---------------------------------------------------------------------------
 
+
 class TestStructureToolAgenticRouting:
     @patch("unstract.sdk1.x2txt.X2Text")
     def test_structure_tool_dispatches_agentic_extract(self, mock_x2text_cls, tmp_path):
         """Verify _run_agentic_extraction sends executor_name='agentic'."""
-
         from file_processing.structure_tool_task import _run_agentic_extraction
 
         mock_dispatcher = MagicMock()
@@ -224,6 +217,7 @@ def test_structure_tool_dispatches_agentic_extract(self, mock_x2text_cls, tmp_pa
             dispatcher=mock_dispatcher,
             shim=MagicMock(),
             file_execution_id="exec-001",
+            execution_id="exec-parent-001",
             organization_id="org-001",
             source_file_name="test.pdf",
             fs=MagicMock(),
@@ -241,6 +235,7 @@ def test_structure_tool_dispatches_agentic_extract(self, mock_x2text_cls, tmp_pa
 # 6. tasks.py log_component for agentic operations
 # ---------------------------------------------------------------------------
 
+
 class TestTasksLogComponent:
     @pytest.mark.parametrize("op", AGENTIC_OPERATIONS)
     def test_agentic_ops_use_default_log_component(self, op):
@@ -262,6 +257,8 @@ def test_agentic_ops_use_default_log_component(self, op):
         # Agentic ops should NOT match ide_index, structure_pipeline,
         # or table_extract/smart_table_extract branches
         assert context.operation not in (
-            "ide_index", "structure_pipeline",
-            "table_extract", "smart_table_extract",
+            "ide_index",
+            "structure_pipeline",
+            "table_extract",
+            "smart_table_extract",
         )