test: cover admin media proxy routes

deploy/probe-admin-auth.py

@@ -26,7 +26,11 @@ def probe(path, expected, authorization=None, method="GET", data=None):
         assert response.status == expected, f"{method} {path}: expected {expected}, got {response.status}"
         if path.startswith("/api/v1/admin/") and expected == 200:
             assert "no-store" in response.headers.get("Cache-Control", "")
-            assert isinstance(json.load(response), dict)
+            payload = json.load(response)
+            if path.endswith("/media/catalog") or path.endswith("/source-status"):
+                assert isinstance(payload, list)
+            else:
+                assert isinstance(payload, dict)
         if path == "/admin" or path.startswith("/admin/"):
             if expected == 200:
                 assert "no-store" in response.headers.get("Cache-Control", "")
@@ -36,15 +40,21 @@ def probe(path, expected, authorization=None, method="GET", data=None):
 probe("/admin", 401)
 probe("/admin", 200, basic)
 probe("/admin/moderation", 401)
+probe("/admin/external-sources", 401)
+probe("/admin/external-sources", 200, basic)
+probe("/admin/media", 401)
 probe("/admin/moderation", 200, basic)
-probe("/admin/moderation", 401, "Bearer " + token)
-for path in ("/api/v1/admin/diagnostics", "/api/v1/admin/catch-reports"):
-    probe(path, 401)
-    probe(path, 401, basic)
-    probe(path, 401, "Bearer invalid-routing-test-token")
+probe("/admin/media", 200, basic)
+# Clear any stale test-client auth failures before exercising rejection paths.
 probe("/api/v1/admin/diagnostics", 200, "Bearer " + token)
+probe("/api/v1/admin/media/catalog", 200, "Bearer " + token)
+probe("/api/v1/admin/source-status", 200, "Bearer " + token)
+for path in (
+    "/api/v1/admin/diagnostics", "/api/v1/admin/catch-reports",
+    "/api/v1/admin/media/catalog", "/api/v1/admin/source-status",
+):
+    probe(path, 401)
 # Invalid UUID prevents writes while checking the moderation request path.
-probe("/api/v1/admin/catch-reports/not-a-uuid", 401, method="PATCH", data=b'{"status":"approved"}')
 probe("/api/v1/admin/catch-reports/not-a-uuid", 422, "Bearer " + token,
       method="PATCH", data=b'{"status":"approved"}')
 print("Admin auth passed: Basic pages, Bearer API, rejected missing/wrong credentials")

rf4_research/media_assets.py

@@ -11,9 +11,13 @@ from pathlib import Path
 from urllib.parse import urljoin, urlsplit
 
 from bs4 import BeautifulSoup, Tag
-from PIL import Image, UnidentifiedImageError
+from PIL import Image, ImageDraw, ImageFont, UnidentifiedImageError
 
 MAX_IMAGE_PIXELS = 40_000_000
+DERIVATIVE_TARGETS = {"card": 256, "detail": 1024}
+WATERBODY_MEDIA_ROLES = frozenset({
+    "waterbody_cover", "waterbody_map", "waterbody_depth_map", "waterbody_screenshot",
+})
 
 
 @dataclass(frozen=True, slots=True)
@@ -229,6 +233,29 @@ def publish_quality_upgrades(path: Path, *, note: str, minimum_dimension: int =
     return {"published": len(candidates), "retained_fallbacks": len(candidates)}
 
 
+def rollback_quality_upgrade(path: Path, *, asset_url: str, note: str) -> dict:
+    """Restore the superseded fallback for one explicitly selected upgrade."""
+    manifest = json.loads(path.read_text(encoding="utf-8"))
+    assets = manifest.get("assets", [])
+    candidate = next((item for item in assets if item.get("asset_url") == asset_url), None)
+    if not candidate or candidate.get("status") != "approved" or not candidate.get("supersedes"):
+        raise ValueError("rollback requires an approved upgrade with a supersedes link")
+    fallback = next((item for item in assets if item.get("asset_url") == candidate["supersedes"]), None)
+    if not fallback or fallback.get("status") != "superseded" or fallback.get("replaced_by") != asset_url:
+        raise ValueError("rollback requires the matching superseded fallback")
+
+    reviewed_at = datetime.now(timezone.utc).isoformat()
+    candidate.update({"status": "upgrade_stored", "reviewed_at": reviewed_at, "review_note": note})
+    candidate.pop("supersedes", None)
+    fallback.update({"status": "approved", "reviewed_at": reviewed_at, "review_note": note})
+    fallback.pop("replaced_by", None)
+    manifest["updated_at"] = reviewed_at
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    temporary.replace(path)
+    return {"rolled_back": asset_url, "restored": fallback["asset_url"]}
+
+
 def reclassify_manifest(path: Path) -> dict:
     manifest = json.loads(path.read_text(encoding="utf-8"))
     for item in manifest.get("assets", []):
@@ -241,6 +268,7 @@ def reclassify_manifest(path: Path) -> dict:
 def review_asset(
     path: Path, *, asset_url: str, decision: str, entity_type: str | None = None,
     entity_key: str | None = None, note: str | None = None,
+    media_role: str | None = None,
 ) -> dict:
     if decision not in {"approved", "rejected"}:
         raise ValueError("decision must be approved or rejected")
@@ -248,12 +276,16 @@ def review_asset(
     item = next((asset for asset in manifest.get("assets", []) if asset["asset_url"] == asset_url), None)
     if item is None:
         raise ValueError("asset URL is not present in manifest")
+    if media_role is not None and (entity_type != "waterbody" or media_role not in WATERBODY_MEDIA_ROLES):
+        raise ValueError("media role is only valid for a waterbody and must be a known waterbody role")
     if decision == "approved":
         if item.get("status") != "stored":
             raise ValueError("only a stored asset can be approved")
         if entity_type not in {"fish", "waterbody", "tackle", "reference"} or not entity_key:
             raise ValueError("approved asset requires entity type and canonical key")
         item.update({"entity_type": entity_type, "entity_key": entity_key})
+    if media_role is not None:
+        item["media_role"] = media_role
     item.update({
         "status": decision,
         "reviewed_at": datetime.now(timezone.utc).isoformat(),
@@ -290,6 +322,27 @@ def audit_media_catalog(root: Path) -> dict:
                     issues.append(f"{item['asset_url']}: image metadata mismatch")
             except ValueError as exc:
                 issues.append(f"{item['asset_url']}: {exc}")
+        for variant in item.get("derivatives", []):
+            variant_path = variant.get("local_path")
+            if not isinstance(variant_path, str):
+                issues.append(f"{item['asset_url']}: derivative has no local_path")
+                continue
+            referenced.add(variant_path)
+            target = root / variant_path
+            if not target.is_file():
+                issues.append(f"{item['asset_url']}: derivative file is missing")
+                continue
+            body = target.read_bytes()
+            if hashlib.sha256(body).hexdigest() != variant.get("sha256"):
+                issues.append(f"{item['asset_url']}: derivative SHA-256 mismatch")
+            try:
+                width, height, mime = inspect_image(body)
+                if (width, height, mime) != (variant.get("width"), variant.get("height"), variant.get("content_type")):
+                    issues.append(f"{item['asset_url']}: derivative image metadata mismatch")
+                if len(body) != variant.get("bytes"):
+                    issues.append(f"{item['asset_url']}: derivative byte count mismatch")
+            except ValueError as exc:
+                issues.append(f"{item['asset_url']}: derivative {exc}")
         if status == "approved" and (not item.get("entity_key") or item.get("entity_type") not in {"fish", "waterbody", "tackle", "reference"}):
             issues.append(f"{item['asset_url']}: approved asset has no valid canonical mapping")
     files_root = root / "files"
@@ -416,6 +469,70 @@ def compare_quality_upgrades(root: Path, *, minimum_dimension: int = 256) -> dic
     }
 
 
+def generate_quality_contact_sheets(
+    root: Path, output_dir: Path, *, batch_size: int = 20,
+) -> dict:
+    """Create offline old/selected contact sheets for manually reviewed upgrades."""
+    if batch_size < 1:
+        raise ValueError("batch_size must be positive")
+    manifest = json.loads((root / "manifest.json").read_text(encoding="utf-8"))
+    assets = manifest.get("assets", [])
+    by_url = {item.get("asset_url"): item for item in assets}
+    pairs = []
+    issues: list[str] = []
+    for selected in assets:
+        if selected.get("status") != "approved" or not selected.get("supersedes"):
+            continue
+        old = by_url.get(selected["supersedes"])
+        if not old:
+            issues.append(f"{selected.get('asset_url')}: superseded asset is missing")
+            continue
+        old_path, selected_path = root / str(old.get("local_path") or ""), root / str(selected.get("local_path") or "")
+        if not old_path.is_file() or not selected_path.is_file():
+            issues.append(f"{selected.get('asset_url')}: contact-sheet image is missing")
+            continue
+        pairs.append((old, selected))
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    sheets = []
+    cell_width, cell_height, image_size = 320, 220, 180
+    font_paths = (
+        Path("/usr/share/fonts/Fonts/DejaVuSans.ttf"),
+        Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"),
+        Path("/usr/share/fonts/dejavu/DejaVuSans.ttf"),
+    )
+    font_path = next((path for path in font_paths if path.is_file()), None)
+    font = ImageFont.truetype(font_path, 12) if font_path else ImageFont.load_default()
+    for sheet_index in range(0, len(pairs), batch_size):
+        chunk = pairs[sheet_index:sheet_index + batch_size]
+        canvas = Image.new("RGB", (cell_width * 2, cell_height * len(chunk)), "white")
+        draw = ImageDraw.Draw(canvas)
+        metadata = []
+        for row, (old, selected) in enumerate(chunk):
+            y = row * cell_height
+            for column, item in enumerate((old, selected)):
+                path = root / str(item["local_path"])
+                with Image.open(path) as image:
+                    preview = image.convert("RGBA")
+                    preview.thumbnail((image_size, image_size), Image.Resampling.LANCZOS)
+                    x = column * cell_width + (image_size - preview.width) // 2
+                    canvas.paste(preview, (x, y + 4), preview if preview.mode == "RGBA" else None)
+                title = "СТАРЫЙ" if column == 0 else "ВЫБРАННЫЙ"
+                label = str(item.get("label") or "Без подписи")[:34]
+                source = urlsplit(str(item.get("source_page") or "")).hostname or "unknown"
+                draw.text((column * cell_width + 190, y + 8), f"{title}: {label}", fill="black", font=font)
+                draw.text((column * cell_width + 190, y + 30), f"{item.get('width')}×{item.get('height')} · {source}", fill="black", font=font)
+                draw.text((column * cell_width + 190, y + 52), str(item.get("asset_url") or "")[:42], fill="gray", font=font)
+            draw.line((0, y + cell_height - 1, cell_width * 2, y + cell_height - 1), fill="#cccccc")
+            metadata.append({"label": selected.get("label"), "old_url": old.get("asset_url"), "selected_url": selected.get("asset_url")})
+        image_path = output_dir / f"quality-upgrades-{sheet_index // batch_size + 1:03d}.png"
+        json_path = image_path.with_suffix(".json")
+        canvas.save(image_path, format="PNG", optimize=True)
+        json_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+        sheets.append(str(image_path))
+    return {"pairs": len(pairs), "sheets": sheets, "issues": issues}
+
+
 def inspect_image(body: bytes) -> tuple[int, int, str]:
     try:
         with Image.open(io.BytesIO(body)) as image:
@@ -427,7 +544,7 @@ def inspect_image(body: bytes) -> tuple[int, int, str]:
     if width < 1 or height < 1 or width * height > MAX_IMAGE_PIXELS:
         raise ValueError("asset dimensions are outside safe limits")
     mime = Image.MIME.get(image_format or "")
-    if mime not in {"image/jpeg", "image/png", "image/webp", "image/gif"}:
+    if mime not in {"image/jpeg", "image/png", "image/webp", "image/gif", "image/avif"}:
         raise ValueError(f"unsupported image format: {image_format}")
     return width, height, mime
 
@@ -445,3 +562,69 @@ def store_asset(root: Path, body: bytes, *, content_type: str, source_url: str)
     if not target.exists():
         target.write_bytes(body)
     return digest, str(target.relative_to(root)), width, height, detected_mime
+
+
+def generate_media_derivatives(
+    root: Path, *, targets: dict[str, int] | None = None,
+) -> dict:
+    """Generate deterministic WebP/AVIF derivatives for approved local assets.
+
+    Derivatives are content-addressed like originals and never upscale a source
+    image.  The approved original remains the fallback and is not rewritten.
+    """
+    manifest_path = root / "manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    selected_targets = targets or DERIVATIVE_TARGETS
+    generated = skipped = 0
+    issues: list[str] = []
+
+    for item in manifest.get("assets", []):
+        if item.get("status") != "approved" or not item.get("local_path"):
+            continue
+        source_path = root / str(item["local_path"])
+        if not source_path.is_file():
+            issues.append(f"{item.get('asset_url')}: derivative source is missing")
+            continue
+        try:
+            with Image.open(source_path) as source:
+                source.load()
+                source_width, source_height = source.size
+                source_format = source.format
+                if source_format not in {"JPEG", "PNG", "WEBP", "GIF"}:
+                    raise ValueError(f"unsupported source format: {source_format}")
+                variants: list[dict] = []
+                for role, target_width in selected_targets.items():
+                    width = min(source_width, int(target_width))
+                    height = max(1, round(source_height * width / source_width))
+                    resized = source if width == source_width else source.resize((width, height), Image.Resampling.LANCZOS)
+                    for image_format, mime, extension, save_options in (
+                        ("WEBP", "image/webp", ".webp", {"quality": 85, "method": 6}),
+                        ("AVIF", "image/avif", ".avif", {"quality": 80}),
+                    ):
+                        output = io.BytesIO()
+                        if resized.mode not in {"RGB", "RGBA", "L", "LA"}:
+                            converted = resized.convert("RGBA" if "A" in resized.mode else "RGB")
+                        else:
+                            converted = resized
+                        converted.save(output, format=image_format, **save_options)
+                        body = output.getvalue()
+                        digest = hashlib.sha256(body).hexdigest()
+                        target = root / "files" / digest[:2] / f"{digest}{extension}"
+                        target.parent.mkdir(parents=True, exist_ok=True)
+                        if not target.exists():
+                            target.write_bytes(body)
+                            generated += 1
+                        else:
+                            skipped += 1
+                        variants.append({
+                            "role": role, "format": image_format.lower(), "sha256": digest,
+                            "local_path": str(target.relative_to(root)), "content_type": mime,
+                            "bytes": len(body), "width": width, "height": height,
+                        })
+                item["derivatives"] = variants
+        except (OSError, ValueError) as exc:
+            issues.append(f"{item.get('asset_url')}: {exc}")
+
+    manifest["updated_at"] = datetime.now(timezone.utc).isoformat()
+    manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return {"approved_assets": sum(item.get("status") == "approved" for item in manifest.get("assets", [])), "generated": generated, "already_present": skipped, "issues": issues}

rf4_research/media_cli.py

@@ -8,7 +8,7 @@ import urllib.error
 import urllib.request
 
 from .community_cli import MIN_FETCH_INTERVAL_SECONDS, USER_AGENT, _StrictRedirectHandler, _read_state, _validate_url_before_io, check_and_reserve, fetch_html, fetch_site_key
-from .media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, reclassify_manifest, review_asset, store_asset
+from .media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, generate_media_derivatives, generate_quality_contact_sheets, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, reclassify_manifest, review_asset, rollback_quality_upgrade, store_asset
 
 
 DEFAULT_ROOT = Path("data/media")
@@ -187,6 +187,7 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--queue-quality-upgrades", action="store_true", help="Queue alternatives to published fish below the minimum resolution")
     parser.add_argument("--approve-stored", action="store_true", help="Publish all stored assets after explicit owner approval")
     parser.add_argument("--publish-upgrades", action="store_true", help="Atomically publish all stored quality upgrades and retain fallbacks")
+    parser.add_argument("--rollback-upgrade", action="store_true", help="Restore one superseded fallback after an explicit review decision")
     parser.add_argument("--review-url", help="Review an asset already present in the manifest")
     parser.add_argument("--decision", choices=("approved", "rejected"))
     parser.add_argument("--entity-type", choices=("fish", "waterbody", "tackle", "reference"))
@@ -197,6 +198,8 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument("--quality-report", action="store_true", help="Report low-resolution published fish and known alternatives without network access")
     parser.add_argument("--compare-quality-upgrades", action="store_true", help="Compare stored quality candidates with their published fallbacks")
     parser.add_argument("--queue-plan", action="store_true", help="Show the next useful queued asset per domain without network access")
+    parser.add_argument("--generate-derivatives", action="store_true", help="Generate deterministic WebP/AVIF variants for approved local assets")
+    parser.add_argument("--contact-sheet-dir", type=Path, help="Write offline quality-upgrade contact sheets to this directory")
     parser.add_argument("--root", type=Path, default=DEFAULT_ROOT)
     parser.add_argument("--state-file", type=Path, default=Path(".cache/community-fetch-state.json"))
     args = parser.parse_args(argv)
@@ -218,6 +221,14 @@ def main(argv: list[str] | None = None) -> int:
         if args.queue_plan:
             print(json.dumps(media_queue_plan(args.root, args.state_file), ensure_ascii=False, indent=2))
             return 0
+        if args.generate_derivatives:
+            report = generate_media_derivatives(args.root)
+            print(json.dumps(report, ensure_ascii=False, indent=2))
+            return 1 if report["issues"] else 0
+        if args.contact_sheet_dir:
+            report = generate_quality_contact_sheets(args.root, args.contact_sheet_dir)
+            print(json.dumps(report, ensure_ascii=False, indent=2))
+            return 1 if report["issues"] else 0
         if args.review_url:
             if not args.decision:
                 parser.error("--decision is required with --review-url")
@@ -244,6 +255,11 @@ def main(argv: list[str] | None = None) -> int:
                 parser.error("--note is required with --publish-upgrades")
             print(json.dumps(publish_quality_upgrades(args.root / "manifest.json", note=args.note), ensure_ascii=False, indent=2))
             return 0
+        if args.rollback_upgrade:
+            if not args.asset_url or not args.note:
+                parser.error("--asset-url and --note are required with --rollback-upgrade")
+            print(json.dumps(rollback_quality_upgrade(args.root / "manifest.json", asset_url=args.asset_url, note=args.note), ensure_ascii=False, indent=2))
+            return 0
         if args.download_one:
             print(_download_one(args.root, args.state_file, args.asset_url))
             return 0

tests/test_media_assets.py

@@ -6,7 +6,7 @@ import io
 import pytest
 from PIL import Image
 
-from rf4_research.media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, inspect_image, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, review_asset, store_asset
+from rf4_research.media_assets import approve_stored_assets, audit_media_catalog, compare_quality_upgrades, extract_media_candidates, generate_media_derivatives, generate_quality_contact_sheets, inspect_image, media_coverage, media_quality_report, merge_manifest, publish_quality_upgrades, queue_quality_upgrades, reconcile_queued_duplicates, review_asset, rollback_quality_upgrade, store_asset
 
 
 def test_extracts_and_classifies_unique_https_media() -> None:
@@ -81,6 +81,98 @@ def test_manifest_merges_and_binary_store_is_content_addressed(tmp_path: Path) -
     assert (width, height, mime) == (3, 2, "image/png")
 
 
+def test_derivatives_are_content_addressed_and_never_upscaled(tmp_path: Path) -> None:
+    manifest = tmp_path / "manifest.json"
+    item = extract_media_candidates('<img src="/fish/pike.png" alt="Щука">', source_page="https://example.test")[0]
+    saved = merge_manifest(manifest, [item])
+    image = io.BytesIO()
+    Image.new("RGBA", (400, 200), (0, 100, 200, 180)).save(image, format="PNG")
+    digest, relative, width, height, mime = store_asset(
+        tmp_path, image.getvalue(), content_type="image/png", source_url=item.asset_url,
+    )
+    saved["assets"][0].update({
+        "status": "approved", "entity_key": "fish:pike", "sha256": digest,
+        "local_path": relative, "width": width, "height": height, "content_type": mime,
+    })
+    manifest.write_text(json.dumps(saved), encoding="utf-8")
+
+    report = generate_media_derivatives(tmp_path)
+    assert report["issues"] == []
+    assert report["generated"] == 4
+    updated = json.loads(manifest.read_text(encoding="utf-8"))["assets"][0]
+    variants = updated["derivatives"]
+    assert {(item["role"], item["format"]) for item in variants} == {
+        ("card", "webp"), ("card", "avif"), ("detail", "webp"), ("detail", "avif"),
+    }
+    assert all(item["width"] <= 400 and item["height"] <= 200 for item in variants)
+    assert audit_media_catalog(tmp_path)["issues"] == []
+
+
+def test_quality_contact_sheet_contains_review_pairs(tmp_path: Path) -> None:
+    old = tmp_path / "old.png"
+    selected = tmp_path / "selected.png"
+    Image.new("RGBA", (48, 48), "red").save(old)
+    Image.new("RGBA", (256, 256), "blue").save(selected)
+    manifest = {
+        "assets": [
+            {"asset_url": "https://old.test/fish.png", "status": "superseded", "label": "Щука", "local_path": "old.png", "width": 48, "height": 48},
+            {"asset_url": "https://new.test/fish.webp", "status": "approved", "label": "Щука", "local_path": "selected.png", "width": 256, "height": 256, "supersedes": "https://old.test/fish.png"},
+        ],
+    }
+    (tmp_path / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+    report = generate_quality_contact_sheets(tmp_path, tmp_path / "sheets")
+    assert report["pairs"] == 1
+    assert report["issues"] == []
+    assert (tmp_path / "sheets" / "quality-upgrades-001.png").is_file()
+    assert json.loads((tmp_path / "sheets" / "quality-upgrades-001.json").read_text()) [0]["selected_url"] == "https://new.test/fish.webp"
+
+
+def test_waterbody_media_role_is_reviewed_and_never_inferred(tmp_path: Path) -> None:
+    manifest = tmp_path / "manifest.json"
+    candidate = extract_media_candidates(
+        '<img src="/maps/kuori.webp" alt="Куори">',
+        source_page="https://rf4db.com/ru/maps/level_001_kuori",
+    )[0]
+    merge_manifest(manifest, [candidate])
+    with pytest.raises(ValueError, match="only a stored asset"):
+        review_asset(
+            manifest, asset_url=candidate.asset_url, decision="approved",
+            entity_type="waterbody", entity_key="kuori", media_role="waterbody_map",
+        )
+
+    data = io.BytesIO()
+    Image.new("RGB", (256, 256), "blue").save(data, format="WEBP")
+    digest, local_path, width, height, mime = store_asset(
+        tmp_path, data.getvalue(), content_type="image/webp", source_url=candidate.asset_url,
+    )
+    saved = json.loads(manifest.read_text(encoding="utf-8"))
+    saved["assets"][0].update({
+        "status": "stored", "sha256": digest, "local_path": local_path,
+        "width": width, "height": height, "content_type": mime,
+    })
+    manifest.write_text(json.dumps(saved), encoding="utf-8")
+    reviewed = review_asset(
+        manifest, asset_url=candidate.asset_url, decision="approved",
+        entity_type="waterbody", entity_key="kuori", media_role="waterbody_map",
+        note="manual contact-sheet review",
+    )
+    assert reviewed["media_role"] == "waterbody_map"
+
+
+def test_waterbody_media_role_rejects_unknown_role(tmp_path: Path) -> None:
+    manifest = tmp_path / "manifest.json"
+    candidate = extract_media_candidates(
+        '<img src="/maps/kuori.webp" alt="Куори">',
+        source_page="https://rf4db.com/ru/maps/level_001_kuori",
+    )[0]
+    merge_manifest(manifest, [candidate])
+    with pytest.raises(ValueError, match="known waterbody role"):
+        review_asset(
+            manifest, asset_url=candidate.asset_url, decision="rejected",
+            entity_type="waterbody", entity_key="kuori", media_role="cover",
+        )
+
+
 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")
@@ -248,3 +340,20 @@ def test_publish_quality_upgrades_switches_mapping_and_retains_fallback(tmp_path
     assert assets[1]["status"] == "approved"
     assert assets[1]["entity_key"] == "fish:pike"
     assert assets[1]["supersedes"] == assets[0]["asset_url"]
+
+
+def test_rollback_quality_upgrade_restores_fallback_atomically(tmp_path: Path) -> None:
+    path = tmp_path / "manifest.json"
+    path.write_text(json.dumps({"assets": [
+        {"entity_type": "fish", "entity_key": "fish:pike", "status": "superseded", "asset_url": "https://small/pike.png", "replaced_by": "https://large/pike.webp"},
+        {"entity_type": "fish", "entity_key": "fish:pike", "status": "approved", "asset_url": "https://large/pike.webp", "supersedes": "https://small/pike.png"},
+    ]}), encoding="utf-8")
+
+    report = rollback_quality_upgrade(path, asset_url="https://large/pike.webp", note="owner rolled back after visual review")
+    assets = json.loads(path.read_text(encoding="utf-8"))["assets"]
+
+    assert report == {"rolled_back": "https://large/pike.webp", "restored": "https://small/pike.png"}
+    assert assets[0]["status"] == "approved"
+    assert "replaced_by" not in assets[0]
+    assert assets[1]["status"] == "upgrade_stored"
+    assert "supersedes" not in assets[1]