390 lines
13 KiB
Python
390 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from backup_tool.db.models import (
|
|
Backup,
|
|
Execution,
|
|
Job,
|
|
NotificationDelivery,
|
|
NotificationEvent,
|
|
NotificationSubscription,
|
|
Repository,
|
|
Schedule,
|
|
Source,
|
|
)
|
|
from backup_tool.execution import record_event
|
|
from backup_tool.notifications.events import EVENT_CATALOG
|
|
from backup_tool.scheduler import deliver_due
|
|
from backup_tool.worker import Worker
|
|
from sqlalchemy import select
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execution_catalog_events_are_produced_and_delivered(app_client) -> None:
|
|
"""Exercise the execution producer, not emit_event(), for each live execution type."""
|
|
client, _ = app_client
|
|
setup = await client.post(
|
|
"/api/v2/setup",
|
|
json={"username": "admin", "password": "correct horse battery staple"},
|
|
)
|
|
assert setup.status_code == 201
|
|
app = client._transport.app
|
|
expected = {
|
|
"execution.queued",
|
|
"execution.started",
|
|
"execution.committed",
|
|
"execution.failed",
|
|
"execution.cancelled",
|
|
"execution.retry_queued",
|
|
"execution.worker_recovered",
|
|
}
|
|
assert expected <= set(EVENT_CATALOG)
|
|
deferred_prefixes = ["source.", "gc.", "reconciliation."]
|
|
assert not any(item.startswith(tuple(deferred_prefixes)) for item in EVENT_CATALOG)
|
|
async with app.state.sessions() as db:
|
|
repository = Repository(
|
|
name="events-repository",
|
|
root="/events-repository",
|
|
format_version=1,
|
|
compression="none",
|
|
encryption="none",
|
|
)
|
|
source = Source(
|
|
name="events-source",
|
|
kind="local",
|
|
public_config={"root": "/events-source"},
|
|
secret_refs=[],
|
|
)
|
|
db.add_all([repository, source])
|
|
await db.flush()
|
|
job = Job(
|
|
name="events-job",
|
|
source_id=source.id,
|
|
repository_id=repository.id,
|
|
exclusions=[],
|
|
retention={},
|
|
requested_mode="full",
|
|
)
|
|
subscription = NotificationSubscription(
|
|
channel="email",
|
|
event_filters=["execution.*"],
|
|
destination_config={"recipients": ["operator@example.test"]},
|
|
rate_limit_per_minute=60,
|
|
rate_tokens=60.0,
|
|
)
|
|
db.add_all([job, subscription])
|
|
await db.flush()
|
|
cases = (
|
|
("queued", 1, None, "execution.queued"),
|
|
("preparing", 1, None, "execution.started"),
|
|
("committed", 1, None, "execution.committed"),
|
|
("failed", 1, "transient_io", "execution.failed"),
|
|
("cancelled", 1, "cancellation_requested", "execution.cancelled"),
|
|
("queued", 2, None, "execution.retry_queued"),
|
|
("queued", 1, "worker_lost", "execution.worker_recovered"),
|
|
)
|
|
for state, attempt, reason, _event_type in cases:
|
|
execution = Execution(
|
|
job_id=job.id,
|
|
trigger="manual",
|
|
state=state,
|
|
attempt=attempt,
|
|
reason_code=reason,
|
|
progress={},
|
|
)
|
|
db.add(execution)
|
|
await db.flush()
|
|
await record_event(db, execution)
|
|
if state in {"queued", "preparing"}:
|
|
execution.state = "failed"
|
|
execution.reason_code = "test_cleanup"
|
|
await db.flush()
|
|
await db.commit()
|
|
event_statement = select(NotificationEvent.type).where(NotificationEvent.type.in_(expected))
|
|
event_types = set((await db.scalars(event_statement)).all())
|
|
deliveries = await db.scalar(
|
|
select(NotificationDelivery.id)
|
|
.join(NotificationEvent, NotificationDelivery.event_id == NotificationEvent.id)
|
|
.where(NotificationEvent.type.in_(expected))
|
|
.limit(1)
|
|
)
|
|
assert event_types == expected
|
|
assert deliveries is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schedule_catalog_events_are_produced_and_delivered(app_client) -> None:
|
|
client, _ = app_client
|
|
setup = await client.post(
|
|
"/api/v2/setup",
|
|
json={"username": "admin", "password": "correct horse battery staple"},
|
|
)
|
|
assert setup.status_code == 201
|
|
csrf = client.cookies["backup_tool_csrf"]
|
|
app = client._transport.app
|
|
expected = {
|
|
"schedule.created",
|
|
"schedule.updated",
|
|
"schedule.deleted",
|
|
"schedule.enabled",
|
|
"schedule.disabled",
|
|
"schedule.occurrence_enqueued",
|
|
"schedule.occurrence_misfired",
|
|
"schedule.occurrence_blocked",
|
|
}
|
|
assert expected <= set(EVENT_CATALOG)
|
|
async with app.state.sessions() as db:
|
|
repository = Repository(
|
|
name="schedule-repository",
|
|
root="/schedule-repository",
|
|
format_version=1,
|
|
compression="none",
|
|
encryption="none",
|
|
)
|
|
source = Source(
|
|
name="schedule-source",
|
|
kind="local",
|
|
public_config={"root": "/schedule-source"},
|
|
secret_refs=[],
|
|
)
|
|
db.add_all([repository, source])
|
|
await db.flush()
|
|
jobs = [
|
|
Job(
|
|
name=f"schedule-job-{number}",
|
|
source_id=source.id,
|
|
repository_id=repository.id,
|
|
exclusions=[],
|
|
retention={},
|
|
requested_mode="full",
|
|
enabled=number not in {2, 3},
|
|
)
|
|
for number in range(1, 5)
|
|
]
|
|
subscription = NotificationSubscription(
|
|
channel="email",
|
|
event_filters=["schedule.*"],
|
|
destination_config={"recipients": ["operator@example.test"]},
|
|
rate_limit_per_minute=60,
|
|
rate_tokens=60.0,
|
|
)
|
|
db.add_all([*jobs, subscription])
|
|
await db.commit()
|
|
job_ids = [job.id for job in jobs]
|
|
|
|
created = await client.post(
|
|
f"/api/v2/jobs/{job_ids[0]}/schedule",
|
|
json={"cron": "0 0 * * *", "timezone": "UTC", "enabled": True},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert created.status_code == 201
|
|
disabled = await client.patch(
|
|
f"/api/v2/jobs/{job_ids[0]}/schedule",
|
|
json={"cron": "0 0 * * *", "timezone": "UTC", "enabled": False},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert disabled.status_code == 200
|
|
enabled = await client.patch(
|
|
f"/api/v2/jobs/{job_ids[0]}/schedule",
|
|
json={"cron": "1 0 * * *", "timezone": "UTC", "enabled": True},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert enabled.status_code == 200
|
|
updated = await client.patch(
|
|
f"/api/v2/jobs/{job_ids[0]}/schedule",
|
|
json={"cron": "2 0 * * *", "timezone": "UTC", "enabled": True},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert (
|
|
await client.delete(f"/api/v2/jobs/{job_ids[0]}/schedule", headers={"X-CSRF-Token": csrf})
|
|
).status_code == 204
|
|
|
|
async with app.state.sessions() as db:
|
|
now = datetime.now(UTC)
|
|
db.add_all(
|
|
[
|
|
Schedule(
|
|
job_id=job_ids[1],
|
|
cron="* * * * *",
|
|
timezone="UTC",
|
|
misfire_grace_seconds=0,
|
|
enabled=True,
|
|
next_nominal_at=now - timedelta(hours=1),
|
|
),
|
|
Schedule(
|
|
job_id=job_ids[2],
|
|
cron="* * * * *",
|
|
timezone="UTC",
|
|
misfire_grace_seconds=60,
|
|
enabled=True,
|
|
next_nominal_at=now,
|
|
),
|
|
Schedule(
|
|
job_id=job_ids[3],
|
|
cron="* * * * *",
|
|
timezone="UTC",
|
|
misfire_grace_seconds=60,
|
|
enabled=True,
|
|
next_nominal_at=now,
|
|
),
|
|
]
|
|
)
|
|
await db.commit()
|
|
assert await deliver_due(db, now=now) == 1
|
|
statement = select(NotificationEvent.type).where(NotificationEvent.type.in_(expected))
|
|
event_types = set((await db.scalars(statement)).all())
|
|
delivery = await db.scalar(
|
|
select(NotificationDelivery.id)
|
|
.join(NotificationEvent, NotificationDelivery.event_id == NotificationEvent.id)
|
|
.where(NotificationEvent.type.in_(expected))
|
|
.limit(1)
|
|
)
|
|
assert event_types == expected
|
|
assert delivery is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backup_restore_and_retention_events_are_produced_and_delivered(
|
|
app_client,
|
|
) -> None:
|
|
client, settings = app_client
|
|
source_root = settings.local_source_roots[0] / "notification-project"
|
|
source_root.mkdir()
|
|
(source_root / "data.txt").write_text("notification data\n", encoding="utf-8")
|
|
setup = await client.post(
|
|
"/api/v2/setup",
|
|
json={"username": "admin", "password": "correct horse battery staple"},
|
|
)
|
|
assert setup.status_code == 201
|
|
headers = {"X-CSRF-Token": client.cookies["backup_tool_csrf"]}
|
|
subscription = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "email",
|
|
"event_filters": ["backup.*", "restore.*", "retention.*"],
|
|
"destination": {"recipients": ["operator@example.test"]},
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert subscription.status_code == 201
|
|
repository = await client.post(
|
|
"/api/v2/repositories",
|
|
json={"name": "notification-repo", "relative_path": "notification-repo"},
|
|
headers=headers,
|
|
)
|
|
source = await client.post(
|
|
"/api/v2/sources",
|
|
json={
|
|
"name": "notification-source",
|
|
"kind": "local",
|
|
"public_config": {"root": str(source_root)},
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert repository.status_code == source.status_code == 201
|
|
job = await client.post(
|
|
"/api/v2/jobs",
|
|
json={
|
|
"name": "notification-job",
|
|
"source_id": source.json()["id"],
|
|
"repository_id": repository.json()["id"],
|
|
"requested_mode": "full",
|
|
"exclusions": [],
|
|
"retention": {"keep_last": 1},
|
|
"allow_empty": False,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert job.status_code == 201
|
|
execution = await client.post(f"/api/v2/jobs/{job.json()['id']}/executions", headers=headers)
|
|
assert execution.status_code == 202
|
|
worker = Worker(settings, owner="notification-backup-worker")
|
|
try:
|
|
assert await worker.run_once()
|
|
finally:
|
|
await worker.engine.dispose()
|
|
app = client._transport.app
|
|
async with app.state.sessions() as db:
|
|
backup = await db.scalar(
|
|
select(Backup).where(Backup.execution_id == execution.json()["id"])
|
|
)
|
|
assert backup is not None
|
|
# A synthetic older catalog entry is valid business state; tombstoning is
|
|
# performed only through the real retention producer below.
|
|
older_execution = Execution(
|
|
job_id=job.json()["id"],
|
|
trigger="manual",
|
|
state="committed",
|
|
progress={},
|
|
)
|
|
db.add(older_execution)
|
|
await db.flush()
|
|
older = Backup(
|
|
execution_id=older_execution.id,
|
|
manifest_id="00000000-0000-7000-8000-000000000001",
|
|
manifest_digest="0" * 64,
|
|
logical_bytes=0,
|
|
stored_bytes=0,
|
|
integrity="verified",
|
|
created_at=datetime.now(UTC) - timedelta(days=1),
|
|
)
|
|
db.add(older)
|
|
await db.commit()
|
|
|
|
retention_worker = Worker(settings, owner="notification-retention-worker")
|
|
try:
|
|
# Retention/GC is executed by worker maintenance, not a direct helper call.
|
|
assert await retention_worker.run_once()
|
|
finally:
|
|
await retention_worker.engine.dispose()
|
|
|
|
restore = await client.post(
|
|
f"/api/v2/backups/{backup.id}/restores",
|
|
json={
|
|
"destination": str(settings.restore_roots[0] / "notification-restore"),
|
|
"selection": [],
|
|
"dry_run": True,
|
|
"overwrite_policy": "fail",
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert restore.status_code == 202
|
|
restore_worker = Worker(settings, owner="notification-restore-worker")
|
|
try:
|
|
assert await restore_worker.run_once()
|
|
finally:
|
|
await restore_worker.engine.dispose()
|
|
|
|
expected = {
|
|
"backup.committed",
|
|
"backup.verification_succeeded",
|
|
"restore.queued",
|
|
"restore.committed",
|
|
"retention.tombstoned",
|
|
}
|
|
async with app.state.sessions() as db:
|
|
types = set(
|
|
(
|
|
await db.scalars(
|
|
select(NotificationEvent.type).where(NotificationEvent.type.in_(expected))
|
|
)
|
|
).all()
|
|
)
|
|
deliveries = list(
|
|
(
|
|
await db.scalars(
|
|
select(NotificationDelivery.id)
|
|
.join(
|
|
NotificationEvent,
|
|
NotificationDelivery.event_id == NotificationEvent.id,
|
|
)
|
|
.where(NotificationEvent.type.in_(expected))
|
|
)
|
|
).all()
|
|
)
|
|
assert types == expected
|
|
assert len(deliveries) >= len(expected)
|