210 lines
11 KiB
Python
210 lines
11 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,
|
||
_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,
|
||
_persist_consistent_image_artifacts,
|
||
_data_image_digest,
|
||
)
|
||
|
||
|
||
class AttributionParserTests(unittest.TestCase):
|
||
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_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()
|