"""Integration tests for Poly Model Asset Collections and ZIP batch extraction. Covers tasks dfe19da5a448 and 840bb736d098. """ import io import zipfile import pytest from fastapi.testclient import TestClient from app.main import app def _register(client: TestClient, username: str) -> tuple[int, dict[str, str]]: resp = client.post( "/api/auth/register", json={ "username": username, "email": f"{username}@example.com", "password": "secret123", "date_of_birth": "1990-01-01", }, ) assert resp.status_code == 201, resp.text token = resp.json()["access_token"] headers = {"Authorization": f"Bearer {token}"} me = client.get("/api/users/me", headers=headers) assert me.status_code == 200, me.text return me.json()["id"], headers def test_collection_crud_and_permissions(): client = TestClient(app) uid1, h1 = _register(client, "col_user_1") uid2, h2 = _register(client, "col_user_2") # 1. Create collection as user 1 resp = client.post( "/api/assets/collections/", headers=h1, json={ "title": "Low Poly Fantasy Props", "description": "Swords, shields, and potions", "theme": "fantasy", "is_public": True, "tags": "fantasy,lowpoly,weapons", }, ) assert resp.status_code == 201, resp.text col_data = resp.json() assert col_data["title"] == "Low Poly Fantasy Props" assert col_data["theme"] == "fantasy" assert col_data["owner_id"] == uid1 col_id = col_data["id"] # 2. List collections with theme filter resp = client.get("/api/assets/collections/", headers=h2, params={"theme": "fantasy"}) assert resp.status_code == 200 cols = resp.json() assert any(c["id"] == col_id for c in cols) # 3. List with non-matching theme filter resp = client.get("/api/assets/collections/", headers=h2, params={"theme": "scifi"}) assert resp.status_code == 200 cols = resp.json() assert not any(c["id"] == col_id for c in cols) # 4. Search filter resp = client.get("/api/assets/collections/", headers=h2, params={"search": "shields"}) assert resp.status_code == 200 assert any(c["id"] == col_id for c in resp.json()) # 5. User 2 views public collection resp = client.get(f"/api/assets/collections/{col_id}", headers=h2) assert resp.status_code == 200 assert resp.json()["id"] == col_id # 6. User 2 cannot modify User 1's collection resp = client.patch( f"/api/assets/collections/{col_id}", headers=h2, json={"title": "Hacked Title"}, ) assert resp.status_code == 403 # 7. User 2 cannot delete User 1's collection resp = client.delete(f"/api/assets/collections/{col_id}", headers=h2) assert resp.status_code == 403 # 8. User 1 updates collection to private resp = client.patch( f"/api/assets/collections/{col_id}", headers=h1, json={"is_public": False, "theme": "medieval"}, ) assert resp.status_code == 200 assert resp.json()["is_public"] is False assert resp.json()["theme"] == "medieval" # 9. User 2 can no longer view private collection resp = client.get(f"/api/assets/collections/{col_id}", headers=h2) assert resp.status_code == 403 # 10. User 1 deletes collection resp = client.delete(f"/api/assets/collections/{col_id}", headers=h1) assert resp.status_code == 204 # 11. Verify gone resp = client.get(f"/api/assets/collections/{col_id}", headers=h1) assert resp.status_code == 404 def test_collection_item_association_and_ordering(): client = TestClient(app) uid, h = _register(client, "col_items_user") # Create 2 assets a1_resp = client.post( "/api/assets/", headers=h, json={"name": "Wooden Crate", "asset_type": "model_3d"}, ) assert a1_resp.status_code == 201, a1_resp.text asset1_id = a1_resp.json()["id"] a2_resp = client.post( "/api/assets/", headers=h, json={"name": "Iron Barrel", "asset_type": "model_3d"}, ) assert a2_resp.status_code == 201, a2_resp.text asset2_id = a2_resp.json()["id"] # Create collection col_resp = client.post( "/api/assets/collections/", headers=h, json={"title": "Industrial Pack", "theme": "industrial"}, ) assert col_resp.status_code == 201 col_id = col_resp.json()["id"] # Add items with explicit ordering add1 = client.post( f"/api/assets/collections/{col_id}/items", headers=h, json={"asset_id": asset1_id, "sort_order": 0}, ) assert add1.status_code == 200 assert add1.json()["items_count"] == 1 assert add1.json()["items"][0]["id"] == asset1_id add2 = client.post( f"/api/assets/collections/{col_id}/items", headers=h, json={"asset_id": asset2_id, "sort_order": 1}, ) assert add2.status_code == 200 assert add2.json()["items_count"] == 2 assert add2.json()["items"][1]["id"] == asset2_id # Remove item 1 rem = client.delete( f"/api/assets/collections/{col_id}/items/{asset1_id}", headers=h, ) assert rem.status_code == 200 assert rem.json()["items_count"] == 1 assert rem.json()["items"][0]["id"] == asset2_id def test_collection_zip_upload_and_extraction(): client = TestClient(app) uid, h = _register(client, "col_zip_user") # 1. Build a valid in-memory zip file with mock GLB content zip_buf = io.BytesIO() with zipfile.ZipFile(zip_buf, "w") as zf: zf.writestr("stone_pillar.glb", b"glTFmockheader_binary_stone_pillar") zf.writestr("wooden_arch.glb", b"glTFmockheader_binary_wooden_arch") zf.writestr("temple_roof.gltf", b'{"asset": {"version": "2.0"}}') zf.writestr("readme.txt", b"Instructions for this pack") zip_bytes = zip_buf.getvalue() # Upload zip pack resp = client.post( "/api/assets/collections/upload-zip", headers=h, data={ "title": "Ancient Ruins Pack", "theme": "historical", "description": "Set of stone pillars and wooden arches", "is_public": "true", "tags": "ruins,ancient,stone", }, files={"file": ("ruins_pack.zip", zip_bytes, "application/zip")}, ) assert resp.status_code == 201, resp.text data = resp.json() assert data["title"] == "Ancient Ruins Pack" assert data["theme"] == "historical" assert data["items_count"] == 3 names = [i["name"] for i in data["items"]] assert "Stone Pillar" in names assert "Wooden Arch" in names assert "Temple Roof" in names def test_collection_zip_error_handling(): client = TestClient(app) uid, h = _register(client, "col_badzip_user") # 1. Invalid zip payload resp = client.post( "/api/assets/collections/upload-zip", headers=h, data={"title": "Corrupt Pack", "theme": "general"}, files={"file": ("bad.zip", b"not-a-valid-zip-file", "application/zip")}, ) assert resp.status_code == 400 assert "Invalid ZIP archive" in resp.json()["detail"] # 2. Valid zip but no 3D models zip_buf = io.BytesIO() with zipfile.ZipFile(zip_buf, "w") as zf: zf.writestr("notes.txt", b"Hello world") zf.writestr("image.png", b"fake image bytes") resp = client.post( "/api/assets/collections/upload-zip", headers=h, data={"title": "No Models Pack", "theme": "general"}, files={"file": ("empty.zip", zip_buf.getvalue(), "application/zip")}, ) assert resp.status_code == 400 assert "no valid 3D models" in resp.json()["detail"]