andrew
e49d12a72b
First automated coverage. 38 tests, run against the built image (which has the app deps); tests/ and requirements-dev.txt are not baked into the runtime image. - tests/test_dedup_gating.py: the safety-critical one. Drives dedup-library.sh's process_group with --apply + --only-paths and asserts ONLY the confirmed path is deleted and the higher-ranked FLAC is kept. - tests/test_pipeline_runner.py: lock skip, use_lock=False bypass, lock release, success/failure/timeout recording. - tests/test_db.py: schema version + idempotent migrations, FK-safe job-run pruning, stale-run reconciliation. - tests/test_credential_service.py: conf-field substitution incl. newline- injection safety, secret/non-secret classification, cookie-expiry parsing. - tests/test_playlist_service.py: cron round-trips, name validation, render. - tests/test_diagnostics.py: /health + /setup checks. - conftest.py points every path setting at throwaway temp dirs and a temp DB, so tests never touch a real library, /config, or the network. See tests/README.md. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
61 lines
1.6 KiB
Python
61 lines
1.6 KiB
Python
import asyncio
|
|
|
|
from app.services import pipeline_runner as pr
|
|
|
|
# One loop for the whole module: pipeline_runner._lock is a module-level
|
|
# asyncio.Lock, and reusing it across different loops (as asyncio.run would
|
|
# create) raises "bound to a different event loop".
|
|
_loop = asyncio.new_event_loop()
|
|
|
|
|
|
def _run(coro):
|
|
return _loop.run_until_complete(coro)
|
|
|
|
|
|
def test_run_job_success_records_row_and_log():
|
|
run = _run(pr.run_job("test:ok", ["true"]))
|
|
assert run.status == "success"
|
|
assert run.exit_code == 0
|
|
assert run.log_path # a log file path was recorded
|
|
|
|
|
|
def test_run_job_failure():
|
|
run = _run(pr.run_job("test:fail", ["false"]))
|
|
assert run.status == "failed"
|
|
assert run.exit_code == 1
|
|
|
|
|
|
def test_lock_skip_when_held():
|
|
async def scenario():
|
|
await pr._lock.acquire()
|
|
try:
|
|
return await pr.run_job("test:locked", ["true"], use_lock=True)
|
|
finally:
|
|
pr._lock.release()
|
|
|
|
run = _run(scenario())
|
|
assert run.status == "skipped_lock"
|
|
|
|
|
|
def test_use_lock_false_bypasses_held_lock():
|
|
async def scenario():
|
|
await pr._lock.acquire()
|
|
try:
|
|
return await pr.run_job("test:bypass", ["true"], use_lock=False)
|
|
finally:
|
|
pr._lock.release()
|
|
|
|
run = _run(scenario())
|
|
assert run.status == "success"
|
|
|
|
|
|
def test_lock_released_after_run():
|
|
_run(pr.run_job("test:release", ["true"]))
|
|
assert not pr._lock.locked()
|
|
|
|
|
|
def test_timeout_kills_and_marks_failed():
|
|
run = _run(pr.run_job("test:timeout", ["sleep", "5"], timeout=0.3))
|
|
assert run.status == "failed"
|
|
assert run.exit_code == -1
|