"""Test the Brand Kit bulk upload endpoint behavior. Validates the real /api/brandkit/{domain}/upload pipeline: - Multi-file upload via multipart form - File type classification (doc vs image vs skipped) - File size and count limits - BrandKit + BrandKitAsset DB creation - Response shape matches what the editor JS expects - Files actually written to disk - Duplicate filename handling - Empty/invalid file rejection """ import os import io import pytest from app.models import BrandKit, BrandKitAsset TEST_DOMAIN = "bulkuploadtest.com" @pytest.fixture(autouse=True) def cleanup_brandkit(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)) os.rmdir(upload_dir) class TestBulkUploadEndpoint: """Validates POST /api/brandkit/{domain}/upload with real multipart files.""" def test_upload_single_text_file(self, client, db_session): files = [("files", ("readme.txt", io.BytesIO(b"Brand guidelines: use blue and gold."), "text/plain"))] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text}" data = resp.json() assert "uploaded" in data, f"Response missing 'uploaded' key: {data}" assert data["uploaded"]["docs"] == 1 assert data["uploaded"]["images"] == 0 assert data["uploaded"]["skipped"] == 0 assert data["status"] == "uploaded" assert data["brand_kit_id"] is not None def test_upload_single_image_file(self, client, db_session): png_header = b'\x89PNG\r\n\x1a\n' + b'\x00' * 100 files = [("files", ("logo.png", io.BytesIO(png_header), "image/png"))] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) assert resp.status_code == 200 data = resp.json() assert data["uploaded"]["images"] == 1 assert data["uploaded"]["docs"] == 0 def test_upload_multiple_mixed_files(self, client, db_session): files = [ ("files", ("brand_guide.txt", io.BytesIO(b"Our brand is about wellness."), "text/plain")), ("files", ("colors.md", io.BytesIO(b"# Colors\n- Blue\n- Gold"), "text/markdown")), ("files", ("hero.jpg", io.BytesIO(b'\xff\xd8\xff\xe0' + b'\x00' * 100), "image/jpeg")), ("files", ("icon.png", io.BytesIO(b'\x89PNG\r\n\x1a\n' + b'\x00' * 100), "image/png")), ] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) assert resp.status_code == 200 data = resp.json() assert data["uploaded"]["docs"] == 2, f"Expected 2 docs, got {data['uploaded']}" assert data["uploaded"]["images"] == 2, f"Expected 2 images, got {data['uploaded']}" assert data["uploaded"]["skipped"] == 0 def test_unsupported_file_types_skipped(self, client, db_session): files = [ ("files", ("script.py", io.BytesIO(b"import os"), "text/x-python")), ("files", ("data.csv", io.BytesIO(b"a,b,c\n1,2,3"), "text/csv")), ("files", ("valid.txt", io.BytesIO(b"This is valid brand content."), "text/plain")), ] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) assert resp.status_code == 200 data = resp.json() assert data["uploaded"]["skipped"] == 2, f"Expected 2 skipped, got {data['uploaded']}" assert data["uploaded"]["docs"] == 1 def test_empty_file_skipped(self, client, db_session): files = [("files", ("empty.txt", io.BytesIO(b""), "text/plain"))] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) assert resp.status_code == 200 data = resp.json() assert data["uploaded"]["skipped"] == 1 assert data["uploaded"]["docs"] == 0 def test_files_written_to_disk(self, client, db_session): content = b"Real brand content for disk write test." files = [("files", ("disktest.txt", io.BytesIO(content), "text/plain"))] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) assert resp.status_code == 200 upload_dir = os.path.join("static", "uploads", "brandkit", TEST_DOMAIN.replace(".", "_")) assert os.path.isdir(upload_dir), f"Upload directory not created: {upload_dir}" written_files = os.listdir(upload_dir) assert len(written_files) >= 1, f"No files written to {upload_dir}" found = any("disktest" in f for f in written_files) assert found, f"Expected disktest.txt in {written_files}" def test_db_records_created(self, client, db_session): files = [ ("files", ("dbtest.txt", io.BytesIO(b"DB record test content."), "text/plain")), ("files", ("dbtest.png", io.BytesIO(b'\x89PNG\r\n\x1a\n' + b'\x00' * 50), "image/png")), ] 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() assert kit is not None, "BrandKit record not created" assert kit.status == "uploaded" assets = db_session.query(BrandKitAsset).filter(BrandKitAsset.brand_kit_id == kit.id).all() assert len(assets) == 2, f"Expected 2 assets, got {len(assets)}" types = {a.asset_type for a in assets} assert "document" in types assert "image" in types def test_duplicate_filenames_handled(self, client, db_session): files = [ ("files", ("same.txt", io.BytesIO(b"First version."), "text/plain")), ] client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) files2 = [ ("files", ("same.txt", io.BytesIO(b"Second version with different content."), "text/plain")), ] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files2) assert resp.status_code == 200 upload_dir = os.path.join("static", "uploads", "brandkit", TEST_DOMAIN.replace(".", "_")) written = os.listdir(upload_dir) assert len(written) >= 2, f"Expected duplicate handling to create 2+ files, got {written}" def test_response_shape_matches_editor_js(self, client, db_session): """The editor JS reads: result.uploaded.docs, result.uploaded.images, result.uploaded.skipped. This test ensures the API contract matches.""" files = [("files", ("contract.txt", io.BytesIO(b"Contract test."), "text/plain"))] resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload", files=files) data = resp.json() assert "brand_kit_id" in data assert "domain" in data assert "uploaded" in data assert "status" in data uploaded = data["uploaded"] assert "docs" in uploaded, f"Missing 'docs' in uploaded: {uploaded}" assert "images" in uploaded, f"Missing 'images' in uploaded: {uploaded}" assert "skipped" in uploaded, f"Missing 'skipped' in uploaded: {uploaded}" assert isinstance(uploaded["docs"], int) assert isinstance(uploaded["images"], int) assert isinstance(uploaded["skipped"], int) def test_no_files_returns_400(self, client): resp = client.post(f"/api/brandkit/{TEST_DOMAIN}/upload") assert resp.status_code in (400, 422), f"Expected 400/422 for no files, got {resp.status_code}"