574 lines
23 KiB
Python
574 lines
23 KiB
Python
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
import shutil
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
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.cleanup_service import CleanupService # noqa: E402
|
|
from service.metadata_store import MetadataStore # noqa: E402
|
|
from service.storage import Storage # noqa: E402
|
|
from service.schemas import JobStatus # 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_detail_contract_includes_images_and_archive_versions(
|
|
product_archive_client, prepared_wordcloud_job
|
|
):
|
|
product = create_product(product_archive_client)
|
|
archived = archive_product_version(
|
|
product_archive_client, product["product_id"], prepared_wordcloud_job.document
|
|
)
|
|
|
|
detail = product_archive_client.get(
|
|
f"/api/products/{product['product_id']}", headers=orders_auth_header()
|
|
).json()
|
|
|
|
assert archived.status_code == 201
|
|
assert detail["product_id"] == product["product_id"]
|
|
assert len(detail["images"]) == 1
|
|
assert detail["images"][0]["image_type"] == "design_preview"
|
|
assert detail["images"][0]["is_cover"] is True
|
|
assert len(detail["versions"]) == 1
|
|
version = detail["versions"][0]
|
|
preview_image = detail["images"][0]
|
|
expected_image_url = f"/api/products/{product['product_id']}/images/{preview_image['image_id']}"
|
|
assert version["version_id"] == archived.json()["version_id"]
|
|
assert version["design_preview_image_id"] == preview_image["image_id"]
|
|
assert version["design_preview_url"] == expected_image_url
|
|
assert version["wordcloud_count"] == 1
|
|
assert version["wordcloud_archives"][0]["source_job_id"] == "job-success"
|
|
assert Path(version["design_preview_path"]).name == "design-preview.png"
|
|
|
|
|
|
def test_product_detail_image_contract_is_authenticated_and_path_safe(
|
|
product_archive_client, prepared_wordcloud_job, tmp_path
|
|
):
|
|
product = create_product(product_archive_client)
|
|
archive_product_version(
|
|
product_archive_client, product["product_id"], prepared_wordcloud_job.document
|
|
)
|
|
detail = product_archive_client.get(
|
|
f"/api/products/{product['product_id']}", headers=orders_auth_header()
|
|
).json()
|
|
image = detail["images"][0]
|
|
|
|
assert image["image_url"] == f"/api/products/{product['product_id']}/images/{image['image_id']}"
|
|
|
|
served = product_archive_client.get(
|
|
f"/api/products/{product['product_id']}/images/{image['image_id']}",
|
|
headers=orders_auth_header(),
|
|
)
|
|
assert served.status_code == 200
|
|
assert served.headers["content-type"] == "image/png"
|
|
assert served.content == product_archive_client.archive_root.joinpath(
|
|
product["product_id"], image["version_id"], "design-preview.png"
|
|
).read_bytes()
|
|
|
|
outside_path = tmp_path / "outside.png"
|
|
outside_path.write_bytes(PNG_BYTES)
|
|
product_archive_client.product_store._execute(
|
|
"UPDATE product_images SET image_path = ? WHERE image_id = ?",
|
|
(str(outside_path), image["image_id"]),
|
|
)
|
|
unsafe = product_archive_client.get(
|
|
f"/api/products/{product['product_id']}/images/{image['image_id']}",
|
|
headers=orders_auth_header(),
|
|
)
|
|
assert unsafe.status_code == 404
|
|
|
|
unknown = product_archive_client.get(
|
|
f"/api/products/{product['product_id']}/images/img_unknown",
|
|
headers=orders_auth_header(),
|
|
)
|
|
assert unknown.status_code == 404
|
|
|
|
unauthenticated_detail = product_archive_client.get(
|
|
f"/api/products/{product['product_id']}"
|
|
)
|
|
unauthenticated_image = product_archive_client.get(
|
|
f"/api/products/{product['product_id']}/images/{image['image_id']}"
|
|
)
|
|
assert unauthenticated_detail.status_code == unauthenticated_image.status_code == 403
|
|
|
|
|
|
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_version_rejects_failed_cleanup_product_and_status_is_visible(
|
|
product_archive_client,
|
|
):
|
|
product = create_product(product_archive_client)
|
|
deleted = product_archive_client.delete(
|
|
f"/api/products/{product['product_id']}", headers=orders_auth_header()
|
|
).json()
|
|
store = product_archive_client.product_store
|
|
purge_after = store.get_product(deleted["product_id"]).purge_after
|
|
store.claim_product_purge(deleted["product_id"], purge_after)
|
|
store.fail_product_purge(deleted["product_id"])
|
|
|
|
listed = product_archive_client.get(
|
|
f"/api/products/{deleted['product_id']}", headers=orders_auth_header()
|
|
).json()
|
|
assert listed["status"] == "failed_cleanup"
|
|
|
|
response = archive_product_version(
|
|
product_archive_client, deleted["product_id"], {"elements": []}
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
restored = product_archive_client.post(
|
|
f"/api/products/{deleted['product_id']}/restore", headers=orders_auth_header()
|
|
).json()
|
|
assert restored["status"] == "active"
|
|
|
|
|
|
def test_delete_rejects_failed_cleanup_product_until_restore(product_archive_client):
|
|
product = create_product(product_archive_client)
|
|
deleted = product_archive_client.delete(
|
|
f"/api/products/{product['product_id']}", headers=orders_auth_header()
|
|
).json()
|
|
store = product_archive_client.product_store
|
|
purge_after = store.get_product(deleted["product_id"]).purge_after
|
|
store.claim_product_purge(deleted["product_id"], purge_after)
|
|
store.fail_product_purge(deleted["product_id"], now=purge_after)
|
|
|
|
rejected = product_archive_client.delete(
|
|
f"/api/products/{deleted['product_id']}", headers=orders_auth_header()
|
|
)
|
|
assert rejected.status_code == 409
|
|
assert store.get_product(deleted["product_id"]).status == "failed_cleanup"
|
|
|
|
restored = product_archive_client.post(
|
|
f"/api/products/{deleted['product_id']}/restore", headers=orders_auth_header()
|
|
)
|
|
assert restored.status_code == 200
|
|
assert restored.json()["status"] == "active"
|
|
|
|
soft_deleted = product_archive_client.delete(
|
|
f"/api/products/{deleted['product_id']}", headers=orders_auth_header()
|
|
)
|
|
assert soft_deleted.status_code == 200
|
|
assert soft_deleted.json()["status"] == "pending_cleanup"
|
|
|
|
|
|
def test_only_visible_inserted_wordcloud_is_retained_after_30_day_cleanup(
|
|
product_archive_client, prepared_wordcloud_job, tmp_path, monkeypatch
|
|
):
|
|
archived_at = datetime.now(timezone.utc)
|
|
created_at = archived_at - timedelta(days=31)
|
|
assets_dir = tmp_path / "e2e-assets"
|
|
assets_dir.mkdir(parents=True)
|
|
monkeypatch.setattr(service_app, "ASSETS_DIR", assets_dir)
|
|
|
|
source_specs = {
|
|
"job-visible": "visible",
|
|
"job-hidden": "hidden",
|
|
"job-detached": None,
|
|
}
|
|
source_assets: dict[str, dict[str, str]] = {}
|
|
source_paths: dict[str, Path] = {}
|
|
metadata_store = MetadataStore(tmp_path / "e2e-metadata" / "app.db")
|
|
storage = Storage(tmp_path / "e2e-workspace")
|
|
|
|
for job_id, placement in source_specs.items():
|
|
asset_id = f"asset_{job_id.replace('-', '_')}"
|
|
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_id}),
|
|
encoding="utf-8",
|
|
)
|
|
source_assets[job_id] = asset_id
|
|
source_paths[job_id] = storage.job_root(job_id) / "output" / "word_locations.sqlite"
|
|
source_paths[job_id].parent.mkdir(parents=True, exist_ok=True)
|
|
with sqlite3.connect(source_paths[job_id]) as connection:
|
|
connection.execute("CREATE TABLE word_locations (id INTEGER PRIMARY KEY, name TEXT)")
|
|
connection.execute("INSERT INTO word_locations (name) VALUES (?)", (job_id,))
|
|
metadata_store.upsert_job(
|
|
JobStatus(
|
|
job_id=job_id,
|
|
status="success",
|
|
stage="done",
|
|
progress_percent=100,
|
|
message="done",
|
|
artifacts={"db": str(source_paths[job_id])},
|
|
created_at=created_at,
|
|
updated_at=created_at,
|
|
)
|
|
)
|
|
|
|
def resolve_job(job_id: str):
|
|
return SimpleNamespace(
|
|
status="success",
|
|
artifacts={"db": str(source_paths[job_id])},
|
|
)
|
|
|
|
monkeypatch.setattr(service_app, "_resolve_job_status", resolve_job)
|
|
product = create_product(product_archive_client)
|
|
response = archive_product_version(product_archive_client, product["product_id"], {
|
|
"layers": [
|
|
{"id": "shown", "visible": True},
|
|
{"id": "hidden", "visible": False},
|
|
],
|
|
"elements": [
|
|
{"type": "sticker", "assetId": source_assets["job-visible"], "layerId": "shown"},
|
|
{"type": "sticker", "assetId": source_assets["job-hidden"], "layerId": "hidden"},
|
|
],
|
|
})
|
|
|
|
assert response.status_code == 201
|
|
archived = response.json()["wordcloud_archives"]
|
|
assert [item["source_job_id"] for item in archived] == ["job-visible"]
|
|
visible_snapshot = Path(archived[0]["db_path"])
|
|
assert visible_snapshot.exists()
|
|
|
|
cleanup = CleanupService(storage, metadata_store, product_archive_client.product_store)
|
|
report = cleanup.apply(now=archived_at + timedelta(days=31))
|
|
|
|
assert "job-visible" not in report.deleted_job_ids
|
|
assert {"job-hidden", "job-detached"} <= set(report.deleted_job_ids)
|
|
assert not storage.job_root("job-hidden").exists()
|
|
assert not storage.job_root("job-detached").exists()
|
|
assert storage.job_root("job-visible").exists()
|
|
assert visible_snapshot.exists()
|
|
|
|
|
|
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") == []
|