108 lines
2.8 KiB
Python
108 lines
2.8 KiB
Python
"""Shared helpers for task and activity API tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import timedelta
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from app.core.security import jwt_service
|
|
from app.models.contact import Contact
|
|
from app.models.deal import Deal
|
|
from app.models.organization import Organization
|
|
from app.models.organization_member import OrganizationMember, OrganizationRole
|
|
from app.models.user import User
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Scenario:
|
|
"""Captures seeded entities for API tests."""
|
|
|
|
user_id: int
|
|
user_email: str
|
|
organization_id: int
|
|
contact_id: int
|
|
deal_id: int
|
|
|
|
|
|
async def prepare_scenario(session_factory: async_sessionmaker[AsyncSession]) -> Scenario:
|
|
async with session_factory() as session:
|
|
user = User(
|
|
email="owner@example.com",
|
|
hashed_password="hashed",
|
|
name="Owner",
|
|
is_active=True,
|
|
)
|
|
org = Organization(name="Acme LLC")
|
|
session.add_all([user, org])
|
|
await session.flush()
|
|
|
|
membership = OrganizationMember(
|
|
organization_id=org.id,
|
|
user_id=user.id,
|
|
role=OrganizationRole.OWNER,
|
|
)
|
|
session.add(membership)
|
|
|
|
contact = Contact(
|
|
organization_id=org.id,
|
|
owner_id=user.id,
|
|
name="John Doe",
|
|
email="john@example.com",
|
|
)
|
|
session.add(contact)
|
|
await session.flush()
|
|
|
|
deal = Deal(
|
|
organization_id=org.id,
|
|
contact_id=contact.id,
|
|
owner_id=user.id,
|
|
title="Website redesign",
|
|
amount=None,
|
|
)
|
|
session.add(deal)
|
|
await session.commit()
|
|
|
|
return Scenario(
|
|
user_id=user.id,
|
|
user_email=user.email,
|
|
organization_id=org.id,
|
|
contact_id=contact.id,
|
|
deal_id=deal.id,
|
|
)
|
|
|
|
|
|
async def create_deal(
|
|
session_factory: async_sessionmaker[AsyncSession],
|
|
*,
|
|
scenario: Scenario,
|
|
title: str,
|
|
) -> int:
|
|
async with session_factory() as session:
|
|
deal = Deal(
|
|
organization_id=scenario.organization_id,
|
|
contact_id=scenario.contact_id,
|
|
owner_id=scenario.user_id,
|
|
title=title,
|
|
amount=None,
|
|
)
|
|
session.add(deal)
|
|
await session.commit()
|
|
return deal.id
|
|
|
|
|
|
def auth_headers(token: str, scenario: Scenario) -> dict[str, str]:
|
|
return {
|
|
"Authorization": f"Bearer {token}",
|
|
"X-Organization-Id": str(scenario.organization_id),
|
|
}
|
|
|
|
|
|
def make_token(user_id: int, email: str) -> str:
|
|
return jwt_service.create_access_token(
|
|
subject=str(user_id),
|
|
expires_delta=timedelta(minutes=30),
|
|
claims={"email": email},
|
|
)
|