path: root/scripts/tests/test_seed_pearl_workspace.py

"""Tests for seed_pearl_workspace.

The seeding logic splits into a pure planner (given a workspace snapshot, decide
what to create, recreate, or reconcile) and a thin Linear GraphQL client (the
only network boundary). The planner carries the behavior and gets exhaustive
pure tests; the client and seed() orchestration get fake-transport tests.

The state reconcile is shaped by what Linear actually allows, verified live:
teamCreate seeds its own default columns, a team must keep an unstarted state,
the Duplicate state is reserved, and a state's type is immutable. Because the
board orders columns by type category first, the columns need specific types
(Icebox and Triage backlog, Backlog unstarted) to render in board order, and a
column found with the wrong type has to be recreated -- create with the right
type, move its issues, archive the old one. Positions are set with distinct
nonzero values, since Linear ignores a 0.0 position and appends new states high.
"""

import os
import sys

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import seed_pearl_workspace as seed


# --- pure: target definitions -------------------------------------------------

def test_target_states_are_the_six_agile_columns_in_order():
    assert [s["name"] for s in seed.TARGET_STATES] == [
        "Icebox", "Triage", "Backlog", "In Progress", "Done", "Canceled"]


def test_target_state_types_put_backlog_last_in_the_pre_work_group():
    by_name = {s["name"]: s for s in seed.TARGET_STATES}
    # Icebox and Triage are backlog-type; Backlog is the unstarted state, which
    # Linear renders after the backlog columns -> board order Icebox, Triage, Backlog.
    assert by_name["Icebox"]["type"] == "backlog"
    assert by_name["Triage"]["type"] == "backlog"
    assert by_name["Backlog"]["type"] == "unstarted"
    assert by_name["In Progress"]["type"] == "started"
    assert by_name["Done"]["type"] == "completed"
    assert by_name["Canceled"]["type"] == "canceled"
    positions = [s["position"] for s in seed.TARGET_STATES]
    assert positions == sorted(positions)
    assert all(p > 0 for p in positions)          # Linear ignores a 0.0 position
    assert len(set(positions)) == 6               # distinct


def test_sources_take_over_the_right_defaults_on_a_fresh_team():
    by_name = {s["name"]: s for s in seed.TARGET_STATES}
    # On a fresh team Icebox takes over the default Backlog and Backlog takes
    # over the unstarted Todo (keeping its type).
    assert "Backlog" in by_name["Icebox"]["sources"]
    assert "Todo" in by_name["Backlog"]["sources"]


def test_view_specs_names_and_project_scope():
    specs = seed.desired_views("proj-1")
    assert [v["name"] for v in specs] == [
        "Pearl Open Issues", "Pearl Icebox", "Pearl Inbox"]
    for v in specs:
        assert v["filterData"]["project"]["id"]["eq"] == "proj-1"


def test_open_view_excludes_completed_and_cancelled():
    open_view = next(v for v in seed.desired_views("p") if v["name"] == "Pearl Open Issues")
    nin = open_view["filterData"]["state"]["type"]["nin"]
    assert "completed" in nin and "canceled" in nin


def test_icebox_and_inbox_views_filter_to_their_column():
    views = {v["name"]: v for v in seed.desired_views("p")}
    assert views["Pearl Icebox"]["filterData"]["state"]["name"]["eq"] == "Icebox"
    assert views["Pearl Inbox"]["filterData"]["state"]["name"]["eq"] == "Triage"


# --- pure: the state reconcile planner ----------------------------------------

LINEAR_DEFAULTS = [
    {"id": "d1", "name": "Backlog", "type": "backlog"},
    {"id": "d2", "name": "Todo", "type": "unstarted"},
    {"id": "d3", "name": "In Progress", "type": "started"},
    {"id": "d4", "name": "Done", "type": "completed"},
    {"id": "d5", "name": "Canceled", "type": "canceled"},
    {"id": "d6", "name": "Duplicate", "type": "duplicate"},
]


def _entries_by_target(existing):
    return {p["target"]["name"]: p for p in seed.reconcile_state_plan(existing)}


def test_reconcile_against_linear_defaults_reuses_defaults_and_creates_triage():
    plan = _entries_by_target(LINEAR_DEFAULTS)
    # Icebox takes the default backlog Backlog; Backlog takes the unstarted Todo
    # (both type-correct -> update). Triage is genuinely new.
    assert plan["Icebox"]["action"] == "update" and plan["Icebox"]["match_id"] == "d1"
    assert plan["Backlog"]["action"] == "update" and plan["Backlog"]["match_id"] == "d2"
    assert plan["Triage"]["action"] == "create"
    assert plan["In Progress"]["match_id"] == "d3"
    assert plan["Done"]["match_id"] == "d4"
    assert plan["Canceled"]["match_id"] == "d5"


def test_reconcile_exact_name_wins_over_a_source_alias():
    # A live "Backlog" must be claimed by the Backlog target, not stolen by
    # Icebox's "Backlog" source alias.
    existing = [{"id": "s1", "name": "Backlog", "type": "unstarted"},
                {"id": "s2", "name": "Icebox", "type": "backlog"}]
    plan = _entries_by_target(existing)
    assert plan["Backlog"]["match_id"] == "s1"
    assert plan["Icebox"]["match_id"] == "s2"


def test_reconcile_marks_a_wrong_type_match_for_recreate():
    # The live workspace before the fix: Triage is unstarted, Backlog is backlog
    # -- both the wrong type, so both must be recreated.
    existing = [
        {"id": "s1", "name": "Icebox", "type": "backlog"},
        {"id": "s2", "name": "Triage", "type": "unstarted"},
        {"id": "s3", "name": "Backlog", "type": "backlog"},
        {"id": "s4", "name": "In Progress", "type": "started"},
        {"id": "s5", "name": "Done", "type": "completed"},
        {"id": "s6", "name": "Canceled", "type": "canceled"},
    ]
    plan = _entries_by_target(existing)
    assert plan["Triage"]["action"] == "recreate" and plan["Triage"]["match_id"] == "s2"
    assert plan["Backlog"]["action"] == "recreate" and plan["Backlog"]["match_id"] == "s3"
    assert plan["Icebox"]["action"] == "update"


def test_reconcile_is_idempotent_against_correctly_typed_columns():
    correct = [("Icebox", "backlog"), ("Triage", "backlog"), ("Backlog", "unstarted"),
               ("In Progress", "started"), ("Done", "completed"), ("Canceled", "canceled")]
    existing = [{"id": f"s{i}", "name": n, "type": t} for i, (n, t) in enumerate(correct)]
    plan = _entries_by_target(existing)
    assert all(e["action"] == "update" for e in plan.values())


def test_reconcile_matches_case_insensitively():
    existing = [{"id": "s1", "name": "icebox", "type": "backlog"}]
    plan = _entries_by_target(existing)
    assert plan["Icebox"]["match_id"] == "s1"


def test_leftover_state_names_reports_unmatched():
    # Duplicate isn't a target and stays; Todo is claimed (by Backlog), so only
    # Duplicate is reported.
    assert seed.leftover_state_names(LINEAR_DEFAULTS) == ["Duplicate"]


# --- boundary: a fake GraphQL transport ---------------------------------------

class FakeTransport:
    """Records GraphQL calls and replies from a scripted op-name -> response map."""

    def __init__(self, responses):
        self.responses = responses
        self.calls = []

    def __call__(self, query, variables):
        op = next(name for name in self.responses if name in query)
        self.calls.append((op, variables))
        resp = self.responses[op]
        return resp(variables) if callable(resp) else resp

    def ops(self):
        return [op for op, _ in self.calls]


def _ok(field, payload):
    return {"data": {field: {"success": True, **payload}}}


def test_seed_on_empty_workspace_reconciles_then_builds():
    fake = FakeTransport({
        "Teams": {"data": {"teams": {"nodes": []}}},
        "TeamCreate": _ok("teamCreate", {"team": {"id": "t1", "key": "PEARL", "name": "Pearl"}}),
        # states re-fetched after create: Linear's defaults
        "TeamStates": {"data": {"team": {"states": {"nodes": LINEAR_DEFAULTS}}}},
        "StateCreate": _ok("workflowStateCreate", {"workflowState": {"id": "new1"}}),
        "StateUpdate": _ok("workflowStateUpdate", {"workflowState": {"id": "u"}}),
        "TeamUpdate": _ok("teamUpdate", {}),
        "Projects": {"data": {"projects": {"nodes": []}}},
        "ProjectCreate": _ok("projectCreate", {"project": {"id": "p1", "name": "Pearl"}}),
        "CustomViews": {"data": {"customViews": {"nodes": []}}},
        "CustomViewCreate": _ok("customViewCreate", {"customView": {"id": "v1"}}),
    })
    summary = seed.seed(seed.LinearClient("key", transport=fake))
    ops = fake.ops()
    assert ops.count("TeamCreate") == 1
    assert ops.count("StateCreate") == 1            # only Triage is new
    assert ops.count("StateUpdate") == 6            # six columns named/positioned
    assert ops.count("StateArchive") == 0           # nothing to recreate on a fresh team
    assert ops.count("TeamUpdate") == 1             # default new-issue state -> Triage
    assert ops.count("ProjectCreate") == 1
    assert ops.count("CustomViewCreate") == 3
    assert summary["team"] == "t1"
    assert summary["project"] == "p1"
    assert summary["states_created"] == ["Triage"]
    assert summary["states_recreated"] == []
    assert summary["leftover_states"] == ["Duplicate"]
    assert summary["default_issue_state"] == "Triage"
    # the default points at the freshly-created Triage
    team_update = next(v for op, v in fake.calls if op == "TeamUpdate")
    assert team_update["input"]["defaultIssueStateId"] == "new1"


def test_seed_heals_type_drift_by_recreating_and_migrating():
    # The live workspace before the fix: Triage unstarted (the old default,
    # empty), Backlog backlog (holds two issues). Both wrong type -> recreate.
    live_states = [
        {"id": "s1", "name": "Icebox", "type": "backlog", "position": 1.0},
        {"id": "s2", "name": "Triage", "type": "unstarted", "position": 2.0},
        {"id": "s3", "name": "Backlog", "type": "backlog", "position": 3.0},
        {"id": "s4", "name": "In Progress", "type": "started", "position": 4.0},
        {"id": "s5", "name": "Done", "type": "completed", "position": 5.0},
        {"id": "s6", "name": "Canceled", "type": "canceled", "position": 6.0},
        {"id": "s7", "name": "Duplicate", "type": "duplicate", "position": 5.0},
    ]
    new_ids = iter(["nTriage", "nBacklog"])

    def make_state(_v):
        return _ok("workflowStateCreate", {"workflowState": {"id": next(new_ids)}})

    def state_issues(v):
        nodes = [{"id": "i1"}, {"id": "i2"}] if v["id"] == "s3" else []
        return {"data": {"workflowState": {"issues": {
            "nodes": nodes, "pageInfo": {"hasNextPage": False, "endCursor": None}}}}}

    fake = FakeTransport({
        "Teams": {"data": {"teams": {"nodes": [
            {"id": "t1", "key": "PEARL", "name": "Pearl",
             "defaultIssueState": {"id": "s2"},
             "states": {"nodes": live_states}}]}}},
        "StateUpdate": _ok("workflowStateUpdate", {"workflowState": {"id": "u"}}),
        "StateCreate": make_state,
        "StateIssues": state_issues,
        "IssueMove": _ok("issueUpdate", {"issue": {"id": "i"}}),
        "StateArchive": _ok("workflowStateArchive", {}),
        "TeamUpdate": _ok("teamUpdate", {}),
        "Projects": {"data": {"projects": {"nodes": [{"id": "p1", "name": "Pearl"}]}}},
        "CustomViews": {"data": {"customViews": {"nodes": [
            {"id": "v1", "name": "Pearl Open Issues"},
            {"id": "v2", "name": "Pearl Icebox"},
            {"id": "v3", "name": "Pearl Inbox"}]}}},
    })
    summary = seed.seed(seed.LinearClient("key", transport=fake))
    ops = fake.ops()
    assert ops.count("StateCreate") == 2            # new Triage + new Backlog
    assert ops.count("StateArchive") == 2           # old Triage + old Backlog
    assert ops.count("IssueMove") == 2              # the two issues off old Backlog
    # every create happens before any archive, so a new unstarted state exists
    # before the old unstarted one is archived
    assert max(i for i, op in enumerate(ops) if op == "StateCreate") \
        < min(i for i, op in enumerate(ops) if op == "StateArchive")
    assert summary["states_recreated"] == ["Triage", "Backlog"]
    assert summary["issues_moved"] == 2
    # the issues land on the new Backlog, and the default moves to the new Triage
    moves = [v for op, v in fake.calls if op == "IssueMove"]
    assert all(m["state"] == "nBacklog" for m in moves)
    team_update = next(v for op, v in fake.calls if op == "TeamUpdate")
    assert team_update["input"]["defaultIssueStateId"] == "nTriage"


def test_seed_is_idempotent_against_a_correctly_seeded_workspace():
    correct = [("Icebox", "backlog"), ("Triage", "backlog"), ("Backlog", "unstarted"),
               ("In Progress", "started"), ("Done", "completed"),
               ("Canceled", "canceled"), ("Duplicate", "duplicate")]
    seeded_states = [{"id": f"s{i}", "name": n, "type": t, "position": float(i + 1)}
                     for i, (n, t) in enumerate(correct)]
    fake = FakeTransport({
        # default already points at Triage (s1), so no team update fires either
        "Teams": {"data": {"teams": {"nodes": [
            {"id": "t1", "key": "PEARL", "name": "Pearl",
             "defaultIssueState": {"id": "s1"},
             "states": {"nodes": seeded_states}}]}}},
        "StateUpdate": _ok("workflowStateUpdate", {"workflowState": {"id": "u"}}),
        "Projects": {"data": {"projects": {"nodes": [{"id": "p1", "name": "Pearl"}]}}},
        "CustomViews": {"data": {"customViews": {"nodes": [
            {"id": "v1", "name": "Pearl Open Issues"},
            {"id": "v2", "name": "Pearl Icebox"},
            {"id": "v3", "name": "Pearl Inbox"}]}}},
    })
    seed.seed(seed.LinearClient("key", transport=fake))
    ops = fake.ops()
    for absent in ("TeamCreate", "StateCreate", "StateArchive", "ProjectCreate",
                   "CustomViewCreate", "TeamUpdate"):
        assert absent not in ops                     # nothing created, recreated, or repointed


def test_client_raises_on_graphql_errors():
    fake = FakeTransport({"Teams": {"errors": [{"message": "bad key"}]}})
    client = seed.LinearClient("key", transport=fake)
    with pytest.raises(seed.LinearError, match="bad key"):
        client.execute("query Teams { teams { nodes { id } } }", {})


# --- key-file permission warning ----------------------------------------------

def test_warn_key_file_world_readable_warns(tmp_path, capsys):
    p = tmp_path / "apikey.txt"
    p.write_text("lin_api_secret")
    os.chmod(p, 0o644)
    assert seed.warn_if_key_file_world_readable(str(p)) is True
    assert "600" in capsys.readouterr().err


def test_warn_key_file_mode_600_is_silent(tmp_path, capsys):
    p = tmp_path / "apikey.txt"
    p.write_text("lin_api_secret")
    os.chmod(p, 0o600)
    assert seed.warn_if_key_file_world_readable(str(p)) is False
    assert capsys.readouterr().err == ""


def test_warn_key_file_absent_is_silent(tmp_path):
    assert seed.warn_if_key_file_world_readable(str(tmp_path / "nope.txt")) is False