"""Test Brand Kit dedup, classification skip logic, and asset resolver. Validates: - SHA-256 file hashing on upload prevents duplicates per brand_kit_id - classified_at timestamps cause skip during reprocessing (unless force=True) - Empty/unreadable docs still get classified_at set (no infinite reprocess loop) - Failed classification still sets classified_at (no retry storm) - Asset resolver maps classified images to correct site sections - Asset resolver handles dict and ORM-style inputs - Asset resolver skips documents - Force mode overrides all skip logic """ import os import io import hashlib import datetime import pytest from unittest.mock import patch, MagicMock from app.models import BrandKit, BrandKitAsset from app.services.brandkit import ( compute_file_hash, resolve_assets_for_sections, CLASSIFICATION_TO_SECTIONS, CLASSIFICATION_LABELS, ) TEST_DOMAIN = "deduptest.com" @pytest.fixture(autouse=True) def cleanup(db_session): yield kit = db_session.query(BrandKit).filter(BrandKit.domain == TEST_DOMAIN).first() if kit: assets = db_session.query(BrandKitAsset).filter(BrandKitAsset.brand_kit_id == kit.id).all() for a in assets: if a.file_path and os.path.isfile(a.file_path): os.remove(a.file_path) db_session.delete(a) db_session.delete(kit) db_session.commit() upload_dir = os.path.join("static", "uploads", "brandkit", TEST_DOMAIN.replace(".", "_")) if os.path.isdir(upload_dir): for f in os.listdir(upload_dir): os.remove(os.path.join(upload_dir, f)) try: os.rmdir(upload_dir) except OSError: pass class TestFileHashing: """Validates SHA-256 hashing utility.""" def test_hash_returns_hex_string(self, tmp_path): f = tmp_path / "test.txt" f.write_bytes(b"hello world") h = compute_file_hash(str(f)) assert len(h) == 64 assert h == hashlib.sha256(b"hello world").hexdigest() def test_same_content_same_hash(self, tmp_path): f1 = tmp_path / "a.txt" f2 = tmp_path / "b.txt" f1.write_bytes(b"identical content") f2.write_bytes(b"identical content") assert compute_file_hash(str(f1)) == compute_file_hash(str(f2)) def test_different_content_different_hash(self, tmp_path): f1 = tmp_path / "a.txt" f2 = tmp_path / "b.txt" f1.write_bytes(b"content A") f2.write_bytes(b"content B") assert compute_file_hash(str(f1)) != compute_file_hash(str(f2)) def test_missing_file_returns_empty(self): h = compute_file_hash("/nonexistent/path/file.txt") assert h == "" def test_empty_file_returns_valid_hash(self, tmp_path): f = tmp_path / "empty.txt" f.write_bytes(b"") h = compute_file_hash(str(f)) assert len(h) == 64 assert h == hashlib.sha256(b"").hexdigest() class TestDuplicateUploadPrevention: """Validates that uploading the same file content twice is caught.""" def test_duplicate_content_skipped(self, client, db_session): content = b"Unique brand content for dedup test." files1 = [("files", ("brand.txt", io.BytesIO(content), "text/plain"))] resp1 = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files1) assert resp1.status_code == 200 assert resp1.json()["uploaded"]["docs"] == 1 files2 = [("files", ("brand_copy.txt", io.BytesIO(content), "text/plain"))] resp2 = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files2) assert resp2.status_code == 200 data2 = resp2.json() assert data2["uploaded"]["skipped"] >= 1 assert "duplicates_skipped" in data2 assert len(data2["duplicates_skipped"]) >= 1 def test_different_content_same_name_accepted(self, client, db_session): files1 = [("files", ("brand.txt", io.BytesIO(b"Version 1 content."), "text/plain"))] resp1 = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files1) assert resp1.status_code == 200 files2 = [("files", ("brand.txt", io.BytesIO(b"Version 2 completely different content."), "text/plain"))] resp2 = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files2) assert resp2.status_code == 200 assert resp2.json()["uploaded"]["docs"] >= 1 def test_file_hash_stored_on_asset(self, client, db_session): content = b"Content with hash tracking." files = [("files", ("hashtest.txt", io.BytesIO(content), "text/plain"))] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) assert resp.status_code == 200 db_session.expire_all() kit = db_session.query(BrandKit).filter(BrandKit.domain == TEST_DOMAIN).first() asset = db_session.query(BrandKitAsset).filter(BrandKitAsset.brand_kit_id == kit.id).first() assert asset is not None assert asset.file_hash is not None assert len(asset.file_hash) == 64 class TestAssetResolver: """Validates resolve_assets_for_sections mapping logic.""" def test_hero_banner_maps_to_hero_and_about(self): assets = [{"classification": "hero_banner", "filename": "hero.jpg", "file_path": "/img/hero.jpg", "id": 1, "asset_type": "image", "ai_description": "A banner", "tags": ["banner"]}] result = resolve_assets_for_sections(assets) assert "hero" in result assert "about" in result assert result["hero"][0]["filename"] == "hero.jpg" def test_logo_maps_to_hero_nav_footer(self): assets = [{"classification": "logo", "filename": "logo.png", "file_path": "/img/logo.png", "id": 2, "asset_type": "image", "ai_description": "Company logo", "tags": ["logo"]}] result = resolve_assets_for_sections(assets) assert "hero" in result assert "nav" in result assert "footer" in result def test_documents_are_skipped(self): assets = [{"classification": "other", "filename": "readme.txt", "file_path": "/docs/readme.txt", "id": 3, "asset_type": "document", "ai_description": "A doc", "tags": []}] result = resolve_assets_for_sections(assets) assert len(result) == 0 def test_multiple_assets_same_section(self): assets = [ {"classification": "hero_banner", "filename": "hero1.jpg", "file_path": "/img/hero1.jpg", "id": 1, "asset_type": "image", "ai_description": "", "tags": []}, {"classification": "lifestyle", "filename": "life.jpg", "file_path": "/img/life.jpg", "id": 2, "asset_type": "image", "ai_description": "", "tags": []}, ] result = resolve_assets_for_sections(assets) hero_filenames = [a["filename"] for a in result.get("hero", [])] assert "hero1.jpg" in hero_filenames assert "life.jpg" in hero_filenames def test_used_in_sections_override(self): assets = [{"classification": "other", "filename": "custom.jpg", "file_path": "/img/custom.jpg", "id": 4, "asset_type": "image", "used_in_sections": ["pricing", "faq"], "ai_description": "", "tags": []}] result = resolve_assets_for_sections(assets) assert "pricing" in result assert "faq" in result def test_empty_assets_returns_empty(self): result = resolve_assets_for_sections([]) assert result == {} def test_classification_labels_populated(self): assets = [{"classification": "team_photo", "filename": "team.jpg", "file_path": "/img/team.jpg", "id": 5, "asset_type": "image", "ai_description": "", "tags": []}] result = resolve_assets_for_sections(assets) team_assets = result.get("team", []) assert len(team_assets) >= 1 assert team_assets[0]["classification_label"] == "Team / People" def test_orm_style_object_input(self): mock_asset = MagicMock() mock_asset.classification = "product_shot" mock_asset.asset_type = "image" mock_asset.used_in_sections = None mock_asset.id = 10 mock_asset.filename = "product.jpg" mock_asset.file_path = "/img/product.jpg" mock_asset.ai_description = "A product" mock_asset.tags = ["product"] result = resolve_assets_for_sections([mock_asset]) assert "features" in result assert "pricing" in result def test_all_classifications_have_labels(self): for cls in CLASSIFICATION_TO_SECTIONS: assert cls in CLASSIFICATION_LABELS, f"Missing label for classification: {cls}" def test_all_classification_mappings_valid(self): for cls, sections in CLASSIFICATION_TO_SECTIONS.items(): assert isinstance(sections, list), f"Sections for {cls} should be a list" for s in sections: assert isinstance(s, str), f"Section name should be string, got {type(s)} for {cls}"