eduard/cobra-boards

Cobra Boards, team task boards you can run from Claude or ChatGPT. FastAPI + DataK3 + appid + Dodil Models + MCP.

• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix eduard/cobra-boards

scripts/live_check.py

"""End-to-end check against REAL Dodil services: DataK3 and Dodil Models.

Run it against a TEST bucket, never the live one:
    BUCKET=cobra-dev DODIL_SERVICE_ACCOUNT_ID=... DODIL_SERVICE_ACCOUNT_SECRET=... \
    python scripts/live_check.py

It plays a small team story through the same use cases the web page and MCP use, then checks
what DataK3 actually stored. Every board it makes is prefixed "live-check" so it is easy to spot.
"""
import os
import sys
import time

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))

from boards.application.use_cases import (
    Boards,  # noqa: E402
    NotFound,  # noqa: E402
)
from boards.domain.model import Conflict, Forbidden, Person  # noqa: E402
from boards.infrastructure.dodil import DodilObjects, ModelsAssistant  # noqa: E402
from boards.infrastructure.repository import SqlBoards  # noqa: E402

OWNER, MEMBER, VIEWER, STRANGER = (Person(f"{n}@live-check.test")
                                   for n in ("owner", "member", "viewer", "stranger"))


def step(name, fn):
    t = time.time()
    out = fn()
    print(f"  ok  {name}  ({(time.time() - t) * 1000:.0f} ms)")
    return out


def expect_error(name, exc, fn):
    try:
        fn()
    except exc:
        print(f"  ok  {name}  (refused)")
        return
    raise AssertionError(f"{name}: expected {exc.__name__}")


def main():
    if os.environ.get("BUCKET", "").endswith("-data"):
        sys.exit("refusing to run against a live '-data' bucket; use the test copy")
    s = Boards(SqlBoards(), ModelsAssistant(), objects=DodilObjects())
    print(f"live check on bucket {os.environ['BUCKET']}")
    b = step("create board", lambda: s.create_board(OWNER, f"live-check {int(time.time())}"))
    bid = b["id"]
    step("invite member + viewer", lambda: (s.invite(OWNER, bid, MEMBER.email, "member"),
                                            s.invite(OWNER, bid, VIEWER.email, "viewer")))
    inv = step("add card", lambda: s.add_card(MEMBER, bid, "Send invoice to Acme",
                                              description="Q4 retainer, net 30",
                                              assignee=MEMBER.email, due_date="2026-01-01",
                                              labels=["finance"]))
    bill = step("add card 2", lambda: s.add_card(OWNER, bid, "Chase the unpaid bill"))
    logo = step("add card 3", lambda: s.add_card(OWNER, bid, "Design the new logo"))
    assert (inv["number"], bill["number"], logo["number"]) == (1, 2, 3), "numbers"
    step("move card by number", lambda: s.move_card(MEMBER, "2", "Doing", board_ref=bid))
    expect_error("viewer can't edit", Forbidden, lambda: s.add_card(VIEWER, bid, "nope"))
    expect_error("stale edit refused", Conflict,
                 lambda: s.update_card(OWNER, inv["id"], expected_version=0, title="x"))
    step("blocker", lambda: s.add_blocker(OWNER, logo["id"], inv["id"]))
    chain = step("what is blocking", lambda: s.what_is_blocking(VIEWER, logo["id"]))
    assert chain["start_with"][0]["number"] == 1, chain
    hist = step("history", lambda: s.card_history(OWNER, bill["id"]))
    assert hist[0]["changes"]["list"] == ["To do", "Doing"], hist
    assert step("overdue", lambda: [c["number"] for c in s.overdue(MEMBER, bid)]) == [1]
    assert s.my_boards(STRANGER) == [] or all(x["id"] != bid for x in s.my_boards(STRANGER))
    sug = step("AI checklist (Dodil Models)", lambda: s.suggest_checklist(MEMBER, inv["id"]))
    print("      ", sug["suggested_steps"])
    step("save 2 steps", lambda: s.add_checklist_items(MEMBER, inv["id"],
                                                       sug["suggested_steps"][:2]))
    sim = step("similar cards (Dodil embeddings + DataK3 vectors)",
               lambda: s.similar_cards(MEMBER, inv["id"], board_ref=bid))
    print("      ", [(c["title"], c["similarity"]) for c in sim["cards"]])
    assert sim["cards"][0]["title"] == "Chase the unpaid bill", sim
    summ = step("daily summary", lambda: s.daily_summary(OWNER, bid))
    print("      ", summ["summary"][:300].replace("\n", " "))

    # v2: comments, notifications, live stamp, files in DataK3 object storage, archive
    stamp = s.board_stamp(VIEWER, bid)["stamp"]
    c = step("comment with @mention", lambda: s.add_comment(OWNER, inv["id"], "@member please send it"))
    assert c["mentioned"] == [MEMBER.email], c
    n = step("notification arrives", lambda: s.my_notifications(MEMBER, unread_only=True))
    assert any(i["kind"] == "mention" for i in n["items"]), n
    assert step("live stamp moved", lambda: s.board_stamp(VIEWER, bid)["stamp"]) != stamp
    step("mark read", lambda: s.mark_notifications_read(MEMBER))
    assert s.my_notifications(MEMBER, unread_only=True)["items"] == []
    att = step("attach a file (object storage)", lambda: s.add_attachment(
        MEMBER, inv["id"], "invoice-draft.txt", b"Invoice draft for Acme", "text/plain"))["attachment"]
    got = step("download it back", lambda: s.download_attachment(VIEWER, att["id"]))
    assert got["data"] == b"Invoice draft for Acme", got
    expect_error("stranger can't fetch it", NotFound, lambda: s.download_attachment(STRANGER, att["id"]))
    step("remove the file", lambda: s.delete_attachment(OWNER, att["id"]))
    step("archive a card", lambda: s.archive_card(OWNER, logo["id"]))
    titles = [x["title"] for lst in s.show_board(VIEWER, bid)["lists"] for x in lst["cards"]]
    assert "Design the new logo" not in titles, titles
    step("restore it", lambda: s.restore_card(OWNER, logo["id"]))
    counts = step("comment counts on the board", lambda: {
        x["number"]: x["comments"] for lst in s.show_board(VIEWER, bid)["lists"] for x in lst["cards"]})
    assert counts[1] == 1, counts
    print("ALL GOOD")


if __name__ == "__main__":
    main()