"""图片 sidecar 元数据服务的 schema、指纹和检索文本测试。""" from __future__ import annotations import json from pathlib import Path import pytest from pydantic import ValidationError from backend.metadata import MemeContext, MetadataError, MetadataService def make_image(tmp_path: Path, name: str = "cat.png", content: bytes = b"image") -> tuple[Path, MetadataService]: """创建隔离图片根目录和测试图片。""" root = tmp_path / "images" root.mkdir(parents=True, exist_ok=True) image = root / name image.write_bytes(content) return image, MetadataService(root) def test_sidecar_preserves_schema_and_unknown_fields(tmp_path: Path): """sidecar 使用完整文件名命名，并保留未知扩展字段。""" image, service = make_image(tmp_path) metadata = service.create_pending(image) payload = metadata.model_dump(mode="json") payload["custom"] = {"owner": "test"} payload["meme_context"]["summary"] = "一只猫正在做出反应" service.write(image, payload) loaded = service.load(image) assert service.sidecar_path(image).name == "cat.png.json" assert loaded.model_extra["custom"] == {"owner": "test"} assert loaded.meme_context.summary == "一只猫正在做出反应" def test_title_only_accepts_letters_numbers_and_spaces(): """标题只接受文字、数字和空格，并将连续空白折叠为单个空格。""" assert MemeContext(title=" 杰瑞 震惊 2 ").title == "杰瑞 震惊 2" assert MemeContext(title="Cafe\u0301\u00a0测试").title == "Café 测试" assert MemeContext(title=" ").title is None for title in ("认真!", "认真！", "猫：认真", "滑稽“认真”", "测试_标题", "开心🙂", "第①次", "版本Ⅷ", "零\u200d宽"): with pytest.raises(ValidationError, match="title_contains_disallowed_characters"): MemeContext(title=title) def test_title_rejects_overlong_and_multiline_values(): """标题按规范化结果限制长度，并优先报告控制字符错误。""" assert MemeContext(title=("x " * 40) + "x").title == ("x " * 40) + "x" with pytest.raises(ValidationError): MemeContext(title="x" * 121) with pytest.raises(ValidationError, match="title_must_be_single_line"): MemeContext(title="第一行\n第二行") def test_same_stem_different_extensions_have_distinct_sidecars(tmp_path: Path): """同名不同扩展名的图片不会共享 sidecar。""" first, service = make_image(tmp_path, "cat.png", b"png") second = first.with_name("cat.gif") second.write_bytes(b"gif") service.create_pending(first) service.create_pending(second) assert service.sidecar_path(first) != service.sidecar_path(second) assert service.sidecar_path(first).is_file() assert service.sidecar_path(second).is_file() def test_image_change_marks_metadata_repair_required(tmp_path: Path): """图片内容变化后，旧 sidecar 不能继续作为有效元数据。""" image, service = make_image(tmp_path) service.create_pending(image) image.write_bytes(b"changed") assert service.status(image)["status"] == "repair_required" with pytest.raises(MetadataError) as error: service.load(image) assert error.value.code == "metadata_image_mismatch" def test_human_fields_are_not_overwritten_by_research(tmp_path: Path): """自动研究更新不得覆盖人工确认的字段。""" image, service = make_image(tmp_path) service.create_pending(image) service.update_context(image, {"title": "人工确认标题", "summary": "人工确认摘要"}, producer="human", status="ready") service.update_context(image, {"title": "模型猜测摘要", "summary": "模型猜测摘要", "keywords": ["模型猜测摘要", "关键词"]}, producer="research", status="ready") loaded = service.load(image) assert loaded.meme_context.title == "人工确认标题" assert loaded.meme_context.summary == "人工确认摘要" assert loaded.meme_context.keywords == ["模型猜测摘要", "关键词"] assert loaded.provenance.field_sources["title"] == "human" def test_embedding_record_skips_pending_and_never_uses_filename(tmp_path: Path): """待研究图片不可索引，文件名不能成为 embedding 输入。""" image, service = make_image(tmp_path, "filename-fallback.png") service.create_pending(image) record = service.embedding_record(image) assert record["indexable"] is False assert record["text"] == "" assert record["skip_reason"] == "metadata_pending" def test_semantic_document_excludes_research_helpers(tmp_path: Path): """embedding 文本只包含允许的语义字段。""" image, service = make_image(tmp_path) service.create_pending(image) service.update_context( image, { "title": "无奈摊手", "summary": "一个人无奈地摊手", "subjects": ["人物"], "visible_text": ["原文"], "references": ["已确认模板"], "meaning": "表示无奈", "keywords": ["无奈"], "search_queries": ["不应进入向量"], "uncertainties": ["不确定出处"], "source_urls": ["https://example.com/source"], }, producer="research", status="ready", ) text = service.embedding_record(image)["text"] assert text.startswith("标题：无奈摊手\n摘要：一个人无奈地摊手") assert "一个人无奈地摊手" in text assert "已确认模板" in text assert "不应进入向量" not in text assert "不确定出处" not in text assert "example.com" not in text def test_repair_materializes_title_for_legacy_sidecar(tmp_path: Path): """修复任务为旧 sidecar 补写空标题，不丢失已有语境。""" image, service = make_image(tmp_path) service.create_pending(image) sidecar = service.sidecar_path(image) payload = json.loads(sidecar.read_text(encoding="utf-8")) payload["meme_context"].pop("title") payload["meme_context"]["summary"] = "旧摘要" sidecar.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") result = service.repair() repaired = json.loads(sidecar.read_text(encoding="utf-8")) assert result["repaired"] == 1 assert repaired["meme_context"]["title"] is None assert repaired["meme_context"]["summary"] == "旧摘要" def test_repair_creates_pending_sidecar_and_reports_orphan(tmp_path: Path): """修复任务补齐缺失 sidecar，并报告而不删除孤立 JSON。""" image, service = make_image(tmp_path) nested = image.parent / "nested" / "other.gif" nested.parent.mkdir() nested.write_bytes(b"other") orphan = image.with_name("orphan.png.json") orphan.write_text(json.dumps({"orphan": True}), encoding="utf-8") result = service.repair() assert result["created"] + result["repaired"] == 2 assert service.sidecar_path(image).is_file() assert service.sidecar_path(nested).is_file() assert result["orphaned"] == 1 assert orphan.is_file() second = service.repair() assert second["processed"] == 2 assert second["repaired"] == 0