path: root/tests/test_sessions.py

import json
import os
import tempfile
import unittest
import unittest.mock

from claude_logbook import sessions as S

from .fixtures import ai_title, assistant, simple_tree, ts, user, write_session


class TempRoot(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory()
        self.root = os.path.join(self._tmp.name, "projects")
        os.makedirs(self.root)
        self.cache = os.path.join(self._tmp.name, "cache.json")
        self.addCleanup(self._tmp.cleanup)

    def load(self, **kw):
        kw.setdefault("cache_path", self.cache)
        return S.load_sessions(root=self.root, **kw)


class TestReadSession(TempRoot):
    def test_basic_conversation(self):
        path = write_session(self.root, "-p", "11111111-1111-1111-1111-111111111111", [
            user("¿por qué falla?", at=ts(0)),
            assistant("Miro.", tools=[("Bash", {"command": "make test"})], at=ts(5)),
        ])
        rec = S.read_session(path)

        self.assertEqual(rec["u"], 1)
        self.assertEqual(rec["a"], 1)
        self.assertEqual(rec["d"], 5)
        self.assertEqual(rec["p"], "/home/u/proj")
        self.assertEqual(rec["b"], "main")
        self.assertFalse(rec["e"])
        self.assertFalse(rec["ai"])
        self.assertEqual(rec["t"], "¿por qué falla?")
        self.assertEqual([m["r"] for m in rec["c"]], ["u", "a", "t"])
        self.assertEqual(rec["c"][2]["x"], "Bash: make test")

    def test_claude_title_beats_the_first_message(self):
        path = write_session(self.root, "-p", "22222222-0000-0000-0000-000000000000", [
            user("arreglá esto", at=ts(0)),
            ai_title("Primer intento"),
            ai_title("Título final"),
        ])
        rec = S.read_session(path)
        self.assertEqual(rec["t"], "Título final")
        self.assertTrue(rec["ai"])

    def test_ignores_harness_noise(self):
        path = write_session(self.root, "-p", "33333333-0000-0000-0000-000000000000", [
            user("<command-name>/clear</command-name>", at=ts(0)),
            user("<system-reminder>ojo</system-reminder>", at=ts(1)),
            user("texto real <system-reminder>ojo</system-reminder>", at=ts(2)),
            user("meta", at=ts(2), isMeta=True),
            user("de un subagente", at=ts(3), isSidechain=True),
            assistant("respuesta de subagente", at=ts(4), isSidechain=True),
        ])
        rec = S.read_session(path)
        self.assertEqual(rec["u"], 1)
        self.assertEqual(rec["a"], 0)
        self.assertEqual(rec["c"][0]["x"], "texto real")

    def test_tolerates_a_line_cut_in_half(self):
        path = write_session(self.root, "-p", "44444444-0000-0000-0000-000000000000", [
            user("primero", at=ts(0)),
        ])
        with open(path, "a", encoding="utf-8") as f:
            f.write('{"type": "user", "message": {"content": [{"type": "te')
        rec = S.read_session(path)
        self.assertEqual(rec["u"], 1)

    def test_session_without_messages(self):
        path = write_session(self.root, "-p", "55555555-0000-0000-0000-000000000000", [
            {"type": "system", "timestamp": ts(0), "cwd": "/home/u/proj"},
        ])
        rec = S.read_session(path)
        self.assertTrue(rec["e"])
        self.assertIsNone(rec["t"])
        self.assertEqual(rec["c"], [])

    def test_single_huge_message_is_claude_p(self):
        largo = "x" * (S.NONINTERACTIVE_CHARS + 1)
        path = write_session(self.root, "-p", "66666666-0000-0000-0000-000000000000", [
            user(largo, at=ts(0)),
        ])
        self.assertTrue(S.read_session(path)["n"])

    def test_short_chat_is_not_claude_p(self):
        path = write_session(self.root, "-p", "77777777-0000-0000-0000-000000000000", [
            user("hola", at=ts(0)),
        ])
        self.assertFalse(S.read_session(path)["n"])


class TestToolSummary(unittest.TestCase):
    def test_uses_the_representative_parameter(self):
        self.assertEqual(
            S.tool_summary({"name": "Read", "input": {"file_path": "/a/b.py", "limit": 5}}),
            "Read: /a/b.py")

    def test_falls_back_to_first_string_for_unknown_tool(self):
        self.assertEqual(
            S.tool_summary({"name": "Rara", "input": {"n": 1, "q": "algo"}}),
            "Rara: algo")

    def test_truncates_long_arguments(self):
        out = S.tool_summary({"name": "Bash", "input": {"command": "a" * 500}})
        self.assertTrue(out.endswith("…"))
        self.assertEqual(len(out), len("Bash: ") + S.TOOL_ARG_MAX + 1)

    def test_no_usable_arguments(self):
        self.assertEqual(S.tool_summary({"name": "X", "input": {"n": 1}}), "X")
        self.assertEqual(S.tool_summary({"name": "X", "input": "no es dict"}), "X")


class TestLoad(TempRoot):
    def test_sorts_by_last_activity(self):
        simple_tree(self.root)
        got = [s["id"][:8] for s in self.load()]
        self.assertEqual(got, ["aaaaaaaa", "bbbbbbbb", "cccccccc"])

    def test_infers_the_path_from_another_session_of_the_project(self):
        write_session(self.root, "-home-u-proj", "aaaaaaaa-0000-0000-0000-000000000001",
                      [user("con cwd", at=ts(0))])
        write_session(self.root, "-home-u-proj", "dddddddd-0000-0000-0000-000000000004",
                      [{"type": "system", "timestamp": ts(30)}])
        orphan = next(s for s in self.load() if s["id"].startswith("dddddddd"))
        self.assertEqual(orphan["p"], "/home/u/proj")
        self.assertTrue(orphan["i"])

    def test_no_known_path_keeps_the_directory_name(self):
        write_session(self.root, "-sin-cwd", "eeeeeeee-0000-0000-0000-000000000005",
                      [{"type": "system", "timestamp": ts(0)}])
        s = self.load()[0]
        self.assertEqual(s["p"], "-sin-cwd")
        self.assertTrue(s["i"])

    def test_ignora_los_jsonl_de_subagentes(self):
        simple_tree(self.root)
        sub = os.path.join(self.root, "-home-u-proj", "aaaaaaaa-0000-0000-0000-000000000001", "subagents")
        os.makedirs(sub)
        with open(os.path.join(sub, "agent-1.jsonl"), "w", encoding="utf-8") as f:
            f.write(json.dumps(user("soy un subagente")) + "\n")
        self.assertEqual(len(self.load()), 3)


class TestCache(TempRoot):
    def test_reuses_what_did_not_change(self):
        simple_tree(self.root)
        self.load()

        # Dirty the cache by hand: if the second run returns the fake
        # title, it did not read the file again.
        with open(self.cache, encoding="utf-8") as f:
            blob = json.load(f)
        for entry in blob["entries"].values():
            entry["rec"]["t"] = "vino del caché"
        with open(self.cache, "w", encoding="utf-8") as f:
            json.dump(blob, f)

        self.assertEqual(self.load()[0]["t"], "vino del caché")

    def test_reparses_if_the_file_changed(self):
        path = write_session(self.root, "-p", "99999999-0000-0000-0000-000000000009",
                             [user("original", at=ts(0))])
        self.load()
        write_session(self.root, "-p", "99999999-0000-0000-0000-000000000009",
                      [user("cambiado", at=ts(0)), user("y otro", at=ts(1))])
        self.assertEqual(self.load()[0]["u"], 2)
        self.assertTrue(os.path.exists(path))

    def test_cache_from_another_version_is_discarded(self):
        simple_tree(self.root)
        self.load()
        with open(self.cache, encoding="utf-8") as f:
            blob = json.load(f)
        blob["v"] = S.CACHE_VERSION - 1
        for entry in blob["entries"].values():
            entry["rec"]["t"] = "no debería verse"
        with open(self.cache, "w", encoding="utf-8") as f:
            json.dump(blob, f)

        self.assertEqual(self.load()[0]["t"], "Arreglar el build")

    def test_broken_cache_breaks_nothing(self):
        simple_tree(self.root)
        with open(self.cache, "w", encoding="utf-8") as f:
            f.write("{esto no es json")
        self.assertEqual(len(self.load()), 3)

    def test_no_cache_writes_nothing(self):
        simple_tree(self.root)
        self.load(use_cache=False)
        self.assertFalse(os.path.exists(self.cache))

    def test_cache_stores_the_path_without_inferring(self):
        write_session(self.root, "-home-u-proj", "aaaaaaaa-0000-0000-0000-000000000001",
                      [user("con cwd", at=ts(0))])
        write_session(self.root, "-home-u-proj", "dddddddd-0000-0000-0000-000000000004",
                      [{"type": "system", "timestamp": ts(30)}])
        self.load()
        with open(self.cache, encoding="utf-8") as f:
            blob = json.load(f)
        recs = {os.path.basename(k): v["rec"] for k, v in blob["entries"].items()}
        self.assertIsNone(recs["dddddddd-0000-0000-0000-000000000004.jsonl"]["p"])

    def test_drop_from_cache_removes_deleted_ones(self):
        simple_tree(self.root)
        self.load()
        with open(self.cache, encoding="utf-8") as f:
            paths = list(json.load(f)["entries"])
        S.drop_from_cache(paths[:1], cache_path=self.cache)
        with open(self.cache, encoding="utf-8") as f:
            self.assertEqual(len(json.load(f