|
# retoor <retoor@molodetz.nl>
|
|
|
|
import time
|
|
import requests
|
|
from tests.conftest import BASE_URL
|
|
|
|
JSON_trash = {"Accept": "application/json"}
|
|
_counter_trash = [0]
|
|
|
|
|
|
def _unique_trash(prefix="tr"):
|
|
_counter_trash[0] += 1
|
|
return f"{prefix}{int(time.time() * 1000)}{_counter_trash[0]}"
|
|
|
|
|
|
def _member_trash():
|
|
name = _unique_trash("trrmem")
|
|
s = requests.Session()
|
|
s.post(
|
|
f"{BASE_URL}/auth/signup",
|
|
data={
|
|
"username": name,
|
|
"email": f"{name}@t.dev",
|
|
"password": "secret123",
|
|
"confirm_password": "secret123",
|
|
},
|
|
allow_redirects=True,
|
|
)
|
|
return s, name
|
|
|
|
|
|
def _admin_trash(seeded_db):
|
|
s = requests.Session()
|
|
creds = seeded_db["alice"]
|
|
s.post(
|
|
f"{BASE_URL}/auth/login",
|
|
data={"email": creds["email"], "password": creds["password"]},
|
|
allow_redirects=True,
|
|
)
|
|
return s
|
|
|
|
|
|
def _create_post_trash(session):
|
|
return session.post(
|
|
f"{BASE_URL}/posts/create",
|
|
headers=JSON_trash,
|
|
data={"title": _unique_trash("trrp"), "content": "restore post body text", "topic": "devlog"},
|
|
).json()["data"]
|
|
|
|
|
|
def _soft_delete_post_trash(session, slug):
|
|
session.post(f"{BASE_URL}/posts/delete/{slug}", headers=JSON_trash, allow_redirects=False)
|
|
|
|
|
|
def _audit_find(admin, event_key, target_uid):
|
|
data = admin.get(
|
|
f"{BASE_URL}/admin/audit-log", headers=JSON_trash, params={"event_key": event_key}
|
|
).json()
|
|
for entry in data["entries"]:
|
|
if entry.get("target_uid") == target_uid:
|
|
return entry
|
|
return None
|
|
|
|
|
|
def test_restore_requires_admin(seeded_db):
|
|
member, _ = _member_trash()
|
|
r = member.post(
|
|
f"{BASE_URL}/admin/trash/posts/anything/restore", allow_redirects=False
|
|
)
|
|
assert r.status_code in (302, 303)
|
|
|
|
|
|
def test_restore_unknown_table_404(seeded_db):
|
|
admin = _admin_trash(seeded_db)
|
|
r = admin.post(
|
|
f"{BASE_URL}/admin/trash/bogus/anything/restore",
|
|
headers=JSON_trash,
|
|
allow_redirects=False,
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_restore_revives_post_and_records_audit(seeded_db):
|
|
member, _ = _member_trash()
|
|
post = _create_post_trash(member)
|
|
_soft_delete_post_trash(member, post["slug"])
|
|
admin = _admin_trash(seeded_db)
|
|
|
|
r = admin.post(
|
|
f"{BASE_URL}/admin/trash/posts/{post['uid']}/restore",
|
|
headers=JSON_trash,
|
|
allow_redirects=False,
|
|
)
|
|
assert r.status_code == 200, r.text[:300]
|
|
|
|
# restored row is live again and gone from trash
|
|
trash = admin.get(
|
|
f"{BASE_URL}/admin/trash", headers=JSON_trash, params={"table": "posts"}
|
|
).json()
|
|
assert all(item["uid"] != post["uid"] for item in trash["items"])
|
|
assert member.get(f"{BASE_URL}/posts/{post['slug']}").status_code == 200
|
|
|
|
entry = _audit_find(admin, "admin.trash.restore", post["uid"])
|
|
assert entry is not None
|
|
assert entry["result"] == "success"
|
|
|
|
|
|
def test_restore_revoked_award_recomputes_stats(seeded_db):
|
|
from datetime import datetime, timezone
|
|
from devplacepy.database.awards import revoke_award
|
|
from devplacepy.utils import generate_uid, make_combined_slug
|
|
|
|
admin = _admin_trash(seeded_db)
|
|
bob = get_table("users").find_one(username="bob_test")
|
|
alice = get_table("users").find_one(username="alice_test")
|
|
uid = generate_uid()
|
|
slug = make_combined_slug("Trash restore", uid)
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
get_table("awards").insert(
|
|
{
|
|
"uid": uid,
|
|
"slug": slug,
|
|
"description": "Trash restore",
|
|
"giver_uid": alice["uid"],
|
|
"receiver_uid": bob["uid"],
|
|
"attachment_uid_512": "",
|
|
"attachment_uid_256": "",
|
|
"attachment_uid_64": "",
|
|
"generated_at": now,
|
|
"created_at": now,
|
|
"job_uid": "",
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
revoke_award(uid, alice["uid"])
|
|
r = admin.post(
|
|
f"{BASE_URL}/admin/trash/awards/{uid}/restore",
|
|
headers=JSON_trash,
|
|
allow_redirects=False,
|
|
)
|
|
assert r.status_code == 200, r.text[:300]
|
|
row = get_table("users").find_one(uid=bob["uid"])
|
|
assert row.get("award_count") == 1
|
|
assert row.get("last_award_uid") == uid
|
|
assert requests.get(f"{BASE_URL}/awards/{slug}/256").status_code in (302, 404)
|