89 lines
5.0 KiB
Python
89 lines
5.0 KiB
Python
from pathlib import Path
|
|
|
|
import io
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from rf4_research.media_assets import audit_media_catalog, extract_media_candidates, inspect_image, merge_manifest, review_asset, store_asset
|
|
|
|
|
|
def test_extracts_and_classifies_unique_https_media() -> None:
|
|
html = '''<article><a href="/fish/pike"><img src="/media/fish/pike.png" alt="Щука"></a></article>
|
|
<figure><img data-src="https://cdn.example/maps/kuori.webp" alt="Карта Куори"></figure>
|
|
<img src="http://unsafe.example/bait.png"><img src="/media/fish/pike.png">'''
|
|
items = extract_media_candidates(html, source_page="https://example.test/guide")
|
|
assert [(item.entity_type, item.asset_url) for item in items] == [
|
|
("fish", "https://example.test/media/fish/pike.png"),
|
|
("waterbody", "https://cdn.example/maps/kuori.webp"),
|
|
]
|
|
assert items[0].external_id == "pike"
|
|
|
|
|
|
def test_does_not_mistake_seafishing_screenshot_for_fish_entity() -> None:
|
|
items = extract_media_candidates('<img src="/res/userguide/seafishing-1.jpeg">', source_page="https://rf4game.de/userguide/")
|
|
assert items[0].entity_type == "reference"
|
|
|
|
|
|
def test_gallery_page_provenance_classifies_generic_asset_urls() -> None:
|
|
fish = extract_media_candidates('<img src="/uploads/gallery/123.jpg" alt="Hecht">', source_page="https://rf4game.de/media/fische/")
|
|
water = extract_media_candidates('<img src="/uploads/gallery/456.jpg" alt="Kuori">', source_page="https://rf4game.de/media/levels/")
|
|
assert fish[0].entity_type == "fish"
|
|
assert water[0].entity_type == "waterbody"
|
|
chrome = extract_media_candidates('<img src="/wp-content/themes/rf4/img/banner_de.png">', source_page="https://rf4game.de/media/fische/")
|
|
assert chrome[0].entity_type == "reference"
|
|
|
|
|
|
def test_manifest_merges_and_binary_store_is_content_addressed(tmp_path: Path) -> None:
|
|
item = extract_media_candidates('<img src="/bait/worm.png" alt="Bait worm">', source_page="https://example.test")[0]
|
|
first = merge_manifest(tmp_path / "manifest.json", [item])
|
|
second = merge_manifest(tmp_path / "manifest.json", [item])
|
|
assert len(first["assets"]) == len(second["assets"]) == 1
|
|
assert second["assets"][0]["source_pages"] == ["https://example.test"]
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (3, 2), "green").save(image, format="PNG")
|
|
digest, relative, width, height, mime = store_asset(tmp_path, image.getvalue(), content_type="image/png", source_url=item.asset_url)
|
|
assert len(digest) == 64
|
|
assert (tmp_path / relative).read_bytes() == image.getvalue()
|
|
assert (width, height, mime) == (3, 2, "image/png")
|
|
|
|
|
|
def test_image_inspection_rejects_invalid_body_and_mime_mismatch(tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError, match="valid raster"):
|
|
inspect_image(b"not an image")
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (1, 1)).save(image, format="PNG")
|
|
with pytest.raises(ValueError, match="MIME mismatch"):
|
|
store_asset(tmp_path, image.getvalue(), content_type="image/jpeg", source_url="https://example.test/a.jpg")
|
|
|
|
|
|
def test_review_requires_stored_asset_and_canonical_mapping(tmp_path: Path) -> None:
|
|
item = extract_media_candidates('<img src="/fish/pike.png" alt="Щука">', source_page="https://example.test")[0]
|
|
path = tmp_path / "manifest.json"
|
|
manifest = merge_manifest(path, [item])
|
|
with pytest.raises(ValueError, match="stored"):
|
|
review_asset(path, asset_url=item.asset_url, decision="approved", entity_type="fish", entity_key="pike")
|
|
manifest["assets"][0]["status"] = "stored"
|
|
path.write_text(__import__("json").dumps(manifest), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="canonical key"):
|
|
review_asset(path, asset_url=item.asset_url, decision="approved", entity_type="fish")
|
|
reviewed = review_asset(path, asset_url=item.asset_url, decision="approved", entity_type="fish", entity_key="pike", note="matched by name")
|
|
assert (reviewed["status"], reviewed["entity_key"]) == ("approved", "pike")
|
|
|
|
|
|
def test_catalog_audit_detects_tampering_and_orphans(tmp_path: Path) -> None:
|
|
item = extract_media_candidates('<img src="/fish/pike.png">', source_page="https://example.test")[0]
|
|
path = tmp_path / "manifest.json"
|
|
manifest = merge_manifest(path, [item])
|
|
image = io.BytesIO()
|
|
Image.new("RGB", (2, 2)).save(image, format="PNG")
|
|
digest, relative, width, height, mime = store_asset(tmp_path, image.getvalue(), content_type="image/png", source_url=item.asset_url)
|
|
manifest["assets"][0].update({"status": "stored", "sha256": digest, "local_path": relative, "width": width, "height": height, "content_type": mime})
|
|
path.write_text(__import__("json").dumps(manifest), encoding="utf-8")
|
|
assert audit_media_catalog(tmp_path)["issues"] == []
|
|
(tmp_path / relative).write_bytes(b"changed")
|
|
(tmp_path / "files" / "orphan.png").write_bytes(b"orphan")
|
|
report = audit_media_catalog(tmp_path)
|
|
assert any("SHA-256 mismatch" in issue for issue in report["issues"])
|
|
assert "files/orphan.png" in report["orphaned_files"]
|