84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
from src.models.drift import DriftItem, DriftLevel, DriftReport, DriftStatus
|
|
from src.services.drift_fingerprint_state_service import (
|
|
build_drift_fingerprint_state,
|
|
)
|
|
from src.services.drift_repeat_state import build_drift_repeat_state
|
|
|
|
|
|
def _report(report_id: str = "drift-1", status: DriftStatus = DriftStatus.PENDING) -> DriftReport:
|
|
return DriftReport(
|
|
report_id=report_id,
|
|
namespace="awoooi-prod",
|
|
high_count=1,
|
|
medium_count=32,
|
|
info_count=23,
|
|
status=status,
|
|
items=[
|
|
DriftItem(
|
|
resource_kind="Deployment",
|
|
resource_name="awoooi-api",
|
|
namespace="awoooi-prod",
|
|
field_path="spec.template.spec.volumes",
|
|
git_value=["a"],
|
|
actual_value=["a", "b"],
|
|
drift_level=DriftLevel.HIGH,
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def test_build_state_marks_repeated_pending_human() -> None:
|
|
current = _report("drift-3")
|
|
repeat = build_drift_repeat_state(
|
|
current,
|
|
[_report("drift-1"), _report("drift-2")],
|
|
)
|
|
|
|
state = build_drift_fingerprint_state(current, repeat)
|
|
|
|
assert state["schema_version"] == "drift_fingerprint_state_v1"
|
|
assert state["fingerprint"].startswith("dfp_")
|
|
assert state["occurrences_12h"] == 3
|
|
assert state["fsm_state"] == "pending_human_repeated"
|
|
assert state["writes_drift_status"] is False
|
|
assert state["writes_incident_state"] is False
|
|
assert state["p0_escalation"]["dedup_window_hours"] == 24
|
|
|
|
|
|
def test_build_state_marks_zero_diff_open_pr() -> None:
|
|
report = _report("drift-4")
|
|
repeat = build_drift_repeat_state(report, [])
|
|
state = build_drift_fingerprint_state(
|
|
report,
|
|
repeat,
|
|
open_pr={
|
|
"number": 145,
|
|
"state": "open",
|
|
"merged": False,
|
|
"file_count": 0,
|
|
"commit_count": 0,
|
|
"is_zero_diff": True,
|
|
"html_url": "http://gitea.local/wooo/awoooi/pulls/145",
|
|
},
|
|
)
|
|
|
|
assert state["fsm_state"] == "pr_open_zero_diff"
|
|
assert state["next_step"] == "close_zero_diff_pr_and_prepare_real_yaml_patch"
|
|
assert state["open_pr"]["number"] == 145
|
|
|
|
|
|
def test_build_state_marks_handoff_recorded() -> None:
|
|
report = _report("drift-5")
|
|
repeat = build_drift_repeat_state(report, [])
|
|
state = build_drift_fingerprint_state(
|
|
report,
|
|
repeat,
|
|
latest_handoff={
|
|
"handoff_kind": "manual_investigation",
|
|
"handoff_status": "recorded",
|
|
},
|
|
)
|
|
|
|
assert state["fsm_state"] == "handoff_recorded"
|
|
assert state["next_step"] == "operator_review_handoff_and_execute_manual_plan"
|