test_gui.py

"""Tests for the knob GUI (tanglebrain/gui).

Hermetic: the view functions and the pure `dispatch` router are exercised directly — no socket is
bound and no network/subprocess runs (run_once is mocked). Covers secret-safety (key_ref is a ref
string, never resolved), the view shapes, run handling, and HTTP routing.
"""
from __future__ import annotations

import http.client
import json
import os
import threading
import unittest
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from unittest.mock import patch

from tanglebrain.gui import server, views
from tanglebrain.roster import Invoke, Roster, RosterEntry, packaged_roster_path
from tanglebrain.router import RouterError


def _entry(eid, tier, *, key_ref=None, model=None, kind="cli", good_at=(), orch=False):
    return RosterEntry(
        id=eid, tier=tier,
        invoke=Invoke(kind=kind, model=model, key_ref=key_ref, cmd=["x"] if kind == "cli" else None),
        cost="free" if tier == "local" else "subscription",
        good_at=list(good_at), can_orchestrate=orch,
    )


class ViewRosterTest(unittest.TestCase):
    def test_packaged_roster_shape(self):
        # Pin to the packaged example (env override) so this is independent of any operator roster
        # at ~/.config/tanglebrain/roster.yaml on the dev machine. R2a: the packaged default ships
        # one active entry — the free local tier (the opt-in sub/paid tiers are commented examples).
        with patch.dict(os.environ, {"TANGLEBRAIN_ROSTER": str(packaged_roster_path())}, clear=False):
            out = views.view_roster()
        ids = {e["id"] for e in out["entries"]}
        self.assertEqual(ids, {"local-ollama"})
        local = next(e for e in out["entries"] if e["id"] == "local-ollama")
        self.assertEqual(local["tier"], "local")
        self.assertIn("kind", local["invoke"])

    def test_key_ref_passed_through_not_resolved(self):
        # The secret-safety guarantee: key_ref is emitted verbatim as the reference string, and
        # no file is ever opened to resolve it.
        roster = Roster([_entry("local", "local", kind="openai-compat", model="m",
                                 key_ref="file:/secret/path.key")])
        with patch("tanglebrain.gui.views.load_roster", return_value=roster), \
             patch("builtins.open", side_effect=AssertionError("must not read key file")):
            out = views.view_roster()
        self.assertEqual(out["entries"][0]["invoke"]["key_ref"], "file:/secret/path.key")

    def test_no_secret_fields_leak(self):
        # Only the documented invoke subset is exposed (no cmd/scrub_env/delegate_args).
        roster = Roster([_entry("claude", "sub", key_ref="env:ANTHROPIC", good_at=["reasoning"], orch=True)])
        with patch("tanglebrain.gui.views.load_roster", return_value=roster):
            inv = views.view_roster()["entries"][0]["invoke"]
        self.assertEqual(set(inv), {"kind", "base_url", "model", "parse", "key_ref"})

    def test_surfaces_enabled_and_budget(self):
        # The panel shows the per-key kill-switch + the display-only monthly budget.
        paid = RosterEntry(
            id="gpt5", tier="api",
            invoke=Invoke(kind="api", base_url="u", model="gpt-5", key_ref="file:/k.key"),
            enabled=False, budget_usd_month=25.0,
        )
        with patch("tanglebrain.gui.views.load_roster", return_value=Roster([paid])):
            e = views.view_roster()["entries"][0]
        self.assertFalse(e["enabled"])
        self.assertEqual(e["budget_usd_month"], 25.0)

    def test_default_entry_enabled_true_no_budget(self):
        with patch("tanglebrain.gui.views.load_roster",
                   return_value=Roster([_entry("claude", "sub")])):
            e = views.view_roster()["entries"][0]
        self.assertTrue(e["enabled"])
        self.assertIsNone(e["budget_usd_month"])


class ViewSettingsTest(unittest.TestCase):
    def test_packaged_gate_is_off(self):
        # The shipped settings.yaml keeps paid billing off — the panel must report that.
        self.assertFalse(views.view_settings()["api_billing_enabled"])

    def test_reports_gate_on_when_enabled(self):
        from tanglebrain.settings import Settings
        with patch("tanglebrain.gui.views.load_settings", return_value=Settings(api_billing_enabled=True)):
            self.assertTrue(views.view_settings()["api_billing_enabled"])


class ViewPricingTest(unittest.TestCase):
    def test_packaged_pricing(self):
        out = views.view_pricing()
        self.assertFalse(out["is_placeholder"])
        self.assertEqual(out["input_per_mtok"], 3.0)
        self.assertEqual(out["output_per_mtok"], 15.0)


class ViewStatsTest(unittest.TestCase):
    def test_rolls_up_records(self):
        recs = [
            {"tier": "local", "in_tokens_est": 10, "out_tokens_est": 20,
             "cloud_equiv_usd": 1.0, "spend_avoided_usd": 1.0},
            {"tier": "sub", "in_tokens_est": 5, "out_tokens_est": 5,
             "cloud_equiv_usd": 0.5, "spend_avoided_usd": 0.5},
        ]
        with patch("tanglebrain.gui.views.read_records", return_value=recs):
            out = views.view_stats()
        self.assertEqual(out["summary"]["tasks"], 2)
        self.assertEqual(out["summary"]["by_tier"], {"local": 1, "sub": 1})
        self.assertAlmostEqual(out["summary"]["spend_avoided_usd"], 1.5)
        self.assertIn("is_placeholder", out)

    def test_includes_delegate_breakdown(self):
        recs = [
            {"kind": "task", "tier": "sub", "in_tokens_est": 5, "out_tokens_est": 5,
             "cloud_equiv_usd": 0.5, "spend_avoided_usd": 0.5},
            {"kind": "delegate", "model": "local-x", "in_tokens_est": 40, "out_tokens_est": 60,
             "cloud_equiv_usd": 2.0},
        ]
        with patch("tanglebrain.gui.views.read_records", return_value=recs):
            out = views.view_stats()
        # Headline stays task-only; delegates surface separately for the panel's fan-out breakdown.
        self.assertEqual(out["summary"]["tasks"], 1)
        delegates = out["summary"]["delegates"]
        self.assertEqual(delegates["count"], 1)
        self.assertEqual(delegates["by_backend"]["local-x"]["count"], 1)

    def test_includes_parent_task_tree(self):
        # The panel's delegate card renders the by_parent tree, so view_stats must carry it through.
        recs = [
            {"kind": "delegate", "model": "local-x", "parent_task_id": "p1"},
            {"kind": "delegate", "model": "local-x", "parent_task_id": "p2"},
            {"kind": "delegate", "model": "local-x"},
        ]
        with patch("tanglebrain.gui.views.read_records", return_value=recs):
            out = views.view_stats()
        by_parent = out["summary"]["delegates"]["by_parent"]
        self.assertEqual({k for k in by_parent if k != "unlinked"}, {"p1", "p2"})
        self.assertEqual(by_parent["unlinked"]["count"], 1)


class RunPromptTest(unittest.TestCase):
    def test_happy_path_reports_served(self):
        served = {"path": "router", "tier": "sub", "model": "claude"}
        with patch("tanglebrain.gui.views.run_once", return_value=("hello back", served)) as run:
            out = views.run_prompt({"prompt": "hi", "task": "code"})
        self.assertTrue(out["ok"])
        self.assertEqual(out["text"], "hello back")
        self.assertEqual(out["served"]["model"], "claude")
        self.assertEqual(run.call_args.kwargs["task"], "code")
        self.assertTrue(run.call_args.kwargs["return_served"])  # uses the returned meta, no log re-read
        self.assertEqual(run.call_args.kwargs["origin"], "gui")  # #74 attribution

    def test_does_not_reread_log(self):
        # The race fix: run_prompt must NOT call read_records (served comes from run_once).
        with patch("tanglebrain.gui.views.run_once", return_value=("x", None)), \
             patch("tanglebrain.gui.views.read_records", side_effect=AssertionError("must not re-read log")):
            out = views.run_prompt({"prompt": "hi"})
        self.assertIsNone(out["served"])

    def test_empty_prompt_rejected(self):
        out = views.run_prompt({"prompt": "   "})
        self.assertFalse(out["ok"])
        self.assertIn("required", out["error"])

    def test_missing_prompt_key_rejected(self):
        self.assertFalse(views.run_prompt({})["ok"])

    def test_backend_error_returned(self):
        with patch("tanglebrain.gui.views.run_once", side_effect=RouterError("all subs failed")):
            out = views.run_prompt({"prompt": "hi"})
        self.assertFalse(out["ok"])
        self.assertIn("all subs failed", out["error"])

    def test_local_flag_threaded(self):
        with patch("tanglebrain.gui.views.run_once", return_value=("x", None)) as run:
            views.run_prompt({"prompt": "hi", "local": True})
        self.assertTrue(run.call_args.kwargs["local"])


class SavePricingViewTest(unittest.TestCase):
    def _payload(self, **over):
        base = {"reference_model": "Test Model", "input_per_mtok": 2.0,
                "output_per_mtok": 8.0, "placeholder": False}
        base.update(over)
        return base

    def test_valid_save_persists_and_returns_view(self):
        with patch("tanglebrain.gui.views.save_pricing") as save, \
             patch("tanglebrain.gui.views.view_pricing", return_value={"reference_model": "Test Model"}):
            out = views.save_pricing_view(self._payload())
        self.assertTrue(out["ok"])
        self.assertEqual(out["pricing"]["reference_model"], "Test Model")
        save.assert_called_once()

    def test_invalid_does_not_save(self):
        with patch("tanglebrain.gui.views.save_pricing") as save:
            out = views.save_pricing_view(self._payload(input_per_mtok=-1))
        self.assertFalse(out["ok"])
        self.assertIn("input_per_mtok", out["error"])
        save.assert_not_called()


class SaveRosterViewTest(unittest.TestCase):
    def test_happy_path_returns_updated_roster(self):
        with patch("tanglebrain.gui.views.save_roster_edits") as save, \
             patch("tanglebrain.gui.views.load_roster",
                   return_value=Roster([_entry("claude", "sub")])):
            out = views.save_roster_view({"id": "claude", "fields": {"enabled": False}})
        save.assert_called_once_with("claude", {"enabled": False})
        self.assertTrue(out["ok"])
        self.assertIn("entries", out["roster"])

    def test_missing_id_or_fields_rejected(self):
        for bad in ({"fields": {"enabled": False}}, {"id": "claude"}, {"id": "claude", "fields": {}}):
            self.assertFalse(views.save_roster_view(bad)["ok"])

    def test_edit_error_returned_not_raised(self):
        from tanglebrain.roster_edit import RosterEditError
        with patch("tanglebrain.gui.views.save_roster_edits", side_effect=RosterEditError("nope")):
            out = views.save_roster_view({"id": "x", "fields": {"enabled": True}})
        self.assertFalse(out["ok"])
        self.assertIn("nope", out["error"])


class DispatchTest(unittest.TestCase):
    def test_get_index_is_html(self):
        status, ctype, body = server.dispatch("GET", "/")
        self.assertEqual(status, 200)
        self.assertIn("text/html", ctype)
        self.assertIn(b"TangleBrain", body)

    def test_get_logo_is_png(self):
        status, ctype, body = server.dispatch("GET", "/logo.png")
        self.assertEqual(status, 200)
        self.assertEqual(ctype, "image/png")
        self.assertTrue(body.startswith(b"\x89PNG\r\n\x1a\n"))  # PNG magic
        self.assertGreater(len(body), 0)

    def test_get_roster_json(self):
        with patch("tanglebrain.gui.views.load_roster",
                   return_value=Roster([_entry("local", "local", kind="openai-compat", model="m")])):
            status, ctype, body = server.dispatch("GET", "/api/roster")
        self.assertEqual(status, 200)
        self.assertIn("application/json", ctype)
        self.assertEqual(json.loads(body)["entries"][0]["id"], "local")

    def test_get_stats_ignores_query_string(self):
        with patch("tanglebrain.gui.views.read_records", return_value=[]):
            status, _, _ = server.dispatch("GET", "/api/stats?t=123")
        self.assertEqual(status, 200)

    def test_get_settings_json(self):
        from tanglebrain.settings import Settings
        with patch("tanglebrain.gui.views.load_settings", return_value=Settings(api_billing_enabled=True)):
            status, ctype, body = server.dispatch("GET", "/api/settings")
        self.assertEqual(status, 200)
        self.assertIn("application/json", ctype)
        self.assertTrue(json.loads(body)["api_billing_enabled"])

    def test_unknown_path_404(self):
        status, _, body = server.dispatch("GET", "/api/nope")
        self.assertEqual(status, 404)
        self.assertIn("not found", json.loads(body)["error"])

    def test_post_run_valid(self):
        body = json.dumps({"prompt": "hi"}).encode()
        with patch("tanglebrain.gui.views.run_once", return_value=("ok", None)):
            status, _, out = server.dispatch("POST", "/api/run", body)
        self.assertEqual(status, 200)
        self.assertTrue(json.loads(out)["ok"])

    def test_post_roster_valid(self):
        body = json.dumps({"id": "claude", "fields": {"enabled": False}}).encode()
        with patch("tanglebrain.gui.views.save_roster_edits"), \
             patch("tanglebrain.gui.views.load_roster",
                   return_value=Roster([_entry("claude", "sub")])):
            status, _, out = server.dispatch("POST", "/api/roster", body)
        self.assertEqual(status, 200)
        self.assertTrue(json.loads(out)["ok"])

    def test_post_roster_bad_request_400(self):
        body = json.dumps({"id": "claude"}).encode()  # no fields
        status, _, out = server.dispatch("POST", "/api/roster", body)
        self.assertEqual(status, 400)
        self.assertFalse(json.loads(out)["ok"])

    def test_post_pricing_valid(self):
        body = json.dumps({"reference_model": "M", "input_per_mtok": 1.0,
                           "output_per_mtok": 2.0, "placeholder": False}).encode()
        with patch("tanglebrain.gui.views.save_pricing"), \
             patch("tanglebrain.gui.views.view_pricing", return_value={"reference_model": "M"}):
            status, _, out = server.dispatch("POST", "/api/pricing", body)
        self.assertEqual(status, 200)
        self.assertTrue(json.loads(out)["ok"])

    def test_post_pricing_invalid_400(self):
        body = json.dumps({"reference_model": "", "input_per_mtok": 1.0, "output_per_mtok": 2.0}).encode()
        with patch("tanglebrain.gui.views.save_pricing") as save:
            status, _, out = server.dispatch("POST", "/api/pricing", body)
        self.assertEqual(status, 400)
        self.assertFalse(json.loads(out)["ok"])
        save.assert_not_called()

    def test_post_run_bad_json_400(self):
        status, _, out = server.dispatch("POST", "/api/run", b"{not json")
        self.assertEqual(status, 400)
        self.assertFalse(json.loads(out)["ok"])

    def test_post_run_non_object_400(self):
        status, _, _ = server.dispatch("POST", "/api/run", b"[1,2,3]")
        self.assertEqual(status, 400)

    def test_post_unknown_path_404(self):
        status, _, _ = server.dispatch("POST", "/api/nope", b"{}")
        self.assertEqual(status, 404)

    def test_empty_prompt_run_is_400(self):
        body = json.dumps({"prompt": ""}).encode()
        status, _, _ = server.dispatch("POST", "/api/run", body)
        self.assertEqual(status, 400)

    def test_unsupported_method_405(self):
        status, _, _ = server.dispatch("DELETE", "/api/roster")
        self.assertEqual(status, 405)

    def test_read_view_error_is_clean_json_500(self):
        # A failing read view returns a JSON 500, not a traceback to the client.
        from tanglebrain.roster import RosterError

        with patch("tanglebrain.gui.views.load_roster", side_effect=RosterError("bad roster yaml")):
            status, ctype, body = server.dispatch("GET", "/api/roster")
        self.assertEqual(status, 500)
        self.assertIn("application/json", ctype)
        self.assertIn("bad roster yaml", json.loads(body)["error"])

    def test_post_non_json_content_type_is_415_view_never_invoked(self):
        # A cross-origin browser fetch can POST text/plain to localhost with no CORS preflight —
        # /api/run spends real backend quota, so non-JSON must never reach a view (issue #72).
        # All three POST endpoints ride the same gate.
        body = json.dumps({"prompt": "hi"}).encode()
        for path in ("/api/run", "/api/pricing", "/api/roster"):
            with patch("tanglebrain.gui.views.run_once") as run, \
                 patch("tanglebrain.gui.views.save_pricing") as pricing, \
                 patch("tanglebrain.gui.views.save_roster_edits") as roster:
                status, ctype, out = server.dispatch(
                    "POST", path, body, content_type="text/plain;charset=UTF-8"
                )
            self.assertEqual(status, 415, path)
            self.assertIn("application/json", ctype)
            self.assertIn("application/json", json.loads(out)["error"])
            for view in (run, pricing, roster):
                view.assert_not_called()

    def test_post_charset_qualified_json_content_type_accepted(self):
        body = json.dumps({"prompt": "hi"}).encode()
        with patch("tanglebrain.gui.views.run_once", return_value=("ok", None)):
            status, _, out = server.dispatch(
                "POST", "/api/run", body, content_type="application/json; charset=utf-8"
            )
        self.assertEqual(status, 200)
        self.assertTrue(json.loads(out)["ok"])


class LiveHandlerTest(unittest.TestCase):
    """Loopback-socket tests proving the real Handler wiring for the #72 hardening."""

    def setUp(self):
        self.server = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
        self.port = self.server.server_address[1]
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.addCleanup(self.thread.join, 2)
        self.addCleanup(self.server.server_close)
        self.addCleanup(self.server.shutdown)

    def test_malformed_content_length_is_400_not_a_reset(self):
        connection = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        self.addCleanup(connection.close)
        connection.putrequest("POST", "/api/run")
        connection.putheader("Content-Type", "application/json")
        connection.putheader("Content-Length", "abc")
        connection.endheaders()
        response = connection.getresponse()
        self.assertEqual(response.status, 400)
        self.assertIn("Content-Length", json.loads(response.read())["error"])

    def test_negative_content_length_clamps_to_empty_body(self):
        # max(0, …) must keep rfile.read(-1) from ever blocking on the socket; the request then
        # proceeds with an empty body and gets the view's own 400, not a hang or a reset.
        connection = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
        self.addCleanup(connection.close)
        connection.putrequest("POST", "/api/run")
        connection.putheader("Content-Type", "application/json")
        connection.putheader("Content-Length", "-5")
        connection.endheaders()
        response = connection.getresponse()
        self.assertEqual(response.status, 400)
        self.assertIn("prompt is required", json.loads(response.read())["error"])

    def test_text_plain_post_is_415_over_the_wire(self):
        # Proves the Handler threads the real Content-Type header into dispatch.
        request = urllib.request.Request(
            f"http://127.0.0.1:{self.port}/api/run",
            data=json.dumps({"prompt": "hi"}).encode(),
            headers={"Content-Type": "text/plain"},
            method="POST",
        )
        with patch("tanglebrain.gui.views.run_once") as run:
            with self.assertRaises(urllib.error.HTTPError) as ctx:
                urllib.request.urlopen(request, timeout=5)
        self.assertEqual(ctx.exception.code, 415)
        run.assert_not_called()


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