Add integration tests for Celery error, state and revoke paths

[berendt]

Extends the Celery integration coverage beyond the noop round-trip with three
lifecycle tests against the live Redis broker and the real worker subprocess.

One commit, e240b64:

• tests/integration/worker_tasks.py (new) registers two test-only tasks on
  the ansible app: itest_fail raises RuntimeError, itest_block sleeps
  and returns its argument. The explicit osism.tasks.ansible.* names route
  them onto the osism-ansible queue the worker consumes; with their
  auto-generated names they would land on the unconsumed default queue.
• tests/integration/conftest.py starts the session worker with
  --include=tests.integration.worker_tasks and pins its cwd to the
  repository root so the module resolves regardless of where pytest was
  launched. A broken include surfaces as the fixture's existing early-exit
  RuntimeError instead of a hang.
• tests/integration/test_celery_lifecycle.py (new) adds the three tests: an
  exception raised in the worker reaches the caller through
  AsyncResult.get() and leaves the task in FAILURE; task_track_started
  reports STARTED while a task runs; osism.utils.revoke_task terminates a
  running task, the state ends REVOKED, get() raises TaskRevokedError,
  and a final noop round-trip proves the prefork pool respawned its child.
  All waits go through a time.monotonic() deadline helper that fails the
  test instead of hanging.

The tests carry pytestmark = pytest.mark.integration, skip without a
reachable Redis, and run for real in the python-osism-integration-tests
job. black --check and flake8 pass locally.

Closes #2404

[sourcery-ai]

Hey - I've left some high level feedback:

• The REPO_ROOT = Path(__file__).resolve().parents[2] assumption in conftest.py ties the worker startup to a specific directory layout; consider deriving the repo root in a more robust way (e.g., via an environment variable or walking up to a marker like .git/pyproject.toml) to avoid breakage if the file is moved.
• The various hard-coded timeouts (_wait_for_state default, 60s/300s in the tests) could be centralized as named constants or made configurable (e.g., via environment variables) to make tuning easier and reduce the risk of flaky tests in slower CI environments.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The `REPO_ROOT = Path(__file__).resolve().parents[2]` assumption in `conftest.py` ties the worker startup to a specific directory layout; consider deriving the repo root in a more robust way (e.g., via an environment variable or walking up to a marker like `.git`/`pyproject.toml`) to avoid breakage if the file is moved.
- The various hard-coded timeouts (`_wait_for_state` default, 60s/300s in the tests) could be centralized as named constants or made configurable (e.g., via environment variables) to make tuning easier and reduce the risk of flaky tests in slower CI environments.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[ideaship]

A bare assert state == "STARTED" misdirects the triage on the one run where it fires. There are two ways to observe a terminal state first — task_track_started stopped working, or this poll loop stalled for the full BLOCK_SHORT window — and the docstring above commits to the first ("would mean STARTED was never reported") while the second is what a loaded CI machine produces. Naming both costs one line:

assert state == "STARTED", (
    f"got {state} first — either task_track_started stopped working, "
    f"or this poll loop stalled for the full {BLOCK_SHORT}s window"
)

The wait set itself is right and shouldn't change: admitting SUCCESS/FAILURE is what turns a miss into an immediate failure instead of a 30 s timeout. It's the docstring's categorical phrasing that wants the same qualification.

[ideaship]

"Keeps the lookup correct if this file moves" promises something WORKER_INCLUDE = "tests.integration.worker_tasks" at :35 doesn't deliver. Move only conftest.py and the dotted path still resolves; move the whole tests/integration/ directory — the more likely change — and the constant breaks while the walk survives. Either derive both from __package__, or trim the sentence to the failure-mode half, which holds unconditionally and is the part worth having: a located error here instead of an opaque ImportError at worker boot.

[ideaship]

Please add a time_limit here. If the first _wait_for_state in test_revoke_running_task times out before the revoke is issued, pytest.fail() leaves a 300 s sleep occupying the single-concurrency worker, and teardown burns its full 30 s warm-shutdown window before killing the process. Today that costs 30 s on an already-red run, because this test happens to collect last among the ones taking celery_worker — but that safety is invisible at the point where it matters, and it disappears the day someone adds a worker-using test file sorting after test_celery_lifecycle.py, which has no second slot to run in.

BLOCK_LONG = 300 is the right value and shortening it would be the wrong fix — "never race the task's natural end" is the correct trade. The bound belongs on the task, not on the duration.