157 lines
5.2 KiB
Python
157 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from src.services.awooop_ansible_learning_writeback import (
|
|
_existing_trust_readback_matches,
|
|
)
|
|
|
|
|
|
def _learning_receipt_fixture() -> (
|
|
tuple[
|
|
SimpleNamespace,
|
|
dict[str, object],
|
|
dict[str, object],
|
|
dict[str, object],
|
|
]
|
|
):
|
|
playbook = SimpleNamespace(
|
|
playbook_id="PB-ANSIBLE-110-DISK-PRESSURE",
|
|
project_id="awoooi",
|
|
version=1,
|
|
success_count=19,
|
|
failure_count=0,
|
|
trust_score=0.9,
|
|
symptom_pattern={"catalog_revision": "disk-pressure-v1"},
|
|
updated_at=datetime(2026, 7, 15, tzinfo=UTC),
|
|
)
|
|
identity: dict[str, object] = {
|
|
"schema_version": "ansible_playbook_identity_v1",
|
|
"identity_fingerprint": "a" * 64,
|
|
"canonical_playbook_id": playbook.playbook_id,
|
|
"catalog_id": "ansible:110-disk-pressure",
|
|
"playbook_path": (
|
|
"infra/ansible/playbooks/host-disk-pressure-bounded-cleanup.yml"
|
|
),
|
|
}
|
|
receipt_input: dict[str, object] = {
|
|
"schema_version": "ansible_learning_writeback_receipt_v3",
|
|
"automation_run_id": "run-1",
|
|
"incident_id": "INC-1",
|
|
"catalog_id": identity["catalog_id"],
|
|
"canonical_playbook_id": identity["canonical_playbook_id"],
|
|
"identity_schema_version": identity["schema_version"],
|
|
"identity_fingerprint": identity["identity_fingerprint"],
|
|
"playbook_path": identity["playbook_path"],
|
|
"apply_op_id": "11111111-1111-4111-8111-111111111111",
|
|
"verification_result": "success",
|
|
"observation_kind": "controlled_apply",
|
|
"trust_mutation_performed": True,
|
|
"learning_repository": "playbooks",
|
|
"stores_raw_logs": False,
|
|
"stores_secret_values": False,
|
|
}
|
|
trust_receipt: dict[str, object] = {
|
|
"schema_version": "ansible_playbook_trust_writeback_v2",
|
|
"identity_schema_version": identity["schema_version"],
|
|
"identity_fingerprint": identity["identity_fingerprint"],
|
|
"canonical_playbook_id": identity["canonical_playbook_id"],
|
|
"catalog_id": identity["catalog_id"],
|
|
"playbook_path": identity["playbook_path"],
|
|
"playbook_row_version": 1,
|
|
"playbook_row_fingerprint": "b" * 64,
|
|
"verification_result": "success",
|
|
"observation_kind": "controlled_apply",
|
|
"learning_recorded": True,
|
|
"trust_observation_recorded": True,
|
|
"trust_updated": True,
|
|
"trust_mutation_performed": True,
|
|
"repository_write_acknowledged": True,
|
|
"repository_readback_verified": True,
|
|
"operation_receipt_readback_verified": True,
|
|
"durable_write_acknowledged": True,
|
|
"raw_log_payload_stored": False,
|
|
"secret_value_stored": False,
|
|
}
|
|
return playbook, identity, receipt_input, trust_receipt
|
|
|
|
|
|
def _matches(
|
|
playbook: SimpleNamespace,
|
|
identity: dict[str, object],
|
|
receipt_input: dict[str, object],
|
|
trust_receipt: dict[str, object],
|
|
) -> bool:
|
|
return _existing_trust_readback_matches(
|
|
receipt_input,
|
|
trust_receipt,
|
|
playbook=playbook,
|
|
identity=identity,
|
|
automation_run_id="run-1",
|
|
incident_id="INC-1",
|
|
catalog_id="ansible:110-disk-pressure",
|
|
apply_op_id="11111111-1111-4111-8111-111111111111",
|
|
verification_result="success",
|
|
)
|
|
|
|
|
|
def test_historical_receipt_survives_later_playbook_mutation() -> None:
|
|
playbook, identity, receipt_input, trust_receipt = _learning_receipt_fixture()
|
|
|
|
assert _matches(playbook, identity, receipt_input, trust_receipt)
|
|
|
|
playbook.version = 3
|
|
playbook.success_count = 21
|
|
playbook.updated_at = datetime(2026, 7, 16, tzinfo=UTC)
|
|
|
|
assert _matches(playbook, identity, receipt_input, trust_receipt)
|
|
|
|
|
|
def test_historical_receipt_accepts_legacy_same_version_row_drift() -> None:
|
|
playbook, identity, receipt_input, trust_receipt = _learning_receipt_fixture()
|
|
playbook.success_count = 20
|
|
playbook.updated_at = datetime(2026, 7, 16, tzinfo=UTC)
|
|
|
|
assert _matches(playbook, identity, receipt_input, trust_receipt)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "wrong_value"),
|
|
[
|
|
("automation_run_id", "wrong-run"),
|
|
("incident_id", "INC-WRONG"),
|
|
("apply_op_id", "22222222-2222-4222-8222-222222222222"),
|
|
("verification_result", "failed"),
|
|
],
|
|
)
|
|
def test_historical_receipt_rejects_wrong_operation_binding(
|
|
field: str,
|
|
wrong_value: str,
|
|
) -> None:
|
|
playbook, identity, receipt_input, trust_receipt = _learning_receipt_fixture()
|
|
receipt_input[field] = wrong_value
|
|
|
|
assert not _matches(playbook, identity, receipt_input, trust_receipt)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "wrong_value"),
|
|
[
|
|
("playbook_row_version", 0),
|
|
("playbook_row_fingerprint", "not-a-sha256"),
|
|
("repository_readback_verified", False),
|
|
("operation_receipt_readback_verified", False),
|
|
],
|
|
)
|
|
def test_historical_receipt_rejects_weak_durable_readback(
|
|
field: str,
|
|
wrong_value: object,
|
|
) -> None:
|
|
playbook, identity, receipt_input, trust_receipt = _learning_receipt_fixture()
|
|
trust_receipt[field] = wrong_value
|
|
|
|
assert not _matches(playbook, identity, receipt_input, trust_receipt)
|