255 lines
9.1 KiB
Python
255 lines
9.1 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
|
|
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 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
|