forked from retoor/devplacepy
81 lines
4.1 KiB
Python
81 lines
4.1 KiB
Python
 
|
# retoor <retoor@molodetz.nl>
|
||
|
|
|
||
|
|
import base64
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from devplacepy.config import ATTACHMENTS_DIR
|
||
|
|
from devplacepy.database import get_table
|
||
|
|
from tests.api.sync.conftest import PNG_1X1, attachment_record, seed_world, sync_ok
|
||
|
|
|
||
|
|
|
||
|
|
def local(origin, kind, origin_uid):
|
||
|
|
return get_table("sync_links").find_one(origin=origin, kind=kind, origin_uid=origin_uid)["local_uid"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_an_attachment_is_stored_and_linked_to_its_target(client, admin, origin):
|
||
|
|
seed_world(client, admin, origin)
|
||
|
|
result = sync_ok(client, admin, "attachments", origin, [attachment_record()])["results"][0]
|
||
|
|
assert result["status"] == "created"
|
||
|
|
row = get_table("attachments").find_one(uid=result["local_uid"])
|
||
|
|
assert (row["target_type"], row["target_uid"]) == ("post", local(origin, "posts", "p1"))
|
||
|
|
assert row["user_uid"] == local(origin, "users", "u1") and row["mime_type"] == "image/png"
|
||
|
|
assert (Path(ATTACHMENTS_DIR) / row["directory"] / row["stored_name"]).read_bytes() == base64.b64decode(PNG_1X1)
|
||
|
|
assert row["created_at"].startswith("2025-01-05")
|
||
|
|
|
||
|
|
|
||
|
|
def test_replay_is_unchanged_and_a_content_change_is_refused(client, admin, origin):
|
||
|
|
seed_world(client, admin, origin)
|
||
|
|
sync_ok(client, admin, "attachments", origin, [attachment_record()])
|
||
|
|
assert sync_ok(client, admin, "attachments", origin, [attachment_record()])["results"][0]["status"] == "unchanged"
|
||
|
|
other = base64.b64encode(b"GIF89a" + b"x" * 40).decode()
|
||
|
|
changed = sync_ok(client, admin, "attachments", origin, [attachment_record(content=other, filename="b.gif", updated_at="2025-02-01T00:00:00Z")])["results"][0]
|
||
|
|
assert changed["status"] == "rejected" and changed["reason"] == "immutable"
|
||
|
|
|
||
|
|
|
||
|
|
def test_tombstone_soft_deletes_and_restores(client, admin, origin):
|
||
|
|
seed_world(client, admin, origin)
|
||
|
|
created = sync_ok(client, admin, "attachments", origin, [attachment_record()])["results"][0]
|
||
|
|
gone = attachment_record(updated_at="2025-02-01T00:00:00Z", deleted=True)
|
||
|
|
assert sync_ok(client, admin, "attachments", origin, [gone])["results"][0]["status"] == "updated"
|
||
|
|
assert get_table("attachments").find_one(uid=created["local_uid"])["deleted_by"] == "sync"
|
||
|
|
back = attachment_record(updated_at="2025-03-01T00:00:00Z")
|
||
|
|
sync_ok(client, admin, "attachments", origin, [back])
|
||
|
|
assert get_table("attachments").find_one(uid=created["local_uid"])["deleted_at"] is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_validation_and_type_rules(client, admin, origin):
|
||
|
|
seed_world(client, admin, origin)
|
||
|
|
body = sync_ok(
|
||
|
|
client, admin, "attachments", origin,
|
||
|
|
[
|
||
|
|
attachment_record("a1", content="not base64!!"),
|
||
|
|
attachment_record("a2", content=""),
|
||
|
|
attachment_record("a3", filename="run.exe", content=base64.b64encode(b"MZ" + b"0" * 10).decode()),
|
||
|
|
attachment_record("a4", filename="../../etc/passwd.png"),
|
||
|
|
],
|
||
|
|
)
|
||
|
|
statuses = [item["status"] for item in body["results"]]
|
||
|
|
assert statuses[:3] == ["rejected", "rejected", "rejected"]
|
||
|
|
stored = get_table("attachments").find_one(uid=body["results"][3].get("local_uid") or "x")
|
||
|
|
assert statuses[3] == "created" and stored["original_filename"] == "passwd.png"
|
||
|
|
|
||
|
|
|
||
|
|
def test_an_oversize_file_is_refused(client, admin, origin):
|
||
|
|
from devplacepy.services.sync.spec import LIMITS
|
||
|
|
|
||
|
|
seed_world(client, admin, origin)
|
||
|
|
big = base64.b64encode(b"\x00" * (LIMITS.max_attachment_bytes + 1)).decode()
|
||
|
|
result = sync_ok(client, admin, "attachments", origin, [attachment_record(content=big)])["results"][0]
|
||
|
|
assert result["status"] == "rejected" and "exceeds" in result["errors"][0]
|
||
|
|
|
||
|
|
|
||
|
|
def test_an_unknown_target_waits_and_is_released_when_the_target_arrives(client, admin, origin):
|
||
|
|
from tests.api.sync.conftest import post_record, user_record
|
||
|
|
|
||
|
|
sync_ok(client, admin, "users", origin, [user_record("u1", "attuser1")])
|
||
|
|
waiting = sync_ok(client, admin, "attachments", origin, [attachment_record(target="p1")])["results"][0]
|
||
|
|
assert waiting["status"] == "deferred"
|
||
|
|
body = sync_ok(client, admin, "posts", origin, [post_record("p1", "u1")])
|
||
|
|
assert body["released"] == {"created": 1}
|
||
|
|
assert get_table("attachments").count(target_uid=local(origin, "posts", "p1")) == 1
|