import asyncio import io import uuid import httpx import pytest import pytest_asyncio from httpx import AsyncClient from starlette.testclient import TestClient from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy import select, text from src.database import Base, get_db from src.main import app from src.user.models import User, UserRole DATABASE_URL = "sqlite+aiosqlite:///:memory:" engine = create_async_engine(DATABASE_URL, future=True, echo=False) AsyncSessionLocal = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) @pytest_asyncio.fixture(scope="session", autouse=True) async def prepare_database(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) yield async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) @pytest_asyncio.fixture() async def db_session(): async with AsyncSessionLocal() as session: yield session await session.rollback() @pytest.fixture() def client(db_session): def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db with TestClient(app) as c: yield c app.dependency_overrides.clear() @pytest_asyncio.fixture() async def user_role(db_session): unique_name = f"test_role_{uuid.uuid4().hex[:6]}" role = UserRole(name=unique_name) db_session.add(role) await db_session.commit() await db_session.refresh(role) return role @pytest_asyncio.fixture(autouse=True) async def cleanup_tables(db_session): # Очищати таблицю user await db_session.execute(text('DELETE FROM user')) await db_session.commit() @pytest.mark.asyncio async def test_create_user_success(client, user_role): file_content = b"fake image content" files = {"logo": ("logo.png", io.BytesIO(file_content), "image/png")} data = { "role_id": str(user_role.role_id), "name": "Test User", "login": "testlogin", "phone": "+1234567890", "password": "testpassword", } response = client.post("/user", data=data, files=files) assert response.status_code == 200 resp_json = response.json() assert resp_json["user_id"] > 0 assert resp_json["name"] == "Test User" assert resp_json["login"] == "testlogin" assert resp_json["role_id"] == user_role.role_id assert resp_json["logo"] is not None @pytest.mark.asyncio async def test_get_user_by_id_success(client, db_session, user_role): # Створюємо користувача напряму в БД для тесту get_by_id user = User( role_id=user_role.role_id, name="User GetById", login="getbyidlogin", phone="123456", password="hashed_password", logo="/some/logo.png" ) db_session.add(user) await db_session.commit() await db_session.refresh(user) response = client.get(f"/user/{user.user_id}") assert response.status_code == 200 data = response.json() assert data["user_id"] == user.user_id assert data["login"] == "getbyidlogin" @pytest.mark.asyncio async def test_get_user_by_id_not_found(client): response = client.get("/user/999999999") assert response.status_code == 404 assert response.json()["detail"] == "User not found" @pytest.mark.asyncio async def test_get_user_list(client, db_session, user_role): # Створимо 3 користувачів for i in range(3): user = User( role_id=user_role.role_id, name=f"User {i}", login=f"user{i}", phone=f"phone{i}", password="hashed", logo="/logo.png" ) db_session.add(user) await db_session.commit() response = client.get("/user?limit=2&page=1") assert response.status_code == 200 data = response.json() assert data["status"] is True assert "items" in data assert len(data["items"]) <= 2 assert data["total"] >= 3 assert data["page"] == 1 @pytest.mark.asyncio async def test_update_user_success(client, db_session, user_role): user = User( role_id=user_role.role_id, name="Old Name", login="oldlogin", phone="000000", password="oldhashed", logo="/oldlogo.png" ) db_session.add(user) await db_session.commit() await db_session.refresh(user) new_name = "New Name" new_phone = "+999999999" data = { "name": new_name, "phone": new_phone } response = client.patch(f"/user/{user.user_id}", data=data) assert response.status_code == 200 resp_json = response.json() assert resp_json["status"] is True assert resp_json["data"]["name"] == new_name assert resp_json["data"]["phone"] == new_phone @pytest.mark.asyncio async def test_delete_user_success(client, db_session, user_role): user = User( role_id=user_role.role_id, name="Delete User", login="deletelogin", phone="111111", password="hashedpassword", logo="/logo.png" ) db_session.add(user) await db_session.commit() await db_session.refresh(user) response = client.delete(f"/user/{user.user_id}") assert response.status_code == 200 resp_json = response.json() assert resp_json["status"] is True assert resp_json["message"] == "User deleted" # Перевіримо, що логін та пароль стерті result = await db_session.execute(select(User).where(User.user_id == user.user_id)) deleted_user = result.scalar_one() assert deleted_user.login == "" assert deleted_user.password == "" @pytest.mark.asyncio async def test_delete_user_not_found(client): response = client.delete("/user/9999999999") assert response.status_code == 404 assert response.json()["detail"] == "User not found" @pytest.mark.asyncio async def test_get_all_roles(db_session, client): # 1. Додати ролі в базу roles = [UserRole(name=f"Role{i}") for i in range(3)] db_session.add_all(roles) await db_session.commit() # 2. Перевизначити залежність get_db для client, щоб використовувати db_session async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db # 3. Викликати API (з урахуванням root_path та префікса /user) response = client.get("/api/user/allRole") # 4. Очистити overrides після тесту app.dependency_overrides.clear() # 5. Перевірки assert response.status_code == 200, f"Response text: {response.text}" data = response.json() assert data["status"] is True assert isinstance(data["data"], list) assert any("id" in r and "name" in r for r in data["data"])