import os
import base64
import io
import sys
import tempfile
import unittest
import urllib.parse
from pathlib import Path
from unittest import mock
from PIL import Image
ROOT = Path(__file__).resolve().parents[1]
AGENT = ROOT / "agent"
RADAR = ROOT / "radar"
sys.path.insert(0, str(AGENT))
sys.path.insert(0, str(RADAR))
from visual_gate import can_generate_visual
from caption_gate import can_generate_caption
class _ImageResponse:
def __init__(self, body: bytes, *, content_type: str = "image/jpeg"):
self._body = io.BytesIO(body)
self.status = 200
self.headers = {"Content-Type": content_type, "Content-Length": str(len(body))}
def read(self, amount=-1):
return self._body.read(amount)
def geturl(self):
return "https://images.example.invalid/topic.jpg"
def __enter__(self):
return self
def __exit__(self, *_args):
return None
class VisualPhotoPipelineTests(unittest.TestCase):
@staticmethod
def _write_image(path: Path, *, image_format: str = "PNG") -> None:
Image.new("RGB", (2, 3), "blue").save(path, format=image_format)
def test_feuchte_signals_require_a_photo_visual(self):
signal = {
"signal_id": "photo-test-1",
"signal_category": "feuchte",
"briefing_freigabe": "geeignet",
"briefing_winkel": "Kondensation am Fenster früh erkennen",
"mallorca_relevance": 25,
"radar_score": 88,
}
result = can_generate_visual(signal)
self.assertTrue(result["allowed"])
self.assertEqual(result["visual_type"], "photo")
self.assertIn("photo", result["reason"])
def test_business_relevance_does_not_block_operator_visual_choice(self):
signal = {
"signal_id": "operator-choice-1",
"signal_category": "allgemein",
"briefing_freigabe": "verwerfen",
"briefing_winkel": "",
"mallorca_relevance": 0,
"radar_score": 1,
"topic": "Lokales Verkehrsthema auf Mallorca",
}
result = can_generate_visual(signal)
self.assertTrue(result["allowed"])
self.assertEqual(result["visual_type"], "photo")
def test_death_and_ethical_safeguards_scan_every_text_field(self):
base = {
"signal_id": "safety-test-1",
"signal_category": "allgemein",
"briefing_freigabe": "geeignet",
"mallorca_relevance": 25,
"radar_score": 80,
}
fields = (
"briefing_risiko", "topic", "topic_de", "short_summary",
"briefing_winkel", "briefing_nutzen", "source_name",
)
for field in fields:
with self.subTest(kind="death", field=field, gate="visual"):
result = can_generate_visual({**base, field: "Todesfall gemeldet"})
self.assertFalse(result["allowed"])
self.assertIn("Todesfall", result["reason"])
with self.subTest(kind="death", field=field, gate="caption"):
result = can_generate_caption({
**base, "briefing_winkel": "Sachliche Einordnung", field: "Todesfall gemeldet"
})
self.assertFalse(result["allowed"])
self.assertIn("Todesfall", result["reason"])
with self.subTest(kind="ethical", field=field, gate="visual"):
result = can_generate_visual({**base, field: "Terroranschlag gemeldet"})
self.assertFalse(result["allowed"])
self.assertIn("Ethisch", result["reason"])
with self.subTest(kind="ethical", field=field, gate="caption"):
result = can_generate_caption({
**base, "briefing_winkel": "Sachliche Einordnung", field: "Terroranschlag gemeldet"
})
self.assertFalse(result["allowed"])
self.assertIn("Ethisch", result["reason"])
both_topics = can_generate_visual({
**base,
"topic_de": "Unverdächtige Übersetzung",
"topic": "Terroranschlag auf Mallorca",
})
self.assertFalse(both_topics["allowed"])
realistic_phrases = (
"Bei dem Unfall wurde ein Mensch getötet",
"Der Fahrer starb noch am Unfallort",
"Zwei Tote bei Unfall auf Mallorca",
"Drei Menschen wurden schwer verletzt",
"El conductor falleció en el accidente",
"Una víctima murió tras el choque",
"One passenger was killed in the crash",
"Two fatalities in Mallorca crash",
"Er erlitt lebensgefährliche Verletzungen",
"Una persona resultó gravemente herida",
"One passenger was critically injured",
"Two people lost their lives",
"Zwei Menschen verloren ihr Leben",
"Drei Menschen wurden schwer\nverletzt",
"Drei Menschen wurden schwer\tverletzt",
"Drei Menschen wurden schwer-verletzt",
"lebensgefährlich",
"Todes\u200bfall",
"d\u200bead",
"schw\u200ber verletzt",
"T.o.t.e",
"tödlicher Unfall",
"tödliche Kollision",
"getötete Person",
"Terroranschläge",
"Vergewaltigungen",
"Kindesmissbrauchsfälle",
"Terroran\u200bschläge",
)
caption_base = {
**base,
"briefing_winkel": "Sachliche Einordnung",
"mallorca_relevance": 25,
}
for phrase in realistic_phrases:
with self.subTest(phrase=phrase, gate="visual"):
self.assertFalse(can_generate_visual({**base, "topic": phrase})["allowed"])
with self.subTest(phrase=phrase, gate="caption"):
self.assertFalse(can_generate_caption({**caption_base, "topic": phrase})["allowed"])
def test_selected_real_photo_is_validated_and_installed_for_signal(self):
import visual_render
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
selected = root / "selected.jpg"
photo_dir = root / "photos"
photo_dir.mkdir()
self._write_image(selected, image_format="JPEG")
installed = visual_render.select_photo_asset(
{"signal_id": "real-photo-1"}, selected, photo_dir
)
data_uri = visual_render.load_photo_data_uri(
{"signal_id": "real-photo-1"}, photo_dir
)
self.assertEqual(installed.name, "real-photo-1.jpg")
self.assertTrue(data_uri.startswith("data:image/jpeg;base64,"))
def test_manual_photo_remains_valid_when_full_signal_context_arrives(self):
import visual_render
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
selected = root / "selected.jpg"
photo_dir = root / "photos"
photo_dir.mkdir()
self._write_image(selected, image_format="JPEG")
visual_render.select_photo_asset({"signal_id": "manual-context-1"}, selected, photo_dir)
with mock.patch.object(visual_render, "_open_image_request") as opener:
result = visual_render.ensure_topic_photo(
{
"signal_id": "manual-context-1",
"topic": "Vollständiger späterer Signalkontext",
"signal_category": "ferienimmobilie",
},
photo_dir,
)
self.assertIsNotNone(result)
opener.assert_not_called()
def test_photo_asset_is_embedded_as_data_uri(self):
import visual_render
with tempfile.TemporaryDirectory() as tmp:
photo_dir = Path(tmp)
self._write_image(photo_dir / "photo-test-1.jpg", image_format="JPEG")
signal = {"signal_id": "photo-test-1"}
data_uri = visual_render.load_photo_data_uri(signal, photo_dir)
self.assertTrue(data_uri.startswith("data:image/jpeg;base64,"))
def test_photo_asset_rejects_unsafe_signal_identifier(self):
import visual_render
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
photo_dir = root / "photos"
photo_dir.mkdir()
self._write_image(root / "outsi.png")
data_uri = visual_render.load_photo_data_uri(
{"signal_id": "../outside"}, photo_dir
)
self.assertIsNone(data_uri)
def test_photo_asset_rejects_symlink(self):
import visual_render
with tempfile.TemporaryDirectory() as tmp:
photo_dir = Path(tmp)
target = photo_dir / "actual.png"
self._write_image(target)
(photo_dir / "photo-test-1.png").symlink_to(target)
data_uri = visual_render.load_photo_data_uri(
{"signal_id": "photo-test-1"}, photo_dir
)
self.assertIsNone(data_uri)
def test_photo_asset_rejects_undecodable_image(self):
import visual_render
with tempfile.TemporaryDirectory() as tmp:
photo_dir = Path(tmp)
(photo_dir / "photo-test-1.jpg").write_bytes(b"not an image")
data_uri = visual_render.load_photo_data_uri(
{"signal_id": "photo-test-1"}, photo_dir
)
self.assertIsNone(data_uri)
def test_unapproved_image_host_is_rejected_even_if_it_resolves_publicly(self):
import visual_render
with mock.patch("socket.getaddrinfo", return_value=[(2, 1, 6, "", ("8.8.8.8", 443))]):
with self.assertRaisesRegex(ValueError, "freigegeben"):
visual_render._validate_public_https_url("https://attacker.example/image.jpg")
def test_image_transport_uses_a_pinned_https_connection(self):
import visual_render
self.assertTrue(
hasattr(visual_render, "_PinnedHTTPSConnection"),
"image transport must pin the validated public IP during TLS connection",
)
def test_topic_photo_is_acquired_automatically_from_signal_image(self):
import visual_render
signal = {
"signal_id": "automatic-photo-1",
"topic": "Direkte Busverbindung zwischen Sóller und Flughafen Palma",
"image_url": "https://images.example.invalid/topic.jpg",
}
image = io.BytesIO()
Image.new("RGB", (1200, 800), "blue").save(image, format="JPEG")
with tempfile.TemporaryDirectory() as tmp:
photo_dir = Path(tmp)
with (
mock.patch.object(visual_render, "_validate_public_https_url"),
mock.patch.object(visual_render, "_open_image_request", return_value=_ImageResponse(image.getvalue())),
):
result = visual_render.ensure_topic_photo(signal, photo_dir)
installed = list(photo_dir.glob("*.jpg"))
self.assertTrue(result.startswith("data:image/jpeg;base64,"))
self.assertEqual(len(installed), 1)
def test_missing_topic_image_is_generated_automatically_from_signal_topic(self):
import visual_render
signal = {
"signal_id": "4f8f77c8-be8e-4a74-8dcf-13f4b2de2db4",
"topic_de": "Sommernebel überrascht Palma während einer Hitzewarnung",
"short_summary": "Dichter Küstennebel liegt am Morgen über Palma de Mallorca.",
"signal_category": "wetter_klima",
}
image = io.BytesIO()
Image.new("RGB", (1200, 1200), "blue").save(image, format="JPEG")
requests = []
def open_generated(request, timeout):
requests.append((request.full_url, timeout))
return _ImageResponse(image.getvalue())
with tempfile.TemporaryDirectory() as tmp:
photo_dir = Path(tmp)
with (
mock.patch.object(visual_render, "_validate_public_https_url"),
mock.patch.object(visual_render, "_open_image_request", side_effect=open_generated),
):
result = visual_render.ensure_topic_photo(signal, photo_dir)
installed = list(photo_dir.glob("*.jpg"))
self.assertIsNotNone(result)
self.assertTrue(result.startswith("data:image/jpeg;base64,"))
self.assertEqual(len(installed), 1)
self.assertEqual(len(requests), 1)
self.assertIn("image.pollinations.ai", requests[0][0])
self.assertIn("Sommernebel", requests[0][0])
self.assertIn("Dense%20summer%20fog", requests[0][0])
self.assertIn("not%20documentary%20evidence", requests[0][0])
self.assertLess(len(requests[0][0]), 350)
query = urllib.parse.parse_qs(urllib.parse.urlparse(requests[0][0]).query)
self.assertLessEqual(int(query["seed"][0]), 2_147_483_647)
def test_generated_cache_is_replaced_when_a_source_image_later_arrives(self):
import visual_render
generated = io.BytesIO()
source = io.BytesIO()
Image.new("RGB", (20, 20), "blue").save(generated, format="JPEG")
Image.new("RGB", (20, 20), "red").save(source, format="JPEG")
responses = [
_ImageResponse(generated.getvalue()),
_ImageResponse(source.getvalue()),
]
signal = {
"signal_id": "source-replaces-generated-1",
"topic": "Nebel über Palma",
"signal_category": "wetter_klima",
}
with tempfile.TemporaryDirectory() as tmp:
photo_dir = Path(tmp)
with (
mock.patch.object(visual_render, "_validate_public_https_url"),
mock.patch.object(visual_render, "_open_image_request", side_effect=responses) as opener,
):
first = visual_render.ensure_topic_photo(signal, photo_dir)
second = visual_render.ensure_topic_photo(
{**signal, "image_url": "https://images.example.invalid/source.jpg"},
photo_dir,
)
self.assertEqual(opener.call_count, 2)
self.assertNotEqual(first, second)
raw = base64.b64decode(second.split(",", 1)[1])
with Image.open(io.BytesIO(raw)) as image:
red, _green, blue = image.convert("RGB").getpixel((10, 10))
self.assertGreater(red, blue)
def test_metadata_less_legacy_cache_is_replaced_fail_closed(self):
import visual_render
generated = io.BytesIO()
Image.new("RGB", (20, 20), "red").save(generated, format="JPEG")
signal = {
"signal_id": "legacy-cache-1",
"topic": "Nebel über Palma",
"signal_category": "wetter_klima",
}
with tempfile.TemporaryDirectory() as tmp:
photo_dir = Path(tmp)
self._write_image(photo_dir / "legacy-cache-1.jpg", image_format="JPEG")
with (
mock.patch.object(visual_render, "_validate_public_https_url"),
mock.patch.object(
visual_render,
"_open_image_request",
return_value=_ImageResponse(generated.getvalue()),
) as opener,
):
result = visual_render.ensure_topic_photo(signal, photo_dir)
metadata = visual_render._read_photo_metadata(signal, photo_dir)
self.assertEqual(opener.call_count, 1)
self.assertEqual(metadata.get("provenance"), "generated")
self.assertIsNotNone(result)
def test_extreme_image_dimensions_are_rejected(self):
import visual_render
image = io.BytesIO()
Image.new("RGB", (20001, 1), "blue").save(image, format="PNG")
with self.assertRaisesRegex(ValueError, "[Aa]bmessungen"):
visual_render._validated_image(image.getvalue())
def test_feuchte_copy_is_factual_and_not_alarmist(self):
import visual_render
signal = {
"signal_category": "feuchte",
"topic_de": "Hohe Luftfeuchtigkeit auf Mallorca: Kondensation in leerstehender Ferienwohnung",
"briefing_winkel": "Feuchteprobleme in Mallorca-Wohnungen: Ursachen und Lösung",
"briefing_nutzen": "Eigentümer erkennt Feuchteprobleme frühzeitig",
}
gate = {"category_class": "cat-feuchte"}
context = visual_render.build_context(signal, "4x5", gate)
self.assertEqual(context["cat_label"], "Feuchtigkeit im Leerstand")
self.assertEqual(context["hook_line"], "Kondensation\nfrüh erkennen.")
self.assertNotIn("Schimmel kommt", " ".join(map(str, context.values())))
def test_unknown_category_uses_neutral_hook_instead_of_action_claim(self):
import visual_render
context = visual_render.build_context(
{
"signal_category": "allgemein",
"topic": "Keine direkte Busverbindung zwischen Sóller und Flughafen Palma",
},
"1x1",
{"category_class": "cat-allgemein"},
)
self.assertEqual(context["hook_line"], "Mallorca\naktuell.")
self.assertNotIn("handeln", context["hook_line"].lower())
def test_weather_fog_uses_topic_specific_hook(self):
import visual_render
context = visual_render.build_context(
{
"signal_category": "wetter_klima",
"topic_de": "Sommernebel überrascht Palma mit voller Hitzewarnung",
},
"1x1",
{"category_class": "cat-wetter-klima"},
)
self.assertIn("Sommernebel", context["hook_line"])
self.assertNotIn("Sturm", context["hook_line"])
def test_upstream_headline_is_html_escaped_before_browser_rendering(self):
import visual_render
html = visual_render.render_template_html(
"<div>{{ headline }}</div>",
{"headline": "Busmeldung <script>fetch('https://evil.invalid')</script>"},
)
self.assertNotIn("<script>", html)
self.assertIn("<script>", html)
def test_templates_use_a_real_photo_and_no_svg_scene(self):
for name in ("visual_square.html", "visual_45.html"):
template = (AGENT / "templates" / name).read_text(encoding="utf-8")
self.assertIn("{{ photo_data_uri }}", template)
self.assertIn("object-fit: cover", template)
self.assertNotIn("<svg", template)
self.assertNotIn("feTurbulence", template)
def test_templates_restore_conditional_risk_warning(self):
for name in ("visual_square.html", "visual_45.html"):
template = (AGENT / "templates" / name).read_text(encoding="utf-8")
self.assertIn("{% if risk_flag %}", template)
self.assertIn("Manuell prüfen — Risiko-Signal", template)
def test_templates_disclose_generated_image_provenance(self):
for name in ("visual_square.html", "visual_45.html"):
template = (AGENT / "templates" / name).read_text(encoding="utf-8")
self.assertIn("{% if ai_generated %}", template)
self.assertIn("KI-generiertes Motiv", template)
class PostingPackageTests(unittest.TestCase):
def test_expired_db60c553_like_signal_can_generate_complete_package(self):
import posting_package
signal = {
"signal_id": "db60c553-expired",
"signal_category": "schimmel_risiko",
"briefing_freigabe": "zurueckstellen",
"briefing_winkel": "Feuchtigkeit an Außenwänden sachlich erklären",
"briefing_nutzen": "Eigentümer erkennt Handlungsbedarf frühzeitig",
"briefing_risiko": "kein besonderes Risiko",
"mallorca_relevance": 25,
"radar_score": 80,
"topic": "Feuchtigkeit in einer Mallorca-Wohnung",
"expires_at": "2020-01-01T00:00:00+00:00",
}
caption = {
"success": True,
"facebook": "Facebook",
"linkedin": "LinkedIn",
"char_count": {"facebook": 8, "linkedin": 8},
}
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
rendered = root / "rendered.png"
Image.new("RGB", (2, 3), "blue").save(rendered, format="PNG")
original_output = posting_package.OUTPUT_DIR
posting_package.OUTPUT_DIR = root / "packages"
posting_package.OUTPUT_DIR.mkdir()
try:
with (
mock.patch.object(posting_package, "load_signal", return_value=signal),
mock.patch.object(posting_package, "render_caption", return_value=caption),
mock.patch.object(posting_package, "render_signal", return_value={"success": True, "path": str(rendered), "reason": "ok"}),
):
result = posting_package.generate_package("db60c553-expired")
files = {path.name for path in (posting_package.OUTPUT_DIR / "db60c553-expired").iterdir()} if result["success"] else set()
finally:
posting_package.OUTPUT_DIR = original_output
self.assertTrue(result["success"], result["reason"])
self.assertEqual(Path(result["package_dir"]).name, "db60c553-expired")
self.assertEqual(
files,
{"facebook.txt", "linkedin.txt", "visual_square.png", "visual_45.png", "metadata.json"},
)
def test_package_rejects_unsafe_signal_id_before_lookup_or_path_use(self):
import posting_package
with mock.patch.object(posting_package, "load_signal") as load_signal:
result = posting_package.generate_package("../outside")
self.assertFalse(result["success"])
self.assertIn("Ungültige Signal-ID", result["reason"])
load_signal.assert_not_called()
def test_package_rejects_unsafe_loaded_signal_id_before_path_use(self):
import posting_package
with mock.patch.object(
posting_package, "load_signal", return_value={"signal_id": "../outside"}
):
result = posting_package.generate_package("safe-id")
self.assertFalse(result["success"])
self.assertIn("Ungültige Signal-ID", result["reason"])
def test_failed_final_swap_preserves_previous_package_and_cleans_temporary_files(self):
import posting_package
signal = {
"signal_id": "swap-test-1",
"signal_category": "feuchte",
"briefing_freigabe": "geeignet",
"radar_score": 80,
"topic": "Test",
}
caption = {
"success": True,
"facebook": "new Facebook",
"linkedin": "new LinkedIn",
"char_count": {"facebook": 12, "linkedin": 12},
}
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
packages = root / "packages"
packages.mkdir()
package_dir = packages / "swap-test-1"
package_dir.mkdir()
(package_dir / "marker.txt").write_text("old", encoding="utf-8")
rendered = root / "rendered.png"
Image.new("RGB", (2, 3), "blue").save(rendered, format="PNG")
original_output = posting_package.OUTPUT_DIR
def fail_exchange(_left, _right):
raise OSError("injected atomic exchange failure")
posting_package.OUTPUT_DIR = packages
try:
with (
mock.patch.object(posting_package, "load_signal", return_value=signal),
mock.patch.object(posting_package, "can_generate_caption", return_value={"allowed": True, "reason": "ok"}),
mock.patch.object(posting_package, "can_generate_visual", return_value={"allowed": True, "reason": "ok"}),
mock.patch.object(posting_package, "render_caption", return_value=caption),
mock.patch.object(posting_package, "render_signal", return_value={"success": True, "path": str(rendered), "reason": "ok"}),
mock.patch.object(posting_package, "_rename_exchange", side_effect=fail_exchange),
):
result = posting_package.generate_package("swap-test-1")
marker = (package_dir / "marker.txt").read_text(encoding="utf-8")
leftovers = [path.name for path in packages.iterdir() if path != package_dir]
finally:
posting_package.OUTPUT_DIR = original_output
self.assertFalse(result["success"])
self.assertEqual(marker, "old")
self.assertEqual(leftovers, [])
def test_required_visual_failure_returns_no_partial_successful_package(self):
import posting_package
signal = {
"signal_id": "package-test-1",
"signal_category": "feuchte",
"briefing_freigabe": "geeignet",
"radar_score": 80,
"topic": "Test",
}
caption = {
"success": True,
"facebook": "Facebook",
"linkedin": "LinkedIn",
"char_count": {"facebook": 8, "linkedin": 8},
}
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
rendered = root / "rendered.png"
Image.new("RGB", (2, 3), "blue").save(rendered, format="PNG")
original_output = posting_package.OUTPUT_DIR
posting_package.OUTPUT_DIR = root / "packages"
posting_package.OUTPUT_DIR.mkdir()
try:
with (
mock.patch.object(posting_package, "load_signal", return_value=signal),
mock.patch.object(posting_package, "can_generate_caption", return_value={"allowed": True, "reason": "ok"}),
mock.patch.object(posting_package, "can_generate_visual", return_value={"allowed": True, "reason": "ok"}),
mock.patch.object(posting_package, "render_caption", return_value=caption),
mock.patch.object(
posting_package,
"render_signal",
side_effect=[
{"success": True, "path": str(rendered), "reason": "ok"},
{"success": False, "path": None, "reason": "broken"},
],
),
):
result = posting_package.generate_package("package-test-1")
package_exists = (posting_package.OUTPUT_DIR / "package").exists()
finally:
posting_package.OUTPUT_DIR = original_output
self.assertFalse(result["success"])
self.assertIsNone(result["package_dir"])
self.assertIsNone(result["captions"])
self.assertFalse(package_exists)
if __name__ == "__main__":
unittest.main()