akif/arclet-boards

Arclet Boards — private Kanban workspaces with AppID, DataK3, shared web and MCP actions.

A remix template · 1 remixes · Remix akif/arclet-boards

tests/test_extras.py

import json
from copy import deepcopy

import pytest

from app.domain import model as m
from tests.support import OWNER, MEMBER, VIEWER, OUTSIDER, call, fixture_board
from tests.test_application import add


class Assistant:
    def __init__(self):
        self.calls = []
        self.hook = None
        self.focus_ids = None

    def checklist(self,title,description):
        self.calls.append(("checklist",title,description))
        return ["Review the audience","Write the first draft","Ask for feedback"]

    def focus(self,cards):
        self.calls.append(("focus",cards))
        if self.hook:
            self.hook()
        return self.focus_ids if self.focus_ids is not None else [row["id"] for row in cards[:2]]

    def embed(self,texts):
        self.calls.append(("embed",texts))
        if self.hook:
            self.hook()
        return [json.dumps(([1.0,0.0] if "alpha" in text else
                            [.9,.1] if "beta" in text else [0.0,1.0]) + [0.0]*510) for text in texts]


@pytest.fixture
def fixture():
    service,workspace,board = fixture_board()
    service.assistant = Assistant()
    return service,workspace,board


def test_dependency_cycle_is_prevented_and_chain_explained(fixture):
    service,_,board = fixture
    a,b,c = [add(service,board,title=title) for title in ("Launch","Review","Draft")]
    call(service,"add_dependency",MEMBER,card_id=a["id"],depends_on_id=b["id"],version=1)
    call(service,"add_dependency",MEMBER,card_id=b["id"],depends_on_id=c["id"],version=1)
    with pytest.raises(m.Invalid,match="cycle"):
        call(service,"add_dependency",card_id=c["id"],depends_on_id=a["id"],version=1)
    chain = call(service,"get_blocking_chain",VIEWER,card_id=a["id"])
    assert chain["blocked"] and [row["title"] for row in chain["chains"][0]] == ["Launch","Review","Draft"]
    assert "waits for" in chain["explanation"]


def test_dependency_self_link_viewer_and_outsider_are_refused(fixture):
    service,_,board = fixture
    a,b = add(service,board),add(service,board,title="Other")
    with pytest.raises(m.Invalid):
        call(service,"add_dependency",card_id=a["id"],depends_on_id=a["id"],version=1)
    with pytest.raises(m.Forbidden):
        call(service,"add_dependency",VIEWER,card_id=a["id"],depends_on_id=b["id"],version=1)
    with pytest.raises(m.NotFound):
        call(service,"get_blocking_chain",OUTSIDER,card_id=a["id"])


def test_cross_board_dependency_is_not_allowed_even_for_owner_of_both(fixture):
    service,workspace,board = fixture
    a = add(service,board)
    other = call(service,"create_board",workspace_id=workspace["id"],title="Other")
    other = call(service,"get_board",board_id=other["id"])
    b = add(service,other)
    with pytest.raises(m.NotFound):
        call(service,"add_dependency",card_id=a["id"],depends_on_id=b["id"],version=1)


def test_blocked_card_cannot_be_completed_until_prerequisite_is_done(fixture):
    service,_,board = fixture
    a,b = add(service,board,title="Launch"),add(service,board,title="Review")
    call(service,"add_dependency",card_id=a["id"],depends_on_id=b["id"],version=1)
    fresh = call(service,"get_board",board_id=board["id"])
    with pytest.raises(m.Invalid,match="prerequisites"):
        call(service,"move_card",card_id=a["id"],list_id=board["lists"][2]["id"],index=0,
             version=2,board_version=fresh["version"])
    call(service,"move_card",card_id=b["id"],list_id=board["lists"][2]["id"],index=0,
         version=1,board_version=fresh["version"])
    fresh = call(service,"get_board",board_id=board["id"])
    a = call(service,"get_card",card_id=a["id"])
    assert not a["blocking_chain"]["blocked"]
    call(service,"move_card",card_id=a["id"],list_id=board["lists"][2]["id"],index=1,
         version=a["version"],board_version=fresh["version"])


def test_dependency_removal_is_versioned_and_logged(fixture):
    service,_,board = fixture
    a,b = add(service,board),add(service,board,title="Other")
    call(service,"add_dependency",card_id=a["id"],depends_on_id=b["id"],version=1)
    call(service,"remove_dependency",card_id=a["id"],depends_on_id=b["id"],version=2)
    fresh = call(service,"get_card",card_id=a["id"])
    assert fresh["dependencies"] == []
    assert fresh["activity"][-1]["action"] == "dependency_removed"


def test_reopening_prerequisite_does_not_silently_block_a_completed_dependent(fixture):
    service,_,board = fixture
    a,b = add(service,board,title="Launch"),add(service,board,title="Review")
    call(service,"add_dependency",card_id=a["id"],depends_on_id=b["id"],version=1)
    for card_id in (b["id"],a["id"]):
        fresh = call(service,"get_board",board_id=board["id"])
        card = call(service,"get_card",card_id=card_id)
        call(service,"move_card",card_id=card_id,list_id=board["lists"][2]["id"],index=0,
             version=card["version"],board_version=fresh["version"])
    fresh = call(service,"get_board",board_id=board["id"])
    b = call(service,"get_card",card_id=b["id"])
    with pytest.raises(m.Invalid,match="dependent"):
        call(service,"move_card",card_id=b["id"],list_id=board["lists"][0]["id"],index=0,
             version=b["version"],board_version=fresh["version"])


def test_checklist_draft_uses_description_without_persisting_until_reviewed(fixture):
    service,_,board = fixture
    card = add(service,board,description="Explain the launch to our readers")
    before = deepcopy(service.store.rows)
    draft = call(service,"suggest_checklist",card_id=card["id"])
    assert service.store.rows == before
    assert service.assistant.calls[0] == ("checklist",card["title"],card["description"])
    assert draft["persisted"] is False
    call(service,"apply_checklist",MEMBER,card_id=card["id"],items=draft["suggested_checklist"],version=1)
    fresh = call(service,"get_card",card_id=card["id"])
    assert len(fresh["checklist"]) == 3 and fresh["activity"][-1]["action"] == "checklist_items_added"


def test_unauthorized_ai_requests_never_reach_the_model(fixture):
    service,_,board = fixture
    card = add(service,board)
    for action,args in (("suggest_checklist",{"card_id":card["id"]}),
        ("daily_summary",{"board_id":board["id"]}),
        ("refresh_semantic_index",{"board_id":board["id"],"version":board["version"]})):
        with pytest.raises(m.NotFound):
            call(service,action,OUTSIDER,**args)
    assert service.assistant.calls == []


def test_viewer_cannot_apply_a_draft_or_refresh_embeddings(fixture):
    service,_,board = fixture
    card = add(service,board)
    with pytest.raises(m.Forbidden):
        call(service,"apply_checklist",VIEWER,card_id=card["id"],items=["No"],version=1)
    with pytest.raises(m.Forbidden):
        call(service,"refresh_semantic_index",VIEWER,board_id=board["id"],version=board["version"])
    assert service.assistant.calls == []


def test_summary_counts_are_grounded_and_unknown_model_ids_are_discarded(fixture):
    service,_,board = fixture
    card = add(service,board,due_date="2020-01-01")
    summary = call(service,"daily_summary",MEMBER,board_id=board["id"])
    assert summary["counts"] == {"total":1,"completed":0,"overdue":1,"changed_today":1}
    assert summary["focus_cards"][0]["id"] == card["id"]
    service.assistant.focus_ids = ["invented-card-id"]
    with pytest.raises(m.Unavailable,match="unknown card"):
        call(service,"daily_summary",board_id=board["id"])


def test_access_is_rechecked_after_model_call(fixture):
    service,_,board = fixture
    add(service,board)
    service.assistant.hook = lambda: call(service,"remove_member",board_id=board["id"],email=MEMBER.email)
    with pytest.raises(m.NotFound):
        call(service,"daily_summary",MEMBER,board_id=board["id"])


def test_semantic_cache_and_search_exclude_stale_cards(fixture):
    service,_,board = fixture
    a,b,c = [add(service,board,title=title) for title in ("alpha launch","beta release","gamma budget")]
    assert call(service,"find_similar_cards",card_id=a["id"])["needs_index"]
    fresh = call(service,"get_board",board_id=board["id"])
    call(service,"refresh_semantic_index",board_id=board["id"],version=fresh["version"])
    matches = call(service,"find_similar_cards",VIEWER,card_id=a["id"])
    assert [row["id"] for row in matches["cards"]] == [b["id"],c["id"]]
    call(service,"update_card",card_id=b["id"],title="Changed",description="Different",assignee=None,
         due_date=None,labels=[],version=1)
    assert b["id"] not in {row["id"] for row in call(service,"find_similar_cards",card_id=a["id"])["cards"]}


def test_changed_source_during_embedding_never_persists_stale_vectors(fixture):
    service,_,board = fixture
    card = add(service,board)
    fresh = call(service,"get_board",board_id=board["id"])
    service.assistant.hook = lambda: call(service,"set_due_date",card_id=card["id"],due_date="2026-12-01",version=1)
    with pytest.raises(m.Conflict):
        call(service,"refresh_semantic_index",board_id=board["id"],version=fresh["version"])
    assert service.store.rows["card_embeddings"] == {}