MisakaNet/tests/test_search_knowledge_stdout.py at main · ghzhost/MisakaNet · GitHub

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
import io
import json
import unittest
from pathlib import Path
from unittest import mock

import search_knowledge


class ReconfigurableStdout(io.StringIO):
    def __init__(self):
        super().__init__()
        self.reconfigure_calls = []

    def reconfigure(self, **kwargs):
        self.reconfigure_calls.append(kwargs)


class TestSearchKnowledgeStdout(unittest.TestCase):
    def test_ensure_utf8_stdout_reconfigures_stream(self):
        stdout = ReconfigurableStdout()

        with mock.patch.object(search_knowledge.sys, "stdout", stdout):
            search_knowledge._ensure_utf8_stdout()

        self.assertEqual(
            stdout.reconfigure_calls,
            [{"encoding": "utf-8", "errors": "replace"}],
        )

    def test_ensure_utf8_stdout_ignores_unsupported_stream(self):
        stdout = io.StringIO()

        with mock.patch.object(search_knowledge.sys, "stdout", stdout):
            search_knowledge._ensure_utf8_stdout()

    def test_json_result_uses_required_schema(self):
        doc = mock.Mock(
            title="Database locked",
            domain="database",
            tags=["sqlite", "locking"],
            filepath=Path(search_knowledge.__file__).parent / "lessons" / "example.md",
            content="---\ntitle: Example\n---\n\nA useful preview.",
        )

        result = search_knowledge._json_result(0.12345678, doc)

        self.assertEqual(
            set(result), {"title", "domain", "tags", "score", "path", "preview"}
        )
        self.assertEqual(result["path"], "lessons/example.md")
        self.assertEqual(result["score"], 0.123457)

    def test_json_result_includes_match_reason_when_query_is_provided(self):
        doc = mock.Mock(
            title="Network timeout",
            domain="devops",
            tags=["network", "retry"],
            filepath=Path(search_knowledge.__file__).parent / "lessons" / "example.md",
            content="---\ntitle: Example\n---\n\nRetry the network timeout with backoff.",
        )

        result = search_knowledge._json_result(0.5, doc, query="timeout network")

        self.assertIn("title keyword 'timeout'", result["match_reason"])
        self.assertIn("tag 'network'", result["match_reason"])
        self.assertIn("[timeout]", result["preview_highlighted"].lower())

    def test_json_result_includes_score_breakdown_when_verbose(self):
        doc = mock.Mock(
            title="Network timeout",
            domain="devops",
            status="published",
            reference="",
            scope="",
            source="manual",
            tags=["network"],
            filepath=Path(search_knowledge.__file__).parent / "lessons" / "example.md",
            content="Network timeout retry.",
            score_baseline=0.1,
            is_draft=False,
            mtime=0.0,
        )

        result = search_knowledge._json_result(0.5, doc, query="timeout", verbose=True)

        self.assertIn("score_breakdown", result)
        self.assertIn("bm25", result["score_breakdown"])
        self.assertIn("metadata", result["score_breakdown"])

    def test_json_error_is_parseable(self):
        stdout = io.StringIO()
        with mock.patch.object(search_knowledge.sys, "stdout", stdout):
            search_knowledge._print_json_error("query failed")

        self.assertEqual(json.loads(stdout.getvalue()), {"error": "query failed"})


if __name__ == "__main__":
    unittest.main()