286 lines
8.5 KiB
Python
286 lines
8.5 KiB
Python
"""API tests for contact endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from app.models.contact import Contact
|
|
from app.models.organization_member import OrganizationMember, OrganizationRole
|
|
from app.models.user import User
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
from tests.api.v1.task_activity_shared import auth_headers, make_token, prepare_scenario
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_contacts_supports_search_and_pagination(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
client: AsyncClient,
|
|
) -> None:
|
|
scenario = await prepare_scenario(session_factory)
|
|
token = make_token(scenario.user_id, scenario.user_email)
|
|
|
|
async with session_factory() as session:
|
|
session.add_all(
|
|
[
|
|
Contact(
|
|
organization_id=scenario.organization_id,
|
|
owner_id=scenario.user_id,
|
|
name="Alpha Lead",
|
|
email="alpha@example.com",
|
|
phone=None,
|
|
),
|
|
Contact(
|
|
organization_id=scenario.organization_id,
|
|
owner_id=scenario.user_id,
|
|
name="Beta Prospect",
|
|
email="beta@example.com",
|
|
phone=None,
|
|
),
|
|
],
|
|
)
|
|
await session.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/contacts/?page=1&page_size=10&search=alpha",
|
|
headers=auth_headers(token, scenario),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 1
|
|
assert data[0]["name"] == "Alpha Lead"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_contact_returns_created_payload(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
client: AsyncClient,
|
|
) -> None:
|
|
scenario = await prepare_scenario(session_factory)
|
|
token = make_token(scenario.user_id, scenario.user_email)
|
|
|
|
response = await client.post(
|
|
"/api/v1/contacts/",
|
|
json={
|
|
"name": "New Contact",
|
|
"email": "new@example.com",
|
|
"phone": "+123",
|
|
"owner_id": scenario.user_id,
|
|
},
|
|
headers=auth_headers(token, scenario),
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
payload = response.json()
|
|
assert payload["name"] == "New Contact"
|
|
assert payload["email"] == "new@example.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_member_cannot_assign_foreign_owner(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
client: AsyncClient,
|
|
) -> None:
|
|
scenario = await prepare_scenario(session_factory)
|
|
token = make_token(scenario.user_id, scenario.user_email)
|
|
|
|
async with session_factory() as session:
|
|
membership = await session.scalar(
|
|
select(OrganizationMember).where(
|
|
OrganizationMember.organization_id == scenario.organization_id,
|
|
OrganizationMember.user_id == scenario.user_id,
|
|
),
|
|
)
|
|
assert membership is not None
|
|
membership.role = OrganizationRole.MEMBER
|
|
|
|
other_user = User(
|
|
email="manager@example.com",
|
|
hashed_password="hashed",
|
|
name="Manager",
|
|
is_active=True,
|
|
)
|
|
session.add(other_user)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
OrganizationMember(
|
|
organization_id=scenario.organization_id,
|
|
user_id=other_user.id,
|
|
role=OrganizationRole.ADMIN,
|
|
),
|
|
)
|
|
await session.commit()
|
|
|
|
response = await client.post(
|
|
"/api/v1/contacts/",
|
|
json={
|
|
"name": "Blocked",
|
|
"email": "blocked@example.com",
|
|
"owner_id": scenario.user_id + 1,
|
|
},
|
|
headers=auth_headers(token, scenario),
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_member_can_view_foreign_contacts(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
client: AsyncClient,
|
|
) -> None:
|
|
scenario = await prepare_scenario(session_factory)
|
|
token = make_token(scenario.user_id, scenario.user_email)
|
|
|
|
async with session_factory() as session:
|
|
membership = await session.scalar(
|
|
select(OrganizationMember).where(
|
|
OrganizationMember.organization_id == scenario.organization_id,
|
|
OrganizationMember.user_id == scenario.user_id,
|
|
),
|
|
)
|
|
assert membership is not None
|
|
membership.role = OrganizationRole.MEMBER
|
|
|
|
other_user = User(
|
|
email="viewer@example.com",
|
|
hashed_password="hashed",
|
|
name="Viewer",
|
|
is_active=True,
|
|
)
|
|
session.add(other_user)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
OrganizationMember(
|
|
organization_id=scenario.organization_id,
|
|
user_id=other_user.id,
|
|
role=OrganizationRole.MANAGER,
|
|
),
|
|
)
|
|
|
|
session.add(
|
|
Contact(
|
|
organization_id=scenario.organization_id,
|
|
owner_id=other_user.id,
|
|
name="Foreign Owner",
|
|
email="foreign@example.com",
|
|
phone=None,
|
|
),
|
|
)
|
|
await session.commit()
|
|
|
|
response = await client.get(
|
|
"/api/v1/contacts/",
|
|
headers=auth_headers(token, scenario),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
names = {contact["name"] for contact in response.json()}
|
|
assert {"John Doe", "Foreign Owner"}.issubset(names)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_member_patch_foreign_contact_forbidden(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
client: AsyncClient,
|
|
) -> None:
|
|
scenario = await prepare_scenario(session_factory)
|
|
token = make_token(scenario.user_id, scenario.user_email)
|
|
|
|
async with session_factory() as session:
|
|
membership = await session.scalar(
|
|
select(OrganizationMember).where(
|
|
OrganizationMember.organization_id == scenario.organization_id,
|
|
OrganizationMember.user_id == scenario.user_id,
|
|
),
|
|
)
|
|
assert membership is not None
|
|
membership.role = OrganizationRole.MEMBER
|
|
|
|
other_user = User(
|
|
email="owner2@example.com",
|
|
hashed_password="hashed",
|
|
name="Owner2",
|
|
is_active=True,
|
|
)
|
|
session.add(other_user)
|
|
await session.flush()
|
|
|
|
session.add(
|
|
OrganizationMember(
|
|
organization_id=scenario.organization_id,
|
|
user_id=other_user.id,
|
|
role=OrganizationRole.MANAGER,
|
|
),
|
|
)
|
|
|
|
contact = Contact(
|
|
organization_id=scenario.organization_id,
|
|
owner_id=other_user.id,
|
|
name="Locked Contact",
|
|
email="locked@example.com",
|
|
phone=None,
|
|
)
|
|
session.add(contact)
|
|
await session.commit()
|
|
contact_id = contact.id
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/contacts/{contact_id}",
|
|
json={"name": "Hacked"},
|
|
headers=auth_headers(token, scenario),
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_contact_updates_fields(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
client: AsyncClient,
|
|
) -> None:
|
|
scenario = await prepare_scenario(session_factory)
|
|
token = make_token(scenario.user_id, scenario.user_email)
|
|
|
|
async with session_factory() as session:
|
|
contact = Contact(
|
|
organization_id=scenario.organization_id,
|
|
owner_id=scenario.user_id,
|
|
name="Old Name",
|
|
email="old@example.com",
|
|
phone="+111",
|
|
)
|
|
session.add(contact)
|
|
await session.commit()
|
|
contact_id = contact.id
|
|
|
|
response = await client.patch(
|
|
f"/api/v1/contacts/{contact_id}",
|
|
json={"name": "Updated", "phone": None},
|
|
headers=auth_headers(token, scenario),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["name"] == "Updated"
|
|
assert payload["phone"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_contact_with_deals_returns_conflict(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
client: AsyncClient,
|
|
) -> None:
|
|
scenario = await prepare_scenario(session_factory)
|
|
token = make_token(scenario.user_id, scenario.user_email)
|
|
|
|
response = await client.delete(
|
|
f"/api/v1/contacts/{scenario.contact_id}",
|
|
headers=auth_headers(token, scenario),
|
|
)
|
|
|
|
assert response.status_code == 409
|