60 lines
1.6 KiB
Python
60 lines
1.6 KiB
Python
"""Tests for database engine — reaping and migrations."""
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reap_stale_analysis_resets_stuck_findings(db_session):
|
|
from guarddog_nexus.db.models import Finding
|
|
|
|
stuck = Finding(
|
|
scan_id=1,
|
|
data={"rule": "test", "severity": "WARNING", "message": "test"},
|
|
report={"status": "analyzing"},
|
|
)
|
|
db_session.add(stuck)
|
|
await db_session.commit()
|
|
|
|
from guarddog_nexus.db.engine import _engine
|
|
|
|
async with _engine.begin() as conn:
|
|
pass # ensure tables exist in _engine too
|
|
|
|
await db_session.execute(
|
|
text(
|
|
"UPDATE findings SET report = NULL "
|
|
"WHERE report IS NOT NULL "
|
|
"AND json_extract(report, '$.status') = 'analyzing'"
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
await db_session.refresh(stuck)
|
|
assert stuck.report is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reap_stale_analysis_spares_completed_reports(db_session):
|
|
from guarddog_nexus.db.models import Finding
|
|
|
|
valid = Finding(
|
|
scan_id=1,
|
|
data={"rule": "test", "severity": "WARNING", "message": "test"},
|
|
report={"verdict": "safe", "summary": "ok"},
|
|
)
|
|
db_session.add(valid)
|
|
await db_session.commit()
|
|
|
|
await db_session.execute(
|
|
text(
|
|
"UPDATE findings SET report = NULL "
|
|
"WHERE report IS NOT NULL "
|
|
"AND json_extract(report, '$.status') = 'analyzing'"
|
|
)
|
|
)
|
|
await db_session.commit()
|
|
|
|
await db_session.refresh(valid)
|
|
assert valid.report == {"verdict": "safe", "summary": "ok"}
|