state-hub/tests/test_repository_rename_cli.py

from __future__ import annotations

import json
import stat
import sys
import uuid

import pytest

import custodian_cli
import repository_rename_cli as rename_cli


REPO_ID = "11111111-1111-4111-8111-111111111111"
OPERATION_ID = "22222222-2222-4222-8222-222222222222"
CONFIRMATION = f"rename:{REPO_ID}:flex-auth:access-engine"


def _repo() -> dict:
    return {
        "id": REPO_ID,
        "slug": "flex-auth",
        "canonical_slug": "flex-auth",
        "requested_slug": "flex-auth",
        "slug_status": "canonical",
    }


def _operation(phase: str, *, no_op: bool = False) -> dict:
    return {
        "id": OPERATION_ID,
        "repo_id": REPO_ID,
        "phase": phase,
        "old_slug": "flex-auth",
        "new_slug": "access-engine",
        "actor": "helixforge",
        "error_code": None,
        "error_message": None,
        "no_op": no_op,
    }


def _preflight() -> dict:
    return {
        "schema_version": "state-hub.repository-rename-preflight.v1",
        "repo_id": REPO_ID,
        "old_slug": "flex-auth",
        "new_slug": "access-engine",
        "safe_to_apply": True,
        "blockers": [],
        "warnings": [],
        "queued_edge_writes": [],
        "preflight_token": "signed-preflight-token",
    }


def _run(monkeypatch, *args: str) -> None:
    monkeypatch.setattr(sys, "argv", ["statehub", *args])
    custodian_cli.main()


def test_preflight_writes_private_token_file_and_redacts_stdout(
    monkeypatch, tmp_path, capsys
):
    calls = []

    def request(_api_base, method, path, body=None):
        calls.append((method, path, body))
        if method == "GET":
            return _repo()
        return _preflight()

    monkeypatch.setattr(rename_cli, "_api_request", request)
    output = tmp_path / "preflight.json"
    _run(
        monkeypatch,
        "repo", "rename", "preflight", "flex-auth", "access-engine",
        "--operation-id", OPERATION_ID,
        "--output", str(output),
        "--json",
    )

    public = json.loads(capsys.readouterr().out)
    private = json.loads(output.read_text())
    assert public["schema_version"] == rename_cli.CLI_SCHEMA_VERSION
    assert public["operation_id"] == OPERATION_ID
    assert public["next_safe_action"]["action"] == "start-operation"
    assert public["result"]["preflight_token"].startswith("[REDACTED")
    assert "signed-preflight-token" not in json.dumps(public)
    assert private["result"]["preflight_token"] == "signed-preflight-token"
    assert stat.S_IMODE(output.stat().st_mode) == 0o600
    assert calls[1][2] == {
        "new_slug": "access-engine",
        "queued_edge_writes": [],
    }


def test_start_uses_client_operation_id_and_retry_safe_api_body(
    monkeypatch, tmp_path, capsys
):
    preflight_file = tmp_path / "preflight.json"
    preflight_file.write_text(
        json.dumps(
            {
                "schema_version": rename_cli.CLI_SCHEMA_VERSION,
                "command": "preflight",
                "operation_id": OPERATION_ID,
                "repo_id": REPO_ID,
                "result": _preflight(),
            }
        )
    )
    preflight_file.chmod(0o600)
    calls = []

    def request(_api_base, method, path, body=None):
        calls.append((method, path, body))
        if path.startswith("/repository-renames/"):
            raise rename_cli.RenameCLIError(
                "not found", code="repository_rename_not_found", status_code=404
            )
        return _repo() if method == "GET" else _operation("preflighted")

    monkeypatch.setattr(rename_cli, "_api_request", request)
    _run(
        monkeypatch,
        "repo", "rename", "start", "flex-auth", "access-engine",
        "--operation-id", OPERATION_ID,
        "--preflight-file", str(preflight_file),
        "--actor", "helixforge",
        "--confirm", CONFIRMATION,
        "--json",
    )

    result = json.loads(capsys.readouterr().out)
    assert result["phase"] == "preflighted"
    assert result["next_safe_action"]["phase"] == "forge-renamed"
    create_body = calls[2][2]
    assert create_body["operation_id"] == OPERATION_ID
    assert create_body["preflight_token"] == "signed-preflight-token"
    assert create_body["actor"] == "helixforge"
    assert "signed-preflight-token" not in json.dumps(result)


def test_start_retry_uses_operation_journal_after_old_slug_becomes_alias(
    monkeypatch, capsys
):
    calls = []

    def request(_api_base, method, path, body=None):
        calls.append((method, path, body))
        return _operation("statehub-rebound")

    monkeypatch.setattr(rename_cli, "_api_request", request)
    _run(
        monkeypatch,
        "repo",
        "rename",
        "start",
        "flex-auth",
        "access-engine",
        "--operation-id",
        OPERATION_ID,
        "--preflight-file",
        "/already-consumed/preflight.json",
        "--actor",
        "helixforge",
        "--confirm",
        CONFIRMATION,
        "--json",
    )

    result = json.loads(capsys.readouterr().out)
    assert result["phase"] == "statehub-rebound"
    assert result["result"]["no_op"] is True
    assert calls == [
        (
            "GET",
            f"/repository-renames/operations/{OPERATION_ID}",
            None,
        )
    ]


def test_apply_uses_files_and_names_the_next_safe_phase(
    monkeypatch, tmp_path, capsys
):
    evidence = tmp_path / "evidence.json"
    evidence.write_text('{"fresh_clone": true}')
    evidence.chmod(0o600)
    calls = []

    def request(_api_base, method, path, body=None):
        calls.append((method, path, body))
        if method == "GET":
            return _operation("statehub-rebound")
        return _operation("source-synced")

    monkeypatch.setattr(rename_cli, "_api_request", request)
    _run(
        monkeypatch,
        "repo", "rename", "apply", OPERATION_ID,
        "--phase", "source-synced",
        "--confirm", CONFIRMATION,
        "--evidence-file", str(evidence),
        "--json",
    )

    result = json.loads(capsys.readouterr().out)
    assert result["state"] == "planned"
    assert result["phase"] == "source-synced"
    assert result["next_safe_action"]["phase"] == "consumers-verified"
    apply_body = calls[1][2]
    assert apply_body["expected_phase"] == "statehub-rebound"
    assert apply_body["evidence"] == {"fresh_clone": True}


def test_rollback_requires_separate_execute_flag(monkeypatch, capsys):
    calls = []

    def request(_api_base, method, path, body=None):
        calls.append((method, path, body))
        if method == "GET":
            return _operation("source-synced")
        return {
            "safe_to_rollback": True,
            "operation": _operation("rollback-preflight"),
        }

    monkeypatch.setattr(rename_cli, "_api_request", request)
    _run(
        monkeypatch,
        "repo", "rename", "rollback", OPERATION_ID,
        "--confirm", f"rollback:{OPERATION_ID}",
        "--json",
    )
    result = json.loads(capsys.readouterr().out)
    assert result["phase"] == "rollback-preflight"
    assert result["next_safe_action"]["action"] == "execute-rollback"
    assert [method for method, _path, _body in calls] == ["GET", "POST"]


def test_rollback_execute_reaches_terminal_state(monkeypatch, capsys):
    calls = []

    def request(_api_base, method, path, body=None):
        calls.append((method, path, body))
        return (
            _operation("rollback-preflight")
            if method == "GET"
            else _operation("rolled-back")
        )

    monkeypatch.setattr(rename_cli, "_api_request", request)
    _run(
        monkeypatch,
        "repo", "rename", "rollback", OPERATION_ID,
        "--confirm", f"rollback:{OPERATION_ID}",
        "--execute",
        "--json",
    )
    result = json.loads(capsys.readouterr().out)
    assert result["state"] == "rolled-back"
    assert result["next_safe_action"]["action"] == "none"
    assert [method for method, _path, _body in calls] == ["GET", "POST"]


def test_json_error_output_redacts_credentials_and_userinfo(monkeypatch, capsys):
    def request(*_args, **_kwargs):
        raise rename_cli.RenameCLIError(
            "failed via https://user:password@forge.example/repo"
            "?X-Amz-Credential=url-credential&X-Amz-Signature=url-secret",
            code="simulated",
            details={"authorization": "Bearer top-secret", "password": "hidden"},
        )

    monkeypatch.setattr(rename_cli, "_api_request", request)
    with pytest.raises(SystemExit) as exc:
        _run(
            monkeypatch,
            "repo", "rename", "status", OPERATION_ID, "--json",
        )
    assert exc.value.code == 1
    output = capsys.readouterr().out
    assert "user:password" not in output
    assert "top-secret" not in output
    assert "hidden" not in output
    assert "url-secret" not in output
    assert "url-credential" not in output
    assert "[REDACTED]" in output


def test_invalid_confirmation_stops_before_mutating_request(monkeypatch, capsys):
    calls = []

    def request(_api_base, method, path, body=None):
        calls.append((method, path, body))
        return _operation("preflighted")

    monkeypatch.setattr(rename_cli, "_api_request", request)
    with pytest.raises(SystemExit) as exc:
        _run(
            monkeypatch,
            "repo", "rename", "apply", OPERATION_ID,
            "--phase", "forge-renamed",
            "--confirm", "yes",
            "--json",
        )
    assert exc.value.code == 1
    assert [method for method, _path, _body in calls] == ["GET"]
    result = json.loads(capsys.readouterr().out)
    assert result["error"]["code"] == "confirmation_mismatch"


def test_unsafe_preflight_is_machine_detectable(monkeypatch, capsys):
    def request(_api_base, method, _path, body=None):
        del body
        if method == "GET":
            return _repo()
        return {
            **_preflight(),
            "safe_to_apply": False,
            "blockers": [{"code": "target_exists"}],
            "preflight_token": None,
        }

    monkeypatch.setattr(rename_cli, "_api_request", request)
    with pytest.raises(SystemExit) as exc:
        _run(
            monkeypatch,
            "repo",
            "rename",
            "preflight",
            "flex-auth",
            "access-engine",
            "--json",
        )
    assert exc.value.code == 2
    result = json.loads(capsys.readouterr().out)
    assert result["state"] == "failed"
    assert result["next_safe_action"]["action"] == "resolve-blockers"


def test_invalid_operation_id_is_stable_json(monkeypatch, capsys):
    monkeypatch.setattr(
        rename_cli,
        "_api_request",
        lambda *_args, **_kwargs: pytest.fail("API should not be called"),
    )
    with pytest.raises(SystemExit) as exc:
        _run(
            monkeypatch,
            "repo",
            "rename",
            "status",
            "not-a-uuid",
            "--json",
        )
    assert exc.value.code == 1
    result = json.loads(capsys.readouterr().out)
    assert result["schema_version"] == rename_cli.CLI_SCHEMA_VERSION
    assert result["state"] == "failed"
    assert result["error"]["code"] == "invalid_operation_id"


@pytest.mark.parametrize(
    ("operation", "label"),
    [
        (_operation("preflighted"), "PLANNED"),
        (_operation("completed"), "ACHIEVED"),
        ({**_operation("source-synced"), "error_code": "failed"}, "FAILED"),
        (_operation("rolled-back"), "ROLLED-BACK"),
    ],
)
def test_human_status_distinguishes_lifecycle_states(
    monkeypatch, capsys, operation, label
):
    monkeypatch.setattr(
        rename_cli,
        "_api_request",
        lambda *_args, **_kwargs: operation,
    )
    if label == "FAILED":
        with pytest.raises(SystemExit) as exc:
            _run(monkeypatch, "repo", "rename", "status", OPERATION_ID)
        assert exc.value.code == 2
    else:
        _run(monkeypatch, "repo", "rename", "status", OPERATION_ID)
    captured = capsys.readouterr()
    assert f"{label}: repository rename status" in captured.out + captured.err