test_search_knowledge_stdout.py

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", "freshness"}
        )
        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"})


class TestRemoteDocs(unittest.TestCase):
    """PRD ④: --remote mode loads lessons from the D1 service."""

    def _mock_response(self, payload):
        resp = mock.Mock()
        resp.read.return_value = json.dumps(payload).encode("utf-8")
        resp.__enter__ = mock.Mock(return_value=resp)
        resp.__exit__ = mock.Mock(return_value=False)
        return resp

    def test_load_remote_docs_builds_cached_docs(self):
        payload = [
            {"id": "pip-mirror", "title": "pip timeout mirror", "domain": "python",
             "status": "published", "tags": ["pip"], "path": "lessons/core/pip-mirror.md",
             "description": "use a mirror", "problem": "pip times out"},
            {"id": "dco", "title": "DCO signoff", "domain": "git",
             "status": "published", "tags": ["dco"], "path": "lessons/core/dco.md",
             "description": "signoff required", "problem": ""},
        ]
        with mock.patch("urllib.request.urlopen", return_value=self._mock_response(payload)):
            docs = search_knowledge._load_remote_docs()

        self.assertEqual(len(docs), 2)
        self.assertEqual(docs[0].filename, "pip-mirror")
        self.assertEqual(docs[0].title, "pip timeout mirror")
        self.assertEqual(docs[0].domain, "python")
        self.assertEqual(docs[0].tags, ["pip"])
        self.assertTrue(docs[0].content)  # description used as content
        # filepath absolute under repo root → _doc_cache_id works
        self.assertTrue(docs[0].filepath.is_absolute())

    def test_load_remote_docs_falls_back_to_problem_field(self):
        payload = [
            {"id": "x", "title": "X", "domain": "devops", "status": "published",
             "tags": [], "path": "lessons/contrib/x.md", "problem": "the failure"},
        ]
        with mock.patch("urllib.request.urlopen", return_value=self._mock_response(payload)):
            docs = search_knowledge._load_remote_docs()
        self.assertIn("the failure", docs[0].content)

    def test_load_remote_docs_rejects_non_list(self):
        with mock.patch("urllib.request.urlopen", return_value=self._mock_response({"error": "x"})):
            with self.assertRaises(RuntimeError):
                search_knowledge._load_remote_docs()


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