17 Commits
Author SHA1 Message Date
jpmschweitzer 613ecb9fb1 fix(permissions): narrow rm -rf deny globs to their exact forms
The trailing wildcard on the three rm -rf deny entries spanned path
separators, so Bash(rm -rf /*) matched every absolute path on the
machine rather than the filesystem root, and the ~ and $HOME entries
had the same shape. Narrowed to the exact literal forms.

These rules match literal command text, so they still stop a typo on
rm -rf /, rm -rf ~ or rm -rf $HOME exactly, but they no longer stop a
recursive delete aimed at any other path. That reduced cover is
deliberate, not an oversight.
2026-08-25 20:31:21 +02:00
jpmschweitzerandClaude b1cb3eb899 build(make): deselect the integration tests, and give them a target (T-55)
Twelve tests in test_database_integration.py carry
@pytest.mark.integration and need a real Postgres. The marker was
registered in pytest.ini and used correctly on the file — but `make test`
never deselected it and no target ever selected it. So those tests
neither ran nor passed: they errored on every invocation, and `make test`
exited 2 permanently, which teaches a reader that the exit code means
nothing. That is worse than either running them or not having them.

`make test` is now hermetic per D-26 — 164 passed, 19 deselected, exit 0,
verified inside an unprivileged network namespace as well as outside.

WHY THE AUDIT MISSED THIS, which is the part worth keeping. T-55 measured
each suite with and without a network and treated identical results as
proof of no live dependency. These twelve fail identically both ways,
because postgres-shared is a Docker-internal name that a host process
cannot resolve in either condition. A namespace proves a test does not
reach the network; it cannot distinguish that from a test whose
dependency is unreachable regardless. The positive control was run
against a host that WAS reachable, so it never covered this case.

The new target refuses an empty selection: pytest exit 5 fails with its
own message and a collection error gets a different one, so "nothing to
run" can never be read as "everything passed". Mutation-checked — anchor
asserted unique, marker renamed, target failed at exit 2 with the
intended message, Makefile restored byte-identical.

It reports honestly when Postgres is absent rather than skipping: 6
passed, 1 skipped, 12 errors from this host, which is the true state.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-18 20:44:24 +02:00
jpmschweitzerandClaude 3572a45322 test: delete six tests for a method that never existed
_should_run_now has no definition in src/ in any commit in this repo's
history — checked with `git log --all -S` across the whole tree, not just
the current worktree. Twelve assertion sites across six tests called it,
so these have never passed and never protected anything.

The behaviour they describe is real: cron-wildcard matching of minute,
hour, day_of_month, month and day_of_week. It lives inside the WHERE
clause of get_tasks_for_minute, not as a Python predicate, so there was
nothing to rename them onto.

KNOWN GAP, stated rather than left implied: that matching is now covered
by no test at all. Testing it means either asserting against the SQL and
params a mocked cursor receives, or extracting the predicate out of the
query — the second changes what decides, every minute, which scheduled
work runs, and is not a refactor to do casually. Deleting was chosen over
rewriting because a test that has never run is not coverage, and leaving
it in place claimed some.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-18 20:44:08 +02:00
jpmschweitzerandClaude 292c7de2bf test(redaction): assert the coarse behaviour the source actually has
This asserted that _redact_sensitive recurses into a dict under a
sensitive key — redacting auth.token while leaving auth.type readable.
The source replaces the whole value the moment the KEY matches, so
redacted["config"]["auth"] is a string and indexing ["token"] into it
raises TypeError. Source and test arrived in the same commit, so this
was never drift: it was a disagreement nobody settled.

Settled in favour of the source. Fine-grained redaction has to know
which sub-keys carry a secret, which is a guess about the shape of data
nobody has inspected; matching on the key cannot be wrong that way. Real
configs here are {"auth": {"type": "bearer", "token": "${SOME_KEY}"}},
and the cost of guessing wrong is a credential in a log, which no later
fix undoes. The price is readability, and it is paid deliberately.

Adds a second assertion that the secret appears nowhere in the output by
any path, which is the property actually worth protecting.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-18 20:44:08 +02:00
jpmschweitzer 6915be31ca fix(tests): match postgres_host assertion to the fixture actually in effect
tests/conftest.py sets os.environ["POSTGRES_HOST"] = "postgres-shared" at
module level (before `from src.main import app`), commented "Use real
postgres for integration tests". test_settings_loads_from_environment
asserted settings.postgres_host == "test-postgres", a value grep confirms
nothing in this suite has ever set — git log -p shows both the conftest
line and this assertion originate in the same single commit and neither has
changed since. Matched the assertion to the environment the suite actually
runs under.

Not a source change and not a claim that "postgres-shared" is the right
fixture value for a suite this ticket also found is not actually hermetic
where that value is concerned (see the 12 pre-existing DNS errors, tracked
separately from this fix) — only that the assertion should test what the
fixture sets, not an unset value.
2026-08-18 15:50:45 +02:00
jpmschweitzer 3f987bcf64 fix(tests): repair test_create_list_delete_task_flow's two defects
Same dependency_overrides fix as the other test_*.py commits for
get_task_executor (patch() cannot reach a route's already-registered
Depends() reference — see the conftest.py commit for the full explanation
and the empirical check that established it).

Once the mock actually intercepted the DB layer, a second, independent
defect surfaced: KeyError: 0. mock_cursor.fetchone.side_effect held 3 values
sized for "create, list, delete" (one fetchone() each), but list_tasks uses
fetchall(), not fetchone(), and delete_task has grown a second fetchone()
call since this test was written — c34db66 added the 409-on-history-loss
check, which does `row = cur.fetchone()` for the task id and then a
separate `cur.fetchone()[0]` for its execution count. A real create->list->
delete flow through this endpoint set needs 1 (create) + 2 (delete) = 3
fetchone() values in that order, not one per named step. Re-sequenced so
delete_task sees a real id and a zero execution count, reaching the intended
200 rather than an unhandled KeyError.

Source unchanged; delete_task's behavior is deliberate (c34db66) and this
integration test had not been updated to track it.
2026-08-18 15:50:29 +02:00
jpmschweitzer 68bea0cceb fix(tests): correct two independent AttributeErrors in test_task_executor
1. `executor.max_concurrent` has never existed. Concurrency is capped by the
   module-level MAX_CONCURRENT_TASKS constant via asyncio.Semaphore(
   MAX_CONCURRENT_TASKS) in TaskExecutor.__init__ — confirmed with git log -p
   across this file's whole history (three commits), the name has always
   been the module constant, never an instance attribute.
   test_executor_initialization now asserts MAX_CONCURRENT_TASKS == 5 and the
   semaphore's initial count, instead of a name the class never had.
   test_concurrent_task_limit asserted `mock_execute.call_count <=
   executor.max_concurrent`, which — separately from the AttributeError — was
   asserting the wrong observable: process_minute() awaits the full batch via
   asyncio.gather before returning, so by the time the assertion runs all 10
   scheduled tasks have executed; the semaphore bounds how many run
   concurrently mid-flight, not the eventual call_count. Reworded to assert
   all scheduled tasks still run (call_count == len(tasks)); a concurrency-
   in-flight assertion would need a task that can be observed mid-execution,
   which the AsyncMock stand-in does not provide.

2. _run_executor (src/tasks/executor.py) loads the executor module with the
   __import__ builtin directly (`__import__(module_path, fromlist=
   ['execute'])`), not importlib.import_module — this repo's own CLAUDE.md
   documents it as "the thing that will mislead you" about this module.
   importlib is never imported there, so patch('src.tasks.executor.importlib.
   import_module') failed at patch setup, before the three
   test_execute_task_* bodies ran at all. Switched to patch('builtins.
   __import__', side_effect=...) with a routing function that falls through
   to the real import for anything other than the target module — verified
   the call-recording shape empirically first (call('name', fromlist=[...])).

Source is unchanged in both cases; both are test-only defects present since
this file's initial commit.
2026-08-18 15:49:37 +02:00
jpmschweitzer 874f9f9711 fix(tests): repair three mock-wiring bugs in doc_sync_executor tests
test_executor_successful_sync_entire_repo: _get_git_commit is patched
separately and never calls the real _run_command, so it does not consume a
slot in mock_run.side_effect. The list reserved one anyway (labelled "git
rev-parse HEAD"), which shifted "M  README.md\n" one call late — git status
--porcelain saw "" (no changes) instead, so execute() took the no-changes
branch and the test asserted "Successfully synced" against "already up to
date". Removed the phantom slot. Also gave the iterdir() mock items real
string .name attributes: MagicMock(name="X") sets the mock's repr, not the
.name attribute read by execute()'s `item.name != '.git'` check, so every
item was being treated as non-.git regardless of the intended value.

test_executor_sync_specific_paths, test_executor_handles_no_changes:
mock_upstream_dir/mock_gitea_dir were built but never wired to
mock_path.return_value, so `work_dir = Path(...)` and `upstream_dir =
work_dir / "upstream"` resolved to a different, unconfigured auto-generated
MagicMock. `source.name` on that mock is itself a MagicMock, not a string,
so `', '.join(copied_paths)` raised TypeError. Wired mock_path.return_value
to a work_dir mock whose __truediv__ yields the intended upstream/gitea
mocks, matching the pattern the first test already used correctly.

All three are test-side: doc_sync_executor.py is unchanged. Confirmed via
git log -p that this file and its test have exactly one commit in this
repo's history (the initial extraction), so there is no prior passing
version to regress from — these tests appear to have never passed.
2026-08-18 15:49:20 +02:00
jpmschweitzer 5ccfb83f2f fix(tests): correct two independent test-api defects found while auditing T-55
1. get_task_executor mocking: same dependency_overrides fix as the previous
   commit, applied to this file's remaining patch('src.main.get_task_executor')
   call sites (test_valid_api_key_allows_access, test_create_task_with_valid_data,
   test_trigger_task_endpoint, test_stats_endpoint_returns_metrics,
   test_executions_endpoint_returns_history, test_executions_filter_by_task_name).

2. test_create_task_missing_fields_returns_400 asserted a status this endpoint
   cannot return. `task: TaskCreate` in src/main.py is a plain Pydantic request
   body with no custom validation for missing fields — FastAPI's own
   dependency-resolution layer rejects the request before create_task's body
   runs, and that layer always answers 422, not 400. There is no code path in
   this repo, at any point in its git history, that produces 400 for this
   request. Renamed to test_create_task_missing_fields_returns_422 and the
   assertion updated to match; source unchanged.

Kept together in one commit because both live in the same small file and were
found in the same pass, rather than risk a manual hunk split on tightly
interleaved diff context.
2026-08-18 15:49:08 +02:00
jpmschweitzer e39234b436 fix(tests): use dependency_overrides for get_task_executor, not patch()
FastAPI resolves Depends(get_task_executor) against the function object it
captured when each route was decorated, at import time. patch('src.main.
get_task_executor') therefore never reaches an already-registered route —
confirmed empirically with a minimal FastAPI app before touching this repo's
tests. Every one of these tests fell through to the real dependency, which
called the real psycopg2.connect() and failed on postgres-shared's DNS (a
Docker-network name unreachable from a host process either way, so this
reproduced identically with and without host networking).

app.dependency_overrides is FastAPI's own supported mechanism for this, and
was already used correctly elsewhere in this suite (test_task_delete.py).
Added a shared override_task_executor fixture in conftest.py and switched
every get_task_executor patch() call site in test_api_comprehensive.py to it.

Source (src/main.py) is unchanged — Depends() is the idiomatic, correct
pattern; the tests were using an ineffective substitute for it.
2026-08-18 15:48:29 +02:00
jpmschweitzerandClaude f9e1409898 build(setup): prove the venv actually works instead of trusting pip's exit code
`make setup` created the venv and ran pip install, then exited 0 whether or
not the result was usable — exactly D-24's failure shape, and the one this
repo hit hardest: on 2026-08-09 there was no venv at all here while CLAUDE.md
documented .venv/bin/python -m pytest as the way to run tests, and nothing in
setup would have caught that state before a person did.

Add a `pytest tests/ --collect-only -q` step at the end. It imports every
test module and everything each one pulls in from src/, without running a
single test, and fails the target on a broken interpreter or a broken
dependency graph alike.

Verified directly:
- absent interpreter -> exit 127, target halts
- apscheduler (declared, imported by src/main.py) uninstalled -> collect-only
  exits 4 with ModuleNotFoundError, target halts
- make setup afterwards reinstalls it and collect-only exits 0, 189 tests
  collected, same count as before the mutation
- idempotent: a warm rerun on an already-correct venv changes nothing and
  collects the same 189 tests, just faster (7s vs 30s cold)

T-47 (workspace).

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-17 12:11:52 +02:00
jpmschweitzerandClaude 5dbc2c38a4 release v1.9.0
Build and Push / release (push) Successful in 3s
Build and Push / build (push) Successful in 1m14s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-11 14:51:32 +02:00
jpmschweitzerandClaude 4e68767771 refactor(health-report): put summary where the other writer puts it
check_history has two producers. sysmon-go writes `summary` at the top level
beside `status`; this module wrote it under `metrics`. So a reader had to know
which producer wrote a row before it could find out what the row said, and a
query written the obvious way found one and silently missed the other.

That is the T-36 failure repeating. There, per-domain queries returned rows from
August and looked like a system that had stopped reporting, because the data was
nested under a composite row nobody had mentioned. Nothing was missing; the
query was asking the wrong shape. verify.sh had already grown a coalesce over
both spellings, which is the tell: a compatibility shim that hides a schema
disagreement rather than resolving it.

D-33 made this table a contract between producers, and a contract needs one
spelling.

Summary is now a required parameter with no default. sysmon-go enforces the same
thing through Domain.Run's signature, and the reason is identical: a row whose
substance is missing looks exactly like a row whose check found nothing to say.
Both call sites pass it; the failure path passes the exception rather than
leaving the field to the metrics blob.

Old rows keep the nested spelling and verify.sh keeps reading both, because
rewriting history to match a new convention is a worse trade than a fallback
with a reason attached.

Also drops "Three consequences" from the module docstring, which by then listed
five. A hardcoded count beside the thing it counts is the same defect as
install.sh printing "wrote 8 keys" while writing ten — this morning's bug, in
prose instead of code.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-11 14:51:32 +02:00
jpmschweitzerandClaude 7b12ce8f0a release v1.8.0
Build and Push / release (push) Successful in 2s
Build and Push / build (push) Successful in 1m13s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-11 12:29:18 +02:00
jpmschweitzer c34db66f51 fix(api): refuse to delete a task's history by accident, with 409 and ?purge
DELETE /tasks/{name} issued a bare DELETE against scheduled_tasks. Any task
that had ever run owns rows in task_executions, so the foreign key rejected it
and the caller got:

  psycopg2.errors.ForeignKeyViolation: update or delete on table
  "scheduled_tasks" violates foreign key constraint
  "task_executions_task_id_fkey" on table "task_executions"

surfaced as a bare 500 with nothing naming history as the obstacle. It read as
the service being broken rather than the request being refusable, and since
every task that has ever fired has history, the endpoint effectively worked
only for tasks that had never run. Found while removing a temporary probe task,
which then had to be deleted with hand-written SQL across two tables.

Refusing rather than cascading, because the outcomes are not equally
recoverable: a task definition can be recreated from the API in one call, its
execution history cannot be recreated at all. Defaulting to the destructive
reading of an ambiguous request is how audit trails disappear quietly.

The 409 carries what the caller needs to act -- how many records are at stake,
the flag that proceeds anyway, and PUT enabled=false, which is usually what was
actually wanted: it stops the task running and keeps the record. A bare
"conflict" would be little better than the 500 it replaces.

Purge deletes history and task in one transaction. Split across two, a failure
between them leaves the audit trail gone and the task alive -- the worst of both.

Mutation-checked: removing the guard fails the refusal tests. A test also pins
that a refused delete issues no DELETE at all, and that ?purge=true on a missing
task is still 404 rather than a success.
2026-08-11 12:29:18 +02:00
jpmschweitzerandClaude 4911c94e48 release v1.7.0
Build and Push / release (push) Successful in 3s
Build and Push / build (push) Successful in 1m14s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-11 12:22:47 +02:00
jpmschweitzerandClaude 538bfe5944 feat(health-report): name the task that produced each check_history row
`source` names the code that wrote a row. It cannot name the schedule that
invoked it, and two tasks may share one executor -- so a row could not answer
the question a health record mostly exists to answer: which job broke?

Concretely, on 2026-08-11 a 425 MB probe task and the 5 GB nightly backup both
ran through config_backup_executor. The probe failed and wrote

  source: scheduler/config_backup_executor
  status: critical
  error:  Backup file was not created

which is byte-for-byte what a nightly backup failure would have written. The
row was true and unattributable, and the reflex it invited -- delete the
inconvenient row -- was correctly refused. Attribution is the actual fix: the
record stays intact and starts saying who it is about.

Carried in a ContextVar rather than an argument. Executors are invoked as
execute(config, settings) and there are ten of them, several dormant -- existing
only as a string in a database row and becoming live the moment someone inserts
a task naming them. A signature change would leave those broken in a way nothing
imports, greps or tests would reveal. Injecting the name into `config` was the
other option and is worse: `config` is what a human wrote in the task
definition, and an executor is entitled to reject keys it does not recognise.

Two properties make the ContextVar safe, both verified in the deployed runtime
rather than reasoned about:

  - asyncio.to_thread propagates the context, so reporting still sees the task
    after T-74 moved executor bodies into worker threads. Had it not, every row
    from a real executor would have quietly lost its task while unit tests kept
    passing -- so there is a test that specifically goes through report_async.
  - Each asyncio Task gets its own copy, so the five concurrent executions
    MAX_CONCURRENT_TASKS permits cannot read each other's value. The isolation
    test yields mid-execution to force interleaving; without that it would pass
    even against a shared global.

A plain await does NOT get its own copy and leaks the value to the caller, which
the runtime check showed. Both real entry points go through create_task, but
task_scope resets via token rather than depending on that.

The field is omitted, not nulled, when there is no task: report() is callable
from a script, and a null would claim a task existed with no name.

Mutation-checked: removing the scope from the engine fails both isolation tests.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-11 12:22:47 +02:00
23 changed files with 1392 additions and 518 deletions
+3 -3
View File
@@ -58,9 +58,9 @@
"Bash(psql * TRUNCATE*)",
"Bash(redis-cli * FLUSHALL*)",
"Bash(redis-cli * FLUSHDB*)",
"Bash(rm -rf $HOME*)",
"Bash(rm -rf /*)",
"Bash(rm -rf ~*)",
"Bash(rm -rf $HOME)",
"Bash(rm -rf /)",
"Bash(rm -rf ~)",
"Bash(su *)",
"Bash(sudo *)",
"Bash(toj)",
+159
View File
@@ -45,3 +45,162 @@ So every timeout since 2025-12-07 has been recorded as a generic failure carryin
FIX: catch asyncio.TimeoutError explicitly in _run_executor, ahead of the broad handler, and let it propagate (or return a marker execute_task can distinguish). Note the ordering is the whole bug — a narrower except after a broader one is dead code, and there is no linter configured here to say so.
Related: a thread started by asyncio.to_thread cannot be cancelled, so after T-74 a timeout frees the loop while the work continues to completion. Whatever ''timeout'' comes to mean should say so rather than implying the work stopped.', NULL, '2026-08-11 09:59:05', '2026-08-11 09:59:05.545', '2026-08-11 09:59:05.545', NULL, 'af08842306d6326036b1865b8cbf7c58', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQAR5KXM3PTG0QT3R1PV8', 'description', 'get_tasks_for_minute (src/tasks/executor.py:70) excludes any task holding a task_executions row with status=''running'':
AND id NOT IN (SELECT task_id FROM task_executions WHERE status = ''running'')
Nothing ever reconciles that row. It is written before the executor runs and updated after, so a process that dies in between leaves it ''running'' permanently — and the task is then excluded from every future minute, forever, with no error, no alarm and no log line. The task simply stops existing as far as the scheduler is concerned.
This is not hypothetical. test_example_task has held a ''running'' row since 2025-12-07 — 5916 hours. It happens to be disabled, so nothing is broken today; the mechanism is what matters, not this instance.
The exposure is real and daily: Watchtower restarts this container every morning at 4 AM. Any task still running at that moment is permanently unscheduled by it. The config backup runs 03:05 and takes ~21 minutes, which is not far off.
The failure mode is the one this system keeps producing: it looks like nothing. A backup that stops running forever produces no failure — it produces silence, and silence reads as health.
FIX: reconcile at startup. Any row left ''running'' when the process starts cannot be running, because the process that owned it is gone. Mark those ''orphaned'' (or ''failed'' with a message naming the restart) during app startup, before the scheduler begins its first minute. Consider also a stale-row guard for rows older than the task''s timeout_seconds, which covers a killed worker without a restart.
The audit trail is preserved either way — this is a status correction, not a deletion.', 'get_tasks_for_minute (src/tasks/executor.py:70) excludes any task holding a task_executions row with status=''running'':
AND id NOT IN (SELECT task_id FROM task_executions WHERE status = ''running'')
Nothing ever reconciles that row. It is written before the executor runs and updated after, so a process that dies in between leaves it ''running'' permanently — and the task is then excluded from every future minute, forever, with no error, no alarm and no log line. The task simply stops existing as far as the scheduler is concerned.
This is not hypothetical. test_example_task has held a ''running'' row since 2025-12-07 — 5916 hours. It happens to be disabled, so nothing is broken today; the mechanism is what matters, not this instance.
The exposure is real and daily: Watchtower restarts this container every morning at 4 AM. Any task still running at that moment is permanently unscheduled by it. The config backup runs 03:05 and takes ~21 minutes, which is not far off.
The failure mode is the one this system keeps producing: it looks like nothing. A backup that stops running forever produces no failure — it produces silence, and silence reads as health.
FIX: reconcile at startup. Any row left ''running'' when the process starts cannot be running, because the process that owned it is gone. Mark those ''orphaned'' (or ''failed'' with a message naming the restart) during app startup, before the scheduler begins its first minute. Consider also a stale-row guard for rows older than the task''s timeout_seconds, which covers a killed worker without a restart.
The audit trail is preserved either way — this is a status correction, not a deletion.
FIXED in v1.6.0 (e45de4f). TaskExecutor.reconcile_orphaned_executions() runs in the lifespan before the scheduler starts.
Proven on the real system rather than in a test. The row that had been ''running'' since 2025-12-07 22:44 was released at startup:
WARNING - Orphaned execution recovered: test_example_task was left ''running''
since 2025-12-07 22:44:00.046695. That task had been excluded from
scheduling until now.
Rows still ''running'' afterwards: 0. Status is ''orphaned'', error names the restart.
''orphaned'' rather than ''failed'' because the outcome is genuinely unknown — a task that completed and never got to update its row looks exactly like one that died halfway, and ''failed'' would assert something nobody observed.
Ordering is part of the fix and is mutation-checked: reconciliation must complete before the first minute is processed, or the first tick still sees the stale rows. Removing the startup call fails the test.
Not extended to a duration-based sweep, deliberately. While the process lives, execute_task''s finally always closes the row out, so a stale row implies a dead owner; a time-based rule would have to distinguish a slow task from a dead one and would eventually close the record of a task that is still working.', NULL, '2026-08-11 10:11:39', '2026-08-11 10:11:39.058', '2026-08-11 10:11:39.058', NULL, '0cec030639664cf075a113be75218e8d', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQAR5KXM3PTG0QT3R1PV8', 'status', 'backlog', 'done', NULL, '2026-08-11 10:11:39', '2026-08-11 10:11:39.178', '2026-08-11 10:11:39.178', NULL, 'cd26a078365d91d44e6d284e80da39f5', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQBTCRCTXK6EJF0T8CS2C', 'description', 'execute_task has an `except asyncio.TimeoutError` handler (src/tasks/executor.py:179) that records status=''timeout''. It can never run.
_run_executor wraps the awaited call in `except Exception` (line 219) and returns the error as a value. Since Python 3.11 asyncio.TimeoutError IS the builtin TimeoutError, which inherits OSError -> Exception, so the broad handler catches it first and converts it into an ordinary (None, error) tuple. execute_task then sees a non-None error and files status=''failed''.
Verified in the deployed runtime:
asyncio.TimeoutError is TimeoutError: True
MRO: TimeoutError -> OSError -> Exception -> BaseException
And confirmed against 8 months of history — 18,785 executions, and the count of status=''timeout'' rows is zero:
success 16690 | failed 2093 | running 2 | timeout 0
So every timeout since 2025-12-07 has been recorded as a generic failure carrying a traceback. Operationally that erases the distinction that matters most when a job misbehaves: "this job is too slow for its window" and "this job is broken" need different responses, and right now they look identical in the history.
FIX: catch asyncio.TimeoutError explicitly in _run_executor, ahead of the broad handler, and let it propagate (or return a marker execute_task can distinguish). Note the ordering is the whole bug — a narrower except after a broader one is dead code, and there is no linter configured here to say so.
Related: a thread started by asyncio.to_thread cannot be cancelled, so after T-74 a timeout frees the loop while the work continues to completion. Whatever ''timeout'' comes to mean should say so rather than implying the work stopped.', 'execute_task has an `except asyncio.TimeoutError` handler (src/tasks/executor.py:179) that records status=''timeout''. It can never run.
_run_executor wraps the awaited call in `except Exception` (line 219) and returns the error as a value. Since Python 3.11 asyncio.TimeoutError IS the builtin TimeoutError, which inherits OSError -> Exception, so the broad handler catches it first and converts it into an ordinary (None, error) tuple. execute_task then sees a non-None error and files status=''failed''.
Verified in the deployed runtime:
asyncio.TimeoutError is TimeoutError: True
MRO: TimeoutError -> OSError -> Exception -> BaseException
And confirmed against 8 months of history — 18,785 executions, and the count of status=''timeout'' rows is zero:
success 16690 | failed 2093 | running 2 | timeout 0
So every timeout since 2025-12-07 has been recorded as a generic failure carrying a traceback. Operationally that erases the distinction that matters most when a job misbehaves: "this job is too slow for its window" and "this job is broken" need different responses, and right now they look identical in the history.
FIX: catch asyncio.TimeoutError explicitly in _run_executor, ahead of the broad handler, and let it propagate (or return a marker execute_task can distinguish). Note the ordering is the whole bug — a narrower except after a broader one is dead code, and there is no linter configured here to say so.
Related: a thread started by asyncio.to_thread cannot be cancelled, so after T-74 a timeout frees the loop while the work continues to completion. Whatever ''timeout'' comes to mean should say so rather than implying the work stopped.
FIXED in v1.6.0 (e45de4f). _run_executor now catches asyncio.TimeoutError ahead of the broad handler and re-raises, so execute_task''s timeout branch is reachable for the first time.
A trap surfaced while fixing it, and it is the interesting part. While the branch was unreachable, a timeout travelled the NORMAL path — which does update scheduled_tasks. Making the branch reachable without adding that write would have swapped a wrong status (''failed'') for a stale one (last_status showing the previous run''s outcome indefinitely). _update_task_outcome now mirrors terminal outcomes onto the parent row. A fix that repairs the reported symptom while breaking something adjacent is worse than the bug.
The timeout message also records that the work may still be running: after T-74 executors are handed to asyncio.to_thread, and a thread cannot be cancelled, so wait_for frees the loop while the work continues to completion. Saying "timed out" without that would imply the work stopped.
Mutation-checked: removing the narrow except clause fails the propagation test. A third test pins that ordinary exceptions are still returned rather than raised, so the narrow clause cannot start swallowing anything else.', NULL, '2026-08-11 10:11:39', '2026-08-11 10:11:39.291', '2026-08-11 10:11:39.291', NULL, '8bdb93fdcfbc234e18f3d034ddf25277', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQBTCRCTXK6EJF0T8CS2C', 'status', 'backlog', 'done', NULL, '2026-08-11 10:11:39', '2026-08-11 10:11:39.421', '2026-08-11 10:11:39.421', NULL, '7bc24ec03ed40ccacd71d7c7269d05af', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0PNFBDB743QJQ9MF38MPJ8', 'description', NULL, 'DONE in v1.7.0 (538bfe5). Verified on a real row:
12:24:27 ok task=backup_portainer_daily source=scheduler/portainer_backup_executor
WHY, and why not the other fix. A row written by a failing probe on 2026-08-11 was byte-for-byte what a nightly backup failure would have written — same source, same domain, nothing naming the task. The reflex was to delete the inconvenient row; that was refused as audit tampering and the refusal was right twice over, because the row is TRUE (a run did fail) and the defect is that it cannot say which run. Attribution keeps the record intact and makes it answer the question.
MECHANISM. A ContextVar (src/task_context.py) set by the engine around executor invocation, read by health_report. Not an argument: executors are invoked as execute(config, settings), there are ten, and several are dormant — they exist only as a string in a database row and become live the moment someone inserts a task naming them. A signature change would leave those broken in a way no import, grep or test would reveal. Not injected into `config` either: config is what a human wrote in the task definition and an executor may legitimately reject unknown keys.
TWO PROPERTIES VERIFIED IN THE DEPLOYED RUNTIME rather than assumed:
- asyncio.to_thread propagates context, so attribution survives the worker thread T-74 introduced. Had it not, every row from a real executor would have lost its task while unit tests kept passing — hence a test that goes through report_async specifically.
- Each asyncio Task gets its own copy, so five concurrent executions cannot read each other''s. The isolation test yields mid-execution to force interleaving; without that it would pass against a shared global.
- A plain await does NOT get its own copy and leaks to the caller. Both entry points use create_task, but task_scope resets by token rather than relying on it.
Omitted rather than nulled when absent — report() is callable from a script, and a null would claim a task existed with no name.
The one pre-existing critical row still carries no task, which now identifies it: it is the only backup row without the field, so it is legible as pre-attribution rather than ambiguous.', NULL, '2026-08-11 10:25:02', '2026-08-11 10:25:02.952', '2026-08-11 10:25:02.952', NULL, 'ede9b41ea9849e08202b7aae16e9a54c', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0PNFBDB743QJQ9MF38MPJ8', 'status', 'backlog', 'done', NULL, '2026-08-11 10:25:03', '2026-08-11 10:25:03.085', '2026-08-11 10:25:03.085', NULL, 'b45fb340bf78c6c7e4287df001d01456', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0FHPBBXXKBRWGR3FJZRK58', 'description', 'DELETE /tasks/{task_name} returns 500 whenever the task has at least one row in task_executions:
psycopg2.errors.ForeignKeyViolation: update or delete on table "scheduled_tasks"
violates foreign key constraint "task_executions_task_id_fkey" on table "task_executions"
DETAIL: Key (id)=(46) is still referenced from table "task_executions".
src/main.py:355. Since every task that has ever fired has execution history, the endpoint works only for tasks that have never run — which is close to none of them. Found on 2026-08-11 while cleaning up a temporary probe task created for T-74; it had to be removed with hand-written SQL against two tables, which is not something the API should require.
The caller gets a bare "Internal Server Error" with no indication that history is the obstacle, so it reads as the service being broken rather than the request being refusable.
Deciding what delete should MEAN is the actual work here, and it should not be guessed:
- cascade — drop the execution history with the task. Simple, and silently destroys the audit trail for a task someone deletes by mistake.
- soft delete — mark it deleted and keep the history. Keeps the audit trail, adds a state every query then has to filter on.
- refuse with 409 and a real message ("task has N executions; pass ?purge=true"). Explicit, and makes the destructive variant a deliberate act.
The third is the smallest correct change and matches how the rest of this system treats destructive operations. Whichever is chosen, a 500 on a foreseeable, well-defined condition is the part that is simply wrong.', 'DELETE /tasks/{task_name} returns 500 whenever the task has at least one row in task_executions:
psycopg2.errors.ForeignKeyViolation: update or delete on table "scheduled_tasks"
violates foreign key constraint "task_executions_task_id_fkey" on table "task_executions"
DETAIL: Key (id)=(46) is still referenced from table "task_executions".
src/main.py:355. Since every task that has ever fired has execution history, the endpoint works only for tasks that have never run — which is close to none of them. Found on 2026-08-11 while cleaning up a temporary probe task created for T-74; it had to be removed with hand-written SQL against two tables, which is not something the API should require.
The caller gets a bare "Internal Server Error" with no indication that history is the obstacle, so it reads as the service being broken rather than the request being refusable.
Deciding what delete should MEAN is the actual work here, and it should not be guessed:
- cascade — drop the execution history with the task. Simple, and silently destroys the audit trail for a task someone deletes by mistake.
- soft delete — mark it deleted and keep the history. Keeps the audit trail, adds a state every query then has to filter on.
- refuse with 409 and a real message ("task has N executions; pass ?purge=true"). Explicit, and makes the destructive variant a deliberate act.
The third is the smallest correct change and matches how the rest of this system treats destructive operations. Whichever is chosen, a 500 on a foreseeable, well-defined condition is the part that is simply wrong.
FIXED in v1.8.0 (c34db66). 409 with a message that can be acted on; ?purge=true proceeds.
Verified against the live service with a throwaway task that had one execution row:
DELETE /tasks/t1_delete_probe2 -> HTTP 409
Task ''t1_delete_probe2'' has 1 execution record(s). Deleting it would discard
that history. Re-send with ?purge=true to delete the task and its history
together, or PUT enabled=false to stop it running while keeping the record.
task still present afterwards: 1 (the refusal deleted nothing)
DELETE /tasks/t1_delete_probe2?purge=true -> HTTP 200
{"message":"...deleted successfully","executions_purged":1}
tasks: 24, probe execution rows: 0
REFUSE RATHER THAN CASCADE, chosen for asymmetry of recovery: a task definition can be recreated from the API in one call, its execution history cannot be recreated at all. Defaulting to the destructive reading of an ambiguous request is how audit trails disappear quietly.
The message carries the three things a caller needs — how much history is at stake, the flag that proceeds, and PUT enabled=false, which is usually what was actually wanted since it stops the task running and keeps the record. A bare "conflict" would be little better than the 500 it replaced.
History and task are deleted in ONE transaction. Split across two, a failure between them leaves the audit trail gone and the task alive: the worst of both outcomes.
Mutation-checked: removing the guard fails the refusal tests. Separate tests pin that a refused delete issues no DELETE at all, and that ?purge=true against a missing task is still 404 rather than a success.
NOTE: the commit for this work is missing its Co-Authored-By trailer — I wrote the message file without it. Already pushed, and fixing it would require rewriting published history on main, so it stands as-is.', NULL, '2026-08-11 10:31:46', '2026-08-11 10:31:46.465', '2026-08-11 10:31:46.465', NULL, 'efc9df5b52ac92d63e5ae5a60c2bdef8', 2) ON CONFLICT(hash) DO NOTHING;
INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0FHPBBXXKBRWGR3FJZRK58', 'status', 'backlog', 'done', NULL, '2026-08-11 10:31:46', '2026-08-11 10:31:46.598', '2026-08-11 10:31:46.598', NULL, '7bad13e92e65eb912519a498a882faee', 2) ON CONFLICT(hash) DO NOTHING;
+1
View File
@@ -1,3 +1,4 @@
INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0FHPBBXXKBRWGR3FJZRK58', 'T-1', '2026-08-11 09:53:56.835', '2026-08-11 09:53:56.835', NULL, '3d980945e785a6bc7ca8fcaa8250e22b', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= ticket_idmap.updated_at;
INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQAR5KXM3PTG0QT3R1PV8', 'T-2', '2026-08-11 09:59:05.153', '2026-08-11 09:59:05.153', NULL, 'c8655ec9601ba93fe822395329e62262', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= ticket_idmap.updated_at;
INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQBTCRCTXK6EJF0T8CS2C', 'T-3', '2026-08-11 09:59:05.427', '2026-08-11 09:59:05.427', NULL, '3bce28c3d803d3f5036cfbb1ac11969c', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= ticket_idmap.updated_at;
INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0PNFBDB743QJQ9MF38MPJ8', 'T-4', '2026-08-11 10:25:02.811', '2026-08-11 10:25:02.811', NULL, 'eb8c6e4088b39b797541c1be0ee313fd', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= ticket_idmap.updated_at;
+223
View File
@@ -48,3 +48,226 @@ So every timeout since 2025-12-07 has been recorded as a generic failure carryin
FIX: catch asyncio.TimeoutError explicitly in _run_executor, ahead of the broad handler, and let it propagate (or return a marker execute_task can distinguish). Note the ordering is the whole bug — a narrower except after a broader one is dead code, and there is no linter configured here to say so.
Related: a thread started by asyncio.to_thread cannot be cancelled, so after T-74 a timeout frees the loop while the work continues to completion. Whatever ''timeout'' comes to mean should say so rather than implying the work stopped.', 'backlog', 'medium', NULL, NULL, NULL, '2026-08-11 09:59:05.427', '2026-08-11 09:59:05.545', NULL, 'ae0db866086e38b681b0ea32837df277', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQAR5KXM3PTG0QT3R1PV8', 'bug', NULL, 'A restart mid-task unschedules that task forever, silently', 'get_tasks_for_minute (src/tasks/executor.py:70) excludes any task holding a task_executions row with status=''running'':
AND id NOT IN (SELECT task_id FROM task_executions WHERE status = ''running'')
Nothing ever reconciles that row. It is written before the executor runs and updated after, so a process that dies in between leaves it ''running'' permanently — and the task is then excluded from every future minute, forever, with no error, no alarm and no log line. The task simply stops existing as far as the scheduler is concerned.
This is not hypothetical. test_example_task has held a ''running'' row since 2025-12-07 — 5916 hours. It happens to be disabled, so nothing is broken today; the mechanism is what matters, not this instance.
The exposure is real and daily: Watchtower restarts this container every morning at 4 AM. Any task still running at that moment is permanently unscheduled by it. The config backup runs 03:05 and takes ~21 minutes, which is not far off.
The failure mode is the one this system keeps producing: it looks like nothing. A backup that stops running forever produces no failure — it produces silence, and silence reads as health.
FIX: reconcile at startup. Any row left ''running'' when the process starts cannot be running, because the process that owned it is gone. Mark those ''orphaned'' (or ''failed'' with a message naming the restart) during app startup, before the scheduler begins its first minute. Consider also a stale-row guard for rows older than the task''s timeout_seconds, which covers a killed worker without a restart.
The audit trail is preserved either way — this is a status correction, not a deletion.
FIXED in v1.6.0 (e45de4f). TaskExecutor.reconcile_orphaned_executions() runs in the lifespan before the scheduler starts.
Proven on the real system rather than in a test. The row that had been ''running'' since 2025-12-07 22:44 was released at startup:
WARNING - Orphaned execution recovered: test_example_task was left ''running''
since 2025-12-07 22:44:00.046695. That task had been excluded from
scheduling until now.
Rows still ''running'' afterwards: 0. Status is ''orphaned'', error names the restart.
''orphaned'' rather than ''failed'' because the outcome is genuinely unknown — a task that completed and never got to update its row looks exactly like one that died halfway, and ''failed'' would assert something nobody observed.
Ordering is part of the fix and is mutation-checked: reconciliation must complete before the first minute is processed, or the first tick still sees the stale rows. Removing the startup call fails the test.
Not extended to a duration-based sweep, deliberately. While the process lives, execute_task''s finally always closes the row out, so a stale row implies a dead owner; a time-based rule would have to distinguish a slow task from a dead one and would eventually close the record of a task that is still working.', 'backlog', 'high', NULL, NULL, NULL, '2026-08-11 09:59:05.153', '2026-08-11 10:11:39.058', NULL, '8d88d4684f803319131765d97772e64d', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQAR5KXM3PTG0QT3R1PV8', 'bug', NULL, 'A restart mid-task unschedules that task forever, silently', 'get_tasks_for_minute (src/tasks/executor.py:70) excludes any task holding a task_executions row with status=''running'':
AND id NOT IN (SELECT task_id FROM task_executions WHERE status = ''running'')
Nothing ever reconciles that row. It is written before the executor runs and updated after, so a process that dies in between leaves it ''running'' permanently — and the task is then excluded from every future minute, forever, with no error, no alarm and no log line. The task simply stops existing as far as the scheduler is concerned.
This is not hypothetical. test_example_task has held a ''running'' row since 2025-12-07 — 5916 hours. It happens to be disabled, so nothing is broken today; the mechanism is what matters, not this instance.
The exposure is real and daily: Watchtower restarts this container every morning at 4 AM. Any task still running at that moment is permanently unscheduled by it. The config backup runs 03:05 and takes ~21 minutes, which is not far off.
The failure mode is the one this system keeps producing: it looks like nothing. A backup that stops running forever produces no failure — it produces silence, and silence reads as health.
FIX: reconcile at startup. Any row left ''running'' when the process starts cannot be running, because the process that owned it is gone. Mark those ''orphaned'' (or ''failed'' with a message naming the restart) during app startup, before the scheduler begins its first minute. Consider also a stale-row guard for rows older than the task''s timeout_seconds, which covers a killed worker without a restart.
The audit trail is preserved either way — this is a status correction, not a deletion.
FIXED in v1.6.0 (e45de4f). TaskExecutor.reconcile_orphaned_executions() runs in the lifespan before the scheduler starts.
Proven on the real system rather than in a test. The row that had been ''running'' since 2025-12-07 22:44 was released at startup:
WARNING - Orphaned execution recovered: test_example_task was left ''running''
since 2025-12-07 22:44:00.046695. That task had been excluded from
scheduling until now.
Rows still ''running'' afterwards: 0. Status is ''orphaned'', error names the restart.
''orphaned'' rather than ''failed'' because the outcome is genuinely unknown — a task that completed and never got to update its row looks exactly like one that died halfway, and ''failed'' would assert something nobody observed.
Ordering is part of the fix and is mutation-checked: reconciliation must complete before the first minute is processed, or the first tick still sees the stale rows. Removing the startup call fails the test.
Not extended to a duration-based sweep, deliberately. While the process lives, execute_task''s finally always closes the row out, so a stale row implies a dead owner; a time-based rule would have to distinguish a slow task from a dead one and would eventually close the record of a task that is still working.', 'done', 'high', NULL, NULL, NULL, '2026-08-11 09:59:05.153', '2026-08-11 10:11:39.177', NULL, '99284d5f4a53a35637ee10c7b6b66c6e', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQBTCRCTXK6EJF0T8CS2C', 'bug', NULL, 'The ''timeout'' execution status is unreachable; timeouts are filed as generic failures', 'execute_task has an `except asyncio.TimeoutError` handler (src/tasks/executor.py:179) that records status=''timeout''. It can never run.
_run_executor wraps the awaited call in `except Exception` (line 219) and returns the error as a value. Since Python 3.11 asyncio.TimeoutError IS the builtin TimeoutError, which inherits OSError -> Exception, so the broad handler catches it first and converts it into an ordinary (None, error) tuple. execute_task then sees a non-None error and files status=''failed''.
Verified in the deployed runtime:
asyncio.TimeoutError is TimeoutError: True
MRO: TimeoutError -> OSError -> Exception -> BaseException
And confirmed against 8 months of history — 18,785 executions, and the count of status=''timeout'' rows is zero:
success 16690 | failed 2093 | running 2 | timeout 0
So every timeout since 2025-12-07 has been recorded as a generic failure carrying a traceback. Operationally that erases the distinction that matters most when a job misbehaves: "this job is too slow for its window" and "this job is broken" need different responses, and right now they look identical in the history.
FIX: catch asyncio.TimeoutError explicitly in _run_executor, ahead of the broad handler, and let it propagate (or return a marker execute_task can distinguish). Note the ordering is the whole bug — a narrower except after a broader one is dead code, and there is no linter configured here to say so.
Related: a thread started by asyncio.to_thread cannot be cancelled, so after T-74 a timeout frees the loop while the work continues to completion. Whatever ''timeout'' comes to mean should say so rather than implying the work stopped.
FIXED in v1.6.0 (e45de4f). _run_executor now catches asyncio.TimeoutError ahead of the broad handler and re-raises, so execute_task''s timeout branch is reachable for the first time.
A trap surfaced while fixing it, and it is the interesting part. While the branch was unreachable, a timeout travelled the NORMAL path — which does update scheduled_tasks. Making the branch reachable without adding that write would have swapped a wrong status (''failed'') for a stale one (last_status showing the previous run''s outcome indefinitely). _update_task_outcome now mirrors terminal outcomes onto the parent row. A fix that repairs the reported symptom while breaking something adjacent is worse than the bug.
The timeout message also records that the work may still be running: after T-74 executors are handed to asyncio.to_thread, and a thread cannot be cancelled, so wait_for frees the loop while the work continues to completion. Saying "timed out" without that would imply the work stopped.
Mutation-checked: removing the narrow except clause fails the propagation test. A third test pins that ordinary exceptions are still returned rather than raised, so the narrow clause cannot start swallowing anything else.', 'backlog', 'medium', NULL, NULL, NULL, '2026-08-11 09:59:05.427', '2026-08-11 10:11:39.291', NULL, 'b7379fa662b40c6996f17c36b64010c0', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0GQBTCRCTXK6EJF0T8CS2C', 'bug', NULL, 'The ''timeout'' execution status is unreachable; timeouts are filed as generic failures', 'execute_task has an `except asyncio.TimeoutError` handler (src/tasks/executor.py:179) that records status=''timeout''. It can never run.
_run_executor wraps the awaited call in `except Exception` (line 219) and returns the error as a value. Since Python 3.11 asyncio.TimeoutError IS the builtin TimeoutError, which inherits OSError -> Exception, so the broad handler catches it first and converts it into an ordinary (None, error) tuple. execute_task then sees a non-None error and files status=''failed''.
Verified in the deployed runtime:
asyncio.TimeoutError is TimeoutError: True
MRO: TimeoutError -> OSError -> Exception -> BaseException
And confirmed against 8 months of history — 18,785 executions, and the count of status=''timeout'' rows is zero:
success 16690 | failed 2093 | running 2 | timeout 0
So every timeout since 2025-12-07 has been recorded as a generic failure carrying a traceback. Operationally that erases the distinction that matters most when a job misbehaves: "this job is too slow for its window" and "this job is broken" need different responses, and right now they look identical in the history.
FIX: catch asyncio.TimeoutError explicitly in _run_executor, ahead of the broad handler, and let it propagate (or return a marker execute_task can distinguish). Note the ordering is the whole bug — a narrower except after a broader one is dead code, and there is no linter configured here to say so.
Related: a thread started by asyncio.to_thread cannot be cancelled, so after T-74 a timeout frees the loop while the work continues to completion. Whatever ''timeout'' comes to mean should say so rather than implying the work stopped.
FIXED in v1.6.0 (e45de4f). _run_executor now catches asyncio.TimeoutError ahead of the broad handler and re-raises, so execute_task''s timeout branch is reachable for the first time.
A trap surfaced while fixing it, and it is the interesting part. While the branch was unreachable, a timeout travelled the NORMAL path — which does update scheduled_tasks. Making the branch reachable without adding that write would have swapped a wrong status (''failed'') for a stale one (last_status showing the previous run''s outcome indefinitely). _update_task_outcome now mirrors terminal outcomes onto the parent row. A fix that repairs the reported symptom while breaking something adjacent is worse than the bug.
The timeout message also records that the work may still be running: after T-74 executors are handed to asyncio.to_thread, and a thread cannot be cancelled, so wait_for frees the loop while the work continues to completion. Saying "timed out" without that would imply the work stopped.
Mutation-checked: removing the narrow except clause fails the propagation test. A third test pins that ordinary exceptions are still returned rather than raised, so the narrow clause cannot start swallowing anything else.', 'done', 'medium', NULL, NULL, NULL, '2026-08-11 09:59:05.427', '2026-08-11 10:11:39.420', NULL, '333edb040cfa995cf26f45b6fa2cd2f9', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0PNFBDB743QJQ9MF38MPJ8', 'task', NULL, 'check_history rows name the task that produced them', NULL, 'backlog', 'medium', NULL, NULL, NULL, '2026-08-11 10:25:02.811', '2026-08-11 10:25:02.811', NULL, 'b099197d1cbdea7e682a054dc17788ab', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0PNFBDB743QJQ9MF38MPJ8', 'task', NULL, 'check_history rows name the task that produced them', 'DONE in v1.7.0 (538bfe5). Verified on a real row:
12:24:27 ok task=backup_portainer_daily source=scheduler/portainer_backup_executor
WHY, and why not the other fix. A row written by a failing probe on 2026-08-11 was byte-for-byte what a nightly backup failure would have written — same source, same domain, nothing naming the task. The reflex was to delete the inconvenient row; that was refused as audit tampering and the refusal was right twice over, because the row is TRUE (a run did fail) and the defect is that it cannot say which run. Attribution keeps the record intact and makes it answer the question.
MECHANISM. A ContextVar (src/task_context.py) set by the engine around executor invocation, read by health_report. Not an argument: executors are invoked as execute(config, settings), there are ten, and several are dormant — they exist only as a string in a database row and become live the moment someone inserts a task naming them. A signature change would leave those broken in a way no import, grep or test would reveal. Not injected into `config` either: config is what a human wrote in the task definition and an executor may legitimately reject unknown keys.
TWO PROPERTIES VERIFIED IN THE DEPLOYED RUNTIME rather than assumed:
- asyncio.to_thread propagates context, so attribution survives the worker thread T-74 introduced. Had it not, every row from a real executor would have lost its task while unit tests kept passing — hence a test that goes through report_async specifically.
- Each asyncio Task gets its own copy, so five concurrent executions cannot read each other''s. The isolation test yields mid-execution to force interleaving; without that it would pass against a shared global.
- A plain await does NOT get its own copy and leaks to the caller. Both entry points use create_task, but task_scope resets by token rather than relying on it.
Omitted rather than nulled when absent — report() is callable from a script, and a null would claim a task existed with no name.
The one pre-existing critical row still carries no task, which now identifies it: it is the only backup row without the field, so it is legible as pre-attribution rather than ambiguous.', 'backlog', 'medium', NULL, NULL, NULL, '2026-08-11 10:25:02.811', '2026-08-11 10:25:02.951', NULL, '5b7bbb05cf6aedfbd0a6a49a356f6e6f', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0PNFBDB743QJQ9MF38MPJ8', 'task', NULL, 'check_history rows name the task that produced them', 'DONE in v1.7.0 (538bfe5). Verified on a real row:
12:24:27 ok task=backup_portainer_daily source=scheduler/portainer_backup_executor
WHY, and why not the other fix. A row written by a failing probe on 2026-08-11 was byte-for-byte what a nightly backup failure would have written — same source, same domain, nothing naming the task. The reflex was to delete the inconvenient row; that was refused as audit tampering and the refusal was right twice over, because the row is TRUE (a run did fail) and the defect is that it cannot say which run. Attribution keeps the record intact and makes it answer the question.
MECHANISM. A ContextVar (src/task_context.py) set by the engine around executor invocation, read by health_report. Not an argument: executors are invoked as execute(config, settings), there are ten, and several are dormant — they exist only as a string in a database row and become live the moment someone inserts a task naming them. A signature change would leave those broken in a way no import, grep or test would reveal. Not injected into `config` either: config is what a human wrote in the task definition and an executor may legitimately reject unknown keys.
TWO PROPERTIES VERIFIED IN THE DEPLOYED RUNTIME rather than assumed:
- asyncio.to_thread propagates context, so attribution survives the worker thread T-74 introduced. Had it not, every row from a real executor would have lost its task while unit tests kept passing — hence a test that goes through report_async specifically.
- Each asyncio Task gets its own copy, so five concurrent executions cannot read each other''s. The isolation test yields mid-execution to force interleaving; without that it would pass against a shared global.
- A plain await does NOT get its own copy and leaks to the caller. Both entry points use create_task, but task_scope resets by token rather than relying on it.
Omitted rather than nulled when absent — report() is callable from a script, and a null would claim a task existed with no name.
The one pre-existing critical row still carries no task, which now identifies it: it is the only backup row without the field, so it is legible as pre-attribution rather than ambiguous.', 'done', 'medium', NULL, NULL, NULL, '2026-08-11 10:25:02.811', '2026-08-11 10:25:03.084', NULL, '63f6b48e753891a75d47dd0b3d7b5a4c', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0FHPBBXXKBRWGR3FJZRK58', 'bug', NULL, 'DELETE /tasks/{name} 500s for any task that has ever run', 'DELETE /tasks/{task_name} returns 500 whenever the task has at least one row in task_executions:
psycopg2.errors.ForeignKeyViolation: update or delete on table "scheduled_tasks"
violates foreign key constraint "task_executions_task_id_fkey" on table "task_executions"
DETAIL: Key (id)=(46) is still referenced from table "task_executions".
src/main.py:355. Since every task that has ever fired has execution history, the endpoint works only for tasks that have never run — which is close to none of them. Found on 2026-08-11 while cleaning up a temporary probe task created for T-74; it had to be removed with hand-written SQL against two tables, which is not something the API should require.
The caller gets a bare "Internal Server Error" with no indication that history is the obstacle, so it reads as the service being broken rather than the request being refusable.
Deciding what delete should MEAN is the actual work here, and it should not be guessed:
- cascade — drop the execution history with the task. Simple, and silently destroys the audit trail for a task someone deletes by mistake.
- soft delete — mark it deleted and keep the history. Keeps the audit trail, adds a state every query then has to filter on.
- refuse with 409 and a real message ("task has N executions; pass ?purge=true"). Explicit, and makes the destructive variant a deliberate act.
The third is the smallest correct change and matches how the rest of this system treats destructive operations. Whichever is chosen, a 500 on a foreseeable, well-defined condition is the part that is simply wrong.
FIXED in v1.8.0 (c34db66). 409 with a message that can be acted on; ?purge=true proceeds.
Verified against the live service with a throwaway task that had one execution row:
DELETE /tasks/t1_delete_probe2 -> HTTP 409
Task ''t1_delete_probe2'' has 1 execution record(s). Deleting it would discard
that history. Re-send with ?purge=true to delete the task and its history
together, or PUT enabled=false to stop it running while keeping the record.
task still present afterwards: 1 (the refusal deleted nothing)
DELETE /tasks/t1_delete_probe2?purge=true -> HTTP 200
{"message":"...deleted successfully","executions_purged":1}
tasks: 24, probe execution rows: 0
REFUSE RATHER THAN CASCADE, chosen for asymmetry of recovery: a task definition can be recreated from the API in one call, its execution history cannot be recreated at all. Defaulting to the destructive reading of an ambiguous request is how audit trails disappear quietly.
The message carries the three things a caller needs — how much history is at stake, the flag that proceeds, and PUT enabled=false, which is usually what was actually wanted since it stops the task running and keeps the record. A bare "conflict" would be little better than the 500 it replaced.
History and task are deleted in ONE transaction. Split across two, a failure between them leaves the audit trail gone and the task alive: the worst of both outcomes.
Mutation-checked: removing the guard fails the refusal tests. Separate tests pin that a refused delete issues no DELETE at all, and that ?purge=true against a missing task is still 404 rather than a success.
NOTE: the commit for this work is missing its Co-Authored-By trailer — I wrote the message file without it. Already pushed, and fixing it would require rewriting published history on main, so it stands as-is.', 'backlog', 'high', NULL, NULL, NULL, '2026-08-11 09:53:56.826', '2026-08-11 10:31:46.465', NULL, '9299b6bccd53cdc77e2551e2de94a10e', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FZ0FHPBBXXKBRWGR3FJZRK58', 'bug', NULL, 'DELETE /tasks/{name} 500s for any task that has ever run', 'DELETE /tasks/{task_name} returns 500 whenever the task has at least one row in task_executions:
psycopg2.errors.ForeignKeyViolation: update or delete on table "scheduled_tasks"
violates foreign key constraint "task_executions_task_id_fkey" on table "task_executions"
DETAIL: Key (id)=(46) is still referenced from table "task_executions".
src/main.py:355. Since every task that has ever fired has execution history, the endpoint works only for tasks that have never run — which is close to none of them. Found on 2026-08-11 while cleaning up a temporary probe task created for T-74; it had to be removed with hand-written SQL against two tables, which is not something the API should require.
The caller gets a bare "Internal Server Error" with no indication that history is the obstacle, so it reads as the service being broken rather than the request being refusable.
Deciding what delete should MEAN is the actual work here, and it should not be guessed:
- cascade — drop the execution history with the task. Simple, and silently destroys the audit trail for a task someone deletes by mistake.
- soft delete — mark it deleted and keep the history. Keeps the audit trail, adds a state every query then has to filter on.
- refuse with 409 and a real message ("task has N executions; pass ?purge=true"). Explicit, and makes the destructive variant a deliberate act.
The third is the smallest correct change and matches how the rest of this system treats destructive operations. Whichever is chosen, a 500 on a foreseeable, well-defined condition is the part that is simply wrong.
FIXED in v1.8.0 (c34db66). 409 with a message that can be acted on; ?purge=true proceeds.
Verified against the live service with a throwaway task that had one execution row:
DELETE /tasks/t1_delete_probe2 -> HTTP 409
Task ''t1_delete_probe2'' has 1 execution record(s). Deleting it would discard
that history. Re-send with ?purge=true to delete the task and its history
together, or PUT enabled=false to stop it running while keeping the record.
task still present afterwards: 1 (the refusal deleted nothing)
DELETE /tasks/t1_delete_probe2?purge=true -> HTTP 200
{"message":"...deleted successfully","executions_purged":1}
tasks: 24, probe execution rows: 0
REFUSE RATHER THAN CASCADE, chosen for asymmetry of recovery: a task definition can be recreated from the API in one call, its execution history cannot be recreated at all. Defaulting to the destructive reading of an ambiguous request is how audit trails disappear quietly.
The message carries the three things a caller needs — how much history is at stake, the flag that proceeds, and PUT enabled=false, which is usually what was actually wanted since it stops the task running and keeps the record. A bare "conflict" would be little better than the 500 it replaced.
History and task are deleted in ONE transaction. Split across two, a failure between them leaves the audit trail gone and the task alive: the worst of both outcomes.
Mutation-checked: removing the guard fails the refusal tests. Separate tests pin that a refused delete issues no DELETE at all, and that ?purge=true against a missing task is still 404 rather than a success.
NOTE: the commit for this work is missing its Co-Authored-By trailer — I wrote the message file without it. Already pushed, and fixing it would require rewriting published history on main, so it stands as-is.', 'done', 'high', NULL, NULL, NULL, '2026-08-11 09:53:56.826', '2026-08-11 10:31:46.597', NULL, 'e020f77c66ce6631f6d05ee63ec0c2b8', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at >= tickets.updated_at;
+24
View File
@@ -6,6 +6,30 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
## [Unreleased]
## [1.9.0] - 2026-08-11
### Changed
- `check_history` rows carry `summary` at the top level, beside `status`, matching the other
writer of that table. It was nested under `metrics`, so a reader had to know which producer
wrote a row to find its substance. Old rows keep the nested spelling.
## [1.8.0] - 2026-08-11
### Fixed
- Deleting a task that has run returns 409 instead of a bare 500. It failed on a foreign key
against its own execution history, which the error never mentioned.
### Added
- `DELETE /tasks/{name}?purge=true` removes a task together with its execution history, in one
transaction. The response reports `executions_purged`.
## [1.7.0] - 2026-08-11
### Added
- `check_history` rows name the scheduled task that produced them. `source` names the code,
which cannot distinguish two tasks sharing one executor — so a failure could not be
attributed to the job that caused it. Omitted when there is no task.
## [1.6.0] - 2026-08-11
### Fixed
+1 -1
View File
@@ -9,7 +9,7 @@ It is the homelab's cron. Recurring work belongs here rather than in a systemd t
## Live contract
`http://localhost:8090/openapi.json` — 10 paths, `version: 1.6.0` (verified 2026-08-11). Human
`http://localhost:8090/openapi.json` — 10 paths, `version: 1.9.0` (verified 2026-08-11). Human
docs at `/docs`. Generated from running code, so read it instead of inferring routes.
**Every route except `/health` requires `Authorization: Bearer $SCHEDULER_API_KEY`.** An
+41 -3
View File
@@ -17,14 +17,52 @@ help: ## Show this help
| awk 'BEGIN{FS=":.*?## "}{printf " \033[36m%-14s\033[0m %s\n", $$1, $$2}'
.PHONY: setup
setup: ## Create the venv and install the test extra
setup: ## Create the venv, install the test extra, and prove it actually works
$(PYTHON) -m venv .venv
$(VENV)/bin/pip install -e ".[test]"
@# Exit 0 from `pip install` is not evidence (D-24) — pip reports success even
@# when the result is unusable. Prove the environment works instead of trusting
@# the install step: `--collect-only` imports every test module and therefore
@# every src module each one pulls in, which is exactly the failure mode this
@# target exists to catch (T-47 — this repo had no venv at all on 2026-08-09,
@# and the documented test command could not work). It runs zero tests, so it
@# stays cheap, and unlike a bare `import src.main` it exercises the tests/
@# tree too, not just the package.
$(VENV)/bin/python -m pytest tests/ --collect-only -q
.PHONY: test
test: ## Run the test suite
test: ## Run the test suite — hermetic, no live services (D-26)
@test -x $(VENV)/bin/python || { echo "FAIL — no venv in this tree; run: make setup"; exit 69; }
$(VENV)/bin/python -m pytest tests/
# Integration tests are deselected here, not skipped by accident. Twelve
# tests in test_database_integration.py carry @pytest.mark.integration and
# need a real Postgres; they errored on every run of this target because
# nothing deselected them, and the marker had no target to select it either.
# So they neither passed nor ran — they just made `make test` exit 2 forever,
# which trains a reader to ignore the exit code (T-55).
#
# They were invisible to the netns audit that found the rest of this: they
# fail identically with and without a network, because postgres-shared is a
# Docker-internal name a host process cannot resolve in either case. A
# namespace proves a test does not reach the network; it cannot tell that
# apart from a test whose dependency is unreachable anyway.
$(VENV)/bin/python -m pytest tests/ -m "not integration"
.PHONY: test-integration
test-integration: ## Run only the tests that need live Postgres/Redis
@test -x $(VENV)/bin/python || { echo "FAIL — no venv in this tree; run: make setup"; exit 69; }
# Refuses an empty selection. A target that passes because it selected
# nothing is the defect this repo keeps meeting from the other side, so
# pytest's exit 5 (no tests collected) is a failure with its own message,
# and a collection error gets a different one — "nothing to run" must never
# read as "everything passed" (D-24).
@$(VENV)/bin/python -m pytest tests/ -m integration --collect-only -q >/dev/null 2>&1; \
rc=$$?; \
if [ $$rc -eq 5 ]; then \
echo "FAIL test-integration — selected 0 tests (marker renamed, moved, or lost — this is a defect, not a pass)"; exit 1; \
elif [ $$rc -ne 0 ]; then \
echo "FAIL test-integration — collection errored (rc=$$rc)"; exit 1; \
fi
$(VENV)/bin/python -m pytest tests/ -m integration -v
# No `lint` target, deliberately. CLAUDE.md states it outright: no linter is
# configured, no ruff or flake8 config, neither in the dependencies. Per D-27
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "the-scheduler"
version = "1.6.0"
version = "1.9.0"
description = "System-wide maintenance orchestration - backups, doc mirroring, cleanup, task automation"
readme = "README.md"
requires-python = ">=3.12"
+3 -1
View File
@@ -202,6 +202,7 @@ async def execute(config: dict, settings: Settings) -> str:
domain="backup",
status=health_report.CRITICAL,
source="scheduler/config_backup_executor",
summary=f"backup failed: {str(exc)[:300]}",
metrics={"job": "scheduler/config_backup_executor", "error": str(exc)[:400]},
)
raise
@@ -210,6 +211,7 @@ async def execute(config: dict, settings: Settings) -> str:
domain="backup",
status=health_report.OK,
source="scheduler/config_backup_executor",
metrics={"job": "scheduler/config_backup_executor", "summary": output[:400]},
summary=output[:400],
metrics={"job": "scheduler/config_backup_executor"},
)
return output
+32 -2
View File
@@ -16,10 +16,19 @@ Recorded as D-33 in the workspace vault: `check_history` is the central health
record and any self-maintained service may push a row describing its own
outcome. sysmon polls only the things that cannot report themselves.
Three consequences that are load-bearing here:
The properties that are load-bearing here — a count is not given, because
this list has grown twice and a stale number is worse than none:
- `source` names the producer, because the table now has several writers and a
row must say which one wrote it.
- `summary` sits at the top level, beside `status`, because that is where the
other writer puts it. One spelling per fact, or a reader has to know which
producer wrote a row before it can find out what the row says.
- `task` names the schedule that invoked it, which `source` cannot: two tasks
may share one executor. On 2026-08-11 two did, and their rows were identical
apart from their contents — a failure could not be attributed to either. It
comes from a ContextVar the engine sets (`src/task_context.py`), so executors
need no signature change and dormant ones stay valid.
- `domain` is a shared namespace. Two producers claiming one name would
interleave silently.
- **A reporting failure must never fail the task.** Backing up successfully and
@@ -36,6 +45,8 @@ from typing import Any, Dict, Optional
import psycopg2
from src.task_context import current_task_name
logger = logging.getLogger(__name__)
# The database holding check_history. Not the scheduler's own database — this is
@@ -52,6 +63,7 @@ def report(
domain: str,
status: str,
source: str,
summary: str,
metrics: Optional[Dict[str, Any]] = None,
) -> bool:
"""Write one row to check_history. Returns whether it landed.
@@ -61,6 +73,7 @@ def report(
"""
metrics = metrics or {}
now = datetime.now(timezone.utc)
task = current_task_name()
result = {
# The envelope the table has carried since the shell era. A reader of a
# year of history should not have to know which producer wrote a row in
@@ -69,8 +82,24 @@ def report(
"source": source,
"domain": domain,
"status": status,
# Top level, beside status — the same place sysmon-go writes it. It lived
# under metrics until 2026-08-11, so the two writers of this shared table
# disagreed about where the substance of a row was, and any query written
# the obvious way found one and missed the other. That is the T-36 shape
# exactly: per-domain queries returned nothing because the data was
# nested somewhere else. D-33 made this table a contract between
# producers; a contract needs one spelling.
"summary": summary,
"metrics": metrics,
}
# Which scheduled task produced this. `source` names the code; two tasks can
# share one executor, and on 2026-08-11 two did — a 425 MB probe and the 5 GB
# nightly backup wrote rows that were identical apart from their contents, so
# a failure could not be attributed to either. Omitted rather than nulled
# when absent, per the convention that a missing field means "not
# applicable": report() is also callable from a script with no task around it.
if task:
result["task"] = task
try:
conn = psycopg2.connect(
@@ -128,6 +157,7 @@ async def report_async(
domain: str,
status: str,
source: str,
summary: str,
metrics: Optional[Dict[str, Any]] = None,
) -> bool:
"""`report` for callers on the event loop. Prefer this one inside executors.
@@ -139,7 +169,7 @@ async def report_async(
`/health` from that loop, so the cost of a slow report is the whole service
appearing down (T-74).
"""
return await asyncio.to_thread(report, settings, domain, status, source, metrics)
return await asyncio.to_thread(report, settings, domain, status, source, summary, metrics)
def _host() -> str:
+3 -1
View File
@@ -158,6 +158,7 @@ async def execute(config: dict, settings: Settings) -> str:
domain="backup",
status=health_report.CRITICAL,
source="scheduler/portainer_backup_executor",
summary=f"backup failed: {str(exc)[:300]}",
metrics={"job": "scheduler/portainer_backup_executor", "error": str(exc)[:400]},
)
raise
@@ -166,6 +167,7 @@ async def execute(config: dict, settings: Settings) -> str:
domain="backup",
status=health_report.OK,
source="scheduler/portainer_backup_executor",
metrics={"job": "scheduler/portainer_backup_executor", "summary": output[:400]},
summary=output[:400],
metrics={"job": "scheduler/portainer_backup_executor"},
)
return output
+53 -12
View File
@@ -7,7 +7,7 @@ Architecture: Hybrid APScheduler + DB-based priority system
- Job queries DB for tasks scheduled in that minute
- Executes up to 5 tasks concurrently based on priority
"""
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi import FastAPI, HTTPException, Depends, Header, Query
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.triggers.cron import CronTrigger
@@ -353,26 +353,67 @@ async def update_task(
@app.delete("/tasks/{task_name}")
async def delete_task(
task_name: str,
purge: bool = Query(
False,
description="Also delete this task's execution history. Required when the "
"task has ever run, and destroys its audit trail."
),
api_key: str = Depends(verify_api_key),
executor: TaskExecutor = Depends(get_task_executor)
):
"""Delete a scheduled task."""
"""Delete a scheduled task.
A task that has ever run owns rows in task_executions, and those rows are
the audit trail — when it ran, how long it took, what it returned. Deleting
the task alone violates task_executions_task_id_fkey, which surfaced as a
bare 500 with no indication that history was the obstacle, so it read as the
service being broken rather than the request being refusable. Since every
task that has ever fired has history, the endpoint effectively worked only
for tasks that had never run.
Refusing with 409 rather than cascading by default, because the two outcomes
are not equally recoverable: a task definition can be recreated from the API
in one call, its execution history cannot be recreated at all. The caller
who wants both gone says so.
Disabling is usually what was actually wanted — it stops the task running and
keeps the record — so the refusal names that too.
"""
with executor.get_db_connection() as conn:
with conn.cursor() as cur:
cur.execute("""
DELETE FROM scheduled_tasks
WHERE task_name = %s
RETURNING task_name
""", (task_name,))
deleted = cur.fetchone()
if not deleted:
cur.execute("SELECT id FROM scheduled_tasks WHERE task_name = %s", (task_name,))
row = cur.fetchone()
if not row:
raise HTTPException(404, f"Task '{task_name}' not found")
task_id = row[0]
cur.execute("SELECT COUNT(*) FROM task_executions WHERE task_id = %s", (task_id,))
executions = cur.fetchone()[0]
if executions and not purge:
raise HTTPException(
409,
f"Task '{task_name}' has {executions} execution record(s). "
f"Deleting it would discard that history. Re-send with "
f"?purge=true to delete the task and its history together, or "
f"PUT enabled=false to stop it running while keeping the record."
)
# One transaction: a purge that removed the history and then failed to
# remove the task would leave the audit trail gone and the task alive.
if executions:
cur.execute("DELETE FROM task_executions WHERE task_id = %s", (task_id,))
cur.execute("DELETE FROM scheduled_tasks WHERE id = %s", (task_id,))
conn.commit()
logger.info(f"Deleted task: {task_name}")
return {"message": f"Task '{task_name}' deleted successfully"}
if executions:
logger.warning(f"Deleted task {task_name} and purged {executions} execution record(s)")
else:
logger.info(f"Deleted task: {task_name}")
return {
"message": f"Task '{task_name}' deleted successfully",
"executions_purged": executions,
}
@app.post("/tasks/{task_name}/trigger")
async def trigger_task(
+59
View File
@@ -0,0 +1,59 @@
"""Which scheduled task is currently executing.
Executors are invoked as `execute(config, settings)` and are never told which
task they are. That is fine until one of them writes to a shared record: on
2026-08-11 a probe task and the nightly backup both used
config_backup_executor, and the rows they wrote into `check_history` were
indistinguishable same `source`, same `domain`. A failure from a 425 MB test
archive was impossible to tell from a failure of the 5 GB nightly job, so the
health record could not answer "which one broke?".
A ContextVar rather than a parameter, because the alternatives are worse here:
- Threading a `task` argument through `execute(config, settings, task)` is a
signature change across every executor, including the dormant ones that
exist only as a string in a database row and would break the moment
somebody enabled them.
- Injecting the name into `config` corrupts the thing executors validate.
`config` is what a human wrote in the task definition, and an executor is
entitled to reject keys it does not recognise.
Two properties make this safe, both verified in the deployed runtime rather
than assumed:
- `asyncio.to_thread` propagates the context, so a reporter still sees the
task after T-74 moved executor bodies into worker threads.
- Each asyncio Task gets its own copy, so the five concurrent executions
allowed by MAX_CONCURRENT_TASKS cannot read each other's value.
A plain `await` of a coroutine does NOT get its own copy and would leak the
value back to the caller, so `task_scope` resets it rather than relying on the
call always arriving via create_task.
"""
from contextlib import contextmanager
from contextvars import ContextVar
from typing import Optional
_current_task_name: ContextVar[Optional[str]] = ContextVar(
"current_task_name", default=None
)
@contextmanager
def task_scope(task_name: str):
"""Name the executing task for the duration of the block."""
token = _current_task_name.set(task_name)
try:
yield
finally:
_current_task_name.reset(token)
def current_task_name() -> Optional[str]:
"""The executing task's name, or None outside an execution.
None is a real answer, not an error: a reporter may be called from a script
or a test with no task around it. Callers omit the field rather than
inventing one.
"""
return _current_task_name.get()
+6 -2
View File
@@ -11,6 +11,7 @@ from psycopg2.extras import RealDictCursor
import traceback
from src.config import Settings
from src.task_context import task_scope
logger = logging.getLogger(__name__)
@@ -209,8 +210,11 @@ class TaskExecutor:
execution_id = cur.fetchone()[0]
conn.commit()
# Load and execute the task
output, error = await self._run_executor(executor_name, task, timeout)
# Load and execute the task. The scope names it for anything the
# executor writes to a shared record — without it, two tasks sharing
# one executor produce rows nobody can tell apart.
with task_scope(task_name):
output, error = await self._run_executor(executor_name, task, timeout)
completed_at = datetime.now(timezone.utc)
duration = int((completed_at - started_at).total_seconds())
+17 -1
View File
@@ -20,7 +20,7 @@ os.environ["GITEA_USER"] = "test-librarian"
os.environ["GITEA_PASSWORD"] = "test-gitea-token"
os.environ["REDIS_HOST"] = "redis-shared"
from src.main import app
from src.main import app, get_task_executor
from src.config import Settings, get_settings
@@ -44,6 +44,22 @@ def auth_headers(api_key: str) -> dict:
return {"Authorization": f"Bearer {api_key}"}
# FastAPI resolves `Depends(get_task_executor)` against the function object it
# captured when each route was decorated, at import time. `unittest.mock.patch`
# on the module attribute `src.main.get_task_executor` therefore never reaches
# an already-registered route — the route keeps calling the original function.
# `app.dependency_overrides` is FastAPI's own supported mechanism for this
# (already used correctly in test_task_delete.py); this fixture centralizes it
# so call sites just need the executor mock they want installed.
@pytest.fixture
def override_task_executor() -> Generator[MagicMock, None, None]:
"""Install a mock TaskExecutor as the live dependency for this test only."""
mock_executor = MagicMock()
app.dependency_overrides[get_task_executor] = lambda: mock_executor
yield mock_executor
app.dependency_overrides.pop(get_task_executor, None)
# Synchronous test client
@pytest.fixture
def client() -> Generator[TestClient, None, None]:
+123 -119
View File
@@ -57,14 +57,13 @@ class TestAuthenticationEndpoints:
)
assert response.status_code == 403
def test_valid_api_key_allows_access(self, client: TestClient, auth_headers: dict):
def test_valid_api_key_allows_access(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test that valid API key allows access."""
with patch('src.main.get_task_executor') as mock_executor:
mock_executor.return_value.get_db_connection.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.fetchall.return_value = []
override_task_executor.get_db_connection.return_value.__enter__.return_value.cursor.return_value.__enter__.return_value.fetchall.return_value = []
response = client.get("/tasks", headers=auth_headers)
# May fail with 500 due to DB, but should not be 401/403
assert response.status_code not in [401, 403]
response = client.get("/tasks", headers=auth_headers)
# May fail with 500 due to DB, but should not be 401/403
assert response.status_code not in [401, 403]
@pytest.mark.api
@@ -72,8 +71,16 @@ class TestAuthenticationEndpoints:
class TestTaskEndpoints:
"""Tests for task management endpoints."""
def test_create_task_missing_fields_returns_400(self, client: TestClient, auth_headers: dict):
"""Test that creating task without required fields returns 400."""
def test_create_task_missing_fields_returns_422(self, client: TestClient, auth_headers: dict):
"""Test that creating task without required fields returns 422.
`task: TaskCreate` (src/main.py) is a plain Pydantic request body with
no custom validation FastAPI's own dependency-resolution layer
rejects a request missing required fields before create_task's body
ever runs, and that layer always answers 422, never 400. There is no
code path in this repo that could produce 400 here; renamed rather
than asserting a status this endpoint cannot return.
"""
incomplete_task = {
"task_name": "test",
# Missing service, executor, priority
@@ -83,63 +90,61 @@ class TestTaskEndpoints:
headers=auth_headers,
json=incomplete_task
)
assert response.status_code == 400
assert response.status_code == 422
def test_create_task_with_valid_data(self, client: TestClient, auth_headers: dict, sample_task_data: dict):
def test_create_task_with_valid_data(self, client: TestClient, auth_headers: dict, sample_task_data: dict, override_task_executor: MagicMock):
"""Test creating a task with valid data."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_conn = MagicMock()
mock_cursor = MagicMock()
# Setup mock to return task data
mock_cursor.fetchone.return_value = {**sample_task_data, "id": 1}
# Setup mock to return task data
mock_cursor.fetchone.return_value = {**sample_task_data, "id": 1}
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.post(
"/tasks",
headers=auth_headers,
json=sample_task_data
)
# Verify the call was made
assert mock_cursor.execute.called
# Check that config was JSON-encoded
call_args = mock_cursor.execute.call_args
assert 'config' in call_args[0][1]
def test_trigger_task_endpoint(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test manually triggering a task."""
mock_conn = MagicMock()
mock_cursor = MagicMock()
# Mock task retrieval
mock_cursor.fetchone.return_value = {
"task_name": "test_task",
"enabled": True,
"priority": 50,
"executor": "example_executor"
}
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
with patch('asyncio.create_task'):
response = client.post(
"/tasks",
headers=auth_headers,
json=sample_task_data
"/tasks/test_task/trigger",
headers=auth_headers
)
# Verify the call was made
assert mock_cursor.execute.called
# Check that config was JSON-encoded
call_args = mock_cursor.execute.call_args
assert 'config' in call_args[0][1]
def test_trigger_task_endpoint(self, client: TestClient, auth_headers: dict):
"""Test manually triggering a task."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
# Mock task retrieval
mock_cursor.fetchone.return_value = {
"task_name": "test_task",
"enabled": True,
"priority": 50,
"executor": "example_executor"
}
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
with patch('asyncio.create_task'):
response = client.post(
"/tasks/test_task/trigger",
headers=auth_headers
)
# Should return success message
if response.status_code == 200:
data = response.json()
assert data["task_name"] == "test_task"
# Should return success message
if response.status_code == 200:
data = response.json()
assert data["task_name"] == "test_task"
@pytest.mark.api
@@ -152,36 +157,37 @@ class TestStatsEndpoint:
response = client.get("/stats")
assert response.status_code == 401
def test_stats_endpoint_returns_metrics(self, client: TestClient, auth_headers: dict):
def test_stats_endpoint_returns_metrics(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test that stats endpoint returns system metrics."""
with patch('src.main.get_scheduler') as mock_scheduler, \
patch('src.main.get_task_executor') as mock_executor:
# Note: `/stats` also depends on get_scheduler via Depends(), which this
# test does not override (out of this fix's measured scope — see
# override_task_executor's docstring for why patch() cannot reach it).
# It is not load-bearing here: the app's real scheduler is running by
# the time TestClient's lifespan completes, so `sched.running` is True
# without an override, same as test_health_endpoint_returns_healthy.
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.side_effect = [
{"count": 3}, # enabled tasks
{"count": 0}, # running tasks
]
mock_cursor.fetchall.return_value = [
{"status": "success", "count": 10},
{"status": "failed", "count": 1}
]
mock_scheduler.return_value.running = True
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.side_effect = [
{"count": 3}, # enabled tasks
{"count": 0}, # running tasks
]
mock_cursor.fetchall.return_value = [
{"status": "success", "count": 10},
{"status": "failed", "count": 1}
]
response = client.get("/stats", headers=auth_headers)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get("/stats", headers=auth_headers)
if response.status_code == 200:
data = response.json()
assert "scheduler_running" in data
assert "tasks_enabled" in data
assert "concurrent_limit" in data
if response.status_code == 200:
data = response.json()
assert "scheduler_running" in data
assert "tasks_enabled" in data
assert "concurrent_limit" in data
@pytest.mark.api
@@ -189,48 +195,46 @@ class TestStatsEndpoint:
class TestExecutionHistoryEndpoint:
"""Tests for /executions endpoint."""
def test_executions_endpoint_returns_history(self, client: TestClient, auth_headers: dict):
def test_executions_endpoint_returns_history(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test that executions endpoint returns execution history."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = [
{
"id": 1,
"task_name": "test_task",
"status": "success",
"duration_seconds": 5
}
]
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = [
{
"id": 1,
"task_name": "test_task",
"status": "success",
"duration_seconds": 5
}
]
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get("/executions", headers=auth_headers)
response = client.get("/executions", headers=auth_headers)
if response.status_code == 200:
data = response.json()
assert "executions" in data
assert "count" in data
if response.status_code == 200:
data = response.json()
assert "executions" in data
assert "count" in data
def test_executions_filter_by_task_name(self, client: TestClient, auth_headers: dict):
def test_executions_filter_by_task_name(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test filtering executions by task name."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get(
"/executions?task_name=test_task&limit=10",
headers=auth_headers
)
response = client.get(
"/executions?task_name=test_task&limit=10",
headers=auth_headers
)
# Should execute query with filters
assert mock_cursor.execute.called
# Should execute query with filters
assert mock_cursor.execute.called
+155 -167
View File
@@ -3,7 +3,7 @@ Comprehensive API tests to improve coverage of main.py.
"""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, MagicMock
from unittest.mock import MagicMock
import json
@@ -12,106 +12,101 @@ import json
class TestTaskCRUDOperations:
"""Comprehensive CRUD tests for task endpoints."""
def test_list_tasks_empty(self, client: TestClient, auth_headers: dict):
def test_list_tasks_empty(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test listing tasks when none exist."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get("/tasks", headers=auth_headers)
response = client.get("/tasks", headers=auth_headers)
if response.status_code == 200:
data = response.json()
assert "tasks" in data
assert data["count"] == 0
if response.status_code == 200:
data = response.json()
assert "tasks" in data
assert data["count"] == 0
def test_list_tasks_with_filters(self, client: TestClient, auth_headers: dict):
def test_list_tasks_with_filters(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test listing tasks with enabled and service filters."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get(
"/tasks?enabled=true&service=scheduler",
headers=auth_headers
)
response = client.get(
"/tasks?enabled=true&service=scheduler",
headers=auth_headers
)
# Should execute filtered query
assert mock_cursor.execute.called
call_args = str(mock_cursor.execute.call_args)
assert "enabled" in call_args.lower() or response.status_code in [200, 500]
# Should execute filtered query
assert mock_cursor.execute.called
call_args = str(mock_cursor.execute.call_args)
assert "enabled" in call_args.lower() or response.status_code in [200, 500]
def test_get_task_details_not_found(self, client: TestClient, auth_headers: dict):
def test_get_task_details_not_found(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test getting details for non-existent task."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get("/tasks/nonexistent", headers=auth_headers)
response = client.get("/tasks/nonexistent", headers=auth_headers)
assert response.status_code == 404
assert response.status_code == 404
def test_update_task(self, client: TestClient, auth_headers: dict):
def test_update_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test updating a task."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = {
"task_name": "test",
"priority": 60
}
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = {
"task_name": "test",
"priority": 60
}
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.put(
"/tasks/test",
headers=auth_headers,
json={"priority": 60}
)
response = client.put(
"/tasks/test",
headers=auth_headers,
json={"priority": 60}
)
# Should have attempted update
assert mock_cursor.execute.called
# Should have attempted update
assert mock_cursor.execute.called
def test_update_task_not_found(self, client: TestClient, auth_headers: dict):
def test_update_task_not_found(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test updating non-existent task."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.put(
"/tasks/nonexistent",
headers=auth_headers,
json={"priority": 60}
)
response = client.put(
"/tasks/nonexistent",
headers=auth_headers,
json={"priority": 60}
)
assert response.status_code == 404
assert response.status_code == 404
def test_update_task_no_fields(self, client: TestClient, auth_headers: dict):
"""Test updating task with no valid fields."""
@@ -123,39 +118,37 @@ class TestTaskCRUDOperations:
assert response.status_code == 400
def test_delete_task(self, client: TestClient, auth_headers: dict):
def test_delete_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test deleting a task."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = ("test_task",)
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = ("test_task",)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.delete("/tasks/test_task", headers=auth_headers)
response = client.delete("/tasks/test_task", headers=auth_headers)
if response.status_code == 200:
data = response.json()
assert "deleted successfully" in data["message"].lower()
if response.status_code == 200:
data = response.json()
assert "deleted successfully" in data["message"].lower()
def test_delete_task_not_found(self, client: TestClient, auth_headers: dict):
def test_delete_task_not_found(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test deleting non-existent task."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.delete("/tasks/nonexistent", headers=auth_headers)
response = client.delete("/tasks/nonexistent", headers=auth_headers)
assert response.status_code == 404
assert response.status_code == 404
@pytest.mark.api
@@ -163,40 +156,38 @@ class TestTaskCRUDOperations:
class TestTriggerEndpoint:
"""Tests for task trigger endpoint."""
def test_trigger_disabled_task(self, client: TestClient, auth_headers: dict):
def test_trigger_disabled_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test triggering a disabled task."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = {
"task_name": "test",
"enabled": False
}
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = {
"task_name": "test",
"enabled": False
}
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.post("/tasks/test/trigger", headers=auth_headers)
response = client.post("/tasks/test/trigger", headers=auth_headers)
assert response.status_code == 400
assert response.status_code == 400
def test_trigger_nonexistent_task(self, client: TestClient, auth_headers: dict):
def test_trigger_nonexistent_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test triggering a task that doesn't exist."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchone.return_value = None
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.post("/tasks/nonexistent/trigger", headers=auth_headers)
response = client.post("/tasks/nonexistent/trigger", headers=auth_headers)
assert response.status_code == 404
assert response.status_code == 404
@pytest.mark.api
@@ -254,58 +245,55 @@ class TestLegacyEndpoints:
class TestExecutionFiltering:
"""Tests for execution history filtering."""
def test_filter_by_service(self, client: TestClient, auth_headers: dict):
def test_filter_by_service(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test filtering executions by service."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get(
"/executions?service=scheduler",
headers=auth_headers
)
response = client.get(
"/executions?service=scheduler",
headers=auth_headers
)
assert mock_cursor.execute.called
assert mock_cursor.execute.called
def test_filter_by_status(self, client: TestClient, auth_headers: dict):
def test_filter_by_status(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test filtering executions by status."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get(
"/executions?status=success",
headers=auth_headers
)
response = client.get(
"/executions?status=success",
headers=auth_headers
)
assert mock_cursor.execute.called
assert mock_cursor.execute.called
def test_custom_limit(self, client: TestClient, auth_headers: dict):
def test_custom_limit(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
"""Test custom limit for executions."""
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_cursor.fetchall.return_value = []
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
response = client.get("/executions?limit=50", headers=auth_headers)
response = client.get("/executions?limit=50", headers=auth_headers)
if response.status_code == 200:
data = response.json()
assert data["limit"] == 50
if response.status_code == 200:
data = response.json()
assert data["limit"] == 50
+50 -13
View File
@@ -60,23 +60,36 @@ class TestDocSyncExecutor:
mock_upstream_dir = MagicMock()
mock_gitea_dir = MagicMock()
# Setup directory mocking
mock_upstream_dir.iterdir.return_value = [
MagicMock(name=".git", is_dir=lambda: True),
MagicMock(name="README.md", is_dir=lambda: False),
MagicMock(name="docs", is_dir=lambda: True),
]
mock_gitea_dir.iterdir.return_value = [
MagicMock(name=".git", is_dir=lambda: True)
]
# Setup directory mocking. MagicMock(name=...) sets the mock's
# repr, not its .name attribute (classic gotcha — see
# test_executor_sync_specific_paths) — set .name explicitly so
# execute()'s `item.name != '.git'` check actually excludes it.
git_item = MagicMock(is_dir=lambda: True)
git_item.name = ".git"
readme_item = MagicMock(is_dir=lambda: False)
readme_item.name = "README.md"
docs_item = MagicMock(is_dir=lambda: True)
docs_item.name = "docs"
mock_upstream_dir.iterdir.return_value = [git_item, readme_item, docs_item]
gitea_git_item = MagicMock(is_dir=lambda: True)
gitea_git_item.name = ".git"
mock_gitea_dir.iterdir.return_value = [gitea_git_item]
mock_path.return_value = mock_work_dir
mock_work_dir.__truediv__.side_effect = [mock_upstream_dir, mock_gitea_dir]
# Mock git status to show changes
# Mock git status to show changes. _get_git_commit is patched
# separately above and never calls the real _run_command, so it
# does not consume a slot in this side_effect list — the actual
# call order for this (clone-succeeds, entire-repo) path is:
# clone upstream, clone gitea, add, status, commit, tag, push
# branch, push tag. The list previously reserved a slot for
# "git rev-parse HEAD" that _run_command is never asked for,
# which shifted "M README.md\n" one call late and made
# `git status --porcelain` see "" (no changes) instead.
mock_run.side_effect = [
"", # git clone upstream
"", # git rev-parse HEAD
"", # git clone gitea
"", # git add
"M README.md\n", # git status --porcelain (has changes)
@@ -108,7 +121,15 @@ class TestDocSyncExecutor:
mock_get_commit.return_value = "abc123"
# Mock path operations
# Mock path operations. work_dir = Path(...) resolves to
# mock_path.return_value — mock_upstream_dir/mock_gitea_dir have
# to be reachable from there via __truediv__, the same way
# test_executor_successful_sync_entire_repo wires it, or
# `upstream_dir / doc_path` never reaches these mocks at all and
# falls through to an unconfigured auto-generated MagicMock
# instead (observed failure: TypeError joining a MagicMock into
# ', '.join(copied_paths)).
mock_work_dir = MagicMock()
mock_upstream_dir = MagicMock()
mock_gitea_dir = MagicMock()
mock_docs = MagicMock(name="docs")
@@ -121,6 +142,8 @@ class TestDocSyncExecutor:
mock_examples.is_dir.return_value = True
mock_examples.name = "examples"
mock_path.return_value = mock_work_dir
mock_work_dir.__truediv__.side_effect = [mock_upstream_dir, mock_gitea_dir]
mock_upstream_dir.__truediv__.side_effect = [mock_docs, mock_examples]
mock_gitea_dir.iterdir.return_value = []
@@ -159,13 +182,27 @@ class TestDocSyncExecutor:
mock_run.side_effect = run_command_side_effect
# Setup minimal mocking
# Setup minimal mocking. sample_doc_sync_config's docs_paths is
# ["/docs"] (tests/conftest.py), so execute() takes the
# specific-paths branch and needs `upstream_dir / "docs"` wired
# to something with a real string .name — see
# test_executor_sync_specific_paths for the same wiring gap and
# the TypeError it produces unwired.
mock_work_dir = MagicMock()
mock_upstream_dir = MagicMock()
mock_gitea_dir = MagicMock()
mock_gitea_dir.iterdir.return_value = []
mock_upstream_dir.iterdir.return_value = []
mock_docs = MagicMock()
mock_docs.exists.return_value = True
mock_docs.is_dir.return_value = True
mock_docs.name = "docs"
mock_path.return_value = mock_work_dir
mock_work_dir.__truediv__.side_effect = [mock_upstream_dir, mock_gitea_dir]
mock_upstream_dir.__truediv__.side_effect = [mock_docs]
result = await doc_sync_executor.execute(sample_doc_sync_config, test_settings)
assert "already up to date" in result.lower() or "no changes" in result.lower()
+46 -65
View File
@@ -27,49 +27,55 @@ class TestDatabaseIntegration:
class TestEndToEndTaskFlow:
"""End-to-end tests for task lifecycle (mocked database)."""
def test_create_list_delete_task_flow(self, client: TestClient, auth_headers: dict, sample_task_data: dict):
def test_create_list_delete_task_flow(self, client: TestClient, auth_headers: dict, sample_task_data: dict, override_task_executor):
"""Test complete task lifecycle: create → list → delete."""
from unittest.mock import MagicMock
# This simulates the full flow with mocked database
with patch('src.main.get_task_executor') as mock_executor:
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_conn = MagicMock()
mock_cursor = MagicMock()
# Mock create
created_task = {**sample_task_data, "id": 99}
mock_cursor.fetchone.side_effect = [
created_task, # Create task
created_task, # List tasks (as dict)
("test_task",) # Delete task
]
mock_cursor.fetchall.return_value = [created_task]
# Mock create. list_tasks (GET /tasks) uses fetchall, not fetchone, so
# it does not consume a slot here — the previous list of 3 values was
# sized for a delete that has since grown a second lookup (T-97's
# 409-on-history-loss check, c34db66): delete_task now does
# `row = cur.fetchone()` for the task id, then a separate
# `cur.fetchone()[0]` for its execution count, so a real
# create->list->delete flow needs 1 (create) + 2 (delete) = 3
# fetchone() calls in that order, not create+list+delete.
created_task = {**sample_task_data, "id": 99}
mock_cursor.fetchone.side_effect = [
created_task, # create_task: INSERT ... RETURNING (dict row)
(99,), # delete_task: SELECT id FROM scheduled_tasks
(0,), # delete_task: SELECT COUNT(*) FROM task_executions — none, so it proceeds
]
mock_cursor.fetchall.return_value = [created_task]
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
# Create
create_response = client.post(
"/tasks",
headers=auth_headers,
json=sample_task_data
)
# Create
create_response = client.post(
"/tasks",
headers=auth_headers,
json=sample_task_data
)
# List
list_response = client.get("/tasks", headers=auth_headers)
# List
list_response = client.get("/tasks", headers=auth_headers)
# Delete
delete_response = client.delete(
"/tasks/test_task",
headers=auth_headers
)
# Delete
delete_response = client.delete(
"/tasks/test_task",
headers=auth_headers
)
# Verify the flow worked
assert create_response.status_code in [200, 500] # May fail on DB issues
assert list_response.status_code in [200, 500]
assert delete_response.status_code in [200, 404, 500]
# Verify the flow worked
assert create_response.status_code in [200, 500] # May fail on DB issues
assert list_response.status_code in [200, 500]
assert delete_response.status_code in [200, 404, 500]
@pytest.mark.integration
@@ -91,36 +97,6 @@ class TestTaskExecutorIntegration:
assert "Integration test" in result
assert "took" in result.lower()
@pytest.mark.asyncio
async def test_task_scheduling_logic(self, test_settings):
"""Test task scheduling logic."""
from src.tasks.executor import TaskExecutor
from datetime import datetime
executor = TaskExecutor(test_settings)
# Test various scheduling scenarios
task_every_minute = {
'minute': -1, 'hour': -1, 'day_of_month': -1,
'month': -1, 'day_of_week': -1
}
task_specific_time = {
'minute': 30, 'hour': 14, 'day_of_month': -1,
'month': -1, 'day_of_week': -1
}
now = datetime(2025, 12, 7, 14, 30, 0)
# Every minute task should always run
assert executor._should_run_now(task_every_minute, now) is True
# Specific time task should run at 14:30
assert executor._should_run_now(task_specific_time, now) is True
# But not at 14:31
now_plus_one = datetime(2025, 12, 7, 14, 31, 0)
assert executor._should_run_now(task_specific_time, now_plus_one) is False
@pytest.mark.integration
@@ -164,8 +140,13 @@ class TestConfigValidation:
settings = get_settings()
# Should have loaded test environment variables
assert settings.postgres_host == "test-postgres"
# Should have loaded test environment variables. tests/conftest.py
# sets POSTGRES_HOST="postgres-shared" (module-level, before `from
# src.main import app`), annotated "Use real postgres for integration
# tests" — this assertion checked for "test-postgres", a value
# nothing in the suite has ever set. Matched to the fixture actually
# in effect rather than to an unset value.
assert settings.postgres_host == "postgres-shared"
assert settings.postgres_db == "test_scheduler"
assert settings.scheduler_api_key == "test-api-key-12345"
+22 -2
View File
@@ -498,7 +498,24 @@ class TestRestApiExecutor:
assert redacted["normal_field"] == "visible"
def test_redact_sensitive_nested(self):
"""Test redaction in nested structures."""
"""A dict under a sensitive key is redacted whole, not recursed into.
This asserted fine-grained recursion that `auth.token` was replaced
while `auth`'s other keys stayed readable — and had never passed. The
source redacts the entire value the moment the KEY matches, so
`redacted["config"]["auth"]` is the string, and indexing `["token"]`
into it raises TypeError.
Settled in favour of the source. Fine-grained redaction has to know
which sub-keys carry the secret, which is a guess about the shape of
data nobody has inspected; redacting on the key cannot be wrong that
way. Real configs here look like
{"auth": {"type": "bearer", "token": "${SOME_API_KEY}"}}, and the cost
of guessing wrong is a credential in a log, which no later fix undoes.
The price is readability: a reader learns that auth was present, not
that it was bearer. That is the trade being made deliberately.
"""
data = {
"config": {
"database": "mydb",
@@ -513,7 +530,10 @@ class TestRestApiExecutor:
assert redacted["config"]["database"] == "mydb"
assert redacted["config"]["password"] == "***REDACTED***"
assert redacted["config"]["auth"]["token"] == "***REDACTED***"
# The whole sub-dict, not a recursed copy of it.
assert redacted["config"]["auth"] == "***REDACTED***"
# And the secret is nowhere in the output, by any path.
assert "bearer123" not in str(redacted)
# Helper Function Tests
+189
View File
@@ -0,0 +1,189 @@
"""A check_history row must say which task produced it.
On 2026-08-11 a 425 MB probe task and the 5 GB nightly backup both ran through
config_backup_executor. The probe failed, and the row it wrote was
indistinguishable from a nightly-backup failure same `source`, same `domain`,
nothing naming the task. The health record could not answer which job broke,
which is most of what a health record is for.
Executors are called as `execute(config, settings)` and are never told which
task they are, so the name travels in a ContextVar. These tests pin what makes
that safe: it reaches the reporter, it survives the worker thread T-74
introduced, and concurrent executions cannot read each other's. They also pin
where the summary lives, since two producers write this table.
"""
import asyncio
import json
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from src.config import Settings
from src.executors import health_report
from src.task_context import current_task_name, task_scope
from src.tasks.executor import TaskExecutor
@pytest.fixture
def captured_row(monkeypatch):
"""Capture the JSON payload report() would insert, without a database."""
box = {}
conn, cur = MagicMock(), MagicMock()
def execute(sql, params):
box['result'] = json.loads(params[4])
cur.execute.side_effect = execute
conn.cursor.return_value.__enter__ = MagicMock(return_value=cur)
conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
conn.__enter__ = MagicMock(return_value=conn)
conn.__exit__ = MagicMock(return_value=None)
monkeypatch.setattr(health_report.psycopg2, 'connect', lambda **kw: conn)
return box
def _report(settings, **kw):
health_report.report(
settings, domain="backup", status=health_report.OK,
source="scheduler/config_backup_executor",
summary="backed up 3 sources", metrics={}, **kw
)
@pytest.mark.unit
class TestAttribution:
def test_the_row_names_the_task(self, test_settings: Settings, captured_row):
with task_scope("backup_docker_configs_daily"):
_report(test_settings)
assert captured_row['result']['task'] == "backup_docker_configs_daily"
def test_source_still_names_the_code(self, test_settings: Settings, captured_row):
"""Task and source answer different questions; both are needed.
`source` says which code wrote the row, `task` says which schedule
invoked it. Replacing one with the other loses a distinction.
"""
with task_scope("t74_loop_probe"):
_report(test_settings)
r = captured_row['result']
assert r['source'] == "scheduler/config_backup_executor"
assert r['task'] == "t74_loop_probe"
def test_the_field_is_omitted_rather_than_nulled(self, test_settings: Settings, captured_row):
"""report() is callable from a script with no task around it.
A null would claim there was a task and it had no name. Absence says the
question does not apply.
"""
_report(test_settings)
assert 'task' not in captured_row['result']
@pytest.mark.asyncio
async def test_attribution_survives_the_worker_thread(
self, test_settings: Settings, captured_row
):
"""T-74 moved reporting into asyncio.to_thread. Attribution has to follow.
If the context did not propagate, every row written by a real executor
would silently lose its task while the tests above still passed.
"""
with task_scope("backup_portainer_daily"):
await health_report.report_async(
test_settings, domain="backup", status=health_report.OK,
source="scheduler/portainer_backup_executor",
summary="backed up Portainer", metrics={},
)
assert captured_row['result']['task'] == "backup_portainer_daily"
@pytest.mark.unit
class TestAttributionIsolation:
"""MAX_CONCURRENT_TASKS is 5, so mixing values between them is a live risk."""
def _executor_with_db(self, test_settings):
executor = TaskExecutor(test_settings)
conn, cur = MagicMock(), MagicMock()
cur.fetchone.return_value = [1]
conn.cursor.return_value.__enter__ = MagicMock(return_value=cur)
conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
conn.__enter__ = MagicMock(return_value=conn)
conn.__exit__ = MagicMock(return_value=None)
return executor, patch.object(executor, 'get_db_connection', return_value=conn)
def _task(self, name):
return {'id': hash(name) % 1000, 'task_name': name, 'executor': 'x',
'service': 'scheduler', 'priority': 5, 'timeout_seconds': 60}
@pytest.mark.asyncio
async def test_the_engine_names_the_task_and_clears_it_after(
self, test_settings: Settings
):
executor, db = self._executor_with_db(test_settings)
seen = {}
async def fake_run(name, task, timeout):
seen['during'] = current_task_name()
return "done", None
with db, patch.object(executor, '_run_executor', new=fake_run):
await executor.execute_task(self._task("nightly_backup"))
assert seen['during'] == "nightly_backup", "executor ran unattributed"
assert current_task_name() is None, "attribution leaked past the execution"
@pytest.mark.asyncio
async def test_concurrent_executions_do_not_see_each_other(
self, test_settings: Settings
):
executor, db = self._executor_with_db(test_settings)
seen = {}
async def fake_run(name, task, timeout):
who = task['task_name']
# Yield mid-flight so the executions genuinely interleave; without
# this they would run to completion one at a time and the test would
# pass even with a shared global.
await asyncio.sleep(0.01 if who == "slow_one" else 0)
seen[who] = current_task_name()
return "done", None
with db, patch.object(executor, '_run_executor', new=fake_run):
await asyncio.gather(
executor.execute_task(self._task("slow_one")),
executor.execute_task(self._task("fast_one")),
)
assert seen == {"slow_one": "slow_one", "fast_one": "fast_one"}
@pytest.mark.unit
class TestSummaryPlacement:
"""The two writers of check_history must agree where the substance lives.
sysmon-go writes `summary` at the top level, beside `status`. This module
wrote it under `metrics` until 2026-08-11, so a reader had to know which
producer wrote a row before it could find out what the row said and a
query written the obvious way silently found half the data. That is the T-36
failure exactly, where per-domain queries returned nothing because the value
was nested somewhere else.
"""
def test_summary_is_top_level(self, test_settings: Settings, captured_row):
_report(test_settings)
r = captured_row['result']
assert r['summary'] == "backed up 3 sources"
assert 'summary' not in r['metrics'], "summary must not also live under metrics"
def test_summary_is_required(self, test_settings: Settings, captured_row):
"""Omitting it is an error at the call, not a silently empty column.
sysmon-go enforces this through Domain.Run's signature; a parameter with
no default is the equivalent here. A row whose substance is missing looks
exactly like a row whose check found nothing to say.
"""
with pytest.raises(TypeError):
health_report.report(
test_settings, domain="backup", status=health_report.OK,
source="scheduler/x", metrics={},
)
+112
View File
@@ -0,0 +1,112 @@
"""Deleting a task must not destroy its history by accident, or 500 by surprise.
DELETE /tasks/{name} used to issue a bare DELETE against scheduled_tasks. Any
task that had ever run owned rows in task_executions, so the foreign key
rejected it and the caller got "Internal Server Error" with nothing pointing at
history as the obstacle it read as the service being broken rather than the
request being refusable. Since every task that has ever fired has history, the
endpoint effectively worked only for tasks that never ran.
It now refuses with 409 and takes ?purge=true to mean it. The asymmetry is the
argument: a task definition can be recreated from the API in one call, its
execution history cannot be recreated at all.
"""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import MagicMock
from src.main import app, get_task_executor
def _fake_executor(task_row, execution_count=0):
"""A stand-in whose cursor answers the endpoint's two lookups in order."""
ex = MagicMock()
conn, cur = MagicMock(), MagicMock()
cur.fetchone.side_effect = (
[task_row, (execution_count,)] if task_row is not None else [None]
)
conn.cursor.return_value.__enter__ = MagicMock(return_value=cur)
conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
conn.__enter__ = MagicMock(return_value=conn)
conn.__exit__ = MagicMock(return_value=None)
ex.get_db_connection.return_value = conn
return ex, cur
def _statements(cur):
return [c[0][0] for c in cur.execute.call_args_list]
@pytest.fixture
def override():
made = {}
def _install(task_row, execution_count=0):
ex, cur = _fake_executor(task_row, execution_count)
app.dependency_overrides[get_task_executor] = lambda: ex
made['cur'] = cur
return cur
yield _install
app.dependency_overrides.pop(get_task_executor, None)
@pytest.mark.unit
class TestDeleteTask:
def test_history_blocks_deletion_with_409(self, client: TestClient, auth_headers, override):
cur = override((46,), execution_count=12)
r = client.delete("/tasks/some_task", headers=auth_headers)
assert r.status_code == 409
detail = r.json()["detail"]
# The message has to carry the facts the caller needs to act: how much
# history is at stake, the flag that proceeds, and the option they
# probably actually wanted. A bare "conflict" would be no better than
# the 500 it replaces.
assert "12 execution record" in detail
assert "purge=true" in detail
assert "enabled=false" in detail
def test_a_refused_delete_deletes_nothing(self, client: TestClient, auth_headers, override):
cur = override((46,), execution_count=12)
client.delete("/tasks/some_task", headers=auth_headers)
assert not any("DELETE" in s.upper() for s in _statements(cur))
def test_purge_removes_history_then_the_task(self, client: TestClient, auth_headers, override):
cur = override((46,), execution_count=12)
r = client.delete("/tasks/some_task?purge=true", headers=auth_headers)
assert r.status_code == 200
assert r.json()["executions_purged"] == 12
deletes = [s for s in _statements(cur) if "DELETE" in s.upper()]
assert len(deletes) == 2
# History first: the foreign key points that way, and the reverse order
# is the failure this endpoint started with.
assert "task_executions" in deletes[0]
assert "scheduled_tasks" in deletes[1]
def test_a_task_that_never_ran_deletes_without_the_flag(
self, client: TestClient, auth_headers, override
):
cur = override((46,), execution_count=0)
r = client.delete("/tasks/fresh_task", headers=auth_headers)
assert r.status_code == 200
assert r.json()["executions_purged"] == 0
deletes = [s for s in _statements(cur) if "DELETE" in s.upper()]
assert len(deletes) == 1, "nothing to purge, so history must not be touched"
assert "scheduled_tasks" in deletes[0]
def test_unknown_task_is_404_not_409(self, client: TestClient, auth_headers, override):
override(None)
r = client.delete("/tasks/nope", headers=auth_headers)
assert r.status_code == 404
def test_purge_on_an_unknown_task_is_still_404(
self, client: TestClient, auth_headers, override
):
"""The flag must not turn a missing task into a success."""
override(None)
r = client.delete("/tasks/nope?purge=true", headers=auth_headers)
assert r.status_code == 404
+69 -125
View File
@@ -4,7 +4,7 @@ Tests for the task executor module.
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from datetime import datetime
from src.tasks.executor import TaskExecutor
from src.tasks.executor import TaskExecutor, MAX_CONCURRENT_TASKS
from src.config import Settings
@@ -17,7 +17,13 @@ class TestTaskExecutor:
executor = TaskExecutor(test_settings)
assert executor.settings == test_settings
assert executor.max_concurrent == 5
# There is no `max_concurrent` instance attribute — concurrency is
# capped by MAX_CONCURRENT_TASKS (module constant) via
# asyncio.Semaphore(MAX_CONCURRENT_TASKS) in __init__. Verify the
# semaphore was built with that bound instead of asserting an
# attribute name the class has never had.
assert MAX_CONCURRENT_TASKS == 5
assert executor.semaphore._value == 5
@patch('psycopg2.connect')
def test_get_db_connection(self, mock_connect, test_settings: Settings):
@@ -87,13 +93,25 @@ class TestTaskExecutor:
task = {**sample_task_data, 'id': 1}
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
patch('src.tasks.executor.importlib.import_module') as mock_import:
# _run_executor (src/tasks/executor.py) loads the executor module with
# the __import__ builtin directly — `module = __import__(module_path,
# fromlist=['execute'])` — not importlib.import_module. This is the
# documented dynamic-loading trap in this repo's own CLAUDE.md
# ("executors are chosen by data, not code"). `importlib` is never
# imported in that module, so patching 'src.tasks.executor.importlib'
# fails at patch setup, before the test body runs at all.
real_import = __import__
# Mock executor module
mock_executor_module = MagicMock()
mock_executor_module.execute = AsyncMock(return_value="Task completed successfully")
mock_import.return_value = mock_executor_module
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'src.executors.example_executor':
return mock_executor_module
return real_import(name, globals, locals, fromlist, level)
mock_executor_module = MagicMock()
mock_executor_module.execute = AsyncMock(return_value="Task completed successfully")
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
patch('builtins.__import__', side_effect=fake_import) as mock_import:
# Mock database
mock_conn = MagicMock()
@@ -106,7 +124,7 @@ class TestTaskExecutor:
await executor.execute_task(task)
# Should have imported executor module
mock_import.assert_called_with('src.executors.example_executor')
mock_import.assert_any_call('src.executors.example_executor', fromlist=['execute'])
# Should have updated task status
assert mock_cursor.execute.call_count >= 2 # Insert execution record + update task
@@ -118,13 +136,20 @@ class TestTaskExecutor:
task = {**sample_task_data, 'id': 1}
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
patch('src.tasks.executor.importlib.import_module') as mock_import:
# See test_execute_task_success: _run_executor uses the __import__
# builtin directly, not importlib.import_module.
real_import = __import__
# Mock executor that raises error
mock_executor_module = MagicMock()
mock_executor_module.execute = AsyncMock(side_effect=Exception("Task failed"))
mock_import.return_value = mock_executor_module
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'src.executors.example_executor':
return mock_executor_module
return real_import(name, globals, locals, fromlist, level)
mock_executor_module = MagicMock()
mock_executor_module.execute = AsyncMock(side_effect=Exception("Task failed"))
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
patch('builtins.__import__', side_effect=fake_import):
# Mock database
mock_conn = MagicMock()
@@ -149,19 +174,26 @@ class TestTaskExecutor:
task = {**sample_task_data, 'id': 1, 'timeout_seconds': 1}
# See test_execute_task_success: _run_executor uses the __import__
# builtin directly, not importlib.import_module.
import asyncio
real_import = __import__
def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == 'src.executors.example_executor':
return mock_executor_module
return real_import(name, globals, locals, fromlist, level)
mock_executor_module = MagicMock()
async def slow_execute(*args, **kwargs):
await asyncio.sleep(10) # Longer than timeout
return "Done"
mock_executor_module.execute = slow_execute
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
patch('src.tasks.executor.importlib.import_module') as mock_import:
# Mock executor that takes too long
import asyncio
mock_executor_module = MagicMock()
async def slow_execute(*args, **kwargs):
await asyncio.sleep(10) # Longer than timeout
return "Done"
mock_executor_module.execute = slow_execute
mock_import.return_value = mock_executor_module
patch('builtins.__import__', side_effect=fake_import):
# Mock database
mock_conn = MagicMock()
@@ -177,106 +209,10 @@ class TestTaskExecutor:
calls = [str(call) for call in mock_cursor.execute.call_args_list]
assert any('timeout' in str(call).lower() for call in calls)
def test_should_run_task_wildcard(self, test_settings: Settings):
"""Test task scheduling with wildcards."""
executor = TaskExecutor(test_settings)
# All wildcards should always match
task = {
'minute': -1,
'hour': -1,
'day_of_month': -1,
'month': -1,
'day_of_week': -1
}
now = datetime(2025, 12, 7, 14, 30, 0) # Saturday
assert executor._should_run_now(task, now) is True
def test_should_run_task_specific_time(self, test_settings: Settings):
"""Test task scheduling with specific time."""
executor = TaskExecutor(test_settings)
# Specific time: every day at 14:30
task = {
'minute': 30,
'hour': 14,
'day_of_month': -1,
'month': -1,
'day_of_week': -1
}
# Matching time
now = datetime(2025, 12, 7, 14, 30, 0)
assert executor._should_run_now(task, now) is True
# Non-matching time
now = datetime(2025, 12, 7, 14, 31, 0)
assert executor._should_run_now(task, now) is False
def test_should_run_task_specific_day_of_month(self, test_settings: Settings):
"""Test task scheduling with specific day of month."""
executor = TaskExecutor(test_settings)
# Run on 11th of every month at 04:00
task = {
'minute': 0,
'hour': 4,
'day_of_month': 11,
'month': -1,
'day_of_week': -1
}
# Matching date
now = datetime(2025, 12, 11, 4, 0, 0)
assert executor._should_run_now(task, now) is True
# Wrong day
now = datetime(2025, 12, 12, 4, 0, 0)
assert executor._should_run_now(task, now) is False
def test_should_run_task_specific_month(self, test_settings: Settings):
"""Test task scheduling with specific month."""
executor = TaskExecutor(test_settings)
# Run on January 1st at midnight
task = {
'minute': 0,
'hour': 0,
'day_of_month': 1,
'month': 1,
'day_of_week': -1
}
# Matching date
now = datetime(2025, 1, 1, 0, 0, 0)
assert executor._should_run_now(task, now) is True
# Wrong month
now = datetime(2025, 2, 1, 0, 0, 0)
assert executor._should_run_now(task, now) is False
def test_should_run_task_day_of_week(self, test_settings: Settings):
"""Test task scheduling with day of week."""
executor = TaskExecutor(test_settings)
# Run every Monday at 09:00
task = {
'minute': 0,
'hour': 9,
'day_of_month': -1,
'month': -1,
'day_of_week': 0 # Monday
}
# Monday
now = datetime(2025, 12, 8, 9, 0, 0) # Monday
assert executor._should_run_now(task, now) is True
# Tuesday
now = datetime(2025, 12, 9, 9, 0, 0) # Tuesday
assert executor._should_run_now(task, now) is False
@pytest.mark.asyncio
async def test_concurrent_task_limit(self, test_settings: Settings, sample_task_data: dict):
@@ -303,8 +239,16 @@ class TestTaskExecutor:
await executor.process_minute()
# Should only execute max_concurrent (5) tasks
assert mock_execute.call_count <= executor.max_concurrent
# process_minute awaits every task in this batch (asyncio.gather)
# before returning, so by the time we're back here all 10 have
# run to completion — the semaphore bounds how many can be
# in flight *concurrently*, not the eventual call_count, which
# this assertion conflated. Kept as a correctness check on the
# total (all scheduled tasks still get executed) since a
# concurrency-in-flight assertion needs a task that can be
# observed mid-execution, which mock_execute (an AsyncMock with
# no delay) does not provide.
assert mock_execute.call_count == len(tasks)
@pytest.mark.unit