aboutsummaryrefslogtreecommitdiffstats
path: root/tests/fixtures.py
blob: aef1d3f02ff59ba289e0b50b9881d88a9e7bd619 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
"""Fake .jsonl sessions for the tests."""

import json
import os

BASE_TS = "2025-08-14T10:00:00.000Z"


def ts(minute=0, hour=10, day=14):
    return f"2025-08-{day:02d}T{hour:02d}:{minute:02d}:00.000Z"


def user(text, at=BASE_TS, **extra):
    ev = {
        "type": "user",
        "timestamp": at,
        "cwd": "/home/u/proj",
        "gitBranch": "main",
        "version": "1.0.0",
        "message": {"role": "user", "content": [{"type": "text", "text": text}]},
    }
    ev.update(extra)
    return ev


def assistant(text=None, tools=(), at=BASE_TS, **extra):
    content = []
    if text is not None:
        content.append({"type": "text", "text": text})
    for name, args in tools:
        content.append({"type": "tool_use", "name": name, "input": args})
    ev = {
        "type": "assistant",
        "timestamp": at,
        "message": {"role": "assistant", "content": content},
    }
    ev.update(extra)
    return ev


def ai_title(title, at=BASE_TS):
    return {"type": "ai-title", "aiTitle": title, "timestamp": at}


def write_session(root, project_dir, session_id, events):
    """Writes a .jsonl and returns its path."""
    d = os.path.join(root, project_dir)
    os.makedirs(d, exist_ok=True)
    path = os.path.join(d, session_id + ".jsonl")
    with open(path, "w", encoding="utf-8") as f:
        for ev in events:
            f.write(json.dumps(ev, ensure_ascii=False) + "\n")
    return path


def simple_tree(root):
    """A small, varied tree: a chat, an empty one and one without cwd."""
    write_session(root, "-home-u-proj", "aaaaaaaa-0000-0000-0000-000000000001", [
        ai_title("Arreglar el build"),
        user("¿por qué falla el build?", at=ts(0)),
        assistant("Miro el log.", tools=[("Bash", {"command": "make"})], at=ts(3)),
        user("gracias", at=ts(12)),
    ])
    write_session(root, "-home-u-proj", "bbbbbbbb-0000-0000-0000-000000000002", [
        {"type": "system", "timestamp": ts(0, hour=9), "cwd": "/home/u/proj"},
    ])
    write_session(root, "-home-u-otro", "cccccccc-0000-0000-0000-000000000003", [
        user("hola", at=ts(0, hour=8), cwd="/home/u/otro"),
    ])
    return root


# ──────────────────────────────── memories ────────────────────────────────

def write_memory(root, project_dir, name, body="cuerpo", desc=None,
                 kind="project", origin=None, frontmatter=True):
    """Writes <project>/memory/<name>.md and returns its path."""
    d = os.path.join(root, project_dir, "memory")
    os.makedirs(d, exist_ok=True)
    path = os.path.join(d, name + ".md")

    parts = []
    if frontmatter:
        campos = ["---", f"name: {name}"]
        if desc is not None:
            campos.append(f"description: {desc}")
        campos += ["metadata:", "  node_type: memory", f"  type: {kind}"]
        if origin:
            campos.append(f"  originSessionId: {origin}")
        campos.append("---")
        parts.append("\n".join(campos))
    parts.append(body)

    with open(path, "w", encoding="utf-8") as f:
        f.write("\n".join(parts) + "\n")
    return path


def write_index(root, project_dir, names, extra=()):
    """Writes the MEMORY.md that links those names."""
    d = os.path.join(root, project_dir, "memory")
    os.makedirs(d, exist_ok=True)
    lines = ["# Memory Index", ""]
    for n in list(names) + list(extra):
        lines.append(f"- [{n}]({n}.md) — pista de {n}")
    path = os.path.join(d, "MEMORY.md")
    with open(path, "w", encoding="utf-8") as f:
        f.write("\n".join(lines) + "\n")
    return path


def memory_tree(root):
    """Varied memories: indexed, unlisted, and a project without an index."""
    write_memory(root, "-home-u-proj", "deploy-docker",
                 body="Se despliega con `make up`.\nVer [[roles-db]] y [[no-existe]].",
                 desc="Cómo se despliega el proyecto",
                 origin="aaaaaaaa-0000-0000-0000-000000000001")
    write_memory(root, "-home-u-proj", "roles-db", body="Roles de la base.",
                 desc="Roles", kind="reference")
    write_memory(root, "-home-u-proj", "suelta", body="No está en el índice.",
                 desc="Huérfana")
    # The index lists two real ones and one that no longer exists.
    write_index(root, "-home-u-proj", ["deploy-docker", "roles-db"],
                extra=["borrada-hace-rato"])

    # Another project with memory but no MEMORY.md.
    write_memory(root, "-home-u-otro", "sin-indice", body="Nadie me indexa.",
                 desc="Sin índice", kind="user",
                 origin="ffffffff-0000-0000-0000-00000000000f")

    # An empty memory/ does not count as a project with memory.
    os.makedirs(os.path.join(root, "-home-u-vacio", "memory"), exist_ok=True)
    return root