akif/arclet-boards

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

• dodil-challenge
• dodil-challenge-boards

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

tests/support.py

"""Test-only transaction double. The deployed app always constructs DataK3Store."""
import base64
from copy import deepcopy
import json
from threading import RLock
import time
import math
from uuid import uuid4

from app.application.boards import Boards
from app.infrastructure.dodil import COLUMNS
from app.domain.model import Person

OWNER = Person("owner-sub", "owner@arclet.test")
MEMBER = Person("member-sub", "member@arclet.test")
VIEWER = Person("viewer-sub", "viewer@arclet.test")
OUTSIDER = Person("outsider-sub", "outsider@arclet.test")


class MemoryTransaction:
    def __init__(self, rows):
        self.rows = rows

    def get(self, table, key, guard=False):
        return deepcopy(self.rows[table].get(key))

    def find(self, table, **filters):
        assert filters, "A query must be explicitly scoped"
        return [deepcopy(row) for row in self.rows[table].values()
                if all(row.get(key) == value for key, value in filters.items())]

    def insert(self, table, row):
        assert row["id"] not in self.rows[table], "Duplicate primary key"
        assert set(row) == COLUMNS[table], (table, set(row) ^ COLUMNS[table])
        self.rows[table][row["id"]] = deepcopy(row)

    def update(self, table, key, changes):
        assert key in self.rows[table]
        assert set(changes) <= COLUMNS[table]
        self.rows[table][key].update(deepcopy(changes))

    def delete(self, table, key):
        self.rows[table].pop(key, None)

    def similar_cards(self,board_id,card_id,embedding):
        vector = json.loads(embedding)
        matches = []
        for row in self.find("card_embeddings",board_id=board_id):
            if row["card_id"] == card_id:
                continue
            other = json.loads(row["embedding"])
            denominator = math.sqrt(sum(v*v for v in vector)*sum(v*v for v in other))
            distance = 1-sum(a*b for a,b in zip(vector,other))/denominator
            matches.append({"card_id":row["card_id"],"content_hash":row["content_hash"],"distance":distance})
        return sorted(matches,key=lambda row:row["distance"])


class MemoryStore:
    def __init__(self):
        self.rows = {table: {} for table in COLUMNS}
        self.lock = RLock()
        self.fail_after = None

    def run(self, fn, write=False):
        with self.lock:
            snapshot = deepcopy(self.rows)
            result = fn(MemoryTransaction(snapshot))
            if self.fail_after:
                raise self.fail_after
            self.rows = snapshot
            return result


def call(service, action, person=OWNER, **args):
    return service.call(action, person, args, uuid4().hex)


def fixture_board(store=None):
    store = store or MemoryStore()
    service = Boards(store)
    workspace = call(service, "create_workspace", title="Studio North")
    board = call(service, "create_board", workspace_id=workspace["id"], title="Launch",
                 description="Bring a useful product to the world", color="indigo")
    for person, role in ((MEMBER, "member"), (VIEWER, "viewer")):
        call(service, "invite_member", board_id=board["id"], email=person.email, role=role)
    board = call(service, "get_board", board_id=board["id"])
    return service, workspace, board


def person_headers(person=OWNER, **overrides):
    claims = {"sub": person.sub, "email": person.email,
              "iss": "https://appid.dodil.io/test/arclet-users", "aud": "pool:arclet-users",
              "amr": ["pwd"], "connection": "local", "exp": time.time() + 600}
    claims.update(overrides)
    body = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
    return {"x-dodil-auth-source": "pool", "x-dodil-user": json.dumps({
        "sub": person.sub, "email": person.email}), "x-dodil-user-jwt": f"test.{body}.test"}