Add lightweight integration tests for Celery/Redis basics in CI

[berendt]

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

[ideaship]

This job can pass green with zero tests executed. tests/integration/conftest.py skips all integration-marked tests when Redis is unreachable, and pytest exits 0 on all-skipped (reproduced locally as 5 skipped, exit 0). With no readiness gate after docker run -d, any Redis-startup failure here surfaces as a passing job instead of a red one, defeating the gate. Please add (a) a Redis-readiness wait before pytest, and (b) a required-Redis mode for CI, e.g. an OSISM_REQUIRE_REDIS=1 env lever that makes conftest.py fail the session instead of skipping when Redis is unreachable. Set it only in this job so local skip behavior is preserved.

[ideaship]

This is a production-path change with no unit test. It replaced two hardcoded constants with real precedence logic (CELERY_BROKER_URL override, and result_backend defaulting to broker_url), but there is no reference to either env var anywhere under tests/; the integration suite only exercises the REDIS_HOST-derived fallback. Please add a unit test asserting: (1) CELERY_BROKER_URL wins over the derived URL, (2) CELERY_RESULT_BACKEND overrides independently, (3) absent CELERY_RESULT_BACKEND, result_backend == broker_url. Note these are class attributes resolved at import time from import-frozen settings.REDIS_*, so the test must importlib.reload(settings) then importlib.reload(osism.tasks) after setting env; monkeypatch.setenv alone will assert the stale cached value and pass for the wrong reason.

[ideaship]

This (True,) to True fix is correct (the 1-tuple only worked by being truthy), but it is an unrelated bug fix folded into a commit about broker-URL derivation; it should have been its own commit so it is traceable independently. Since the commit's suite advertises end-to-end "task core" coverage, also consider asserting the corrected value: either a direct Config.task_track_started is True check, or a STARTED-state transition in the Celery round-trip test.

[ideaship]

This test reads/writes/deletes the fixed global key osism:task_lock on db 0 (see osism/utils/__init__.py:500-554); it cannot use a per-run UUID like the redlock/stream tests do. Against a disposable CI Redis that is fine, but the README example points REDIS_HOST=localhost at a local box, where this would clobber a real operator lock. Please either run the integration suite against a dedicated REDIS_DB, or add an explicit note in the README that the suite mutates live state on db 0 and must target a disposable Redis.

[ideaship]

Readiness breaks on any worker pinging the broker, not ci-worker specifically. On an isolated CI Redis this is fine, but on a shared local broker it can pass before this worker is up and hide a startup failure. Consider gating on the expected node, e.g. any(n.startswith("ci-worker@") for n in (celery_app.control.inspect().ping() or {})), mirroring what test_worker_is_visible already does.