AFFAANhandClaude Opus 5
f6a54d6f93
Independent FastAPI backend for Maskan CRM. Owns contacts, organizations, leads, pipelines, activities, products, quotes, users, permissions, audit records and first-party integration credentials, with Alembic migrations against PostgreSQL. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
131 lines
3.7 KiB
Python
131 lines
3.7 KiB
Python
from collections.abc import Generator
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, event, select
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.core.database import Base, get_db
|
|
from app.core.security import hash_password
|
|
from app.main import app
|
|
from app.models import LeadSource, LeadType, Pipeline, Stage, Tenant, User
|
|
|
|
TEST_PASSWORD = "StrongPassword123!"
|
|
engine = create_engine(
|
|
"sqlite+pysqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
TestingSession = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
|
|
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def enable_sqlite_foreign_keys(connection, _record) -> None:
|
|
cursor = connection.cursor()
|
|
cursor.execute("PRAGMA foreign_keys=ON")
|
|
cursor.close()
|
|
|
|
|
|
@pytest.fixture()
|
|
def db() -> Generator[Session, None, None]:
|
|
Base.metadata.create_all(bind=engine)
|
|
database = TestingSession()
|
|
try:
|
|
seed_test_workspace(database)
|
|
yield database
|
|
finally:
|
|
database.close()
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
|
|
def seed_test_workspace(db: Session) -> None:
|
|
tenant = Tenant(slug="maskan", name="Maskan Technologies")
|
|
db.add(tenant)
|
|
db.flush()
|
|
owner = User(
|
|
tenant_id=tenant.id,
|
|
email="owner@example.com",
|
|
full_name="Test Owner",
|
|
password_hash=hash_password(TEST_PASSWORD),
|
|
role="owner",
|
|
)
|
|
viewer = User(
|
|
tenant_id=tenant.id,
|
|
email="viewer@example.com",
|
|
full_name="Test Viewer",
|
|
password_hash=hash_password(TEST_PASSWORD),
|
|
role="viewer",
|
|
)
|
|
db.add_all([owner, viewer])
|
|
pipeline = Pipeline(tenant_id=tenant.id, name="Sales Pipeline", is_default=True)
|
|
db.add(pipeline)
|
|
db.flush()
|
|
db.add_all(
|
|
[
|
|
Stage(
|
|
tenant_id=tenant.id,
|
|
pipeline_id=pipeline.id,
|
|
name="New",
|
|
position=1,
|
|
probability=10,
|
|
color="#64748b",
|
|
),
|
|
Stage(
|
|
tenant_id=tenant.id,
|
|
pipeline_id=pipeline.id,
|
|
name="Qualified",
|
|
position=2,
|
|
probability=40,
|
|
color="#2563eb",
|
|
),
|
|
LeadSource(tenant_id=tenant.id, name="LinkedIn"),
|
|
LeadType(tenant_id=tenant.id, name="New Business"),
|
|
],
|
|
)
|
|
db.commit()
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(db: Session) -> Generator[TestClient, None, None]:
|
|
def override_db():
|
|
yield db
|
|
|
|
app.dependency_overrides[get_db] = override_db
|
|
with TestClient(app) as test_client:
|
|
yield test_client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def login(client: TestClient, email: str = "owner@example.com") -> dict[str, str]:
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"workspace": "maskan",
|
|
"email": email,
|
|
"password": TEST_PASSWORD,
|
|
},
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return {"Authorization": f"Bearer {response.json()['access_token']}"}
|
|
|
|
|
|
@pytest.fixture()
|
|
def auth_headers(client: TestClient) -> dict[str, str]:
|
|
return login(client)
|
|
|
|
|
|
@pytest.fixture()
|
|
def viewer_headers(client: TestClient) -> dict[str, str]:
|
|
return login(client, "viewer@example.com")
|
|
|
|
|
|
@pytest.fixture()
|
|
def pipeline_ids(db: Session) -> tuple[str, str, str]:
|
|
pipeline = db.scalar(select(Pipeline).where(Pipeline.is_default.is_(True)))
|
|
stages = db.scalars(
|
|
select(Stage).where(Stage.pipeline_id == pipeline.id).order_by(Stage.position),
|
|
).all()
|
|
return pipeline.id, stages[0].id, stages[1].id
|
|
|