1285 lines
56 KiB
Python
1285 lines
56 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.auth.models import EmailVerificationToken
|
|
from app.chats.models import Chat, ChatMember, ChatMemberRole, ChatType
|
|
from app.messages.models import Message
|
|
|
|
|
|
async def _create_verified_user(client, db_session, email: str, username: str, password: str) -> dict:
|
|
await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "name": username, "username": username, "password": password},
|
|
)
|
|
token_row = await db_session.execute(select(EmailVerificationToken).order_by(EmailVerificationToken.id.desc()))
|
|
verify_token = token_row.scalar_one().token
|
|
await client.post("/api/v1/auth/verify-email", json={"token": verify_token})
|
|
login_response = await client.post("/api/v1/auth/login", json={"email": email, "password": password})
|
|
return login_response.json()
|
|
|
|
|
|
async def test_private_chat_message_lifecycle(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "u1@example.com", "user_one", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "u2@example.com", "user_two", "strongpass123")
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
create_chat_response = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_response.status_code == 200
|
|
chat_id = create_chat_response.json()["id"]
|
|
|
|
send_message_response = await client.post(
|
|
"/api/v1/messages",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"chat_id": chat_id, "type": "text", "text": "hello @user_two"},
|
|
)
|
|
assert send_message_response.status_code == 201
|
|
message_id = send_message_response.json()["id"]
|
|
|
|
list_messages_response = await client.get(
|
|
f"/api/v1/messages/{chat_id}",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
)
|
|
assert list_messages_response.status_code == 200
|
|
assert len(list_messages_response.json()) == 1
|
|
|
|
edit_message_response = await client.put(
|
|
f"/api/v1/messages/{message_id}",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"text": "edited text"},
|
|
)
|
|
assert edit_message_response.status_code == 200
|
|
assert edit_message_response.json()["text"] == "edited text"
|
|
|
|
delete_message_response = await client.delete(
|
|
f"/api/v1/messages/{message_id}",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert delete_message_response.status_code == 204
|
|
|
|
|
|
async def test_edit_message_older_than_7_days_is_forbidden(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "edit_older_u1@example.com", "edit_older_u1", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "edit_older_u2@example.com", "edit_older_u2", "strongpass123")
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
create_chat_response = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_response.status_code == 200
|
|
chat_id = create_chat_response.json()["id"]
|
|
|
|
send_message_response = await client.post(
|
|
"/api/v1/messages",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"chat_id": chat_id, "type": "text", "text": "original"},
|
|
)
|
|
assert send_message_response.status_code == 201
|
|
message_id = send_message_response.json()["id"]
|
|
|
|
message_row = await db_session.execute(select(Message).where(Message.id == message_id))
|
|
message = message_row.scalar_one()
|
|
message.created_at = datetime.now(timezone.utc) - timedelta(days=8)
|
|
await db_session.commit()
|
|
|
|
edit_message_response = await client.put(
|
|
f"/api/v1/messages/{message_id}",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"text": "edited text"},
|
|
)
|
|
assert edit_message_response.status_code == 403
|
|
assert "7 days" in edit_message_response.json().get("detail", "")
|
|
|
|
|
|
async def test_delete_saved_messages_chat_clears_messages_but_keeps_chat(client, db_session):
|
|
user = await _create_verified_user(
|
|
client,
|
|
db_session,
|
|
"saved_clear_user@example.com",
|
|
"saved_clear_user",
|
|
"strongpass123",
|
|
)
|
|
auth = {"Authorization": f"Bearer {user['access_token']}"}
|
|
|
|
saved_chat_response = await client.get("/api/v1/chats/saved", headers=auth)
|
|
assert saved_chat_response.status_code == 200
|
|
saved_chat = saved_chat_response.json()
|
|
saved_chat_id = saved_chat["id"]
|
|
|
|
send_message_response = await client.post(
|
|
"/api/v1/messages",
|
|
headers=auth,
|
|
json={"chat_id": saved_chat_id, "type": "text", "text": "saved note"},
|
|
)
|
|
assert send_message_response.status_code == 201
|
|
|
|
delete_chat_response = await client.delete(f"/api/v1/chats/{saved_chat_id}", headers=auth)
|
|
assert delete_chat_response.status_code == 204
|
|
|
|
saved_chat_after_delete = await client.get("/api/v1/chats/saved", headers=auth)
|
|
assert saved_chat_after_delete.status_code == 200
|
|
assert saved_chat_after_delete.json()["id"] == saved_chat_id
|
|
|
|
messages_after_delete = await client.get(f"/api/v1/messages/{saved_chat_id}", headers=auth)
|
|
assert messages_after_delete.status_code == 200
|
|
assert messages_after_delete.json() == []
|
|
|
|
|
|
async def test_chat_list_hides_duplicate_saved_chats_and_returns_single_saved_entry(client, db_session):
|
|
user = await _create_verified_user(
|
|
client,
|
|
db_session,
|
|
"saved_duplicate_user@example.com",
|
|
"saved_duplicate_user",
|
|
"strongpass123",
|
|
)
|
|
auth = {"Authorization": f"Bearer {user['access_token']}"}
|
|
me_response = await client.get("/api/v1/auth/me", headers=auth)
|
|
user_id = me_response.json()["id"]
|
|
|
|
# Force-create an extra saved chat row to emulate historical duplicate data.
|
|
duplicate_saved_chat = Chat(
|
|
type=ChatType.PRIVATE,
|
|
title="Saved Messages",
|
|
description="Personal cloud chat",
|
|
is_public=False,
|
|
is_saved=True,
|
|
)
|
|
db_session.add(duplicate_saved_chat)
|
|
await db_session.flush()
|
|
db_session.add(
|
|
ChatMember(chat_id=duplicate_saved_chat.id, user_id=user_id, role=ChatMemberRole.OWNER)
|
|
)
|
|
await db_session.commit()
|
|
|
|
chats_response = await client.get("/api/v1/chats", headers=auth)
|
|
assert chats_response.status_code == 200
|
|
rows = chats_response.json()
|
|
saved_rows = [row for row in rows if row.get("is_saved") is True]
|
|
assert len(saved_rows) == 1
|
|
|
|
|
|
async def test_chat_list_includes_notification_muted_flag(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "muted_flag_u1@example.com", "muted_flag_u1", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "muted_flag_u2@example.com", "muted_flag_u2", "strongpass123")
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
create_chat_response = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_response.status_code == 200
|
|
chat_id = create_chat_response.json()["id"]
|
|
|
|
mute_response = await client.put(
|
|
f"/api/v1/chats/{chat_id}/notifications",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"muted": True},
|
|
)
|
|
assert mute_response.status_code == 200
|
|
assert mute_response.json()["muted"] is True
|
|
|
|
chats_response = await client.get("/api/v1/chats", headers={"Authorization": f"Bearer {u1['access_token']}"})
|
|
assert chats_response.status_code == 200
|
|
rows = chats_response.json()
|
|
row = next((item for item in rows if item["id"] == chat_id), None)
|
|
assert row is not None
|
|
assert row["muted"] is True
|
|
|
|
|
|
async def test_media_upload_url_accepts_mp4_voice_mime_types(client, db_session):
|
|
user = await _create_verified_user(
|
|
client,
|
|
db_session,
|
|
"media_mime_user@example.com",
|
|
"media_mime_user",
|
|
"strongpass123",
|
|
)
|
|
auth = {"Authorization": f"Bearer {user['access_token']}"}
|
|
|
|
mp4_response = await client.post(
|
|
"/api/v1/media/upload-url",
|
|
headers=auth,
|
|
json={"file_name": "voice.m4a", "file_type": "audio/mp4", "file_size": 2048},
|
|
)
|
|
assert mp4_response.status_code == 200
|
|
assert "upload_url" in mp4_response.json()
|
|
|
|
m4a_response = await client.post(
|
|
"/api/v1/media/upload-url",
|
|
headers=auth,
|
|
json={"file_name": "voice-alt.m4a", "file_type": "audio/x-m4a", "file_size": 2048},
|
|
)
|
|
assert m4a_response.status_code == 200
|
|
assert "upload_url" in m4a_response.json()
|
|
|
|
|
|
async def test_archive_and_pin_chat_are_user_scoped(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "scope_u1@example.com", "scope_u1", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "scope_u2@example.com", "scope_u2", "strongpass123")
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
create_chat_response = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_response.status_code == 200
|
|
chat_id = create_chat_response.json()["id"]
|
|
|
|
pin_response = await client.post(
|
|
f"/api/v1/chats/{chat_id}/pin-chat",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert pin_response.status_code == 200
|
|
assert pin_response.json()["pinned"] is True
|
|
|
|
u1_chats_after_pin = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert u1_chats_after_pin.status_code == 200
|
|
u1_row_after_pin = next((item for item in u1_chats_after_pin.json() if item["id"] == chat_id), None)
|
|
assert u1_row_after_pin is not None
|
|
assert u1_row_after_pin["pinned"] is True
|
|
|
|
u2_chats_after_u1_pin = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
)
|
|
assert u2_chats_after_u1_pin.status_code == 200
|
|
u2_row_after_u1_pin = next((item for item in u2_chats_after_u1_pin.json() if item["id"] == chat_id), None)
|
|
assert u2_row_after_u1_pin is not None
|
|
assert u2_row_after_u1_pin["pinned"] is False
|
|
|
|
archive_response = await client.post(
|
|
f"/api/v1/chats/{chat_id}/archive",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert archive_response.status_code == 200
|
|
assert archive_response.json()["archived"] is True
|
|
|
|
u1_active_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert u1_active_chats.status_code == 200
|
|
assert all(item["id"] != chat_id for item in u1_active_chats.json())
|
|
|
|
u1_archived_chats = await client.get(
|
|
"/api/v1/chats",
|
|
params={"archived": True},
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert u1_archived_chats.status_code == 200
|
|
u1_archived_row = next((item for item in u1_archived_chats.json() if item["id"] == chat_id), None)
|
|
assert u1_archived_row is not None
|
|
assert u1_archived_row["archived"] is True
|
|
|
|
u2_active_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
)
|
|
assert u2_active_chats.status_code == 200
|
|
u2_row_after_u1_archive = next((item for item in u2_active_chats.json() if item["id"] == chat_id), None)
|
|
assert u2_row_after_u1_archive is not None
|
|
assert u2_row_after_u1_archive["archived"] is False
|
|
|
|
|
|
async def test_create_private_chat_is_visible_to_other_member_in_chat_list(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "sync_create_u1@example.com", "sync_create_u1", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "sync_create_u2@example.com", "sync_create_u2", "strongpass123")
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
create_chat_response = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_response.status_code == 200
|
|
chat_id = create_chat_response.json()["id"]
|
|
|
|
u2_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
)
|
|
assert u2_chats.status_code == 200
|
|
assert any(item["id"] == chat_id for item in u2_chats.json())
|
|
|
|
|
|
async def test_clear_chat_hides_messages_only_for_requesting_user(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "sync_clear_u1@example.com", "sync_clear_u1", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "sync_clear_u2@example.com", "sync_clear_u2", "strongpass123")
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
create_chat_response = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_response.status_code == 200
|
|
chat_id = create_chat_response.json()["id"]
|
|
|
|
send_message_response = await client.post(
|
|
"/api/v1/messages",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"chat_id": chat_id, "type": "text", "text": "sync clear message"},
|
|
)
|
|
assert send_message_response.status_code == 201
|
|
|
|
clear_response = await client.post(
|
|
f"/api/v1/chats/{chat_id}/clear",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert clear_response.status_code == 204
|
|
|
|
u1_messages = await client.get(
|
|
f"/api/v1/messages/{chat_id}",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
)
|
|
assert u1_messages.status_code == 200
|
|
assert u1_messages.json() == []
|
|
|
|
u2_messages = await client.get(
|
|
f"/api/v1/messages/{chat_id}",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
)
|
|
assert u2_messages.status_code == 200
|
|
assert len(u2_messages.json()) == 1
|
|
assert u2_messages.json()[0]["text"] == "sync clear message"
|
|
|
|
|
|
async def test_private_chat_respects_contacts_only_policy(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "pm_u1@example.com", "pm_user_one", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "pm_u2@example.com", "pm_user_two", "strongpass123")
|
|
|
|
me_u1 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u1['access_token']}"})
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u1_id = me_u1.json()["id"]
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
update_privacy = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
json={"privacy_private_messages": "contacts"},
|
|
)
|
|
assert update_privacy.status_code == 200
|
|
|
|
create_chat_blocked = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_blocked.status_code == 403
|
|
|
|
add_contact = await client.post(
|
|
f"/api/v1/users/{u1_id}/contacts",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
)
|
|
assert add_contact.status_code == 204
|
|
|
|
create_chat_allowed = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_allowed.status_code == 200
|
|
|
|
|
|
async def test_private_chat_respects_nobody_policy(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "pm_nobody_u1@example.com", "pm_nobody_u1", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "pm_nobody_u2@example.com", "pm_nobody_u2", "strongpass123")
|
|
|
|
me_u1 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u1['access_token']}"})
|
|
u1_id = me_u1.json()["id"]
|
|
|
|
set_nobody = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
json={"privacy_private_messages": "nobody"},
|
|
)
|
|
assert set_nobody.status_code == 200
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
create_chat_blocked = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_blocked.status_code == 403
|
|
|
|
add_contact = await client.post(
|
|
f"/api/v1/users/{u1_id}/contacts",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
)
|
|
assert add_contact.status_code == 204
|
|
|
|
create_chat_still_blocked = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_still_blocked.status_code == 403
|
|
|
|
|
|
async def test_private_chat_respects_everyone_policy_without_contacts(client, db_session):
|
|
u1 = await _create_verified_user(client, db_session, "pm_everyone_u1@example.com", "pm_everyone_u1", "strongpass123")
|
|
u2 = await _create_verified_user(client, db_session, "pm_everyone_u2@example.com", "pm_everyone_u2", "strongpass123")
|
|
|
|
me_u2 = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {u2['access_token']}"})
|
|
u2_id = me_u2.json()["id"]
|
|
|
|
set_everyone = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {u2['access_token']}"},
|
|
json={"privacy_private_messages": "everyone"},
|
|
)
|
|
assert set_everyone.status_code == 200
|
|
|
|
create_chat_allowed = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {u1['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [u2_id]},
|
|
)
|
|
assert create_chat_allowed.status_code == 200
|
|
|
|
|
|
async def test_group_ban_blocks_rejoin(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "ban_owner@example.com", "ban_owner", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "ban_member@example.com", "ban_member", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Test group", "member_ids": [member_id], "is_public": True, "handle": "ban_test_group"},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
ban_response = await client.post(
|
|
f"/api/v1/chats/{chat_id}/bans/{member_id}",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert ban_response.status_code == 204
|
|
|
|
rejoin_response = await client.post(
|
|
f"/api/v1/chats/{chat_id}/join",
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert rejoin_response.status_code == 403
|
|
|
|
|
|
async def test_group_ban_list_visible_to_admin_and_hidden_from_member(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "ban_list_owner@example.com", "ban_list_owner", "strongpass123")
|
|
admin = await _create_verified_user(client, db_session, "ban_list_admin@example.com", "ban_list_admin", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "ban_list_member@example.com", "ban_list_member", "strongpass123")
|
|
target = await _create_verified_user(client, db_session, "ban_list_target@example.com", "ban_list_target", "strongpass123")
|
|
|
|
me_admin = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {admin['access_token']}"})
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
me_target = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {target['access_token']}"})
|
|
admin_id = me_admin.json()["id"]
|
|
member_id = me_member.json()["id"]
|
|
target_id = me_target.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Ban list group", "member_ids": [admin_id, member_id, target_id]},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
promote_admin = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/members/{admin_id}/role",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"role": "admin"},
|
|
)
|
|
assert promote_admin.status_code == 200
|
|
|
|
ban_target = await client.post(
|
|
f"/api/v1/chats/{chat_id}/bans/{target_id}",
|
|
headers={"Authorization": f"Bearer {admin['access_token']}"},
|
|
)
|
|
assert ban_target.status_code == 204
|
|
|
|
list_by_admin = await client.get(
|
|
f"/api/v1/chats/{chat_id}/bans",
|
|
headers={"Authorization": f"Bearer {admin['access_token']}"},
|
|
)
|
|
assert list_by_admin.status_code == 200
|
|
bans = list_by_admin.json()
|
|
assert any(item["user_id"] == target_id and item["banned_by_user_id"] == admin_id for item in bans)
|
|
|
|
list_by_member = await client.get(
|
|
f"/api/v1/chats/{chat_id}/bans",
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert list_by_member.status_code == 403
|
|
|
|
|
|
async def test_channel_member_delete_chat_behaves_as_leave(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "channel_owner@example.com", "channel_owner", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "channel_member@example.com", "channel_member", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_channel = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.CHANNEL.value, "title": "Test channel", "member_ids": [member_id]},
|
|
)
|
|
assert create_channel.status_code == 200
|
|
chat_id = create_channel.json()["id"]
|
|
|
|
delete_by_member = await client.delete(
|
|
f"/api/v1/chats/{chat_id}",
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert delete_by_member.status_code == 204
|
|
|
|
member_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert member_chats.status_code == 200
|
|
assert all(chat["id"] != chat_id for chat in member_chats.json())
|
|
|
|
owner_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert owner_chats.status_code == 200
|
|
assert any(chat["id"] == chat_id for chat in owner_chats.json())
|
|
|
|
|
|
async def test_channel_member_cannot_delete_for_all(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "channel_owner2@example.com", "channel_owner2", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "channel_member2@example.com", "channel_member2", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_channel = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.CHANNEL.value, "title": "Test channel 2", "member_ids": [member_id]},
|
|
)
|
|
assert create_channel.status_code == 200
|
|
chat_id = create_channel.json()["id"]
|
|
|
|
delete_for_all_by_member = await client.delete(
|
|
f"/api/v1/chats/{chat_id}",
|
|
params={"for_all": True},
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert delete_for_all_by_member.status_code == 403
|
|
|
|
|
|
async def test_channel_member_is_read_only_for_posting(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "channel_post_owner@example.com", "channel_post_owner", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "channel_post_member@example.com", "channel_post_member", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_channel = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.CHANNEL.value, "title": "Read only channel", "member_ids": [member_id]},
|
|
)
|
|
assert create_channel.status_code == 200
|
|
chat_id = create_channel.json()["id"]
|
|
|
|
member_post = await client.post(
|
|
"/api/v1/messages",
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
json={"chat_id": chat_id, "type": "text", "text": "member post"},
|
|
)
|
|
assert member_post.status_code == 403
|
|
|
|
owner_post = await client.post(
|
|
"/api/v1/messages",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"chat_id": chat_id, "type": "text", "text": "owner post"},
|
|
)
|
|
assert owner_post.status_code == 201
|
|
|
|
|
|
async def test_channel_admin_can_delete_channel_for_all(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "channel_admin_owner@example.com", "channel_admin_owner", "strongpass123")
|
|
admin_user = await _create_verified_user(client, db_session, "channel_admin_user@example.com", "channel_admin_user", "strongpass123")
|
|
|
|
me_admin = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {admin_user['access_token']}"})
|
|
admin_id = me_admin.json()["id"]
|
|
|
|
create_channel = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.CHANNEL.value, "title": "Admin deletable channel", "member_ids": [admin_id]},
|
|
)
|
|
assert create_channel.status_code == 200
|
|
chat_id = create_channel.json()["id"]
|
|
|
|
promote_admin = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/members/{admin_id}/role",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"role": "admin"},
|
|
)
|
|
assert promote_admin.status_code == 200
|
|
|
|
delete_by_admin = await client.delete(
|
|
f"/api/v1/chats/{chat_id}",
|
|
headers={"Authorization": f"Bearer {admin_user['access_token']}"},
|
|
)
|
|
assert delete_by_admin.status_code == 204
|
|
|
|
owner_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert owner_chats.status_code == 200
|
|
assert all(chat["id"] != chat_id for chat in owner_chats.json())
|
|
|
|
|
|
async def test_channel_message_delete_for_all_permissions(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "channel_msg_owner@example.com", "channel_msg_owner", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "channel_msg_member@example.com", "channel_msg_member", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_channel = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.CHANNEL.value, "title": "Channel message permissions", "member_ids": [member_id]},
|
|
)
|
|
assert create_channel.status_code == 200
|
|
chat_id = create_channel.json()["id"]
|
|
|
|
owner_message = await client.post(
|
|
"/api/v1/messages",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"chat_id": chat_id, "type": "text", "text": "to be deleted"},
|
|
)
|
|
assert owner_message.status_code == 201
|
|
message_id = owner_message.json()["id"]
|
|
|
|
member_delete_for_all = await client.delete(
|
|
f"/api/v1/messages/{message_id}",
|
|
params={"for_all": True},
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert member_delete_for_all.status_code == 403
|
|
|
|
promote_admin = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/members/{member_id}/role",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"role": "admin"},
|
|
)
|
|
assert promote_admin.status_code == 200
|
|
|
|
admin_delete_for_all = await client.delete(
|
|
f"/api/v1/messages/{message_id}",
|
|
params={"for_all": True},
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert admin_delete_for_all.status_code == 204
|
|
|
|
owner_messages = await client.get(
|
|
f"/api/v1/messages/{chat_id}",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert owner_messages.status_code == 200
|
|
assert all(item["id"] != message_id for item in owner_messages.json())
|
|
|
|
|
|
async def test_group_member_cannot_edit_chat_profile(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "group_profile_owner@example.com", "group_profile_owner", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "group_profile_member@example.com", "group_profile_member", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Editable group", "member_ids": [member_id]},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
member_edit = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/profile",
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
json={"title": "Member changed title"},
|
|
)
|
|
assert member_edit.status_code == 403
|
|
|
|
|
|
async def test_group_admin_can_edit_chat_profile(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "group_profile_owner2@example.com", "group_profile_owner2", "strongpass123")
|
|
admin_user = await _create_verified_user(client, db_session, "group_profile_admin2@example.com", "group_profile_admin2", "strongpass123")
|
|
|
|
me_admin = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {admin_user['access_token']}"})
|
|
admin_id = me_admin.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Admin editable group", "member_ids": [admin_id]},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
promote_admin = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/members/{admin_id}/role",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"role": "admin"},
|
|
)
|
|
assert promote_admin.status_code == 200
|
|
|
|
admin_edit = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/profile",
|
|
headers={"Authorization": f"Bearer {admin_user['access_token']}"},
|
|
json={"title": "Admin changed title", "description": "Updated by admin"},
|
|
)
|
|
assert admin_edit.status_code == 200
|
|
body = admin_edit.json()
|
|
assert body["title"] == "Admin changed title"
|
|
assert body["description"] == "Updated by admin"
|
|
|
|
|
|
async def test_chat_members_endpoint_returns_user_profile_fields(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "members_owner@example.com", "members_owner", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "members_user@example.com", "members_user", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Members profile group", "member_ids": [member_id]},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
members_response = await client.get(
|
|
f"/api/v1/chats/{chat_id}/members",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert members_response.status_code == 200
|
|
rows = members_response.json()
|
|
member_row = next((item for item in rows if item["user_id"] == member_id), None)
|
|
assert member_row is not None
|
|
assert member_row["username"] == "members_user"
|
|
assert "name" in member_row
|
|
assert "avatar_url" in member_row
|
|
|
|
|
|
async def test_group_admin_cannot_change_member_roles(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "roles_owner@example.com", "roles_owner", "strongpass123")
|
|
admin_user = await _create_verified_user(client, db_session, "roles_admin@example.com", "roles_admin", "strongpass123")
|
|
member_user = await _create_verified_user(client, db_session, "roles_member@example.com", "roles_member", "strongpass123")
|
|
|
|
me_admin = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {admin_user['access_token']}"})
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member_user['access_token']}"})
|
|
admin_id = me_admin.json()["id"]
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Role restrictions", "member_ids": [admin_id, member_id]},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
promote_admin = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/members/{admin_id}/role",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"role": "admin"},
|
|
)
|
|
assert promote_admin.status_code == 200
|
|
|
|
admin_try_promote_member = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/members/{member_id}/role",
|
|
headers={"Authorization": f"Bearer {admin_user['access_token']}"},
|
|
json={"role": "admin"},
|
|
)
|
|
assert admin_try_promote_member.status_code == 403
|
|
|
|
|
|
async def test_group_owner_cannot_demote_self_from_owner_role(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "roles_self_owner@example.com", "roles_self_owner", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "roles_self_member@example.com", "roles_self_member", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Owner self demote", "member_ids": [member_id]},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
me_owner = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {owner['access_token']}"})
|
|
owner_id = me_owner.json()["id"]
|
|
owner_try_self_demote = await client.patch(
|
|
f"/api/v1/chats/{chat_id}/members/{owner_id}/role",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"role": "admin"},
|
|
)
|
|
assert owner_try_self_demote.status_code == 422
|
|
|
|
|
|
async def test_group_invite_link_allows_join_by_token(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "invite_owner@example.com", "invite_owner", "strongpass123")
|
|
joiner = await _create_verified_user(client, db_session, "invite_joiner@example.com", "invite_joiner", "strongpass123")
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Invite group", "member_ids": []},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
invite_link = await client.post(
|
|
f"/api/v1/chats/{chat_id}/invite-link",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert invite_link.status_code == 200
|
|
token = invite_link.json()["token"]
|
|
|
|
join_response = await client.post(
|
|
"/api/v1/chats/join-by-invite",
|
|
headers={"Authorization": f"Bearer {joiner['access_token']}"},
|
|
json={"token": token},
|
|
)
|
|
assert join_response.status_code == 200
|
|
assert join_response.json()["id"] == chat_id
|
|
|
|
joiner_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {joiner['access_token']}"},
|
|
)
|
|
assert joiner_chats.status_code == 200
|
|
assert any(chat["id"] == chat_id for chat in joiner_chats.json())
|
|
|
|
|
|
async def test_group_member_cannot_create_invite_link(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "invite_owner2@example.com", "invite_owner2", "strongpass123")
|
|
member = await _create_verified_user(client, db_session, "invite_member2@example.com", "invite_member2", "strongpass123")
|
|
|
|
me_member = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {member['access_token']}"})
|
|
member_id = me_member.json()["id"]
|
|
|
|
create_group = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Invite rights", "member_ids": [member_id]},
|
|
)
|
|
assert create_group.status_code == 200
|
|
chat_id = create_group.json()["id"]
|
|
|
|
member_invite_link = await client.post(
|
|
f"/api/v1/chats/{chat_id}/invite-link",
|
|
headers={"Authorization": f"Bearer {member['access_token']}"},
|
|
)
|
|
assert member_invite_link.status_code == 403
|
|
|
|
|
|
async def test_join_by_invite_with_invalid_token_returns_not_found(client, db_session):
|
|
user = await _create_verified_user(client, db_session, "invite_invalid_user@example.com", "invite_invalid_user", "strongpass123")
|
|
|
|
join_response = await client.post(
|
|
"/api/v1/chats/join-by-invite",
|
|
headers={"Authorization": f"Bearer {user['access_token']}"},
|
|
json={"token": "invalid-token-value"},
|
|
)
|
|
assert join_response.status_code == 404
|
|
|
|
|
|
async def test_group_invite_privacy_contacts_only(client, db_session):
|
|
inviter = await _create_verified_user(client, db_session, "invite_u1@example.com", "invite_u1", "strongpass123")
|
|
target = await _create_verified_user(client, db_session, "invite_u2@example.com", "invite_u2", "strongpass123")
|
|
|
|
me_inviter = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {inviter['access_token']}"})
|
|
me_target = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {target['access_token']}"})
|
|
inviter_id = me_inviter.json()["id"]
|
|
target_id = me_target.json()["id"]
|
|
|
|
set_contacts_only = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {target['access_token']}"},
|
|
json={"privacy_group_invites": "contacts"},
|
|
)
|
|
assert set_contacts_only.status_code == 200
|
|
|
|
create_group_blocked = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {inviter['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Blocked group", "member_ids": [target_id]},
|
|
)
|
|
assert create_group_blocked.status_code == 403
|
|
|
|
add_contact = await client.post(
|
|
f"/api/v1/users/{inviter_id}/contacts",
|
|
headers={"Authorization": f"Bearer {target['access_token']}"},
|
|
)
|
|
assert add_contact.status_code == 204
|
|
|
|
create_group_allowed = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {inviter['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Allowed group", "member_ids": [target_id]},
|
|
)
|
|
assert create_group_allowed.status_code == 200
|
|
|
|
|
|
async def test_group_invite_privacy_nobody_blocks_invites_even_from_contacts(client, db_session):
|
|
inviter = await _create_verified_user(client, db_session, "invite_nobody_u1@example.com", "invite_nobody_u1", "strongpass123")
|
|
target = await _create_verified_user(client, db_session, "invite_nobody_u2@example.com", "invite_nobody_u2", "strongpass123")
|
|
|
|
me_inviter = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {inviter['access_token']}"})
|
|
me_target = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {target['access_token']}"})
|
|
inviter_id = me_inviter.json()["id"]
|
|
target_id = me_target.json()["id"]
|
|
|
|
set_nobody = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {target['access_token']}"},
|
|
json={"privacy_group_invites": "nobody"},
|
|
)
|
|
assert set_nobody.status_code == 200
|
|
|
|
add_contact = await client.post(
|
|
f"/api/v1/users/{inviter_id}/contacts",
|
|
headers={"Authorization": f"Bearer {target['access_token']}"},
|
|
)
|
|
assert add_contact.status_code == 204
|
|
|
|
create_group_blocked = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {inviter['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Nobody blocked group", "member_ids": [target_id]},
|
|
)
|
|
assert create_group_blocked.status_code == 403
|
|
|
|
|
|
async def test_group_invite_privacy_everyone_allows_invites_without_contacts(client, db_session):
|
|
inviter = await _create_verified_user(client, db_session, "invite_everyone_u1@example.com", "invite_everyone_u1", "strongpass123")
|
|
target = await _create_verified_user(client, db_session, "invite_everyone_u2@example.com", "invite_everyone_u2", "strongpass123")
|
|
|
|
me_target = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {target['access_token']}"})
|
|
target_id = me_target.json()["id"]
|
|
|
|
set_everyone = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {target['access_token']}"},
|
|
json={"privacy_group_invites": "everyone"},
|
|
)
|
|
assert set_everyone.status_code == 200
|
|
|
|
create_group_allowed = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {inviter['access_token']}"},
|
|
json={"type": ChatType.GROUP.value, "title": "Everyone allowed group", "member_ids": [target_id]},
|
|
)
|
|
assert create_group_allowed.status_code == 200
|
|
|
|
|
|
async def test_add_contact_by_email_success_and_not_found(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "contact_email_owner@example.com", "contact_email_owner", "strongpass123")
|
|
target = await _create_verified_user(client, db_session, "contact_email_target@example.com", "contact_email_target", "strongpass123")
|
|
|
|
add_contact = await client.post(
|
|
"/api/v1/users/contacts/by-email",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"email": "contact_email_target@example.com"},
|
|
)
|
|
assert add_contact.status_code == 204
|
|
|
|
contacts = await client.get(
|
|
"/api/v1/users/contacts",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert contacts.status_code == 200
|
|
assert any(item["email"] == "contact_email_target@example.com" for item in contacts.json())
|
|
|
|
add_missing_contact = await client.post(
|
|
"/api/v1/users/contacts/by-email",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"email": "unknown_contact@example.com"},
|
|
)
|
|
assert add_missing_contact.status_code == 404
|
|
|
|
|
|
async def test_add_contact_by_email_blocked_conflict(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "contact_email_block_owner@example.com", "contact_email_block_owner", "strongpass123")
|
|
target = await _create_verified_user(client, db_session, "contact_email_block_target@example.com", "contact_email_block_target", "strongpass123")
|
|
|
|
me_owner = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {owner['access_token']}"})
|
|
owner_id = me_owner.json()["id"]
|
|
|
|
block_owner = await client.post(
|
|
f"/api/v1/users/{owner_id}/block",
|
|
headers={"Authorization": f"Bearer {target['access_token']}"},
|
|
)
|
|
assert block_owner.status_code == 204
|
|
|
|
add_contact_blocked = await client.post(
|
|
"/api/v1/users/contacts/by-email",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"email": "contact_email_block_target@example.com"},
|
|
)
|
|
assert add_contact_blocked.status_code == 409
|
|
|
|
|
|
async def test_avatar_privacy_hidden_from_other_users_search(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "avatar_owner@example.com", "avatar_owner", "strongpass123")
|
|
viewer = await _create_verified_user(client, db_session, "avatar_viewer@example.com", "avatar_viewer", "strongpass123")
|
|
|
|
set_avatar_and_privacy = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"avatar_url": "https://cdn.example.com/avatar-owner.png", "privacy_avatar": "nobody"},
|
|
)
|
|
assert set_avatar_and_privacy.status_code == 200
|
|
assert set_avatar_and_privacy.json()["avatar_url"] == "https://cdn.example.com/avatar-owner.png"
|
|
|
|
search_response = await client.get(
|
|
"/api/v1/users/search",
|
|
params={"query": "avatar_owner", "limit": 20},
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert search_response.status_code == 200
|
|
rows = search_response.json()
|
|
owner_row = next((item for item in rows if item["username"] == "avatar_owner"), None)
|
|
assert owner_row is not None
|
|
assert owner_row["avatar_url"] is None
|
|
|
|
|
|
async def test_private_chat_hides_counterpart_presence_and_avatar_by_privacy(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "privacy_owner@example.com", "privacy_owner", "strongpass123")
|
|
viewer = await _create_verified_user(client, db_session, "privacy_viewer@example.com", "privacy_viewer", "strongpass123")
|
|
|
|
me_owner = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {owner['access_token']}"})
|
|
owner_id = me_owner.json()["id"]
|
|
|
|
set_privacy = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={
|
|
"avatar_url": "https://cdn.example.com/privacy-owner.png",
|
|
"privacy_avatar": "nobody",
|
|
"privacy_last_seen": "nobody",
|
|
},
|
|
)
|
|
assert set_privacy.status_code == 200
|
|
|
|
create_chat = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [owner_id]},
|
|
)
|
|
assert create_chat.status_code == 200
|
|
chat_id = create_chat.json()["id"]
|
|
|
|
viewer_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert viewer_chats.status_code == 200
|
|
row = next((chat for chat in viewer_chats.json() if chat["id"] == chat_id), None)
|
|
assert row is not None
|
|
assert row["counterpart_avatar_url"] is None
|
|
assert row["counterpart_is_online"] is None
|
|
assert row["counterpart_last_seen_at"] is None
|
|
|
|
|
|
async def test_private_chat_contacts_privacy_reveals_avatar_and_presence_for_allowed_viewer(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "privacy_contacts_owner@example.com", "privacy_contacts_owner", "strongpass123")
|
|
viewer = await _create_verified_user(client, db_session, "privacy_contacts_viewer@example.com", "privacy_contacts_viewer", "strongpass123")
|
|
|
|
me_owner = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {owner['access_token']}"})
|
|
me_viewer = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {viewer['access_token']}"})
|
|
owner_id = me_owner.json()["id"]
|
|
viewer_id = me_viewer.json()["id"]
|
|
|
|
set_privacy = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={
|
|
"avatar_url": "https://cdn.example.com/privacy-contacts-owner.png",
|
|
"privacy_avatar": "contacts",
|
|
"privacy_last_seen": "contacts",
|
|
},
|
|
)
|
|
assert set_privacy.status_code == 200
|
|
|
|
create_chat = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [owner_id]},
|
|
)
|
|
assert create_chat.status_code == 200
|
|
chat_id = create_chat.json()["id"]
|
|
|
|
viewer_chats_before = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert viewer_chats_before.status_code == 200
|
|
row_before = next((chat for chat in viewer_chats_before.json() if chat["id"] == chat_id), None)
|
|
assert row_before is not None
|
|
assert row_before["counterpart_avatar_url"] is None
|
|
assert row_before["counterpart_is_online"] is None
|
|
|
|
add_contact = await client.post(
|
|
f"/api/v1/users/{viewer_id}/contacts",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert add_contact.status_code == 204
|
|
|
|
viewer_chats_after = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert viewer_chats_after.status_code == 200
|
|
row_after = next((chat for chat in viewer_chats_after.json() if chat["id"] == chat_id), None)
|
|
assert row_after is not None
|
|
assert row_after["counterpart_avatar_url"] == "https://cdn.example.com/privacy-contacts-owner.png"
|
|
assert row_after["counterpart_is_online"] is not None
|
|
|
|
|
|
async def test_avatar_privacy_contacts_in_search_visible_only_for_contacts(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "avatar_contacts_owner@example.com", "avatar_contacts_owner", "strongpass123")
|
|
viewer = await _create_verified_user(client, db_session, "avatar_contacts_viewer@example.com", "avatar_contacts_viewer", "strongpass123")
|
|
|
|
me_viewer = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {viewer['access_token']}"})
|
|
viewer_id = me_viewer.json()["id"]
|
|
|
|
set_avatar_and_privacy = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"avatar_url": "https://cdn.example.com/avatar-contacts-owner.png", "privacy_avatar": "contacts"},
|
|
)
|
|
assert set_avatar_and_privacy.status_code == 200
|
|
|
|
search_before_contact = await client.get(
|
|
"/api/v1/users/search",
|
|
params={"query": "avatar_contacts_owner", "limit": 20},
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert search_before_contact.status_code == 200
|
|
rows_before = search_before_contact.json()
|
|
owner_before = next((item for item in rows_before if item["username"] == "avatar_contacts_owner"), None)
|
|
assert owner_before is not None
|
|
assert owner_before["avatar_url"] is None
|
|
|
|
add_contact = await client.post(
|
|
f"/api/v1/users/{viewer_id}/contacts",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
)
|
|
assert add_contact.status_code == 204
|
|
|
|
search_after_contact = await client.get(
|
|
"/api/v1/users/search",
|
|
params={"query": "avatar_contacts_owner", "limit": 20},
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert search_after_contact.status_code == 200
|
|
rows_after = search_after_contact.json()
|
|
owner_after = next((item for item in rows_after if item["username"] == "avatar_contacts_owner"), None)
|
|
assert owner_after is not None
|
|
assert owner_after["avatar_url"] == "https://cdn.example.com/avatar-contacts-owner.png"
|
|
|
|
|
|
async def test_avatar_privacy_everyone_in_search_visible_without_contacts(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "avatar_everyone_owner@example.com", "avatar_everyone_owner", "strongpass123")
|
|
viewer = await _create_verified_user(client, db_session, "avatar_everyone_viewer@example.com", "avatar_everyone_viewer", "strongpass123")
|
|
|
|
set_avatar_and_privacy = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={"avatar_url": "https://cdn.example.com/avatar-everyone-owner.png", "privacy_avatar": "everyone"},
|
|
)
|
|
assert set_avatar_and_privacy.status_code == 200
|
|
|
|
search_response = await client.get(
|
|
"/api/v1/users/search",
|
|
params={"query": "avatar_everyone_owner", "limit": 20},
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert search_response.status_code == 200
|
|
rows = search_response.json()
|
|
owner_row = next((item for item in rows if item["username"] == "avatar_everyone_owner"), None)
|
|
assert owner_row is not None
|
|
assert owner_row["avatar_url"] == "https://cdn.example.com/avatar-everyone-owner.png"
|
|
|
|
|
|
async def test_private_chat_everyone_privacy_reveals_avatar_and_presence_without_contacts(client, db_session):
|
|
owner = await _create_verified_user(client, db_session, "privacy_everyone_owner@example.com", "privacy_everyone_owner", "strongpass123")
|
|
viewer = await _create_verified_user(client, db_session, "privacy_everyone_viewer@example.com", "privacy_everyone_viewer", "strongpass123")
|
|
|
|
me_owner = await client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {owner['access_token']}"})
|
|
owner_id = me_owner.json()["id"]
|
|
|
|
set_privacy = await client.put(
|
|
"/api/v1/users/profile",
|
|
headers={"Authorization": f"Bearer {owner['access_token']}"},
|
|
json={
|
|
"avatar_url": "https://cdn.example.com/privacy-everyone-owner.png",
|
|
"privacy_avatar": "everyone",
|
|
"privacy_last_seen": "everyone",
|
|
},
|
|
)
|
|
assert set_privacy.status_code == 200
|
|
|
|
create_chat = await client.post(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
json={"type": ChatType.PRIVATE.value, "title": None, "member_ids": [owner_id]},
|
|
)
|
|
assert create_chat.status_code == 200
|
|
chat_id = create_chat.json()["id"]
|
|
|
|
viewer_chats = await client.get(
|
|
"/api/v1/chats",
|
|
headers={"Authorization": f"Bearer {viewer['access_token']}"},
|
|
)
|
|
assert viewer_chats.status_code == 200
|
|
row = next((chat for chat in viewer_chats.json() if chat["id"] == chat_id), None)
|
|
assert row is not None
|
|
assert row["counterpart_avatar_url"] == "https://cdn.example.com/privacy-everyone-owner.png"
|
|
assert row["counterpart_is_online"] is not None
|