349 lines
18 KiB
Python
349 lines
18 KiB
Python
"""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,
|
||
_build_compact_character_render_prompts,
|
||
_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_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_glombo_stays_human_and_the_scene_does_not_cue_cameras(self):
|
||
"""Glombo rendered as a dwarf holding modern cameras despite being an adult human siege expert."""
|
||
sheet = {
|
||
"gender": "male",
|
||
"race_species": "Mensch",
|
||
"profession": "Belagerungsexperte",
|
||
"physical": "Groß gewachsener, breitschultriger Mann mit braunem Haar und Vollbart.",
|
||
"clothing": "Wolltunika, Lederwams und Köcher für Armbrustbolzen.",
|
||
}
|
||
portrait, triptych, _side, _back, scene, _props = _build_compact_character_render_prompts(
|
||
"Glombo Brohm", "Buch", sheet, {}, "cinematic", "", "",
|
||
)
|
||
|
||
for prompt in (portrait, triptych):
|
||
self.assertIn("Natural adult human body proportions", prompt)
|
||
self.assertNotIn("Natural adult dwarf body proportions", prompt)
|
||
self.assertNotIn("photographer", scene.lower())
|
||
self.assertIn("Every visible object belongs to the established book world", scene)
|
||
|
||
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):
|
||
self.assertEqual(_data_image_digest("data:image/png;base64,AA=="), _data_image_digest("data:image/png;base64,AA=="))
|
||
self.assertTrue(_data_image_digest("data:image/png;base64,AA=="))
|
||
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"])
|