1
0

fix: test DB isolation, Discord recovery, and over-mocked tests

Test data was bleeding into production tasks.db because
swarm.task_queue.models.DB_PATH (relative path) was never patched in
conftest.clean_database. Fixed by switching to absolute paths via
settings.repo_root and adding the missing module to the patching list.

Discord bot could leak orphaned clients on retry after ERROR state.
Added _cleanup_stale() to close stale client/task before each start()
attempt, with improved logging in the token watcher.

Rewrote test_paperclip_client.py to use httpx.MockTransport instead of
patching _get/_post/_delete — tests now exercise real HTTP status codes,
error handling, and JSON parsing. Added end-to-end test for
capture_error → create_task DB isolation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Trip T
2026-03-11 20:33:59 -04:00
parent 9d9449cdcf
commit ea2dbdb4b5
8 changed files with 253 additions and 66 deletions

View File

@@ -1,5 +1,7 @@
"""Tests for infrastructure.error_capture module."""
import sqlite3
from infrastructure.error_capture import (
_dedup_cache,
_get_git_context,
@@ -117,5 +119,37 @@ class TestCaptureError:
except RuntimeError as e:
capture_error(e, source="test_module", context={"path": "/api/foo"})
def test_capture_creates_task_in_temp_db(self):
"""capture_error should write a task to the isolated temp DB, not production.
This validates that conftest.clean_database properly redirects
swarm.task_queue.models.DB_PATH to the per-test temp directory.
"""
_dedup_cache.clear()
try:
raise ValueError("Test error for DB isolation check")
except ValueError as e:
task_id = capture_error(e, source="test")
# task_id can be None if error_feedback_enabled is False — that's fine,
# but if it was created, verify it landed in the temp DB
if task_id:
import swarm.task_queue.models as tq_models
db_path = tq_models.DB_PATH
conn = sqlite3.connect(str(db_path))
try:
row = conn.execute(
"SELECT id, title FROM tasks WHERE id = ?", (task_id,)
).fetchone()
assert row is not None, (
f"Task {task_id} not found in DB at {db_path}"
"check conftest.clean_database patches swarm.task_queue.models.DB_PATH"
)
assert "[BUG]" in row[1]
finally:
conn.close()
def teardown_method(self):
_dedup_cache.clear()