TASK-001: Setup FastAPI project structure

This commit is contained in:
OpenClaw Agent
2026-04-16 04:40:40 +00:00
parent a3eca3b7da
commit ef5b32143a
26 changed files with 399 additions and 415 deletions

View File

@@ -1,129 +1,49 @@
"""Pytest fixtures for SocialPhoto tests."""
import os
import sqlite3
from pathlib import Path
from typing import Generator
"""Pytest configuration and fixtures."""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
# Set test database path before importing app
TEST_DB = Path(__file__).parent / "test_socialphoto.db"
from app.main import app
from app.db.database import Base, get_db
def override_db_path():
"""Override database path for tests."""
return TEST_DB
# Create in-memory SQLite database for testing
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="function")
def db() -> Generator[sqlite3.Connection, None, None]:
def db_session():
"""Create a fresh database for each test."""
# Remove existing test database
if TEST_DB.exists():
TEST_DB.unlink()
# Patch the database path before importing
import app.database as db_module
original_path = db_module.DATABASE_PATH
db_module.DATABASE_PATH = TEST_DB
# Initialize database with new path
db_module.init_db()
conn = db_module.get_db_connection()
conn.row_factory = sqlite3.Row
yield conn
conn.close()
db_module.DATABASE_PATH = original_path
if TEST_DB.exists():
TEST_DB.unlink()
Base.metadata.create_all(bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
Base.metadata.drop_all(bind=engine)
@pytest.fixture(scope="function")
def client(db: sqlite3.Connection) -> Generator[TestClient, None, None]:
"""Create a test client."""
from app.main import app
from app.database import get_db
def client(db_session):
"""Create a test client with fresh database."""
# Override database dependency
def override_get_db():
yield db
try:
yield db_session
finally:
pass
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
def get_user_id_from_token(token: str) -> int:
"""Extract user ID from JWT token (for test fixtures only)."""
from jose import jwt
from app.auth import SECRET_KEY, ALGORITHM
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return int(payload["sub"])
@pytest.fixture
def registered_user(client: TestClient) -> dict:
"""Register a test user and return user data with token."""
response = client.post(
"/auth/register",
json={
"username": "testuser",
"email": "test@example.com",
"password": "testpass123",
},
)
assert response.status_code == 201
data = response.json()
token = data["access_token"]
user_id = get_user_id_from_token(token)
return {
"id": user_id,
"username": "testuser",
"email": "test@example.com",
"password": "testpass123",
"token": token,
}
@pytest.fixture
def two_users(client: TestClient) -> dict:
"""Create two test users."""
# User 1
response = client.post(
"/auth/register",
json={
"username": "user1",
"email": "user1@example.com",
"password": "password123",
},
)
assert response.status_code == 201
token1 = response.json()["access_token"]
# User 2
response = client.post(
"/auth/register",
json={
"username": "user2",
"email": "user2@example.com",
"password": "password123",
},
)
assert response.status_code == 201
token2 = response.json()["access_token"]
return {
"user1_id": get_user_id_from_token(token1),
"user2_id": get_user_id_from_token(token2),
"user1_token": token1,
"user2_token": token2,
"user1_username": "user1",
"user2_username": "user2",
}