"""Focused attribution parser regressions.""" from __future__ import annotations import os import json import tempfile import shutil import unittest os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-conversation-test-config-")) from routes.conversation import ( _attribution_parse, _character_form_sheet, _character_identity_fingerprint, _build_character_portrait_prompt, _build_character_sheet_prompt, _consistent_image_result_metadata, _CHARACTER_VISUAL_PROMPT_VERSION, _cij_check_cancelled, _cij_new, _consistent_image_cancel_events, _consistent_image_jobs, _fallback_attribute_response, _bundled_comfyui_workflow, _comfyui_run, _CastJobRequest, match_characters_voices, _request_timeout_seconds, _response_error_text, _llm_chat_completion_urls, _persist_consistent_image_artifacts, _data_image_digest, ) class AttributionParserTests(unittest.TestCase): def test_visual_prompts_share_canonical_identity_anchors(self): sheet = { "gender": "female", "age_estimate": "late thirties", "race_species": "human", "physical": "olive skin, angular face, shoulder-length copper hair, green eyes", "clothing": "faded blue wool coat", "profession": "cartographer", "marks": "crescent scar on left cheek", } portrait = _build_character_portrait_prompt("Mara Venn", "The Survey", sheet, {"era": "early Renaissance"}) reference = _build_character_sheet_prompt("Mara Venn", "The Survey", sheet, {"era": "early Renaissance"}) for anchor in ("female", "late thirties", "human", "olive skin", "copper hair", "green eyes", "blue wool coat", "cartographer"): self.assertIn(anchor, portrait) self.assertIn(anchor, reference) def test_visual_prompts_forbid_generated_writing_including_chinese_characters(self): sheet = {"gender": "male", "physical": "short dark hair"} portrait = _build_character_portrait_prompt("Marcian", "The Trial", sheet, {}) reference = _build_character_sheet_prompt("Marcian", "The Trial", sheet, {}) for prompt in (portrait, reference): self.assertIn("Chinese characters", prompt) self.assertIn("text of any kind", prompt) def test_visual_prompt_version_changes_when_render_constraints_change(self): self.assertEqual(_CHARACTER_VISUAL_PROMPT_VERSION, "v4") def test_visual_pair_metadata_persists_shared_prompt_version_and_identity(self): sheet = {"name": "Mara Venn", "gender": "female", "physical": "copper hair and green eyes"} profile = "data:image/png;base64,AA==" reference = "data:image/png;base64,AQ==" result = _consistent_image_result_metadata(sheet, {"profile": profile, "reference": reference}) self.assertEqual(result["prompt_version"], _CHARACTER_VISUAL_PROMPT_VERSION) self.assertEqual(result["identity_fingerprint"], _character_identity_fingerprint(sheet)) self.assertTrue(result["identity_consistent"]) self.assertTrue(result["profile_asset_digest"]) self.assertTrue(result["reference_asset_digest"]) def test_consistent_image_persistence_keeps_live_payload_and_adds_artifact_refs(self): from core.database import _BACKGROUND_ARTIFACT_DIR job_id = "live-image-artifact-test" data_uri = "data:image/png;base64," + "AA==" try: result = _persist_consistent_image_artifacts(job_id, {"profile": data_uri, "reference": data_uri}) self.assertEqual(result["profile"], data_uri) self.assertEqual(result["reference"], data_uri) self.assertEqual([item["kind"] for item in result["artifacts"]], ["profile", "reference"]) finally: shutil.rmtree(_BACKGROUND_ARTIFACT_DIR / job_id, ignore_errors=True) def test_generated_image_digest_is_stable_and_rejects_invalid_data_uri(self): uri = "data:image/png;base64,AA==" first = _data_image_digest(uri) second = _data_image_digest(uri) self.assertEqual(first, second) self.assertTrue(first) self.assertEqual(_data_image_digest("not-an-image"), "") def test_active_character_form_overlays_visual_identity_without_changing_record(self): sheet = { "physical": "base form", "image_prompt": "base prompt", "forms": [{"id": "wolf", "name": "Wolf form", "physical": "black wolf", "image_prompt": "black wolf portrait"}], "active_form_id": "wolf", } resolved = _character_form_sheet(sheet) self.assertEqual(resolved["physical"], "black wolf") self.assertEqual(resolved["image_prompt"], "black wolf portrait") self.assertEqual(sheet["active_form_id"], "wolf") self.assertNotEqual(_character_identity_fingerprint(sheet), _character_identity_fingerprint({**sheet, "active_form_id": ""})) def test_fallback_accepts_mixed_guillemets(self): result = _fallback_attribute_response("Intro. »Ein langer Satz endet hier.‹ Danach.") self.assertEqual([segment["type"] for segment in result["segments"]], ["narration", "dialogue", "narration"]) self.assertIn("Ein langer Satz endet hier.", result["segments"][1]["text"]) def test_attribution_normalization_prevents_phantom_narrator_and_preserves_text_on_bad_json(self): raw = json.dumps({"segments": [ {"speaker": "Invented Narrator", "type": "narration", "text": "A road disappeared.", "emotion": "dramatic"}, {"speaker": "Mira", "type": "dialogue", "text": "We keep walking.", "emotion": ""}, ]}) result = _attribution_parse(raw, "unused") self.assertEqual(result["segments"][0]["speaker"], "Narrator") self.assertEqual(result["characters"], ["Mira"]) fallback = _attribution_parse("not valid JSON", "Original prose must survive") self.assertEqual(fallback["segments"], [{"speaker": "Narrator", "type": "narration", "text": "Original prose must survive", "emotion": ""}]) def test_api_timeout_and_error_text_are_bounded(self): self.assertEqual(_request_timeout_seconds(-10), 5.0) self.assertEqual(_request_timeout_seconds(99999), 600.0) self.assertEqual(_request_timeout_seconds("not-a-number", default=12), 12.0) class ResponseStub: text = "x" * 5000 reason = "Bad Gateway" self.assertEqual(len(_response_error_text(ResponseStub())), 2000) def test_chat_completion_urls_fall_back_from_versioned_gateway_route(self): # Some OpenAI-compatible routing gateways return 405 for /v1 but # accept their chat endpoint at the root. Keep the configured route # first, then try that compatible alternative. self.assertEqual( _llm_chat_completion_urls("http://gateway:28080/v1"), [ "http://gateway:28080/v1/chat/completions", "http://gateway:28080/chat/completions", ], ) self.assertEqual( _llm_chat_completion_urls("http://gateway:28080"), [ "http://gateway:28080/v1/chat/completions", "http://gateway:28080/chat/completions", ], ) def test_repairs_epigraph_split_before_mixed_guillemet_close(self): raw = '{"segments":[' \ '{"speaker":"Schwester Larissa","type":"dialogue","text":"»Von dem Windhager Felsgesteyn bis zur Kueste von Mendena ziehet es sich hin, von Ifirns Eisgebirg bis zu dem dampfenden Waldt der Mohas, und"},' \ '{"speaker":"Narrator","type":"narration","text":"fürwahr, es ist ein seltsam Land. Lieblich bisweilen, mild und voll der guten Menschen. ‹ Zitiert aus dem Buch."}' \ ']}' result = _attribution_parse(raw, "unused") self.assertEqual(len(result["segments"]), 2) dialogue, narration = result["segments"] self.assertEqual(dialogue["speaker"], "Schwester Larissa") self.assertEqual(dialogue["type"], "dialogue") self.assertTrue(dialogue["text"].endswith("Land. Lieblich bisweilen, mild und voll der guten Menschen. ‹")) self.assertEqual(narration["text"], "Zitiert aus dem Buch.") def test_book2_neutral_fixture_preserves_multiple_speakers(self): fixture_path = os.path.join(os.path.dirname(__file__), "fixtures", "book2_attribution.json") with open(fixture_path, encoding="utf-8") as handle: fixture = json.load(handle) result = _attribution_parse(fixture["raw"], " ".join(fixture["known_characters"])) self.assertEqual( [segment["speaker"] for segment in result["segments"]], ["Narrator", "Captain Vale", "The Ferryman", "Mira"], ) self.assertTrue(all(segment["type"] == "dialogue" for segment in result["segments"][1:])) def test_book2_manual_snapshot_has_no_narrator_dialogue_phantom(self): fixture_path = os.path.join(os.path.dirname(__file__), "..", "config", "regression_snapshots", "book2", "cast_manual.json") with open(fixture_path, encoding="utf-8") as handle: data = json.load(handle) result = _attribution_parse(json.dumps({"segments": data["segments"]}, ensure_ascii=False), " ".join(data.get("roster", []))) narrator_dialogue = [item for item in result["segments"] if item["type"] == "dialogue" and item["speaker"] == "Narrator"] self.assertEqual(narrator_dialogue, []) self.assertEqual(len(result["segments"]), len(data["segments"])) def test_bundled_comfyui_workflows_have_required_prompt_and_save_nodes(self): for image_kind in ("portrait", "concept_art"): raw, prompt_node, prompt_field, save_node = _bundled_comfyui_workflow(image_kind) graph = json.loads(raw) self.assertIn(prompt_node, graph) self.assertEqual(graph[prompt_node]["class_type"], "CLIPTextEncode") self.assertIn(prompt_field, graph[prompt_node]["inputs"]) self.assertIn(save_node, graph) self.assertIn("filename_prefix", graph[save_node]["inputs"]) def test_comfyui_submission_injects_prompt_polls_and_returns_image(self): from unittest.mock import patch class ResponseStub: def __init__(self, payload=None, content=b"", headers=None): self._payload, self.content, self.headers = payload, content, headers or {} self.ok = True self.status_code = 200 self.text = "" def json(self): return self._payload def raise_for_status(self): return None graph = { "2": {"class_type": "CLIPTextEncode", "inputs": {"text": "old"}}, "5": {"class_type": "KSampler", "inputs": {"seed": 0}}, "7": {"class_type": "SaveImage", "inputs": {"filename_prefix": "test"}}, } post = ResponseStub({"prompt_id": "p1"}) history = ResponseStub({"p1": {"status": {"completed": True}, "outputs": {"7": {"images": [{"filename": "out.png"}]}}}}) view = ResponseStub(content=b"png", headers={"Content-Type": "image/png"}) with patch("routes.conversation.requests.post", return_value=post) as post_call, \ patch("routes.conversation.requests.get", side_effect=[history, view]), \ patch("routes.conversation.time.sleep"): result = _comfyui_run("http://comfyui:8188", json.dumps(graph), "2", "text", "7", "new prompt", "stable") self.assertTrue(result.startswith("data:image/png;base64,")) submitted = post_call.call_args.kwargs["json"]["prompt"] self.assertEqual(submitted["2"]["inputs"]["text"], "new prompt") self.assertNotEqual(submitted["5"]["inputs"]["seed"], 0) def test_voice_matching_rejects_invented_and_unnecessarily_duplicate_ids(self): from unittest.mock import patch class ResponseStub: status_code = 200 ok = True text = "" def raise_for_status(self): return None def json(self): return {"choices": [{"message": {"content": json.dumps({"assignments": [ {"name": "A", "voice_id": "v1", "reason": "match"}, {"name": "B", "voice_id": "invented", "reason": "bad"}, {"name": "C", "voice_id": "v1", "reason": "duplicate"}, ]})}}]} data = {"names": ["A", "B", "C"], "voices": [{"id": "v1", "gender": "F"}, {"id": "v2", "gender": "M"}], "script": "short", "llm_url": "http://llm/v1"} with patch("routes.conversation.requests.post", return_value=ResponseStub()): result = __import__("asyncio").run(match_characters_voices(_CastJobRequest(data, {}))) self.assertEqual([item["voice_id"] for item in result["assignments"]], ["v1"]) def test_consistent_image_job_can_be_cancelled(self): from fastapi.testclient import TestClient from server import app job_id = "cancel-test" _cij_new(job_id, "Test character") try: with TestClient(app) as client: response = client.post( f"/api/character-generate-consistent-images/cancel/{job_id}", headers={"origin": "http://testserver"}, ) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()["cancelled"]) with self.assertRaises(Exception): _cij_check_cancelled(job_id) finally: _consistent_image_jobs.pop(job_id, None) _consistent_image_cancel_events.pop(job_id, None) from core.database import _open with _open() as conn: conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,)) conn.commit() if __name__ == "__main__": unittest.main() class CharacterSheetsEmptyAnswerTests(unittest.TestCase): """A reasoning model that spends its budget thinking returns no content; the endpoint retries with thinking off.""" def _client(self): from fastapi import FastAPI from fastapi.testclient import TestClient import routes.conversation as conv app = FastAPI(); app.include_router(conv.router) return TestClient(app), conv def _resp(self, content, status=200): from unittest import mock r = mock.Mock(); r.status_code = status; r.raise_for_status = lambda: None r.json.return_value = {"choices": [{"message": {"content": content, "reasoning_content": "thinking..."}}]} return r def test_retries_once_with_thinking_disabled_when_the_answer_is_empty(self): from unittest import mock client, conv = self._client() good = '{"sheets":[{"name":"Linosch","physical":"alter Zwerg"}],"characters":["Linosch"]}' with mock.patch.object(conv, "_post_llm_chat_completion", side_effect=[self._resp(""), self._resp(good)]) as post: r = client.post("/api/character-sheets", json={"text": "Linosch trat ein.", "known_characters": ["Linosch"]}) self.assertEqual(r.status_code, 200) self.assertEqual(r.json()["sheets"][0]["physical"], "alter Zwerg") self.assertEqual(post.call_count, 2) self.assertEqual(post.call_args_list[1].args[1].get("chat_template_kwargs"), {"enable_thinking": False}) def test_no_retry_when_the_first_answer_has_content(self): from unittest import mock client, conv = self._client() good = '{"sheets":[{"name":"A","physical":"x"}],"characters":["A"]}' with mock.patch.object(conv, "_post_llm_chat_completion", return_value=self._resp(good)) as post: client.post("/api/character-sheets", json={"text": "A kam.", "known_characters": ["A"]}) self.assertEqual(post.call_count, 1) class LlmModelAliasTests(unittest.TestCase): def _get(self, responses): from unittest import mock from fastapi import FastAPI from fastapi.testclient import TestClient import routes.conversation as conv app = FastAPI(); app.include_router(conv.router) def fake_get(url, **kw): r = mock.Mock() status, body = responses.get(url.split("://", 1)[1].split("/", 1)[1], (404, {})) r.status_code = status; r.json.return_value = body; r.text = str(body) return r with mock.patch.object(conv.requests, "get", side_effect=fake_get): return TestClient(app).get("/api/conversation/llm-models", params={"url": "http://192.168.1.5:4000"}).json() def test_router_aliases_are_listed_before_the_models(self): out = self._get({ "models": (200, {"data": [{"id": "gpt-x"}]}), "router/settings": (200, {"current_values": {"model_group_alias": {"ocr": "a", "default": "b"}}}), }) self.assertEqual(out["models"], ["ocr", "default", "gpt-x"]) self.assertEqual(out["aliases"], ["ocr", "default"]) def test_a_gateway_without_alias_settings_still_lists_its_models(self): out = self._get({"models": (200, {"data": [{"id": "m1"}, {"id": "m2"}]})}) self.assertEqual(out["models"], ["m1", "m2"]) def test_an_alias_that_is_already_a_model_is_not_repeated(self): out = self._get({ "models": (200, {"data": [{"id": "reasoning"}]}), "router/settings": (200, {"model_group_alias": {"reasoning": "x", "cheap": "y"}}), }) self.assertEqual(out["models"], ["cheap", "reasoning"])