151 lines
4.3 KiB
Python
151 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import AbstractContextManager
|
|
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from govoplan_core.core.recovery import RecoveryStatus
|
|
from govoplan_campaign.backend import retention
|
|
|
|
|
|
class _EvidenceSession(AbstractContextManager):
|
|
def __init__(self, job: object | None) -> None:
|
|
self.job = job
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args) -> None:
|
|
return None
|
|
|
|
def get(self, _model, _object_id):
|
|
return self.job
|
|
|
|
|
|
def _recovery(*, storage, storage_key: str | None, local_path: str | None):
|
|
return retention._GeneratedArtifactRecovery(
|
|
operation=Mock(),
|
|
job_id="job-1",
|
|
storage_key=storage_key,
|
|
local_path=local_path,
|
|
storage=storage,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("outcome", "expected_method", "expected_status"),
|
|
[
|
|
("succeeded", "succeed", None),
|
|
("failed", "reject", None),
|
|
("outcome_unknown", "unresolved", RecoveryStatus.OUTCOME_UNKNOWN),
|
|
("recovery_required", "unresolved", RecoveryStatus.RECOVERY_REQUIRED),
|
|
],
|
|
)
|
|
def test_retention_recovery_maps_verified_artifact_state(
|
|
monkeypatch,
|
|
outcome: str,
|
|
expected_method: str,
|
|
expected_status: RecoveryStatus | None,
|
|
) -> None:
|
|
recovery = _recovery(
|
|
storage=Mock(),
|
|
storage_key="campaigns/message.eml",
|
|
local_path=None,
|
|
)
|
|
monkeypatch.setattr(
|
|
retention,
|
|
"_generated_artifact_recovery_evidence",
|
|
lambda _recovery: (
|
|
outcome,
|
|
{
|
|
"verified": outcome != "outcome_unknown",
|
|
"checks": {"artifact_locations_probed": True},
|
|
},
|
|
),
|
|
)
|
|
|
|
retention._finish_generated_artifact_recovery(recovery)
|
|
|
|
method = getattr(recovery.operation, expected_method)
|
|
method.assert_called_once()
|
|
if expected_status is not None:
|
|
assert method.call_args.kwargs["status"] == expected_status
|
|
for other in {"succeed", "reject", "unresolved"} - {expected_method}:
|
|
getattr(recovery.operation, other).assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("object_exists", "metadata_key", "expected"),
|
|
[
|
|
(False, None, "succeeded"),
|
|
(True, "campaigns/message.eml", "failed"),
|
|
(False, "campaigns/message.eml", "recovery_required"),
|
|
(True, None, "recovery_required"),
|
|
],
|
|
)
|
|
def test_retention_recovery_compares_storage_and_database_independently(
|
|
monkeypatch,
|
|
object_exists: bool,
|
|
metadata_key: str | None,
|
|
expected: str,
|
|
) -> None:
|
|
storage = Mock()
|
|
storage.exists.return_value = object_exists
|
|
job = SimpleNamespace(
|
|
eml_storage_key=metadata_key,
|
|
eml_local_path=None,
|
|
)
|
|
monkeypatch.setattr(
|
|
retention,
|
|
"get_database",
|
|
lambda: SimpleNamespace(
|
|
SessionLocal=lambda: _EvidenceSession(job),
|
|
),
|
|
)
|
|
|
|
outcome, evidence = retention._generated_artifact_recovery_evidence(
|
|
_recovery(
|
|
storage=storage,
|
|
storage_key="campaigns/message.eml",
|
|
local_path=None,
|
|
)
|
|
)
|
|
|
|
assert outcome == expected
|
|
assert evidence["verified"] is True
|
|
|
|
|
|
def test_campaign_retention_commits_before_recovery_verification(monkeypatch) -> None:
|
|
events: list[str] = []
|
|
session = Mock()
|
|
session.commit.side_effect = lambda: events.append("commit")
|
|
recovery = Mock()
|
|
|
|
monkeypatch.setattr(retention, "_apply_raw_json_retention", lambda *_args, **_kwargs: {})
|
|
monkeypatch.setattr(retention, "_apply_report_detail_retention", lambda *_args, **_kwargs: {})
|
|
|
|
def apply_eml(*_args, **kwargs):
|
|
kwargs["recovery_operations"].append(recovery)
|
|
return {"metadata_cleared": 1}
|
|
|
|
monkeypatch.setattr(retention, "_apply_eml_retention", apply_eml)
|
|
monkeypatch.setattr(
|
|
retention,
|
|
"_finish_generated_artifact_recoveries",
|
|
lambda _recoveries: events.append("verify"),
|
|
)
|
|
|
|
result = retention.apply_campaign_retention(
|
|
session,
|
|
dry_run=False,
|
|
now=datetime.now(timezone.utc),
|
|
policy_for_campaign_id=lambda _campaign_id: object(),
|
|
)
|
|
|
|
assert result["generated_eml"] == {"metadata_cleared": 1}
|
|
assert events == ["commit", "verify"]
|
|
session.rollback.assert_not_called()
|