674 lines
24 KiB
Python
674 lines
24 KiB
Python
"""上传案例：会员提交住宅，管理端只读，不进案例库文章。"""
|
|
|
|
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()
|