120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from govoplan_core.core.recovery import RecoveryStatus
|
|
from govoplan_campaign.backend.sending import jobs
|
|
|
|
|
|
@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_delivery_recovery_maps_verified_job_state(
|
|
monkeypatch,
|
|
outcome: str,
|
|
expected_method: str,
|
|
expected_status: RecoveryStatus | None,
|
|
) -> None:
|
|
evidence = {
|
|
"verified": outcome != "recovery_required",
|
|
"checks": {"job_state_reloaded": True},
|
|
"send_status": outcome,
|
|
}
|
|
monkeypatch.setattr(
|
|
jobs,
|
|
"_job_delivery_recovery_evidence",
|
|
lambda _job_id: (outcome, evidence),
|
|
)
|
|
operation = Mock()
|
|
|
|
jobs._finish_job_delivery_recovery(operation, job_id="job-1")
|
|
|
|
method = getattr(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(operation, other).assert_not_called()
|
|
|
|
|
|
@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_imap_recovery_maps_verified_append_state(
|
|
monkeypatch,
|
|
outcome: str,
|
|
expected_method: str,
|
|
expected_status: RecoveryStatus | None,
|
|
) -> None:
|
|
evidence = {
|
|
"verified": outcome != "recovery_required",
|
|
"checks": {"job_state_reloaded": True},
|
|
"imap_status": outcome,
|
|
}
|
|
monkeypatch.setattr(
|
|
jobs,
|
|
"_imap_recovery_evidence",
|
|
lambda _job_id: (outcome, evidence),
|
|
)
|
|
operation = Mock()
|
|
|
|
jobs._finish_imap_append_recovery(operation, job_id="job-1")
|
|
|
|
method = getattr(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(operation, other).assert_not_called()
|
|
|
|
|
|
@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_single_action_recovery_maps_verified_action_state(
|
|
monkeypatch,
|
|
outcome: str,
|
|
expected_method: str,
|
|
expected_status: RecoveryStatus | None,
|
|
) -> None:
|
|
evidence = {
|
|
"verified": outcome != "recovery_required",
|
|
"checks": {"action_state_reloaded": True},
|
|
"action_status": outcome,
|
|
}
|
|
monkeypatch.setattr(
|
|
jobs,
|
|
"_single_action_recovery_evidence",
|
|
lambda _action_id: (outcome, evidence),
|
|
)
|
|
operation = Mock()
|
|
|
|
jobs._finish_single_action_delivery_recovery(operation, action_id="action-1")
|
|
|
|
method = getattr(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(operation, other).assert_not_called()
|