"""Pair custody stays preview-only until a writer identity is admitted."""

import importlib.util
import json
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location(
    "pair_custody", ROOT / "scripts/application_backup_pair_custody.py"
)
custody = importlib.util.module_from_spec(spec)
spec.loader.exec_module(custody)

PAIR = {
    "AWS_ACCESS_KEY_ID": "SYNTHETICACCESSKEY123",
    "AWS_SECRET_ACCESS_KEY": "SYNTHETICSECRETKEY4567",
}


class Transport:
    def __init__(self):
        self.calls = []
        self.metadata_result = None
        self.version = 1

    def metadata(self, path):
        self.calls.append(("metadata", path))
        return self.metadata_result

    def put_cas0(self, path, body):
        self.calls.append(("put", path, body["options"]["cas"], tuple(body["data"])))
        return {"version": self.version, "request_id": "synthetic-request"}

    def revoke(self):
        self.calls.append(("revoke",))


def identity():
    return {
        "policies": [custody.WRITER_POLICY],
        "identity_policies": [],
        "ttl": 300,
        "num_uses": 2,
    }


@pytest.mark.parametrize("application", sorted(custody.PROFILES))
def test_preview_reads_recorded_contract_and_writes_nothing(application, capsys, monkeypatch):
    monkeypatch.setattr(custody, "provision", lambda *_: pytest.fail("preview provisioned"))
    assert custody.main(["--application", application]) == 0
    receipt = json.loads(capsys.readouterr().out)
    recorded = json.loads(custody.ccr_path(application).read_text(encoding="utf-8"))
    assert receipt["status"] == "preview"
    assert receipt["ccr_status"] == recorded["status"]
    assert receipt["writer_admitted"] is True
    assert receipt["openbao_write"] is False
    assert receipt["provider_keys"] == "not_created"
    assert receipt["effects"] == []
    assert receipt["fields"] == list(custody.FIELDS)
    rendered = json.dumps(receipt)
    assert "SYNTHETIC" not in rendered
    assert "AKIA" not in rendered


def test_unreviewed_writer_makes_no_transport_call(monkeypatch):
    monkeypatch.setattr(custody, "WRITER_ADMITTED", False)
    transport = Transport()
    with pytest.raises(custody.CustodyError, match="writer_identity_unreviewed"):
        custody.provision("email-connect", PAIR, identity(), transport)
    assert transport.calls == []


def test_admitted_writer_uses_one_cas0_write_and_revokes(monkeypatch):
    monkeypatch.setattr(custody, "WRITER_ADMITTED", True)
    transport = Transport()
    receipt = custody.provision("email-connect", dict(PAIR), identity(), transport)
    assert receipt["kv_version"] == 1
    assert receipt["secret_material"] is False
    assert PAIR["AWS_SECRET_ACCESS_KEY"] not in json.dumps(receipt)
    assert transport.calls == [
        ("metadata", "platform/metadata/workloads/email-connect/backup-primary"),
        (
            "put",
            "platform/data/workloads/email-connect/backup-primary",
            0,
            custody.FIELDS,
        ),
        ("revoke",),
    ]


def test_reader_policy_and_existing_version_are_refused(monkeypatch):
    monkeypatch.setattr(custody, "WRITER_ADMITTED", True)
    reader = dict(identity(), policies=["workload-kv-read-email-connect-backup-primary"])
    with pytest.raises(custody.CustodyError, match="writer_identity_unreviewed"):
        custody.provision("email-connect", PAIR, reader, Transport())
    transport = Transport()
    transport.metadata_result = {"current_version": 1}
    with pytest.raises(custody.CustodyError, match="existing_custody_requires_rotation_review"):
        custody.provision("email-connect", PAIR, identity(), transport)
    assert [call[0] for call in transport.calls] == ["metadata"]


def test_writer_policy_is_pair_scoped_and_cannot_read_or_delete():
    text = (ROOT / "openbao/policies/backup-pair-writer.hcl").read_text(encoding="utf-8")
    for application in custody.PROFILES:
        assert f'path "platform/data/workloads/{application}/backup-primary"' in text
        assert f'path "platform/metadata/workloads/{application}/backup-primary"' in text
    capabilities = [
        line.split("=", 1)[1]
        for line in text.splitlines()
        if line.strip().startswith("capabilities")
    ]
    assert capabilities
    assert all("delete" not in line and "list" not in line for line in capabilities)
    assert 'capabilities = ["read"]' in text
    data_blocks = [
        block
        for block in text.split("path ")
        if "platform/data/" in block
    ]
    assert data_blocks
    assert all('["create", "update"]' in block and "read" not in block for block in data_blocks)


@pytest.mark.parametrize(
    "values",
    [
        {"AWS_ACCESS_KEY_ID": PAIR["AWS_ACCESS_KEY_ID"]},
        dict(PAIR, AWS_SECRET_ACCESS_KEY="short"),
        dict(PAIR, AWS_ACCESS_KEY_ID=PAIR["AWS_SECRET_ACCESS_KEY"]),
        dict(PAIR, EXTRA="SYNTHETICEXTRAVALUE1"),
    ],
)
def test_partial_or_embedded_pairs_are_refused(values):
    with pytest.raises(custody.CustodyError):
        custody.validate_pair(values)