102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
|
# 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 _unique(prefix="mconv"):
|
||
|
|
_counter[0] += 1
|
||
|
|
return f"{prefix}{int(time.time() * 1000)}{_counter[0]}"
|
||
|
|
|
||
|
|
|
||
|
|
def _signup():
|
||
|
|
name = _unique()
|
||
|
|
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 _uid(name):
|
||
|
|
return get_table("users").find_one(username=name)["uid"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_conversations_requires_auth_401_for_json(app_server):
|
||
|
|
r = requests.get(
|
||
|
|
f"{BASE_URL}/messages/conversations", headers=JSON, allow_redirects=False
|
||
|
|
)
|
||
|
|
assert r.status_code == 401
|
||
|
|
|
||
|
|
|
||
|
|
def test_conversations_requires_auth_redirects_browser(app_server):
|
||
|
|
r = requests.get(f"{BASE_URL}/messages/conversations", allow_redirects=False)
|
||
|
|
assert r.status_code == 303
|
||
|
|
|
||
|
|
|
||
|
|
def test_conversations_matches_messages_page_shape(app_server):
|
||
|
|
sender, _ = _signup()
|
||
|
|
_, other_name = _signup()
|
||
|
|
other_uid = _uid(other_name)
|
||
|
|
|
||
|
|
sent = sender.post(
|
||
|
|
f"{BASE_URL}/messages/send",
|
||
|
|
headers=JSON,
|
||
|
|
data={"receiver_uid": other_uid, "content": "hello from parity test"},
|
||
|
|
)
|
||
|
|
assert sent.status_code == 200, sent.text[:300]
|
||
|
|
|
||
|
|
page = sender.get(f"{BASE_URL}/messages", headers=JSON)
|
||
|
|
assert page.status_code == 200, page.text[:300]
|
||
|
|
page_conversations = page.json()["conversations"]
|
||
|
|
|
||
|
|
flat = sender.get(f"{BASE_URL}/messages/conversations")
|
||
|
|
assert flat.status_code == 200, flat.text[:300]
|
||
|
|
flat_conversations = flat.json()["conversations"]
|
||
|
|
|
||
|
|
assert page_conversations == flat_conversations
|
||
|
|
|
||
|
|
match = next(
|
||
|
|
c for c in flat_conversations if c["other_user"]["uid"] == other_uid
|
||
|
|
)
|
||
|
|
assert match["last_message"] == "hello from parity test"
|
||
|
|
assert match["unread"] is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_conversations_excludes_blocked_user(app_server):
|
||
|
|
viewer, _ = _signup()
|
||
|
|
blocked, blocked_name = _signup()
|
||
|
|
blocked_uid = _uid(blocked_name)
|
||
|
|
|
||
|
|
reply = viewer.post(
|
||
|
|
f"{BASE_URL}/messages/send",
|
||
|
|
headers=JSON,
|
||
|
|
data={"receiver_uid": blocked_uid, "content": "before block"},
|
||
|
|
)
|
||
|
|
assert reply.status_code == 200, reply.text[:300]
|
||
|
|
|
||
|
|
before = viewer.get(f"{BASE_URL}/messages/conversations")
|
||
|
|
assert any(
|
||
|
|
c["other_user"]["uid"] == blocked_uid for c in before.json()["conversations"]
|
||
|
|
)
|
||
|
|
|
||
|
|
viewer.post(f"{BASE_URL}/block/{blocked_name}", allow_redirects=False)
|
||
|
|
|
||
|
|
after = viewer.get(f"{BASE_URL}/messages/conversations")
|
||
|
|
assert after.status_code == 200
|
||
|
|
assert not any(
|
||
|
|
c["other_user"]["uid"] == blocked_uid for c in after.json()["conversations"]
|
||
|
|
)
|