激光材料数据平台 1.3.0
公开数据看板、四位 Key 录入面板、管理后台与三级权限、动态字段配置、 PostgreSQL/SQLite 双支持、数据库备份，以及本版新增的成品图上传与预览。 图片相关： - 录入表单支持相册选图与手机端直接拍照，每条记录最多 6 张 - 浏览器内压缩到最长边 1600 并剥离 EXIF，入库统一为 JPG/PNG - 二进制直接入库，现有备份自动覆盖图片，部署无需新增卷 - 详情页缩略图宫格与全屏 lightbox，公开看板同样可见 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
@@ -0,0 +1,72 @@
|
||||
"""接口测试用的独立环境：必须在导入 app 之前写好环境变量，配置是冻结的 dataclass。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import struct
|
||||
import tempfile
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("APP_SECRET", "test-secret-0123456789abcdefghijkl")
|
||||
os.environ.setdefault("ADMIN_USERNAME", "admin")
|
||||
os.environ.setdefault("ADMIN_PASSWORD", "TestAdmin12345")
|
||||
os.environ["DATABASE_URL"] = f"sqlite:///{Path(tempfile.mkdtemp(prefix='laser-test-')) / 'test.db'}"
|
||||
|
||||
|
||||
def png_bytes(width: int = 24, height: int = 16) -> bytes:
|
||||
"""不依赖 Pillow 手工拼一张真实可解码的 PNG。"""
|
||||
def chunk(tag: bytes, payload: bytes) -> bytes:
|
||||
return struct.pack(">I", len(payload)) + tag + payload + struct.pack(">I", zlib.crc32(tag + payload))
|
||||
raw = b"".join(b"\x00" + b"\x20\x60\x90" * width for _ in range(height))
|
||||
header = chunk(b"IHDR", struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0))
|
||||
return b"\x89PNG\r\n\x1a\n" + header + chunk(b"IDAT", zlib.compress(raw)) + chunk(b"IEND", b"")
|
||||
|
||||
|
||||
def jpeg_bytes(width: int = 32, height: int = 20) -> bytes:
|
||||
"""只含 SOI/APP0/SOF0/EOI 的最小 JPEG，足以走完服务端的文件头校验。"""
|
||||
app0 = b"\xff\xe0" + struct.pack(">H", 16) + b"JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00"
|
||||
sof0 = b"\xff\xc0" + struct.pack(">HBHHB", 11, 8, height, width, 1) + b"\x01\x11\x00"
|
||||
return b"\xff\xd8" + app0 + sof0 + b"\xff\xd9"
|
||||
|
||||
|
||||
def record_payload(experiment_id: str = "LAS-TEST-001", **overrides) -> dict:
|
||||
payload = {
|
||||
"experiment_id": experiment_id, "test_date": "2026-07-01", "material_category": "木材", "material_name": "椴木板",
|
||||
"material_thickness": 3, "actual_output_power": 8.5, "display_current": 2, "scanning_speed": 100,
|
||||
"pulse_frequency": 20, "pulse_width": 5, "defocus_amount": 0, "scan_line_spacing": 0.1,
|
||||
"filling_method": "双向填充", "processing_size": "20×20", "is_cut_through": True,
|
||||
"carbonized_edge_width": 0.2, "etching_depth": 30, "is_fire_smolder": False,
|
||||
"pattern_clarity_score": 9, "presentation_balance_score": 8,
|
||||
"finished_image_filename": "LAS-TEST-001.jpg", "remarks": "测试记录",
|
||||
}
|
||||
payload.update(overrides)
|
||||
return payload
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def client():
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def admin(client):
|
||||
"""返回带管理员会话的客户端和对应 CSRF 令牌。"""
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
from app.config import settings
|
||||
from app.main import SESSION_COOKIE
|
||||
|
||||
client.cookies.clear()
|
||||
response = client.post("/api/auth/login", data={"username": settings.admin_username, "password": settings.admin_password})
|
||||
assert response.status_code == 200, response.text
|
||||
token = client.cookies.get(SESSION_COOKIE)
|
||||
csrf = hmac.new(settings.secret.encode(), token.encode(), hashlib.sha256).hexdigest()
|
||||
return client, {"X-CSRF-Token": csrf}
|
||||
@@ -0,0 +1,11 @@
|
||||
from app.backup_service import postgres_cli_url
|
||||
|
||||
|
||||
def test_sqlalchemy_postgres_url_is_accepted_by_cli():
|
||||
value = "postgresql+psycopg://user:pass@db:5432/laser_data"
|
||||
assert postgres_cli_url(value) == "postgresql://user:pass@db:5432/laser_data"
|
||||
|
||||
|
||||
def test_standard_postgres_url_is_unchanged():
|
||||
value = "postgresql://user:pass@db:5432/laser_data"
|
||||
assert postgres_cli_url(value) == value
|
||||
@@ -0,0 +1,43 @@
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.dynamic_fields import render_custom_value, validate_raw_value
|
||||
from app.models import CustomFieldDefinition, CustomFieldValue
|
||||
|
||||
|
||||
def definition(input_type="text", required=False, options=None):
|
||||
return CustomFieldDefinition(
|
||||
field_key="custom_test", label="测试字段", input_type=input_type,
|
||||
is_required=required, is_active=True, is_public=True,
|
||||
options_json=json.dumps(options or []), sort_order=0, created_by=1,
|
||||
)
|
||||
|
||||
|
||||
def test_required_dynamic_field_rejects_empty_value():
|
||||
with pytest.raises(HTTPException) as error:
|
||||
validate_raw_value(definition(required=True), None)
|
||||
assert error.value.status_code == 422
|
||||
|
||||
|
||||
def test_select_rejects_value_outside_current_options():
|
||||
with pytest.raises(HTTPException):
|
||||
validate_raw_value(definition("select", options=["A", "B"]), "历史选项")
|
||||
|
||||
|
||||
def test_type_change_preserves_incompatible_raw_value_for_display():
|
||||
item = CustomFieldValue(raw_value="High")
|
||||
item.definition = definition("number")
|
||||
rendered = render_custom_value(item)
|
||||
assert rendered["raw_value"] == "High"
|
||||
assert rendered["display_value"] == "High"
|
||||
assert rendered["compatible"] is False
|
||||
|
||||
|
||||
def test_compatible_number_is_processed_for_display():
|
||||
item = CustomFieldValue(raw_value="12.5")
|
||||
item.definition = definition("number")
|
||||
rendered = render_custom_value(item)
|
||||
assert rendered["value"] == 12.5
|
||||
assert rendered["compatible"] is True
|
||||
@@ -0,0 +1,113 @@
|
||||
"""四位 Key 录入面板的图片链路：这是手机端最主要的入口，必须整条走通。"""
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
import pytest
|
||||
|
||||
from conftest import png_bytes, record_payload
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def entry(client, admin):
|
||||
"""用管理员账号配好四位 Key，再换成录入会话。"""
|
||||
from app.config import settings
|
||||
from app.main import ENTRY_COOKIE
|
||||
|
||||
admin_client, admin_headers = admin
|
||||
response = admin_client.put("/api/me/entry-key", headers=admin_headers, json={"key": "1379", "password": settings.admin_password})
|
||||
assert response.status_code == 200, response.text
|
||||
client.cookies.clear()
|
||||
assert client.post("/api/entry/auth", json={"key": "1379"}).status_code == 200
|
||||
token = client.cookies.get(ENTRY_COOKIE)
|
||||
csrf = hmac.new(settings.secret.encode(), f"entry:{token}".encode(), hashlib.sha256).hexdigest()
|
||||
return client, {"X-CSRF-Token": csrf}
|
||||
|
||||
|
||||
def entry_upload(client, headers, **kwargs):
|
||||
files = {"file": ("field.png", png_bytes(48, 36), "image/png"), "thumbnail": ("t.png", png_bytes(12, 9), "image/png")}
|
||||
return client.post("/api/entry/uploads/images", headers=headers, files=files, **kwargs)
|
||||
|
||||
|
||||
def test_entry_upload_requires_the_entry_session(client):
|
||||
client.cookies.clear()
|
||||
assert client.post("/api/entry/uploads/images", files={"file": ("a.png", png_bytes(), "image/png")}).status_code == 401
|
||||
|
||||
|
||||
def test_entry_upload_requires_the_entry_csrf_token(entry):
|
||||
client, _ = entry
|
||||
assert client.post("/api/entry/uploads/images", files={"file": ("a.png", png_bytes(), "image/png")}).status_code == 403
|
||||
|
||||
|
||||
def test_admin_csrf_token_is_not_accepted_on_the_entry_endpoint(entry, admin):
|
||||
"""两套会话的 CSRF 派生方式不同，管理端令牌不能拿来打录入接口。"""
|
||||
client, _ = entry
|
||||
_, admin_headers = admin
|
||||
assert client.post("/api/entry/uploads/images", headers=admin_headers, files={"file": ("a.png", png_bytes(), "image/png")}).status_code == 403
|
||||
|
||||
|
||||
def test_entry_record_is_saved_with_its_images(entry):
|
||||
from app.config import settings
|
||||
|
||||
client, headers = entry
|
||||
token = entry_upload(client, headers).json()["data"]["token"]
|
||||
payload = record_payload("LAS-ENTRY-001", image_tokens=[token], confirm_username=settings.admin_username)
|
||||
created = client.post("/api/entry/records", headers=headers, json=payload)
|
||||
assert created.status_code == 200, created.text
|
||||
|
||||
images = client.get(f"/api/public/records/{created.json()['data']['id']}").json()["data"]["images"]
|
||||
assert len(images) == 1 and images[0]["width"] == 48
|
||||
|
||||
|
||||
def test_entry_record_rejects_a_mismatched_username(entry):
|
||||
client, headers = entry
|
||||
token = entry_upload(client, headers).json()["data"]["token"]
|
||||
payload = record_payload("LAS-ENTRY-002", image_tokens=[token], confirm_username="someone-else")
|
||||
assert client.post("/api/entry/records", headers=headers, json=payload).status_code == 403
|
||||
|
||||
|
||||
def test_entry_pending_image_can_be_discarded(entry):
|
||||
client, headers = entry
|
||||
token = entry_upload(client, headers).json()["data"]["token"]
|
||||
assert client.delete(f"/api/entry/uploads/images/{token}", headers=headers).status_code == 200
|
||||
assert client.delete(f"/api/entry/uploads/images/{token}", headers=headers).status_code == 404
|
||||
|
||||
|
||||
def test_pending_image_of_another_account_cannot_be_claimed(client, admin):
|
||||
"""令牌必须同时校验归属，否则拿到别人的令牌就能把图挂到自己的记录上。"""
|
||||
from app.config import settings
|
||||
from app.main import SESSION_COOKIE
|
||||
|
||||
admin_client, admin_headers = admin
|
||||
created = admin_client.post("/api/users", headers=admin_headers, json={"username": "uploader9", "display_name": "录入九", "password": "UploaderPass123", "role": "uploader"})
|
||||
assert created.status_code in (200, 409)
|
||||
|
||||
# 换成另一个账号上传，拿到属于它的令牌
|
||||
client.cookies.clear()
|
||||
client.post("/api/auth/login", data={"username": "uploader9", "password": "UploaderPass123"})
|
||||
other_csrf = hmac.new(settings.secret.encode(), client.cookies.get(SESSION_COOKIE).encode(), hashlib.sha256).hexdigest()
|
||||
stolen = client.post("/api/uploads/images", headers={"X-CSRF-Token": other_csrf}, files={"file": ("mine.png", png_bytes(20, 20), "image/png")}).json()["data"]["token"]
|
||||
|
||||
# 换回管理员，用别人的令牌建记录必须失败
|
||||
client.cookies.clear()
|
||||
client.post("/api/auth/login", data={"username": settings.admin_username, "password": settings.admin_password})
|
||||
admin_csrf = hmac.new(settings.secret.encode(), client.cookies.get(SESSION_COOKIE).encode(), hashlib.sha256).hexdigest()
|
||||
response = client.post("/api/records", headers={"X-CSRF-Token": admin_csrf}, json=record_payload("LAS-STEAL-001", image_tokens=[stolen]))
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_thumbnail_variant_reports_its_own_media_type(entry):
|
||||
from app.config import settings
|
||||
|
||||
client, headers = entry
|
||||
body = client.post("/api/entry/uploads/images", headers=headers, files={
|
||||
"file": ("photo.jpg", open("/System/Library/CoreServices/DefaultBackground.jpg", "rb").read() if False else png_bytes(64, 64), "image/png"),
|
||||
"thumbnail": ("t.png", png_bytes(16, 16), "image/png"),
|
||||
}).json()
|
||||
token = body["data"]["token"]
|
||||
payload = record_payload("LAS-ENTRY-004", image_tokens=[token], confirm_username=settings.admin_username)
|
||||
record_id = client.post("/api/entry/records", headers=headers, json=payload).json()["data"]["id"]
|
||||
image = client.get(f"/api/public/records/{record_id}").json()["data"]["images"][0]
|
||||
assert "variant=thumb" in image["thumb_url"]
|
||||
response = client.get(image["thumb_url"])
|
||||
assert response.status_code == 200 and response.headers["content-type"] == "image/png"
|
||||
assert response.headers["etag"].endswith('-thumb"')
|
||||
@@ -0,0 +1,19 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.main import clear_entry_key_attempts, enforce_entry_key_rate_limit, record_failed_entry_key
|
||||
|
||||
|
||||
def test_entry_key_is_rate_limited_after_five_failures():
|
||||
address = f"test-{uuid.uuid4()}"
|
||||
try:
|
||||
for _ in range(5):
|
||||
enforce_entry_key_rate_limit(address)
|
||||
record_failed_entry_key(address)
|
||||
with pytest.raises(HTTPException) as error:
|
||||
enforce_entry_key_rate_limit(address)
|
||||
assert error.value.status_code == 429
|
||||
finally:
|
||||
clear_entry_key_attempts(address)
|
||||
@@ -0,0 +1,324 @@
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.images import MAX_IMAGES_PER_RECORD, inspect_image, sniff
|
||||
from conftest import jpeg_bytes, png_bytes, record_payload
|
||||
|
||||
|
||||
def upload(client, headers, data=None, name="shot.png", content_type="image/png", thumbnail=True):
|
||||
files = {"file": (name, data if data is not None else png_bytes(), content_type)}
|
||||
if thumbnail:
|
||||
files["thumbnail"] = ("thumb.png", png_bytes(8, 6), "image/png")
|
||||
return client.post("/api/uploads/images", headers=headers, files=files)
|
||||
|
||||
|
||||
def test_headers_identify_supported_formats():
|
||||
assert sniff(png_bytes(40, 25)) == ("image/png", 40, 25)
|
||||
assert sniff(jpeg_bytes(640, 480)) == ("image/jpeg", 640, 480)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("blob", [
|
||||
b'<svg xmlns="http://www.w3.org/2000/svg"><script>alert(1)</script></svg>',
|
||||
b"<!doctype html><h1>not an image</h1>",
|
||||
b"GIF89a\x01\x00\x01\x00\x00\x00\x00", # GIF 不在白名单内
|
||||
png_bytes()[:12], # 截断文件
|
||||
b"\xff\xd8\xff" + b"\x00" * 40, # 伪造 JPEG 头但没有 SOF
|
||||
])
|
||||
def test_non_image_payloads_are_rejected(blob):
|
||||
with pytest.raises(HTTPException) as error:
|
||||
inspect_image(blob)
|
||||
assert error.value.status_code in (400, 413, 415)
|
||||
|
||||
|
||||
def test_oversized_payload_is_rejected():
|
||||
with pytest.raises(HTTPException) as error:
|
||||
inspect_image(png_bytes(), limit=10)
|
||||
assert error.value.status_code == 413
|
||||
|
||||
|
||||
def test_absurd_dimensions_are_rejected():
|
||||
with pytest.raises(HTTPException) as error:
|
||||
inspect_image(jpeg_bytes(30000, 30000))
|
||||
assert error.value.status_code == 413
|
||||
|
||||
|
||||
def test_upload_requires_authentication(client):
|
||||
client.cookies.clear()
|
||||
assert client.post("/api/uploads/images", files={"file": ("a.png", png_bytes(), "image/png")}).status_code == 401
|
||||
|
||||
|
||||
def test_upload_requires_csrf_token(admin):
|
||||
client, _ = admin
|
||||
assert client.post("/api/uploads/images", files={"file": ("a.png", png_bytes(), "image/png")}).status_code == 403
|
||||
|
||||
|
||||
def test_upload_returns_metadata_and_token(admin):
|
||||
client, headers = admin
|
||||
body = upload(client, headers, png_bytes(64, 48)).json()
|
||||
assert body["ok"] and body["data"]["width"] == 64 and body["data"]["height"] == 48
|
||||
assert body["data"]["content_type"] == "image/png" and body["data"]["token"]
|
||||
|
||||
|
||||
def test_disguised_html_upload_is_rejected(admin):
|
||||
client, headers = admin
|
||||
response = upload(client, headers, b"<script>alert(1)</script>", name="evil.png")
|
||||
assert response.status_code == 415
|
||||
|
||||
|
||||
def test_pending_image_is_private_until_attached(admin):
|
||||
client, headers = admin
|
||||
image_id = upload(client, headers).json()["data"]["id"]
|
||||
assert client.get(f"/api/images/{image_id}").status_code == 200
|
||||
client.cookies.clear()
|
||||
assert client.get(f"/api/images/{image_id}").status_code == 404
|
||||
|
||||
|
||||
def test_attached_image_is_public_and_cacheable(admin):
|
||||
client, headers = admin
|
||||
token = upload(client, headers, png_bytes(50, 40)).json()["data"]["token"]
|
||||
created = client.post("/api/records", headers=headers, json=record_payload("LAS-PUB-001", image_tokens=[token]))
|
||||
assert created.status_code == 200, created.text
|
||||
record_id = created.json()["data"]["id"]
|
||||
images = client.get(f"/api/public/records/{record_id}").json()["data"]["images"]
|
||||
assert len(images) == 1
|
||||
|
||||
client.cookies.clear()
|
||||
response = client.get(images[0]["url"])
|
||||
assert response.status_code == 200
|
||||
assert response.headers["content-type"] == "image/png"
|
||||
assert response.headers["x-content-type-options"] == "nosniff"
|
||||
assert "immutable" in response.headers["cache-control"]
|
||||
assert client.get(response.url, headers={"If-None-Match": response.headers["etag"]}).status_code == 304
|
||||
|
||||
|
||||
def test_record_detail_never_leaks_binary_payload(admin):
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-BLOB-001", image_tokens=[token])).json()["data"]["id"]
|
||||
data = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]
|
||||
assert "data" not in data and "thumbnail" not in data
|
||||
assert data["images"][0]["url"].startswith("/api/images/")
|
||||
|
||||
|
||||
def test_more_than_six_images_are_refused(admin):
|
||||
client, headers = admin
|
||||
tokens = [upload(client, headers).json()["data"]["token"] for _ in range(MAX_IMAGES_PER_RECORD + 1)]
|
||||
response = client.post("/api/records", headers=headers, json=record_payload("LAS-MANY-001", image_tokens=tokens))
|
||||
assert response.status_code == 422
|
||||
assert "6" in response.json()["message"]
|
||||
|
||||
|
||||
def test_unknown_or_reused_token_is_refused(admin):
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
assert client.post("/api/records", headers=headers, json=record_payload("LAS-TOK-001", image_tokens=[token])).status_code == 200
|
||||
reused = client.post("/api/records", headers=headers, json=record_payload("LAS-TOK-002", image_tokens=[token]))
|
||||
assert reused.status_code == 404
|
||||
assert client.post("/api/records", headers=headers, json=record_payload("LAS-TOK-003", image_tokens=["nope"])).status_code == 404
|
||||
|
||||
|
||||
def test_editing_drops_images_left_out_of_the_keep_list(admin):
|
||||
client, headers = admin
|
||||
tokens = [upload(client, headers).json()["data"]["token"] for _ in range(2)]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-EDIT-001", image_tokens=tokens)).json()["data"]["id"]
|
||||
images = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"]
|
||||
assert len(images) == 2
|
||||
|
||||
kept = images[1]["id"]
|
||||
updated = client.put(f"/api/records/{record_id}", headers=headers, json=record_payload("LAS-EDIT-001", image_ids=[kept]))
|
||||
assert updated.status_code == 200, updated.text
|
||||
remaining = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"]
|
||||
assert [item["id"] for item in remaining] == [kept]
|
||||
assert client.get(f"/api/images/{images[0]['id']}").status_code == 404
|
||||
|
||||
|
||||
def test_images_of_another_record_cannot_be_claimed(admin):
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
first = client.post("/api/records", headers=headers, json=record_payload("LAS-OWN-001", image_tokens=[token])).json()["data"]["id"]
|
||||
image_id = client.get(f"/api/records/{first}", headers=headers).json()["data"]["images"][0]["id"]
|
||||
second = client.post("/api/records", headers=headers, json=record_payload("LAS-OWN-002")).json()["data"]["id"]
|
||||
response = client.put(f"/api/records/{second}", headers=headers, json=record_payload("LAS-OWN-002", image_ids=[image_id]))
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_deleting_a_record_removes_its_images(admin):
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-DEL-001", image_tokens=[token])).json()["data"]["id"]
|
||||
image_id = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"][0]["id"]
|
||||
assert client.delete(f"/api/records/{record_id}", headers=headers).status_code == 200
|
||||
assert client.get(f"/api/images/{image_id}").status_code == 404
|
||||
|
||||
|
||||
def test_expired_pending_images_are_purged(admin):
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from app.db import SessionLocal
|
||||
from app.models import RecordImage
|
||||
|
||||
client, headers = admin
|
||||
stale_token = upload(client, headers).json()["data"]["token"]
|
||||
with SessionLocal() as db:
|
||||
stale = db.query(RecordImage).filter_by(token=stale_token).one()
|
||||
stale.created_at = datetime.now().astimezone() - timedelta(hours=5)
|
||||
db.commit()
|
||||
assert upload(client, headers).status_code == 200 # 任何一次新上传都会顺手清理孤儿
|
||||
with SessionLocal() as db: # 按 token 判断：SQLite 会复用被删除行的 rowid，按 id 判断会误判
|
||||
assert db.query(RecordImage).filter_by(token=stale_token).count() == 0
|
||||
|
||||
|
||||
def test_viewer_role_cannot_upload(admin):
|
||||
client, headers = admin
|
||||
created = client.post("/api/users", headers=headers, json={"username": "readonly1", "display_name": "只读", "password": "ViewerPass123", "role": "viewer"})
|
||||
assert created.status_code in (200, 409)
|
||||
client.cookies.clear()
|
||||
client.post("/api/auth/login", data={"username": "readonly1", "password": "ViewerPass123"})
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
from app.config import settings
|
||||
from app.main import SESSION_COOKIE
|
||||
|
||||
csrf = hmac.new(settings.secret.encode(), client.cookies.get(SESSION_COOKIE).encode(), hashlib.sha256).hexdigest()
|
||||
assert upload(client, {"X-CSRF-Token": csrf}).status_code == 403
|
||||
|
||||
|
||||
|
||||
def test_filename_injection_characters_are_stripped(admin):
|
||||
client, headers = admin
|
||||
files = {"file": ('x" onload="alert(1)".png', png_bytes(), "image/png")}
|
||||
body = client.post("/api/uploads/images", headers=headers, files=files).json()
|
||||
stored = body["data"]["original_name"]
|
||||
assert '"' not in stored and "<" not in stored and ">" not in stored and "'" not in stored
|
||||
|
||||
|
||||
def test_payload_appended_after_image_end_marker_is_discarded(admin):
|
||||
client, headers = admin
|
||||
smuggled = png_bytes() + b"<html><script>alert(1)</script></html>" * 20
|
||||
body = client.post("/api/uploads/images", headers=headers, files={"file": ("ok.png", smuggled, "image/png")}).json()
|
||||
assert body["data"]["size_bytes"] == len(png_bytes())
|
||||
|
||||
token = body["data"]["token"]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-TRIM-001", image_tokens=[token])).json()["data"]["id"]
|
||||
url = client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"][0]["url"]
|
||||
assert b"<script>" not in client.get(url).content
|
||||
|
||||
|
||||
def test_webp_is_no_longer_accepted(admin):
|
||||
"""WebP 的 RIFF 头几乎不校验内容，16 字节垃圾就能冒充一张图，索性不收。"""
|
||||
import struct
|
||||
|
||||
client, headers = admin
|
||||
fake = b"RIFF" + struct.pack("<I", 0) + b"WEBPVP8X" + b"\x00" * 20
|
||||
assert client.post("/api/uploads/images", headers=headers, files={"file": ("a.webp", fake, "image/webp")}).status_code == 415
|
||||
|
||||
|
||||
def test_update_without_image_ids_keeps_existing_images(admin):
|
||||
"""省略 image_ids 必须是"不动图片"，否则任何不了解该字段的旧页面或脚本都会静默删光图。"""
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-KEEP-001", image_tokens=[token])).json()["data"]["id"]
|
||||
payload = record_payload("LAS-KEEP-001")
|
||||
assert "image_ids" not in payload
|
||||
assert client.put(f"/api/records/{record_id}", headers=headers, json=payload).status_code == 200
|
||||
assert len(client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"]) == 1
|
||||
|
||||
|
||||
def test_explicit_empty_image_ids_clears_images(admin):
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-CLEAR-001", image_tokens=[token])).json()["data"]["id"]
|
||||
client.put(f"/api/records/{record_id}", headers=headers, json=record_payload("LAS-CLEAR-001", image_ids=[]))
|
||||
assert client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"] == []
|
||||
|
||||
|
||||
def test_upload_without_content_length_is_refused(admin):
|
||||
"""没有 Content-Length 就无法预判体积，而 multipart 在鉴权之前就已落盘，必须直接拒绝。"""
|
||||
client, headers = admin
|
||||
response = client.post("/api/uploads/images", headers={**headers, "Content-Type": "multipart/form-data; boundary=B", "Transfer-Encoding": "chunked"}, content=iter([b"--B--\r\n"]))
|
||||
assert response.status_code == 411
|
||||
|
||||
|
||||
def test_detail_endpoints_do_not_load_image_binaries(admin):
|
||||
"""图片二进制是 deferred 列：公开详情是匿名接口，一次请求把几 MB 图片读进内存会被用来打内存。"""
|
||||
from sqlalchemy import event
|
||||
|
||||
from app.db import engine
|
||||
|
||||
client, headers = admin
|
||||
token = upload(client, headers, png_bytes(120, 90)).json()["data"]["token"]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-DEFER-001", image_tokens=[token])).json()["data"]["id"]
|
||||
|
||||
statements = []
|
||||
listener = lambda conn, cursor, statement, *rest: statements.append(statement)
|
||||
event.listen(engine, "before_cursor_execute", listener)
|
||||
try:
|
||||
client.get(f"/api/public/records/{record_id}")
|
||||
finally:
|
||||
event.remove(engine, "before_cursor_execute", listener)
|
||||
selected = " ".join(statements).lower()
|
||||
assert "record_images.data" not in selected and "record_images.thumbnail\n" not in selected
|
||||
|
||||
|
||||
def test_public_detail_hides_original_filename(admin):
|
||||
"""原始文件名可能带姓名、车间、项目代号，公开看板不返回。"""
|
||||
client, headers = admin
|
||||
token = upload(client, headers, name="张三-三车间-复检.png").json()["data"]["token"]
|
||||
record_id = client.post("/api/records", headers=headers, json=record_payload("LAS-PRIV-001", image_tokens=[token])).json()["data"]["id"]
|
||||
public_image = client.get(f"/api/public/records/{record_id}").json()["data"]["images"][0]
|
||||
assert "original_name" not in public_image and "size_bytes" not in public_image
|
||||
assert set(public_image) == {"id", "width", "height", "url", "thumb_url"}
|
||||
assert "original_name" in client.get(f"/api/records/{record_id}", headers=headers).json()["data"]["images"][0]
|
||||
|
||||
|
||||
def test_broken_thumbnail_does_not_fail_the_upload(admin):
|
||||
"""缩略图只是加速手段，坏了就回落到原图，不能把整次上传拖垮。"""
|
||||
client, headers = admin
|
||||
files = {"file": ("ok.png", png_bytes(), "image/png"), "thumbnail": ("bad.png", b"not an image at all", "image/png")}
|
||||
body = client.post("/api/uploads/images", headers=headers, files=files)
|
||||
assert body.status_code == 200
|
||||
assert body.json()["data"]["thumb_url"] == body.json()["data"]["url"]
|
||||
|
||||
|
||||
def test_pending_image_can_be_discarded_to_free_its_slot(admin):
|
||||
from app.db import SessionLocal
|
||||
from app.models import RecordImage
|
||||
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
assert client.delete(f"/api/uploads/images/{token}", headers=headers).status_code == 200
|
||||
with SessionLocal() as db:
|
||||
assert db.query(RecordImage).filter_by(token=token).count() == 0
|
||||
assert client.delete(f"/api/uploads/images/{token}", headers=headers).status_code == 404
|
||||
|
||||
|
||||
def test_attached_image_cannot_be_discarded_through_the_upload_endpoint(admin):
|
||||
client, headers = admin
|
||||
token = upload(client, headers).json()["data"]["token"]
|
||||
client.post("/api/records", headers=headers, json=record_payload("LAS-DISC-001", image_tokens=[token]))
|
||||
assert client.delete(f"/api/uploads/images/{token}", headers=headers).status_code == 404
|
||||
|
||||
|
||||
def test_backup_size_column_holds_more_than_two_gigabytes():
|
||||
"""图片入库后备份很容易超过 2 GiB，PostgreSQL 的 integer 到那里就溢出了。"""
|
||||
from sqlalchemy import BigInteger
|
||||
|
||||
from app.models import BackupRecord
|
||||
|
||||
assert isinstance(BackupRecord.__table__.c.size_bytes.type, BigInteger)
|
||||
|
||||
|
||||
def test_unauthenticated_upload_never_buffers_the_request_body(client):
|
||||
"""鉴权必须发生在 multipart 缓冲之前，否则未登录请求也能驱动服务器往磁盘写临时文件。"""
|
||||
consumed = []
|
||||
|
||||
class Probe:
|
||||
def __iter__(self):
|
||||
consumed.append(True)
|
||||
yield b'--B\r\nContent-Disposition: form-data; name="file"; filename="a.png"\r\nContent-Type: image/png\r\n\r\n' + b"x" * 200_000 + b"\r\n--B--\r\n"
|
||||
|
||||
client.cookies.clear()
|
||||
response = client.post("/api/uploads/images", headers={"Content-Type": "multipart/form-data; boundary=B", "Content-Length": "200200"}, content=Probe())
|
||||
assert response.status_code == 401
|
||||
assert not consumed, "未登录请求的 body 不应被读取"
|
||||
@@ -0,0 +1,45 @@
|
||||
from datetime import date
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.schemas import EntryRecordSubmission, MaterialRecordInput
|
||||
|
||||
|
||||
def valid_payload():
|
||||
return {
|
||||
"experiment_id": "LAS-2026-001", "test_date": date(2026, 7, 17), "material_category": "木材", "material_name": "椴木板",
|
||||
"material_thickness": 3, "actual_output_power": 8.5, "display_current": 2, "scanning_speed": 100,
|
||||
"pulse_frequency": 20, "pulse_width": 5, "defocus_amount": 0, "scan_line_spacing": .1,
|
||||
"filling_method": "双向", "processing_size": "20×20", "is_cut_through": True,
|
||||
"carbonized_edge_width": .2, "etching_depth": 30, "is_fire_smolder": False,
|
||||
"pattern_clarity_score": 9, "presentation_balance_score": 8, "finished_image_filename": "LAS-2026-001.jpg", "remarks": "正常",
|
||||
}
|
||||
|
||||
|
||||
def test_all_optional_physical_properties_may_be_empty():
|
||||
record = MaterialRecordInput(**valid_payload())
|
||||
assert record.apparent_density is None
|
||||
assert record.hardness is None
|
||||
assert len(record.model_fields) - 3 == 32 # 32 fixed fields plus the dynamic-field map and the two image lists
|
||||
|
||||
|
||||
def test_scores_are_limited_to_ten():
|
||||
payload = valid_payload()
|
||||
payload["pattern_clarity_score"] = 10.1
|
||||
with pytest.raises(ValidationError):
|
||||
MaterialRecordInput(**payload)
|
||||
|
||||
|
||||
def test_percentage_is_limited_to_hundred():
|
||||
payload = valid_payload()
|
||||
payload["moisture_content"] = 101
|
||||
with pytest.raises(ValidationError):
|
||||
MaterialRecordInput(**payload)
|
||||
|
||||
|
||||
def test_entry_submission_requires_confirmation_username():
|
||||
payload = valid_payload()
|
||||
with pytest.raises(ValidationError):
|
||||
EntryRecordSubmission(**payload)
|
||||
assert EntryRecordSubmission(**payload, confirm_username="operator").confirm_username == "operator"
|
||||
@@ -0,0 +1,21 @@
|
||||
from app.security import entry_key_hash, hash_password, validate_password, verify_password
|
||||
|
||||
|
||||
def test_password_hash_round_trip():
|
||||
encoded = hash_password("StrongPass123")
|
||||
assert "StrongPass123" not in encoded
|
||||
assert verify_password("StrongPass123", encoded)
|
||||
assert not verify_password("wrong", encoded)
|
||||
|
||||
|
||||
def test_password_policy():
|
||||
assert validate_password("short")
|
||||
assert validate_password("onlyletterslong")
|
||||
assert validate_password("StrongPass123") is None
|
||||
|
||||
|
||||
def test_entry_key_hash_is_deterministic_without_storing_plaintext():
|
||||
first = entry_key_hash("1234")
|
||||
assert first == entry_key_hash("1234")
|
||||
assert first != entry_key_hash("4321")
|
||||
assert "1234" not in first
|
||||
Reference in New Issue
Block a user