tests/test_console.py

import json
import html
import unittest
import re
from html.parser import HTMLParser
from unittest.mock import patch

import app
from grounding_demo.console import compile_query, query_from_prompt, targets_html


class ConsoleTest(unittest.TestCase):
    def test_request_ids_distinguish_identical_results_and_errors(self):
        path = str(app.ROOT / app.CASES[0]["image"])
        raw = "<|object_ref_start|>car<|object_ref_end|><|box_start|><1><2><30><40><|box_end|>"
        runs = []
        for error in (False, False, True):
            with patch.object(app.GroundingClient, "infer", return_value=(raw, .1, {}),
                              side_effect=TimeoutError("test") if error else None):
                states = list(app.run_console(path, "car", "grounding"))
            ids = [re.search("data-run-id='([^']+)'", state[5]).group(1) for state in states]
            self.assertEqual(ids[0], ids[-1])
            self.assertIn("data-phase='busy'", states[0][5])
            self.assertNotIn("data-phase='busy'", states[-1][5])
            self.assertTrue(states[-1][-1]["interactive"])
            runs.append(ids[0])
        self.assertEqual(len(set(runs)), 3)

    def test_all_case_prompts_roundtrip_exactly(self):
        for row in app.CASES:
            with self.subTest(case=row["id"]):
                query = query_from_prompt(row["prompt"], row["route"])
                self.assertEqual(compile_query(query, row["route"]), row["prompt"])

    def test_category_separator(self):
        self.assertIn("car</c>person.", compile_query("car， person", "grounding"))

    def test_every_gallery_route_has_loadable_replacement_examples(self):
        self.assertEqual(len(app.CASES), 22)
        self.assertEqual({row["route"] for row in app.CASES}, set(app.ROUTE_META))
        for route in app.ROUTE_META:
            with self.subTest(route=route):
                choices, gallery, *loaded = app.change_route(route)
                self.assertEqual(len(choices["choices"]), len(app.BY_ROUTE[route]))
                self.assertEqual(len(gallery["value"]), len(app.BY_ROUTE[route]))
                self.assertEqual(loaded[8], "[]")
                self.assertIn("Image ready", loaded[10])

    def test_case_switch_resets_prediction(self):
        output = app.load_case(app.CASES[1]["id"])
        self.assertEqual(output[8], "[]")
        self.assertIn("Image ready", output[10])

    def test_upload_cannot_reveal_unrelated_gt(self):
        row = app.CASES[0]
        path = str(app.ROOT / row["image"])
        uploaded = app.upload_image(path, row["route"], row["prompt"])
        self.assertIsNone(uploaded[1])
        self.assertEqual(uploaded[2], path)
        self.assertEqual(uploaded[8], "[]")
        self.assertFalse(hasattr(app, "switch_view"))
        self.assertFalse(any(c.get("props", {}).get("elem_id") == "view-toggle"
                             for c in app.demo.config["components"]))

    def test_invalid_protocol_is_not_shown_as_valid_overlay(self):
        row = app.CASES[0]
        valid = "<|object_ref_start|>car<|object_ref_end|><|box_start|><1><2><30><40><|box_end|>"
        invalid = "<|object_ref_start|>car<|object_ref_end|><|box_start|><1><2><3><|box_end|>"
        result = app.result_bundle(
            str(app.ROOT / row["image"]), row["prompt"], row["route"], valid + invalid
        )
        self.assertEqual(result["normalized"], "[]")
        self.assertIn("Protocol error", result["metrics"])

    def test_actual_request_uses_latest_query_without_reference(self):
        row = app.CASES[0]
        path = str(app.ROOT / row["image"])
        response = "<|object_ref_start|>car<|object_ref_end|><|box_start|><1><2><30><40><|box_end|>"
        with patch.object(app.GroundingClient, "infer", return_value=(response, 1.2, {})) as infer:
            output = app.run_inference(path, "car", row["route"])
        infer.assert_called_once_with(path, compile_query("car", row["route"]))
        self.assertIsInstance(output[0], app.Image.Image)
        self.assertEqual(len(output), 6)
        self.assertIn("1 visual targets", output[-1])

    def test_labels_are_html_escaped(self):
        self.assertIn(
            "&lt;script&gt;", targets_html([{"label": "<script>", "coordinates": [1, 2]}])
        )

    def test_chinese_ocr_result_preserves_unicode_and_coordinates(self):
        label = "测试文字：中文 OCR（2026年）"
        raw = f"<|object_ref_start|>{label}<|object_ref_end|><|box_start|><100><200><800><400><|box_end|>"
        result = app.result_bundle(str(app.ROOT / app.CASES[0]["image"]), "OCR", "ocr", raw)
        self.assertEqual(json.loads(result["normalized"]), [
            {"label": label, "coordinates": [100, 200, 800, 400]}
        ])
        self.assertIn(label, result["normalized"])
        self.assertIn(label, result["targets"])
        self.assertIn("Complete", result["status"])

    def test_focus_buttons_carry_exact_safe_geometry(self):
        class Buttons(HTMLParser):
            def __init__(self):
                super().__init__()
                self.items = []

            def handle_starttag(self, tag, attrs):
                if tag == "button":
                    self.items.append(dict(attrs))

        label = "a' onclick='unsafe<>&\""
        parser = Buttons()
        parser.feed(
            targets_html(
                [
                    {"label": label, "coordinates": [10, 20, 300, 400]},
                    {"label": label, "coordinates": [80, 90]},
                ]
            )
        )
        button = parser.items[0]
        self.assertNotIn("onclick", button)
        self.assertEqual(button["data-label"], label)
        self.assertEqual(json.loads(button["data-coordinates"]), [[10, 20, 300, 400], [80, 90]])
        self.assertEqual(button["aria-pressed"], "false")

    def test_stream_wait_state_precedes_real_request(self):
        row = app.CASES[0]
        path = str(app.ROOT / row["image"])
        response = "<|object_ref_start|>car<|object_ref_end|><|box_start|><1><2><30><40><|box_end|>"
        with patch.object(app.GroundingClient, "infer", return_value=(response, 1.2, {})) as infer:
            stream = app.run_inference_stream(path, "car", row["route"])
            pending = next(stream)
            infer.assert_not_called()
            self.assertEqual(pending[0], path)
            self.assertFalse(pending[-1]["interactive"])
            self.assertIn("data-phase='busy'", pending[-2])
            done = next(stream)
            self.assertTrue(done[-1]["interactive"])
            self.assertIn("1 visual targets", done[-2])
            infer.assert_called_once_with(path, compile_query("car", row["route"]))

    def test_stream_error_releases_action_without_stale_prediction(self):
        row = app.CASES[0]
        with patch.object(app.GroundingClient, "infer", side_effect=TimeoutError("timeout")):
            states = list(
                app.run_inference_stream(str(app.ROOT / row["image"]), "car", row["route"])
            )
        self.assertEqual(len(states), 2)
        self.assertTrue(states[-1][-1]["interactive"])
        self.assertIn("data-phase='error'", states[-1][-2])
        self.assertEqual(states[-1][3], "[]")
        self.assertEqual(len(states[-1]), 7)

    def test_fonts_are_self_contained(self):
        self.assertNotIn("__MANROPE_DATA__", app.CSS)
        self.assertIn("prefers-reduced-motion", app.CSS)

    def test_task_guides_and_placeholders_cover_nine_routes(self):
        self.assertEqual(set(app.TASK_HELP), set(app.ROUTE_META))
        for route in app.ROUTE_META:
            self.assertIn(app.TASK_HELP[route][0], html.unescape(app.task_help_html(route)))
            self.assertTrue(app.TASK_HELP[route][3])
        self.assertNotIn("逗号", app.task_help_html("visual_prompt"))

    def test_visual_input_overlay_is_not_sent_to_model(self):
        row = app.BY_ROUTE["visual_prompt"][0]
        path = str(app.ROOT / row["image"])
        loaded = app.load_case(row["id"])
        self.assertEqual(loaded[0], path)
        self.assertIsInstance(loaded[2], app.Image.Image)
        with patch.object(app.GroundingClient, "infer", return_value=(row["expected"], 1.0, {})) as infer:
            app.run_inference(path, query_from_prompt(row["prompt"], row["route"]), row["route"])
        infer.assert_called_once_with(path, row["prompt"])

    def test_error_has_recovery_and_preserves_details(self):
        row = app.CASES[0]
        with patch.object(app.GroundingClient, "infer", side_effect=ConnectionError("test detail")):
            result = app.run_inference(str(app.ROOT / row["image"]), "car", "grounding")
        self.assertIn("Check the connection and retry", result[-1])
        self.assertNotIn("test detail", result[-1])
        self.assertIn("test detail", result[-2])

    def test_connection_status_explicit_and_escaped(self):
        with patch.object(app.GroundingClient, "health", return_value=(False, "<unsafe>")):
            output = app.service_status()
        self.assertIn("Inference offline", output)
        self.assertIn("&lt;unsafe&gt;", output)

    def test_both_run_buttons_share_real_stream(self):
        with patch.object(app.GroundingClient, "infer", side_effect=TimeoutError("test")):
            states = list(app.run_console(str(app.ROOT / app.CASES[0]["image"]), "car", "grounding"))
        self.assertEqual(len(states), 2)
        for state in states:
            self.assertEqual(state[-1], state[-2])
        self.assertTrue(states[-1][-1]["interactive"])


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