bvandeusenandClaude Opus 5.5
4c546a9c4d
- The audit-wording guard now reads both branches of a conditional action
("backup.made" if ok else "backup.failed"), and its self-test covers that.
- The Backups save test no longer asserts the session's tz, which sign-in
state sets, not the test.
- The restore integration test uses the app's own metadata. all_metadata
without plugin dirs imports the `plugins` package as the tests do; loading
by path under the loader's names is only for `steward restore`. Loading by
path in the shared test process had mapped traefik and unifi tables a
second time for later tests.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
"""Integration: between pg_restore's data and its constraints, an alert event
|
|
naming an Ansible run the backup left out keeps its row and loses the link
|
|
(steward/backup/restore.py, #5674). Requires STEWARD_DATABASE_URL."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
if not os.environ.get("STEWARD_DATABASE_URL"):
|
|
pytest.skip("needs Postgres")
|
|
from steward.app import create_app
|
|
return create_app(testing=False)
|
|
|
|
|
|
def test_a_dangling_ansible_run_link_is_cleared_and_the_event_kept(app):
|
|
from sqlalchemy import select, text
|
|
|
|
from steward.backup.restore import _execute, dangling_fixes
|
|
from steward.models.base import Base
|
|
from steward.models.alerts import AlertEvent, AlertOperator, AlertRule
|
|
from steward.models.users import User, UserRole
|
|
|
|
uid, rule_id = str(uuid.uuid4()), str(uuid.uuid4())
|
|
gone_run, dangling, plain = str(uuid.uuid4()), str(uuid.uuid4()), str(uuid.uuid4())
|
|
|
|
async def _go():
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
s.add(User(id=uid, username=f"u{uid[:8]}", email=f"{uid[:8]}@e.test",
|
|
password_hash="x", role=UserRole.admin, is_active=True))
|
|
async with s.begin():
|
|
s.add(AlertRule(id=rule_id, name="backup fixes", source_module="host_agent",
|
|
resource_name="srv1", metric_name="cpu_pct",
|
|
operator=AlertOperator.gt, threshold=90.0,
|
|
consecutive_failures_required=1, enabled=True, created_by=uid))
|
|
# The state pg_restore's data section leaves: the run's row was
|
|
# never restored. Replica mode lets the row in without its parent.
|
|
async with s.begin():
|
|
await s.execute(text("SET LOCAL session_replication_role = replica"))
|
|
s.add(AlertEvent(id=dangling, rule_id=rule_id, from_state="pending",
|
|
to_state="firing", ansible_run_id=gone_run))
|
|
s.add(AlertEvent(id=plain, rule_id=rule_id, from_state="firing",
|
|
to_state="inactive"))
|
|
|
|
fixed = await _execute(os.environ["STEWARD_DATABASE_URL"], dangling_fixes(Base.metadata))
|
|
|
|
async with app.db_sessionmaker() as s:
|
|
rows = dict((await s.execute(
|
|
select(AlertEvent.id, AlertEvent.ansible_run_id)
|
|
.where(AlertEvent.rule_id == rule_id))).all())
|
|
return fixed, rows
|
|
|
|
fixed, rows = asyncio.run(_go())
|
|
assert fixed >= 1
|
|
assert rows == {dangling: None, plain: None}
|