838 lines
37 KiB
Python
838 lines
37 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.auth.models import EmailVerificationToken
|
|
from app.chats.models import 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_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_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_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
|