Tests for async result processing

[carlosgjs]

Summary

This pull request adds tests for the result processing of async tasks.

It also adds a helper script and launch target for debugging tests.

Checklist

• I have tested these changes appropriately.
• I have added and/or modified relevant tests.
• I updated relevant documentation or comments.
• I have verified that this PR follows the project's coding standards.
• Any dependent changes have already been merged to main.

Summary by CodeRabbit

• Tests

  • Added comprehensive error-handling test coverage for pipeline processing, including edge cases and API integration.

• Chores

  • Added a developer script to run tests with debugger attach support for local debugging.

• Refactor

  • Improved task progress handling and reporting for more accurate, resilient progress snapshots.

[netlify]

✅ Deploy Preview for antenna-ssec canceled.

Name	Link
🔨 Latest commit	79cdd73
🔍 Latest deploy log	https://app.netlify.com/projects/antenna-ssec/deploys/698e58b26139630008585d1d

[netlify]

✅ Deploy Preview for antenna-preview canceled.

Name	Link
🔨 Latest commit	79cdd73
🔍 Latest deploy log	https://app.netlify.com/projects/antenna-preview/deploys/698e58b2c6e4f90008d2e0b6

[coderabbitai]

📝 Walkthrough

Walkthrough

Adds a VS Code debug attach config and a helper script for running tests with debugpy, a new test module covering error scenarios in pipeline result processing, and small TaskStateManager changes (lock key helper and a get_progress read-only view plus in-place pending updates).

Changes

Cohort / File(s)	Summary
Debug configuration & helper /.vscode/launch.json, scripts/debug_tests.sh	Adds a Debugpy "Attach: Tests" configuration (localhost:5680) and a shell script to run Django tests inside Docker with debugpy listening on 5680.
Pipeline result error tests ami/jobs/test_tasks.py	New comprehensive E2E/unit tests for process_nats_pipeline_result covering error cases, mixed results, concurrent lock contention, deleted-job handling, and an API endpoint integration test.
Task state manager updates ami/ml/orchestration/task_state.py	Introduces _lock_key(job_id) helper, adds TaskStateManager.get_progress(stage) (read-only snapshot), and updates _get_progress to remove processed IDs from pending and persist the updated pending set.

Sequence Diagram(s)

sequenceDiagram
    participant NATS
    participant Worker as Celery Worker
    participant State as TaskStateManager (Redis)
    participant DB
    participant API

    NATS->>Worker: publish pipeline result (success/error)
    Worker->>State: acquire lock (using _lock_key(job_id))
    Worker->>State: _get_progress -> update pending, compute progress
    State-->>Worker: progress snapshot
    alt result has image_id and detections
        Worker->>DB: save detections
    else result is error or missing image_id
        Worker->>DB: record failure / skip save
    end
    Worker->>State: update_state / persist progress
    Worker->>API: (optionally) call result endpoint / enqueue follow-up task
    Worker-->>NATS: ack

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Possibly related PRs

• Distributed Worker Architecture for ML Processing (Processing Service V2) pt. 1 #987: Adds tests and touches the same pipeline result processing paths and TaskStateManager behavior referenced here.
• Update debugging support for Django and Celery #1048: Also modifies VS Code launch configurations to add debugpy attach entries (related debugging infra).

Suggested reviewers

• mihow

Poem

🐰 I hopped in with debug port aglow,
Tests lined up in tidy row,
Locks now named and progress shown,
Pipelines tamed, no longer blown,
🥕 Debug, run, and let carrots grow!

🚥 Pre-merge checks | ✅ 3 | ❌ 1

❌ Failed checks (1 inconclusive)

Check name	Status	Explanation	Resolution
Description check	❓ Inconclusive	The description is incomplete; it covers the summary and checklist but omits the template's required sections for detailed explanation, list of changes, related issues, how to test, and deployment notes.	Expand the description to include detailed explanation of changes, list of modified/added components, related issue references, testing instructions, and any deployment considerations required by the template.

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'Tests for async result processing' accurately reflects the primary focus of the changeset, which adds comprehensive E2E tests for error handling in async task result processing.
Docstring Coverage	✅ Passed	Docstring coverage is 88.24% which is sufficient. The required threshold is 80.00%.
Merge Conflict Detection	✅ Passed	✅ No merge conflicts detected when merging into main

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment

---

No actionable comments were generated in the recent review. 🎉

🧹 Recent nitpick comments

ami/ml/orchestration/task_state.py (1)

89-98: Consider extracting the shared progress-computation logic.

get_progress and _get_progress (lines 114–127) duplicate the remaining/processed/percentage calculation and the TaskProgress construction. A small private helper (e.g., _build_progress(pending_count, total)) would keep them in sync and reduce future drift risk.

Otherwise the read-only snapshot method is clean and correctly guarded against None cache values and division by zero.

♻️ Optional: extract shared computation

+    def _build_progress(self, remaining: int, total: int) -> TaskProgress:
+        processed = total - remaining
+        percentage = float(processed) / total if total > 0 else 1.0
+        return TaskProgress(remaining=remaining, total=total, processed=processed, percentage=percentage)
+
     def get_progress(self, stage: str) -> TaskProgress | None:
         """Read-only progress snapshot for the given stage. Does not acquire a lock or mutate state."""
         pending_images = cache.get(self._get_pending_key(stage))
         total_images = cache.get(self._total_key)
         if pending_images is None or total_images is None:
             return None
-        remaining = len(pending_images)
-        processed = total_images - remaining
-        percentage = float(processed) / total_images if total_images > 0 else 1.0
-        return TaskProgress(remaining=remaining, total=total_images, processed=processed, percentage=percentage)
+        return self._build_progress(len(pending_images), total_images)

And similarly in _get_progress, replace lines 114–127 with a call to _build_progress(len(remaining_images), total_images) plus the existing logger call.

ami/jobs/test_tasks.py (1)

273-300: Consider asserting on the EagerResult for stronger guarantees.

The comment says "Should NOT raise exception" but .apply() returns an EagerResult that swallows exceptions by default. The acknowledge_task assertion on line 300 provides some signal, but explicitly checking the result status would make the intent clearer:

result = process_nats_pipeline_result.apply(...)
self.assertIsNone(result.result)  # or self.assertTrue(result.successful())

This is a minor strengthening — the current test still validates the key behavior (acknowledgment).

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[Copilot]

Pull request overview

This pull request adds comprehensive test coverage for async result processing in the jobs module, specifically focusing on error handling paths when PipelineResultsError is received. The PR also includes developer tooling improvements with a debug helper script and VS Code launch configuration for debugging tests.

Changes:

• Added comprehensive E2E tests for error handling in process_nats_pipeline_result task
• Refactored lock key generation in TaskStateManager to a separate function for testability
• Added debugging utilities (shell script and VS Code configuration) to facilitate test debugging

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
scripts/debug_tests.sh	New helper script to launch tests with debugpy for VS Code debugging
ami/ml/orchestration/task_state.py	Extracted _lock_key function for reusability in tests
ami/jobs/test_tasks.py	New comprehensive test suite covering error handling scenarios in pipeline result processing
.vscode/launch.json	Added "Attach: Tests" debug configuration for test debugging

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[coderabbitai]

Actionable comments posted: 2

🤖 Fix all issues with AI agents

In `@ami/jobs/test_tasks.py`:
- Around line 396-404: The test incorrectly assumes apply_async positional args;
update the assertions to extract kwargs robustly from mock_apply_async.call_args
by handling both calling conventions: inspect call =
mock_apply_async.call_args[0] and call_kwargs = mock_apply_async.call_args[1],
then determine the task kwargs either from call_kwargs.get('kwargs') or (if the
old convention used) from the second positional element; assert on
process_nats_pipeline_result.delay parameters by checking that the resolved
task_kwargs contains job_id == self.job.pk, reply_subject ==
"test.reply.error.1", and that "error" is in task_kwargs["result_data"]
(referencing mock_apply_async, process_nats_pipeline_result.delay, call_args,
and task_kwargs to locate the code).

In `@scripts/debug_tests.sh`:
- Around line 1-7: Add a shebang to the top of the debug_tests.sh script to
define the shell interpreter and replace the unquoted "$*" usage with the safe
"$@" form in the manage.py test invocation; update the script that defines the
docker run command (the file debug_tests.sh and the command invoking python -m
debugpy ... manage.py test $*) to start with a proper shebang (e.g.,
#!/usr/bin/env bash) and pass arguments as "$@" so arguments containing spaces
are preserved.

🧹 Nitpick comments (2)

ami/jobs/test_tasks.py (2)

240-240: Unused mock_manager_class parameter — consider suppressing or documenting.

The @patch decorator injects this argument, but the test doesn't use NATS (it tests lock contention before NATS is reached). Either add a brief comment explaining why it's unused or rename to _mock_manager_class to signal intent.

Proposed fix

     `@patch`("ami.jobs.tasks.TaskQueueManager")
-    def test_process_nats_pipeline_result_error_concurrent_locking(self, mock_manager_class):
+    def test_process_nats_pipeline_result_error_concurrent_locking(self, _mock_manager_class):

---

89-97: _get_progress has side effects — calling it in assertions mutates Redis state.

_get_progress internally calls cache.set(...) to update the pending images list (Line 106 in task_state.py). While the empty set passed here means the list is effectively unchanged, this couples the assertion helper to internal implementation details and could cause subtle ordering issues if tests are extended later.

Consider adding a brief comment in _assert_progress_updated noting this side effect, so future test authors are aware.

[mihow]

yay! thanks for the end to end tests