"""上传案例：会员提交住宅，管理端只读，不进案例库文章。""" import time from datetime import date import pytest from fastapi.testclient import TestClient from app.core.config import get_settings from app.core.security import create_access_token from app.services.email_code import _key from app.services.redis_client import get_redis @pytest.fixture def client(monkeypatch): monkeypatch.setenv("JWT_SECRET", "test-secret-must-be-32-characters!") monkeypatch.setenv("API_PAYLOAD_ENCRYPT", "false") monkeypatch.setenv("SMTP_HOST", "") get_settings.cache_clear() if not get_settings().database_url.strip(): pytest.skip("DATABASE_URL 为空") try: get_redis().ping() except Exception: pytest.skip("Redis 不可达") monkeypatch.setattr("app.services.email_code.send_email", lambda **kwargs: None) from app.main import create_app with TestClient(create_app()) as api: yield api get_settings.cache_clear() def _minio_ready() -> bool: try: from app.services.storage import ensure_buckets ensure_buckets() return True except Exception: return False def _png() -> bytes: return ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" b"\x08\x02\x00\x00\x00\x90wS\xde\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f" b"\x00\x00\x01\x01\x00\x05\x18\xd8N\x00\x00\x00\x00IEND\xaeB`\x82" ) def _code(purpose: str, email: str) -> str: stored = get_redis().get(_key(purpose, email)) assert stored is not None if isinstance(stored, bytes): stored = stored.decode("utf-8") return stored def _register(client, email: str) -> str: assert client.post("/api/site/auth/email-code", json={"email": email, "purpose": "register"}).status_code == 200 registered = client.post( "/api/site/auth/register", json={"email": email, "code": _code("register", email), "password": "Passw0rd1", "privacyAccepted": True}, ) assert registered.status_code == 200, registered.text return registered.json()["accessToken"] def _commercial_payload(estate: str, **overrides) -> dict: body = { "kind": "commercial", "caseType": "拍卖成交", "province": "甘肃省", "city": "兰州市", "district": "城关区", "estateName": estate, "structure": "砖混", "caseDate": "2025-06-01", "location": "张掖路 8 号", "builtYear": date.today().year, "totalFloors": "6", "floorNo": "1", "layout": "大开间", "decoration": "高档", "streetFrontage": "单面临街", "area": "120", "orientation": "南", "footTraffic": "一般", "shopType": "住宅底商", "businessFormats": ["餐饮"], "unitPrice": "80", "valueIncludes": [], "remark": "", "institutionName": "甘肃示例评估", "photoFileIds": [], } body.update(overrides) return body def _payload(estate: str, **overrides) -> dict: body = { "kind": "residential", "caseType": "买卖成交", "province": "甘肃省", "city": "兰州市", "district": "城关区", "estateName": estate, "structure": "钢混", "caseDate": "2025-08-16", "location": "张掖路 1 号", "builtYear": date.today().year, "totalFloors": "18", "floorNo": "-1", "purpose": "普通住宅", "layoutStructure": "平面", "decoration": "中档", "valueIncludes": ["装饰装修"], "area": "88.5", "layout": "两室一厅", "orientation": "南北", "unitPrice": "12000", "remark": "一层带院", "institutionName": "甘肃示例评估", "photoFileIds": [], } body.update(overrides) return body def test_residential_upload_is_private_and_admin_readable(client): if not _minio_ready(): pytest.skip("MinIO 不可达") email = f"case-{int(time.time() * 1000)}@example.com" token = _register(client, email) headers = {"Authorization": f"Bearer {token}"} staff = {"Authorization": f"Bearer {create_access_token('1', 'staff')}"} estate = f"上传案例样例-{int(time.time())}" created_ids: list[int] = [] options = client.get("/api/site/uploaded-case-options") assert options.status_code == 200, options.text assert date.today().year in options.json()["residential"]["builtYears"] assert 1979 not in options.json()["residential"]["builtYears"] assert any(item["name"] == "甘肃省" for item in options.json()["regions"]) denied = client.post("/api/site/member/uploaded-cases", json=_payload(estate)) assert denied.status_code == 401 staff_submit = client.post("/api/site/member/uploaded-cases", headers=staff, json=_payload(estate)) assert staff_submit.status_code == 403 uploaded = client.post( "/api/site/member/files", headers=headers, files={"file": ("位置图.png", _png(), "image/png")}, ) assert uploaded.status_code == 200, uploaded.text photo_id = uploaded.json()["id"] missing = client.post("/api/site/member/uploaded-cases", headers=headers, json=_payload(estate, photoFileIds=[])) assert missing.status_code == 400 assert "照片" in missing.text bad_year = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_payload(estate, builtYear=1979, photoFileIds=[photo_id]), ) assert bad_year.status_code == 400 assert "建筑年代" in bad_year.text bad_region = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_payload(estate, city="天水市", photoFileIds=[photo_id]), ) assert bad_region.status_code == 400 closed = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_payload(estate, kind="other", photoFileIds=[photo_id]), ) assert closed.status_code == 400 assert "案例种类不正确" in closed.text me = client.get("/api/site/member/me", headers=headers) member_id = me.json()["id"] from app.db import SessionLocal from app.models.content import AdminFile db = SessionLocal() try: huge = AdminFile( bucket="private", object_key="tests/huge-photo", filename="过大.png", content_type="image/png", size=5 * 1024 * 1024 + 1, created_by=member_id, ) db.add(huge) db.commit() db.refresh(huge) huge_id = huge.id finally: db.close() too_big = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_payload(estate, photoFileIds=[huge_id]), ) assert too_big.status_code == 400 assert "5MB" in too_big.text created = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_payload(estate, photoFileIds=[photo_id]), ) assert created.status_code == 200, created.text assert created.json()["message"] == "提交成功" case_id = created.json()["id"] created_ids.append(case_id) second = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_payload(estate + "-新", photoFileIds=[photo_id]), ) assert second.status_code == 200, second.text created_ids.append(second.json()["id"]) hidden = client.get("/api/site/member/uploaded-cases", headers=headers) assert hidden.status_code in (404, 405) public = client.get("/api/site/about/articles/case_library") assert estate not in public.text searched = client.get("/api/site/about/articles/case_library", params={"q": estate}) assert estate not in searched.text member_admin = client.get("/api/admin/uploaded-cases", headers=headers) assert member_admin.status_code == 403 member_detail = client.get(f"/api/admin/uploaded-cases/{case_id}", headers=headers) assert member_detail.status_code == 403 listed = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "residential"}) assert listed.status_code == 200, listed.text names = [item["estateName"] for item in listed.json()["items"]] assert names.index(estate + "-新") < names.index(estate) newer = listed.json()["items"][names.index(estate + "-新")] assert newer["memberEmail"] == email older = next(item for item in listed.json()["items"] if item["estateName"] == estate) assert "城关区" in older["address"] matched = client.get( "/api/admin/uploaded-cases", headers=staff, params={"kind": "residential", "q": estate + "-新"}, ) assert matched.status_code == 200, matched.text matched_names = [item["estateName"] for item in matched.json()["items"]] assert matched_names == [estate + "-新"] by_email = client.get( "/api/admin/uploaded-cases", headers=staff, params={"kind": "residential", "q": email}, ) assert by_email.status_code == 200, by_email.text assert {item["estateName"] for item in by_email.json()["items"]} >= {estate, estate + "-新"} wildcard = client.get( "/api/admin/uploaded-cases", headers=staff, params={"kind": "residential", "q": "%"}, ) assert wildcard.status_code == 200, wildcard.text assert estate not in [item["estateName"] for item in wildcard.json()["items"]] detail = client.get(f"/api/admin/uploaded-cases/{case_id}", headers=staff) assert detail.status_code == 200, detail.text body = detail.json() assert body["purpose"] == "普通住宅" assert body["builtYear"] == date.today().year assert body["floorNo"] == "-1" assert body["photos"][0]["filename"] == "位置图.png" assert body["memberEmail"] == email other = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "commercial"}) assert other.status_code == 200 assert other.json()["items"] == [] from sqlalchemy import select from app.models.uploaded_case import UploadedCase, UploadedCasePhoto db = SessionLocal() try: for item_id in created_ids: for photo in db.scalars(select(UploadedCasePhoto).where(UploadedCasePhoto.case_id == item_id)).all(): db.delete(photo) row = db.get(UploadedCase, item_id) if row is not None: db.delete(row) huge_row = db.get(AdminFile, huge_id) if huge_row is not None: db.delete(huge_row) db.commit() finally: db.close() def test_commercial_upload_stays_in_its_admin_tab(client): if not _minio_ready(): pytest.skip("MinIO 不可达") email = f"shop-{int(time.time() * 1000)}@example.com" token = _register(client, email) headers = {"Authorization": f"Bearer {token}"} staff = {"Authorization": f"Bearer {create_access_token('1', 'staff')}"} estate = f"商业上传样例-{int(time.time())}" created_ids: list[int] = [] options = client.get("/api/site/uploaded-case-options") assert options.status_code == 200, options.text assert "commercial" in options.json()["openKinds"] commercial = options.json()["commercial"] assert commercial["layouts"] == ["大开间", "隔档"] assert "单面临街" in commercial["streetFrontages"] assert commercial["shopTypes"][0] == "住宅底商" assert "餐饮" in commercial["businessFormats"] uploaded = client.post( "/api/site/member/files", headers=headers, files={"file": ("商铺.png", _png(), "image/png")}, ) assert uploaded.status_code == 200, uploaded.text photo_id = uploaded.json()["id"] extra = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_commercial_payload(estate, purpose="普通住宅", photoFileIds=[photo_id]), ) assert extra.status_code == 400 assert "没有的字段" in extra.text empty_formats = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_commercial_payload(estate, businessFormats=[], photoFileIds=[photo_id]), ) assert empty_formats.status_code == 400 assert "经营业态" in empty_formats.text too_many = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_commercial_payload( estate, businessFormats=["餐饮", "百货", "酒吧", "其他"], photoFileIds=[photo_id], ), ) assert too_many.status_code == 400 assert "经营业态" in too_many.text created = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_commercial_payload(estate, businessFormats=["餐饮", "百货"], photoFileIds=[photo_id]), ) assert created.status_code == 200, created.text assert created.json()["message"] == "提交成功" case_id = created.json()["id"] created_ids.append(case_id) hidden = client.get("/api/site/member/uploaded-cases", headers=headers) assert hidden.status_code in (404, 405) public = client.get("/api/site/about/articles/case_library") assert estate not in public.text shops = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "commercial"}) assert shops.status_code == 200, shops.text assert any(item["estateName"] == estate and item["memberEmail"] == email for item in shops.json()["items"]) homes = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "residential"}) assert all(item["estateName"] != estate for item in homes.json()["items"]) detail = client.get(f"/api/admin/uploaded-cases/{case_id}", headers=staff) assert detail.status_code == 200, detail.text body = detail.json() assert body["kind"] == "commercial" assert body["layout"] == "大开间" assert body["streetFrontage"] == "单面临街" assert body["footTraffic"] == "一般" assert body["shopType"] == "住宅底商" assert body["businessFormats"] == ["餐饮", "百货"] assert "purpose" not in body assert body["photos"][0]["filename"] == "商铺.png" from sqlalchemy import select from app.db import SessionLocal from app.models.uploaded_case import UploadedCase, UploadedCasePhoto db = SessionLocal() try: for item_id in created_ids: for photo in db.scalars(select(UploadedCasePhoto).where(UploadedCasePhoto.case_id == item_id)).all(): db.delete(photo) row = db.get(UploadedCase, item_id) if row is not None: db.delete(row) db.commit() finally: db.close() def _office_payload(estate: str, **overrides) -> dict: body = { "kind": "office", "caseType": "评估案例", "province": "甘肃省", "city": "兰州市", "district": "城关区", "estateName": estate, "structure": "钢混", "caseDate": "2025-04-18", "location": "东岗西路 12 号", "builtYear": date.today().year, "totalFloors": "22", "floorNo": "8", "layoutStructure": "写字楼（大开间）", "decoration": "中档", "area": "360", "orientation": "东南", "unitPrice": "45", "valueIncludes": [], "remark": "", "institutionName": "甘肃示例评估", "photoFileIds": [], } body.update(overrides) return body def test_office_upload_stays_in_its_admin_tab(client): if not _minio_ready(): pytest.skip("MinIO 不可达") email = f"office-{int(time.time() * 1000)}@example.com" token = _register(client, email) headers = {"Authorization": f"Bearer {token}"} staff = {"Authorization": f"Bearer {create_access_token('1', 'staff')}"} estate = f"办公上传样例-{int(time.time())}" created_ids: list[int] = [] options = client.get("/api/site/uploaded-case-options") assert options.status_code == 200, options.text assert "office" in options.json()["openKinds"] office = options.json()["office"] assert office["layoutStructures"] == ["办公楼（中间过道两边办公室）", "写字楼（大开间）"] assert office["totalFloors"][-1] == "60" assert office["floorNos"][0] == "-4" assert "东南" in office["orientations"] assert "layouts" not in office uploaded = client.post( "/api/site/member/files", headers=headers, files={"file": ("办公.png", _png(), "image/png")}, ) assert uploaded.status_code == 200, uploaded.text photo_id = uploaded.json()["id"] extra = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_office_payload(estate, layout="大开间", photoFileIds=[photo_id]), ) assert extra.status_code == 400 assert "没有的字段" in extra.text created = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_office_payload(estate, photoFileIds=[photo_id]), ) assert created.status_code == 200, created.text assert created.json()["message"] == "提交成功" case_id = created.json()["id"] created_ids.append(case_id) hidden = client.get("/api/site/member/uploaded-cases", headers=headers) assert hidden.status_code in (404, 405) public = client.get("/api/site/about/articles/case_library") assert estate not in public.text offices = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "office"}) assert offices.status_code == 200, offices.text assert any(item["estateName"] == estate and item["memberEmail"] == email for item in offices.json()["items"]) homes = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "residential"}) assert all(item["estateName"] != estate for item in homes.json()["items"]) shops = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "commercial"}) assert all(item["estateName"] != estate for item in shops.json()["items"]) detail = client.get(f"/api/admin/uploaded-cases/{case_id}", headers=staff) assert detail.status_code == 200, detail.text body = detail.json() assert body["kind"] == "office" assert body["layoutStructure"] == "写字楼（大开间）" assert body["orientation"] == "东南" assert body["totalFloors"] == 22 assert "layout" not in body assert "purpose" not in body assert "streetFrontage" not in body assert body["photos"][0]["filename"] == "办公.png" from sqlalchemy import select from app.db import SessionLocal from app.models.uploaded_case import UploadedCase, UploadedCasePhoto db = SessionLocal() try: for item_id in created_ids: for photo in db.scalars(select(UploadedCasePhoto).where(UploadedCasePhoto.case_id == item_id)).all(): db.delete(photo) row = db.get(UploadedCase, item_id) if row is not None: db.delete(row) db.commit() finally: db.close() def _industrial_payload(estate: str, **overrides) -> dict: body = { "kind": "industrial", "caseType": "买卖成交", "province": "甘肃省", "city": "兰州市", "district": "西固区", "estateName": estate, "structure": "钢结构", "caseDate": "2024-11-02", "location": "西固工业园 3 号", "builtYear": date.today().year, "totalFloors": "3", "layout": "开间", "heading": "南北", "decoration": "简易", "roadFrontage": "主干道", "area": "2400", "orientation": "东西", "industryType": "机械制造", "unitPrice": "18", "valueIncludes": ["装饰装修"], "remark": "", "institutionName": "甘肃示例评估", "photoFileIds": [], } body.update(overrides) return body def test_industrial_upload_stays_in_its_admin_tab(client): if not _minio_ready(): pytest.skip("MinIO 不可达") email = f"plant-{int(time.time() * 1000)}@example.com" token = _register(client, email) headers = {"Authorization": f"Bearer {token}"} staff = {"Authorization": f"Bearer {create_access_token('1', 'staff')}"} estate = f"工业上传样例-{int(time.time())}" created_ids: list[int] = [] options = client.get("/api/site/uploaded-case-options") assert options.status_code == 200, options.text assert "industrial" in options.json()["openKinds"] industrial = options.json()["industrial"] assert industrial["totalFloors"] == [str(item) for item in range(1, 9)] assert industrial["layouts"] == ["开间"] assert industrial["headings"] == ["南北", "东西"] assert industrial["orientations"] == ["南北", "东西"] assert industrial["roadFrontages"] == ["主干道", "次干道", "街巷"] assert industrial["industryTypes"][0] == "食品饮料加工" assert industrial["valueIncludes"] == ["装饰装修", "家具用具"] assert "floorNos" not in industrial uploaded = client.post( "/api/site/member/files", headers=headers, files={"file": ("厂房.png", _png(), "image/png")}, ) assert uploaded.status_code == 200, uploaded.text photo_id = uploaded.json()["id"] extra = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_industrial_payload(estate, floorNo="1", photoFileIds=[photo_id]), ) assert extra.status_code == 400 assert "没有的字段" in extra.text too_tall = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_industrial_payload(estate, totalFloors="9", photoFileIds=[photo_id]), ) assert too_tall.status_code == 400 assert "总层数" in too_tall.text created = client.post( "/api/site/member/uploaded-cases", headers=headers, json=_industrial_payload(estate, photoFileIds=[photo_id]), ) assert created.status_code == 200, created.text assert created.json()["message"] == "提交成功" case_id = created.json()["id"] created_ids.append(case_id) hidden = client.get("/api/site/member/uploaded-cases", headers=headers) assert hidden.status_code in (404, 405) public = client.get("/api/site/about/articles/case_library") assert estate not in public.text plants = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": "industrial"}) assert plants.status_code == 200, plants.text assert any(item["estateName"] == estate and item["memberEmail"] == email for item in plants.json()["items"]) for other in ("residential", "commercial", "office"): listed = client.get("/api/admin/uploaded-cases", headers=staff, params={"kind": other}) assert all(item["estateName"] != estate for item in listed.json()["items"]) detail = client.get(f"/api/admin/uploaded-cases/{case_id}", headers=staff) assert detail.status_code == 200, detail.text body = detail.json() assert body["kind"] == "industrial" assert body["layout"] == "开间" assert body["heading"] == "南北" assert body["roadFrontage"] == "主干道" assert body["orientation"] == "东西" assert body["industryType"] == "机械制造" assert body["totalFloors"] == 3 assert body["valueIncludes"] == ["装饰装修"] assert "floorNo" not in body assert "purpose" not in body assert "streetFrontage" not in body assert body["photos"][0]["filename"] == "厂房.png" from sqlalchemy import select from app.db import SessionLocal from app.models.uploaded_case import UploadedCase, UploadedCasePhoto db = SessionLocal() try: for item_id in created_ids: for photo in db.scalars(select(UploadedCasePhoto).where(UploadedCasePhoto.case_id == item_id)).all(): db.delete(photo) row = db.get(UploadedCase, item_id) if row is not None: db.delete(row) db.commit() finally: db.close()