@@ -0,0 +1,118 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table
|
||||
|
||||
JSON = {"Accept": "application/json"}
|
||||
_counter = [0]
|
||||
|
||||
|
||||
def _signup():
|
||||
_counter[0] += 1
|
||||
name = f"blk{int(time.time() * 1000)}{_counter[0]}"
|
||||
session = requests.Session()
|
||||
session.post(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data={
|
||||
"username": name,
|
||||
"email": f"{name}@t.dev",
|
||||
"password": "secret123",
|
||||
"confirm_password": "secret123",
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return session, name
|
||||
|
||||
|
||||
def _uid(name):
|
||||
return get_table("users").find_one(username=name)["uid"]
|
||||
|
||||
|
||||
def _key(name):
|
||||
return get_table("users").find_one(username=name)["api_key"]
|
||||
|
||||
|
||||
def _relation(actor_uid, target_uid, kind):
|
||||
return get_table("user_relations").count(
|
||||
user_uid=actor_uid, target_uid=target_uid, kind=kind, deleted_at=None
|
||||
)
|
||||
|
||||
|
||||
def test_block_creates_relation(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
_, target = _signup()
|
||||
r = blocker_session.post(
|
||||
f"{BASE_URL}/block/{target}", allow_redirects=False
|
||||
)
|
||||
assert r.status_code in (302, 200)
|
||||
assert _relation(_uid(blocker), _uid(target), "block") == 1
|
||||
|
||||
|
||||
def test_block_via_api_key_header(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
_, target = _signup()
|
||||
r = requests.post(
|
||||
f"{BASE_URL}/block/{target}",
|
||||
headers={"X-API-KEY": _key(blocker)},
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code in (302, 200)
|
||||
assert _relation(_uid(blocker), _uid(target), "block") == 1
|
||||
|
||||
|
||||
def test_block_is_idempotent(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
_, target = _signup()
|
||||
blocker_session.post(f"{BASE_URL}/block/{target}", allow_redirects=False)
|
||||
blocker_session.post(f"{BASE_URL}/block/{target}", allow_redirects=False)
|
||||
assert _relation(_uid(blocker), _uid(target), "block") == 1
|
||||
|
||||
|
||||
def test_self_block_rejected(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
blocker_session.post(f"{BASE_URL}/block/{blocker}", allow_redirects=False)
|
||||
assert _relation(_uid(blocker), _uid(blocker), "block") == 0
|
||||
|
||||
|
||||
def test_block_nonexistent_user_redirects(app_server):
|
||||
blocker_session, _ = _signup()
|
||||
r = blocker_session.post(
|
||||
f"{BASE_URL}/block/no_such_user_zzz", allow_redirects=False
|
||||
)
|
||||
assert r.status_code == 302
|
||||
|
||||
|
||||
def test_block_no_credentials_redirects(app_server):
|
||||
_, target = _signup()
|
||||
r = requests.post(f"{BASE_URL}/block/{target}", allow_redirects=False)
|
||||
assert r.status_code == 303
|
||||
|
||||
|
||||
def test_block_invalid_api_key_returns_401(app_server):
|
||||
_, target = _signup()
|
||||
r = requests.post(
|
||||
f"{BASE_URL}/block/{target}",
|
||||
headers={"X-API-KEY": "not-a-real-key"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_block_recorded_in_audit(seeded_db):
|
||||
blocker_session, blocker = _signup()
|
||||
blocker_session.post(
|
||||
f"{BASE_URL}/block/bob_test", headers=JSON, allow_redirects=False
|
||||
)
|
||||
admin = requests.Session()
|
||||
admin.headers.update({"X-API-KEY": _key("alice_test")})
|
||||
data = admin.get(
|
||||
f"{BASE_URL}/admin/audit-log",
|
||||
headers=JSON,
|
||||
params={"event_key": "relation.block"},
|
||||
).json()
|
||||
assert any(
|
||||
e.get("target_label") == "bob_test" and blocker in (e.get("summary") or "")
|
||||
for e in data["entries"]
|
||||
)
|
||||
@@ -0,0 +1,71 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table
|
||||
|
||||
_counter = [0]
|
||||
|
||||
|
||||
def _signup():
|
||||
_counter[0] += 1
|
||||
name = f"unblk{int(time.time() * 1000)}{_counter[0]}"
|
||||
session = requests.Session()
|
||||
session.post(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data={
|
||||
"username": name,
|
||||
"email": f"{name}@t.dev",
|
||||
"password": "secret123",
|
||||
"confirm_password": "secret123",
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return session, name
|
||||
|
||||
|
||||
def _uid(name):
|
||||
return get_table("users").find_one(username=name)["uid"]
|
||||
|
||||
|
||||
def _active_blocks(actor_uid, target_uid):
|
||||
return get_table("user_relations").count(
|
||||
user_uid=actor_uid, target_uid=target_uid, kind="block", deleted_at=None
|
||||
)
|
||||
|
||||
|
||||
def test_unblock_reverses_block(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
_, target = _signup()
|
||||
blocker_session.post(f"{BASE_URL}/block/{target}", allow_redirects=False)
|
||||
assert _active_blocks(_uid(blocker), _uid(target)) == 1
|
||||
|
||||
r = blocker_session.post(
|
||||
f"{BASE_URL}/block/unblock/{target}", allow_redirects=False
|
||||
)
|
||||
assert r.status_code in (302, 200)
|
||||
assert _active_blocks(_uid(blocker), _uid(target)) == 0
|
||||
|
||||
|
||||
def test_unblock_then_reblock_revives_one_row(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
_, target = _signup()
|
||||
blocker_session.post(f"{BASE_URL}/block/{target}", allow_redirects=False)
|
||||
blocker_session.post(f"{BASE_URL}/block/unblock/{target}", allow_redirects=False)
|
||||
blocker_session.post(f"{BASE_URL}/block/{target}", allow_redirects=False)
|
||||
|
||||
assert _active_blocks(_uid(blocker), _uid(target)) == 1
|
||||
total = get_table("user_relations").count(
|
||||
user_uid=_uid(blocker), target_uid=_uid(target), kind="block"
|
||||
)
|
||||
assert total == 1
|
||||
|
||||
|
||||
def test_unblock_when_not_blocked_is_noop(app_server):
|
||||
blocker_session, _ = _signup()
|
||||
_, target = _signup()
|
||||
r = blocker_session.post(
|
||||
f"{BASE_URL}/block/unblock/{target}", allow_redirects=False
|
||||
)
|
||||
assert r.status_code in (302, 200)
|
||||
@@ -0,0 +1,206 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import time
|
||||
from uuid import uuid4
|
||||
from datetime import datetime, timezone
|
||||
import requests
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table
|
||||
from devplacepy.utils import make_combined_slug
|
||||
|
||||
JSON = {"Accept": "application/json"}
|
||||
_counter = [0]
|
||||
|
||||
|
||||
def _signup():
|
||||
_counter[0] += 1
|
||||
name = f"blkvis{int(time.time() * 1000)}{_counter[0]}"
|
||||
session = requests.Session()
|
||||
session.post(
|
||||
f"{BASE_URL}/auth/signup",
|
||||
data={
|
||||
"username": name,
|
||||
"email": f"{name}@t.dev",
|
||||
"password": "secret123",
|
||||
"confirm_password": "secret123",
|
||||
},
|
||||
allow_redirects=True,
|
||||
)
|
||||
return session, name
|
||||
|
||||
|
||||
def _uid(name):
|
||||
return get_table("users").find_one(username=name)["uid"]
|
||||
|
||||
|
||||
def _new_post(session, content="blocked author post body"):
|
||||
return session.post(
|
||||
f"{BASE_URL}/posts/create",
|
||||
headers=JSON,
|
||||
data={"title": f"blkpost{uuid4().hex[:8]}", "content": content, "topic": "devlog"},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def _new_project(session):
|
||||
return session.post(
|
||||
f"{BASE_URL}/projects/create",
|
||||
headers=JSON,
|
||||
data={
|
||||
"title": f"blkproj{uuid4().hex[:8]}",
|
||||
"description": "blocked author project description",
|
||||
"project_type": "software",
|
||||
"status": "In Development",
|
||||
"platforms": "",
|
||||
},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def _new_gist(session):
|
||||
return session.post(
|
||||
f"{BASE_URL}/gists/create",
|
||||
headers=JSON,
|
||||
data={
|
||||
"title": f"blkgist{uuid4().hex[:8]}",
|
||||
"description": "blocked author gist",
|
||||
"source_code": "print('x')",
|
||||
"language": "python",
|
||||
},
|
||||
).json()["data"]
|
||||
|
||||
|
||||
def test_blocked_author_post_hidden_from_feed(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
author_session, author = _signup()
|
||||
post = _new_post(author_session)
|
||||
|
||||
before = blocker_session.get(f"{BASE_URL}/feed", headers=JSON).json()
|
||||
assert post["uid"] in [i["post"]["uid"] for i in before["posts"]]
|
||||
|
||||
blocker_session.post(f"{BASE_URL}/block/{author}", allow_redirects=False)
|
||||
|
||||
after = blocker_session.get(f"{BASE_URL}/feed", headers=JSON).json()
|
||||
assert post["uid"] not in [i["post"]["uid"] for i in after["posts"]]
|
||||
|
||||
|
||||
def test_blocked_author_gist_hidden_from_listing(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
author_session, author = _signup()
|
||||
gist = _new_gist(author_session)
|
||||
blocker_session.post(f"{BASE_URL}/block/{author}", allow_redirects=False)
|
||||
|
||||
data = blocker_session.get(f"{BASE_URL}/gists", headers=JSON).json()
|
||||
assert gist["uid"] not in [i["gist"]["uid"] for i in data["gists"]]
|
||||
|
||||
|
||||
def test_blocked_author_project_hidden_from_listing(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
author_session, author = _signup()
|
||||
project = _new_project(author_session)
|
||||
blocker_session.post(f"{BASE_URL}/block/{author}", allow_redirects=False)
|
||||
|
||||
data = blocker_session.get(f"{BASE_URL}/projects", headers=JSON).json()
|
||||
assert project["uid"] not in [i["uid"] for i in data["projects"]]
|
||||
|
||||
|
||||
def test_blocked_author_post_detail_404(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
author_session, author = _signup()
|
||||
post = _new_post(author_session)
|
||||
blocker_session.post(f"{BASE_URL}/block/{author}", allow_redirects=False)
|
||||
|
||||
r = blocker_session.get(f"{BASE_URL}/posts/{post['slug']}")
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_blocked_author_comment_hidden_on_detail(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
host_session, host = _signup()
|
||||
commenter_session, commenter = _signup()
|
||||
post = _new_post(host_session)
|
||||
|
||||
commenter_session.post(
|
||||
f"{BASE_URL}/comments/create",
|
||||
headers=JSON,
|
||||
data={
|
||||
"target_type": "post",
|
||||
"target_uid": post["uid"],
|
||||
"content": "comment from a blocked user",
|
||||
},
|
||||
)
|
||||
|
||||
detail = blocker_session.get(
|
||||
f"{BASE_URL}/posts/{post['slug']}", headers=JSON
|
||||
).json()
|
||||
assert any(
|
||||
c["comment"]["content"] == "comment from a blocked user"
|
||||
for c in detail["comments"]
|
||||
)
|
||||
|
||||
blocker_session.post(f"{BASE_URL}/block/{commenter}", allow_redirects=False)
|
||||
|
||||
detail2 = blocker_session.get(
|
||||
f"{BASE_URL}/posts/{post['slug']}", headers=JSON
|
||||
).json()
|
||||
assert not any(
|
||||
c["comment"]["content"] == "comment from a blocked user"
|
||||
for c in detail2["comments"]
|
||||
)
|
||||
|
||||
|
||||
def test_blocked_user_cannot_send_dm(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
sender_session, sender = _signup()
|
||||
blocker_session.post(f"{BASE_URL}/block/{sender}", allow_redirects=False)
|
||||
|
||||
sender_session.post(
|
||||
f"{BASE_URL}/messages/send",
|
||||
headers=JSON,
|
||||
data={"receiver_uid": _uid(blocker), "content": "you blocked me"},
|
||||
allow_redirects=False,
|
||||
)
|
||||
|
||||
assert (
|
||||
get_table("messages").count(
|
||||
sender_uid=_uid(sender), receiver_uid=_uid(blocker)
|
||||
)
|
||||
== 0
|
||||
)
|
||||
|
||||
|
||||
def test_block_suppresses_notifications(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
actor_session, actor = _signup()
|
||||
blocker_session.post(f"{BASE_URL}/block/{actor}", allow_redirects=False)
|
||||
|
||||
actor_session.post(f"{BASE_URL}/follow/{blocker}", allow_redirects=False)
|
||||
|
||||
assert (
|
||||
get_table("notifications").count(
|
||||
user_uid=_uid(blocker), related_uid=_uid(actor), type="follow"
|
||||
)
|
||||
== 0
|
||||
)
|
||||
|
||||
|
||||
def test_profile_json_reports_is_blocked(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
_, target = _signup()
|
||||
blocker_session.post(f"{BASE_URL}/block/{target}", allow_redirects=False)
|
||||
|
||||
data = blocker_session.get(
|
||||
f"{BASE_URL}/profile/{target}", headers=JSON
|
||||
).json()
|
||||
assert data["is_blocked"] is True
|
||||
assert data["is_muted"] is False
|
||||
|
||||
|
||||
def test_blocked_user_profile_still_shows_their_posts(app_server):
|
||||
blocker_session, blocker = _signup()
|
||||
author_session, author = _signup()
|
||||
post = _new_post(author_session, content="visible on my own profile")
|
||||
blocker_session.post(f"{BASE_URL}/block/{author}", allow_redirects=False)
|
||||
|
||||
data = blocker_session.get(
|
||||
f"{BASE_URL}/profile/{author}", headers=JSON
|
||||
).json()
|
||||
assert post["uid"] in [p["post"]["uid"] for p in data["posts"]]
|
||||
Reference in New Issue
Block a user