350 lines
14 KiB
Python
350 lines
14 KiB
Python
import json
|
|
import hashlib
|
|
import sqlite3
|
|
import sys
|
|
import shutil
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
|
if str(BACKEND_DIR) not in sys.path:
|
|
sys.path.insert(0, str(BACKEND_DIR))
|
|
|
|
from service.product_archive import ( # noqa: E402
|
|
WordcloudSource,
|
|
copy_word_locations_snapshot,
|
|
find_visible_wordcloud_sources,
|
|
validate_word_locations_db,
|
|
)
|
|
from service.product_archive_store import ProductArchiveStore # noqa: E402
|
|
from service import app as service_app # noqa: E402
|
|
|
|
|
|
def _png_bytes() -> bytes:
|
|
image = BytesIO()
|
|
Image.new("RGB", (1, 1), "white").save(image, format="PNG")
|
|
return image.getvalue()
|
|
|
|
|
|
PNG_BYTES = _png_bytes()
|
|
|
|
|
|
def orders_auth_header() -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {service_app._orders_token()}"}
|
|
|
|
|
|
def create_product(client):
|
|
response = client.post(
|
|
"/api/products", json={"name": "笔盒", "source": "manual"}, headers=orders_auth_header()
|
|
)
|
|
assert response.status_code == 201
|
|
return response.json()
|
|
|
|
|
|
@pytest.fixture
|
|
def product_archive_client(tmp_path, monkeypatch):
|
|
archive_root = tmp_path / "product_archives"
|
|
store = ProductArchiveStore(archive_root / "metadata")
|
|
monkeypatch.setattr(service_app, "PRODUCT_ARCHIVES_DIR", archive_root, raising=False)
|
|
monkeypatch.setattr(service_app, "product_archive_store", store, raising=False)
|
|
client = TestClient(service_app.app)
|
|
client.archive_root = archive_root
|
|
client.product_store = store
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def prepared_wordcloud_job(tmp_path, monkeypatch):
|
|
workspace = tmp_path / "source-workspace"
|
|
db_path = workspace / "output" / "word_locations.sqlite"
|
|
db_path.parent.mkdir(parents=True)
|
|
with sqlite3.connect(db_path) as connection:
|
|
connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)")
|
|
connection.execute("INSERT INTO word_locations (name) VALUES ('hello')")
|
|
|
|
asset_id = "asset_wordcloud"
|
|
assets_dir = tmp_path / "assets"
|
|
asset_dir = assets_dir / asset_id[:2] / asset_id
|
|
asset_dir.mkdir(parents=True)
|
|
(asset_dir / "meta.json").write_text(
|
|
json.dumps({"asset_id": asset_id, "type": "wordcloud", "job_id": "job-success"}),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(service_app, "ASSETS_DIR", assets_dir)
|
|
monkeypatch.setattr(
|
|
service_app,
|
|
"_resolve_job_status",
|
|
lambda job_id: SimpleNamespace(status="success", artifacts={"db": str(db_path)}),
|
|
)
|
|
return SimpleNamespace(
|
|
workspace=workspace,
|
|
asset_id=asset_id,
|
|
document={"elements": [{"type": "sticker", "assetId": asset_id}]},
|
|
db_path=db_path,
|
|
)
|
|
|
|
|
|
def archive_product_version(client, product_id, document, preview=PNG_BYTES):
|
|
return client.post(
|
|
f"/api/products/{product_id}/versions",
|
|
data={"document_json": json.dumps(document)},
|
|
files={"preview": ("design-preview.png", preview, "image/png")},
|
|
headers=orders_auth_header(),
|
|
)
|
|
|
|
|
|
def test_scanner_keeps_only_visible_wordcloud_assets_and_deduplicates():
|
|
document = {
|
|
"layers": [{"id": "shown", "visible": True}, {"id": "hidden", "visible": False}],
|
|
"elements": [
|
|
{"type": "sticker", "assetId": "wc-a", "layerId": "shown"},
|
|
{"type": "sticker", "assetId": "wc-a", "layerId": "shown"},
|
|
{"type": "sticker", "assetId": "wc-b", "layerId": "hidden"},
|
|
{"type": "sticker", "assetId": "photo", "layerId": "shown"},
|
|
],
|
|
}
|
|
assets = {
|
|
"wc-a": {"type": "wordcloud", "job_id": "a" * 32},
|
|
"wc-b": {"type": "wordcloud", "job_id": "b" * 32},
|
|
"photo": {"type": "upload", "job_id": ""},
|
|
}
|
|
|
|
assert find_visible_wordcloud_sources(document, assets.__getitem__) == [
|
|
WordcloudSource("wc-a", "a" * 32)
|
|
]
|
|
|
|
|
|
def test_scanner_treats_elements_as_visible_without_layers():
|
|
document = {"elements": [{"type": "sticker", "assetId": "wc-a"}]}
|
|
assets = {"wc-a": {"type": "wordcloud", "job_id": "a" * 32}}
|
|
|
|
assert find_visible_wordcloud_sources(document, assets.__getitem__) == [
|
|
WordcloudSource("wc-a", "a" * 32)
|
|
]
|
|
|
|
|
|
def test_scanner_returns_empty_list_without_elements():
|
|
assert find_visible_wordcloud_sources({"layers": []}, lambda _: {}) == []
|
|
|
|
|
|
def test_scanner_skips_wordcloud_without_job_id():
|
|
document = {"elements": [{"type": "sticker", "assetId": "wc-a"}]}
|
|
assets = {"wc-a": {"type": "wordcloud"}}
|
|
|
|
assert find_visible_wordcloud_sources(document, assets.__getitem__) == []
|
|
|
|
|
|
def test_snapshot_rejects_non_sqlite_file_without_creating_destination(tmp_path):
|
|
source = tmp_path / "not-a-database.sqlite"
|
|
destination = tmp_path / "archive" / "word_locations.sqlite"
|
|
source.write_text("not a SQLite database", encoding="utf-8")
|
|
|
|
with pytest.raises(ValueError):
|
|
copy_word_locations_snapshot(source, destination)
|
|
|
|
assert not destination.exists()
|
|
assert not destination.with_name(f"{destination.name}.tmp").exists()
|
|
|
|
|
|
def test_snapshot_validates_table_copies_atomically_and_returns_checksum(tmp_path):
|
|
source = tmp_path / "word_locations.sqlite"
|
|
destination = tmp_path / "archive" / "word_locations.sqlite"
|
|
with sqlite3.connect(source) as connection:
|
|
connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)")
|
|
connection.execute("INSERT INTO word_locations (name) VALUES ('hello')")
|
|
|
|
validate_word_locations_db(source)
|
|
checksum = copy_word_locations_snapshot(source, destination)
|
|
|
|
assert checksum == hashlib.sha256(destination.read_bytes()).hexdigest()
|
|
with sqlite3.connect(destination) as connection:
|
|
assert connection.execute("SELECT name FROM word_locations").fetchone() == ("hello",)
|
|
|
|
|
|
def test_archive_version_copies_db_after_source_workspace_is_removed(
|
|
product_archive_client, prepared_wordcloud_job
|
|
):
|
|
product = create_product(product_archive_client)
|
|
response = product_archive_client.post(
|
|
f"/api/products/{product['product_id']}/versions",
|
|
data={"document_json": json.dumps(prepared_wordcloud_job.document)},
|
|
files={"preview": ("design-preview.png", PNG_BYTES, "image/png")},
|
|
headers=orders_auth_header(),
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
archive = response.json()["wordcloud_archives"][0]
|
|
shutil.rmtree(prepared_wordcloud_job.workspace)
|
|
assert Path(archive["db_path"]).exists()
|
|
|
|
|
|
def test_archive_version_rejects_preview_with_non_png_mime_type(
|
|
product_archive_client, prepared_wordcloud_job
|
|
):
|
|
product = create_product(product_archive_client)
|
|
|
|
response = product_archive_client.post(
|
|
f"/api/products/{product['product_id']}/versions",
|
|
data={"document_json": json.dumps(prepared_wordcloud_job.document)},
|
|
files={"preview": ("design-preview.jpg", PNG_BYTES, "image/jpeg")},
|
|
headers=orders_auth_header(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_archive_version_rejects_unavailable_source_db(product_archive_client, prepared_wordcloud_job):
|
|
prepared_wordcloud_job.db_path.unlink()
|
|
product = create_product(product_archive_client)
|
|
|
|
response = product_archive_client.post(
|
|
f"/api/products/{product['product_id']}/versions",
|
|
data={"document_json": json.dumps(prepared_wordcloud_job.document)},
|
|
files={"preview": ("design-preview.png", PNG_BYTES, "image/png")},
|
|
headers=orders_auth_header(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert not list(product_archive_client.archive_root.glob("prod_*"))
|
|
|
|
|
|
def test_archive_version_rejects_non_success_source_job(product_archive_client, prepared_wordcloud_job, monkeypatch):
|
|
monkeypatch.setattr(
|
|
service_app,
|
|
"_resolve_job_status",
|
|
lambda job_id: SimpleNamespace(status="running", artifacts={"db": str(prepared_wordcloud_job.db_path)}),
|
|
)
|
|
product = create_product(product_archive_client)
|
|
|
|
response = product_archive_client.post(
|
|
f"/api/products/{product['product_id']}/versions",
|
|
data={"document_json": json.dumps(prepared_wordcloud_job.document)},
|
|
files={"preview": ("design-preview.png", PNG_BYTES, "image/png")},
|
|
headers=orders_auth_header(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_archive_version_ignores_hidden_wordclouds(product_archive_client, prepared_wordcloud_job):
|
|
document = {
|
|
"layers": [{"id": "hidden", "visible": False}],
|
|
"elements": [{"type": "sticker", "assetId": prepared_wordcloud_job.asset_id, "layerId": "hidden"}],
|
|
}
|
|
product = create_product(product_archive_client)
|
|
|
|
response = product_archive_client.post(
|
|
f"/api/products/{product['product_id']}/versions",
|
|
data={"document_json": json.dumps(document)},
|
|
files={"preview": ("design-preview.png", PNG_BYTES, "image/png")},
|
|
headers=orders_auth_header(),
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["wordcloud_count"] == 0
|
|
|
|
|
|
def test_archive_version_handles_document_without_wordcloud_sources(product_archive_client):
|
|
product = create_product(product_archive_client)
|
|
|
|
response = product_archive_client.post(
|
|
f"/api/products/{product['product_id']}/versions",
|
|
data={"document_json": json.dumps({"elements": []})},
|
|
files={"preview": ("design-preview.png", PNG_BYTES, "image/png")},
|
|
headers=orders_auth_header(),
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["wordcloud_count"] == 0
|
|
|
|
|
|
def test_archive_version_rejects_invalid_png_bytes_with_png_mime(product_archive_client):
|
|
product = create_product(product_archive_client)
|
|
|
|
response = archive_product_version(product_archive_client, product["product_id"], {"elements": []}, b"\x89PNG\r\n\x1a\nnot-an-image")
|
|
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_archive_version_persists_each_multi_source_provenance(product_archive_client, prepared_wordcloud_job, monkeypatch):
|
|
second_db = prepared_wordcloud_job.workspace / "output" / "second.sqlite"
|
|
with sqlite3.connect(second_db) as connection:
|
|
connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)")
|
|
second_asset = "asset_second"
|
|
second_dir = service_app._asset_dir(second_asset)
|
|
second_dir.mkdir(parents=True)
|
|
(second_dir / "meta.json").write_text(json.dumps({"type": "wordcloud", "job_id": "job-second"}), encoding="utf-8")
|
|
monkeypatch.setattr(
|
|
service_app, "_resolve_job_status",
|
|
lambda job_id: SimpleNamespace(status="success", artifacts={"db": str(prepared_wordcloud_job.db_path if job_id == "job-success" else second_db)}),
|
|
)
|
|
product = create_product(product_archive_client)
|
|
response = archive_product_version(product_archive_client, product["product_id"], {
|
|
"elements": [
|
|
{"type": "sticker", "assetId": prepared_wordcloud_job.asset_id},
|
|
{"type": "sticker", "assetId": second_asset},
|
|
]
|
|
})
|
|
|
|
assert response.status_code == 201
|
|
archives = {item["source_job_id"]: item for item in response.json()["wordcloud_archives"]}
|
|
assert archives["job-success"]["source_asset_id"] == prepared_wordcloud_job.asset_id
|
|
assert archives["job-second"]["source_asset_id"] == second_asset
|
|
assert archives["job-success"]["db_checksum"] == hashlib.sha256(Path(archives["job-success"]["db_path"]).read_bytes()).hexdigest()
|
|
assert archives["job-second"]["db_checksum"] == hashlib.sha256(Path(archives["job-second"]["db_path"]).read_bytes()).hexdigest()
|
|
|
|
|
|
def test_later_archive_version_becomes_current_cover_in_product_contract(product_archive_client):
|
|
product = create_product(product_archive_client)
|
|
first = archive_product_version(product_archive_client, product["product_id"], {"elements": []})
|
|
first_cover = product_archive_client.get(f"/api/products/{product['product_id']}", headers=orders_auth_header()).json()["cover_image_id"]
|
|
second = archive_product_version(product_archive_client, product["product_id"], {"elements": []})
|
|
detail = product_archive_client.get(f"/api/products/{product['product_id']}", headers=orders_auth_header()).json()
|
|
listed = product_archive_client.get("/api/products", headers=orders_auth_header()).json()[0]
|
|
|
|
assert first.status_code == second.status_code == 201
|
|
assert detail["cover_image_id"] == listed["cover_image_id"]
|
|
assert detail["cover_image_id"] != first_cover
|
|
|
|
|
|
def test_product_routes_require_orders_auth(product_archive_client):
|
|
product = create_product(product_archive_client)
|
|
requests = [
|
|
product_archive_client.post("/api/products", json={"name": "未授权", "source": "manual"}),
|
|
product_archive_client.get("/api/products"),
|
|
product_archive_client.get(f"/api/products/{product['product_id']}"),
|
|
product_archive_client.post(f"/api/products/{product['product_id']}/versions"),
|
|
product_archive_client.delete(f"/api/products/{product['product_id']}"),
|
|
product_archive_client.post(f"/api/products/{product['product_id']}/restore"),
|
|
]
|
|
|
|
assert [response.status_code for response in requests] == [403] * 6
|
|
|
|
|
|
def test_archive_failure_during_final_move_removes_staging_files(product_archive_client, prepared_wordcloud_job, monkeypatch):
|
|
product = create_product(product_archive_client)
|
|
service = service_app._product_archive_service()
|
|
monkeypatch.setattr(service, "_move_staging", lambda *_: (_ for _ in ()).throw(RuntimeError("move failed")))
|
|
|
|
with pytest.raises(RuntimeError, match="move failed"):
|
|
service.archive_version(product["product_id"], prepared_wordcloud_job.document, PNG_BYTES)
|
|
|
|
assert not list((product_archive_client.archive_root / product["product_id"]).iterdir())
|
|
|
|
|
|
def test_archive_failure_during_metadata_transaction_removes_final_files(product_archive_client, prepared_wordcloud_job, monkeypatch):
|
|
product = create_product(product_archive_client)
|
|
monkeypatch.setattr(product_archive_client.product_store, "_before_archive_commit", lambda: (_ for _ in ()).throw(RuntimeError("metadata failed")))
|
|
|
|
with pytest.raises(RuntimeError, match="metadata failed"):
|
|
service_app._product_archive_service().archive_version(product["product_id"], prepared_wordcloud_job.document, PNG_BYTES)
|
|
|
|
assert not list((product_archive_client.archive_root / product["product_id"]).iterdir())
|
|
assert product_archive_client.product_store._fetchall("SELECT * FROM product_versions") == []
|