87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
import hashlib
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
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",)
|