Add lightweight integration tests for Celery/Redis basics in CI

[berendt]

Implements #2368.

Adds a lightweight integration-test job that covers the Celery/Redis
task-processing core (broker, queue routing, worker, result backend, Redis
streams for task output, distributed locks) end-to-end in CI, plus the
prerequisite that makes the broker URL configurable.

Change set (in commit order)

1. Derive Celery broker URL from settings and environment

osism/tasks/__init__.py, osism/commands/service.py

• Config.broker_url / Config.result_backend were hardcoded to
  redis://redis, with the same literals duplicated in the beat and
  flower service commands. They now derive from REDIS_HOST /
  REDIS_PORT / REDIS_DB (the same settings osism.utils._init_redis
  already honors), overridable via CELERY_BROKER_URL /
  CELERY_RESULT_BACKEND. The default of REDIS_HOST is redis, so this
  is backwards compatible.
• service.py now references Config.broker_url (single source of truth)
  for beat/flower, removing the duplicated literals.
• Fixed task_track_started, which was the tuple (True,) instead of the
  intended bool (it only worked because a non-empty tuple is truthy).

This is what lets the tests point Celery at a local Redis via
REDIS_HOST=localhost.

2. Add Celery/Redis integration tests

tests/integration/, setup.cfg, README.md

• test_celery.py — round-trip: ansible.noop.delay() →
  result.get(timeout=60) is True (broker + osism-ansible routing +
  worker + result backend); worker visibility: a fresh Celery
  client configured from Config sees the worker via
  inspect().ping() (the mechanism behind osism get status workers).
• test_redis_streams.py — push_task_output / finish_task_output /
  fetch_task_output round-trip (the path osism apply uses to stream
  logs).
• test_locking.py — create_redlock acquire/release and
  set_task_lock / is_task_locked / remove_task_lock.
• conftest.py starts the worker as a session-scoped fixture from the
  same venv, running only the ansible Celery app (no import-time
  dependency on NetBox/OpenStack/ansible-core) with
  GATHER_FACTS_SCHEDULE=0 so the periodic gather_facts task is not
  registered. It polls inspect().ping() until the worker is ready and
  terminates it on teardown.
• Tests carry an integration marker (registered in setup.cfg for the
  --strict-markers run) and are skipped when Redis is not reachable,
  so a local pytest is unaffected.

3. Add Zuul integration-test job

.zuul.yaml, playbooks/test-integration.yml

• New python-osism-integration-tests job added to the check and
  periodic-daily pipelines. It reuses playbooks/pre.yml (which already
  sets up Docker via ensure-docker), starts a redis:7-alpine
  container, installs the package with pipenv exactly like the unit-test
  job, and runs pytest tests/integration with REDIS_HOST=localhost.

Notes / deviations

• The unit-test job is already restricted to tests/unit via
  testpaths = tests/unit in setup.cfg, so a bare pytest never
  collects tests/integration — the issue's first concern is already
  satisfied. The integration marker + Redis-reachability skip are still
  added (the marker is required by --strict-markers, and the skip keeps
  pytest tests/integration green locally without Redis).
• No CHANGELOG.md entry: this repo maintains the changelog in batched,
  post-release commits that map merged PR numbers to release versions, so
  per-PR entries are out of convention here.
• Follow-ups listed in the issue (compose-based variant, API/beat/revoke
  coverage) are intentionally out of scope.

Local verification

• black --check — clean (6 files)
• flake8 — clean
• pytest tests/integration — not run locally (no Redis); exercised by
  the new CI job.

🤖 Generated with Claude Code

[sourcery-ai]

Hey - I've found 2 security issues, and 1 other issue

Security issues:

• Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)
• Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'. (link)

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="osism/commands/service.py" line_range="63" />
<code_context>
         elif service == "flower":
             p = subprocess.Popen(
-                "celery --broker=redis://redis flower",
+                f"celery --broker={Config.broker_url} flower",
                 shell=True,
             )
</code_context>
<issue_to_address>
**issue (bug_risk):** Handle special characters in broker URL for the flower invocation.

Because this uses `shell=True`, any `&`, `?`, or other shell metacharacters in `Config.broker_url` can be split or reinterpreted by the shell. Please use a list-based `Popen` without `shell=True`, e.g. `Popen(["celery", "--broker", Config.broker_url, "flower"])`, or otherwise ensure the URL is safely quoted/escaped.
</issue_to_address>

### Comment 2
<location path="osism/commands/service.py" line_range="51-54" />
<code_context>
                subprocess.Popen(
                    f"celery -A {t} --broker={Config.broker_url} beat -s /tmp/celerybeat-schedule-{t}.db",
                    shell=True,
                )
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

*Source: opengrep*
</issue_to_address>

### Comment 3
<location path="osism/commands/service.py" line_range="62-65" />
<code_context>
            p = subprocess.Popen(
                f"celery --broker={Config.broker_url} flower",
                shell=True,
            )
</code_context>
<issue_to_address>
**security (python.lang.security.audit.dangerous-subprocess-use-audit):** Detected subprocess function 'Popen' without a static string. If this data can be controlled by a malicious actor, it may be an instance of command injection. Audit the use of this call to ensure it is not controllable by an external resource. You may consider using 'shlex.escape()'.

*Source: opengrep*
</issue_to_address>

---

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.}