aps-agent/tests/golden/test_branch_tree.py

# ============================================================
# 项目 -> 会话 -> 分支三级树黄金测试（plan.md §4.x / 矩阵 51 行）
# 覆盖：BranchStore 生命周期（fork/rename/switch/discard/merge）、持久化、
#       gateway 分支 API + 审计、主干保护。
# ============================================================
from __future__ import annotations

from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from server.state.branches import (
    BranchConflictError,
    BranchStore,
)
from tests.auth_provider import install_test_auth


@pytest.fixture()
def secure_app(tmp_path, monkeypatch):
    monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "tenant.db"))
    monkeypatch.setenv("APS_WORLD_PATH", str(tmp_path / "world.json"))
    monkeypatch.setenv("APS_BRANCH_DIR", str(tmp_path / "branches"))
    from server.db.database import reset_engine
    from server.state import store as world_store

    install_test_auth(monkeypatch, "tenant-a-000000000000000000000001")
    world_store._stores.clear()
    reset_engine()
    from server.gateway.app import create_app
    app = create_app()
    yield app
    reset_engine()
    world_store._stores.clear()


def _login(client: TestClient, username: str = "planner") -> None:
    r = client.post("/api/auth/login", json={
        "method": "password", "username": username, "password": "test",
    })
    assert r.status_code == 200


def test_branch_store_lifecycle(tmp_path: Path):
    """分支树生命周期：fork/rename/switch/merge/discard + 主干保护 + 持久化。"""
    path = str(tmp_path / "branches.json")
    bs = BranchStore(path)
    bs.ensure_session("s1")
    t = bs.tree("s1")
    root_id = t["treeRoot"]
    assert t["activeNode"] == root_id

    b1 = bs.fork("s1", "方案A", checkpoint_id="ckpt-1")
    assert b1["parentId"] == root_id
    assert bs.active("s1") == b1["id"]

    bs.rename("s1", b1["id"], "方案A-改")
    assert bs._branch("s1", b1["id"])["name"] == "方案A-改"

    bs.switch("s1", root_id)
    assert bs.active("s1") == root_id

    merged = bs.merge("s1", b1["id"], root_id)
    assert merged["source"]["status"] == "merged"
    assert bs.active("s1") == root_id

    with pytest.raises(BranchConflictError):
        bs.discard("s1", root_id)                     # 主干不可丢弃

    b2 = bs.fork("s1", "临时分支")
    bs.switch("s1", root_id)
    bs.discard("s1", b2["id"])
    assert bs._branch("s1", b2["id"])["status"] == "discarded"

    # 持久化：重新加载
    bs2 = BranchStore(path)
    t2 = bs2.tree("s1")
    assert len(t2["branches"]) == 3                   # root + b1(merged) + b2(discarded)


def test_branch_api_full_flow(secure_app):
    """gateway 分支 API：建会话 -> 读树 -> fork -> switch -> merge -> discard + 审计。"""
    client = TestClient(secure_app)
    _login(client)

    created = client.post("/api/projects", json={"id": "proj-br", "name": "分支项目"})
    assert created.status_code == 200
    sid = created.json()["session"]["id"]

    tree = client.get(f"/api/sessions/{sid}/branches")
    assert tree.status_code == 200
    root_id = tree.json()["tree"]["treeRoot"]

    forked = client.post(f"/api/sessions/{sid}/branches", json={"name": "方案B"})
    assert forked.status_code == 200
    bid = forked.json()["branch"]["id"]

    switched = client.post(f"/api/sessions/{sid}/branches/{bid}/switch")
    assert switched.status_code == 200

    merged = client.post(f"/api/sessions/{sid}/branches/{bid}/merge", json={"targetId": root_id})
    assert merged.status_code == 200
    assert merged.json()["source"]["status"] == "merged"

    f2 = client.post(f"/api/sessions/{sid}/branches", json={"name": "临时"})
    b2 = f2.json()["branch"]["id"]
    client.post(f"/api/sessions/{sid}/branches/{root_id}/switch")
    discarded = client.post(f"/api/sessions/{sid}/branches/{b2}/discard")
    assert discarded.status_code == 200
    assert discarded.json()["branch"]["status"] == "discarded"

    # 分支操作写审计
    audit = client.get("/api/gov/audit?limit=50").json()["events"]
    branch_audits = [e for e in audit if e["action"].startswith("branch.")]
    actions = {e["action"] for e in branch_audits}
    assert {"branch.fork", "branch.switch", "branch.merge", "branch.discard"} <= actions


def test_branch_api_rejects_discard_active(tmp_path, secure_app):
    """API 层：丢弃当前活动分支被拒（409）。"""
    client = TestClient(secure_app)
    _login(client)
    created = client.post("/api/projects", json={"id": "proj-x", "name": "项目X"})
    sid = created.json()["session"]["id"]
    forked = client.post(f"/api/sessions/{sid}/branches", json={"name": "活动分支"})
    bid = forked.json()["branch"]["id"]
    r = client.post(f"/api/sessions/{sid}/branches/{bid}/discard")
    assert r.status_code == 409