feat: add unit and API tests for activities and tasks, including shared fixtures and scenarios

tests/api/v1/task_activity_shared.py

@@ -0,0 +1,101 @@
+"""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},
+    )