1
0
mirror of https://gitlab.com/MoonTestUse1/AdministrationItDepartmens.git synced 2025-08-14 00:25:46 +02:00

Fix database

This commit is contained in:
MoonTestUse1
2025-01-07 05:26:33 +06:00
parent d9e276ad6b
commit bdf4ae9d70
29 changed files with 727 additions and 531 deletions

View File

@@ -1,110 +1,72 @@
"""Test fixtures"""
"""Test configuration"""
import os
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
import fakeredis.aioredis
from typing import Generator
from fastapi.testclient import TestClient
from typing import Generator, Any
# Устанавливаем флаг тестирования
# Устанавливаем переменную окружения для тестов
os.environ["TESTING"] = "True"
from app.database import Base
from app.main import app
from app.database import Base, get_db
from app.models.employee import Employee
from app.utils.auth import get_password_hash
from app.core.test_config import test_settings
from app.dependencies import get_db
from .fixtures import * # импортируем все фикстуры
# Создаем тестовую базу данных в памяти222
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
# Создаем тестовый движок базы данных
engine = create_engine(test_settings.DATABASE_URL)
# Создаем тестовую фабрику сессий
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Создаем тестовую базу данных
Base.metadata.create_all(bind=engine)
@pytest.fixture(scope="session", autouse=True)
def setup_test_db() -> Generator[None, Any, None]:
"""Setup test database"""
# Пробуем создать базу данных test_app
default_engine = create_engine("postgresql://postgres:postgres@localhost:5432/postgres")
with default_engine.connect() as conn:
conn.execute(text("COMMIT")) # Завершаем текущую транзакцию
try:
conn.execute(text("DROP DATABASE IF EXISTS test_app"))
conn.execute(text("CREATE DATABASE test_app"))
except Exception as e:
print(f"Error creating database: {e}")
# Создаем все таблицы
Base.metadata.create_all(bind=engine)
yield
# Удаляем все таблицы
Base.metadata.drop_all(bind=engine)
# Закрываем соединение с тестовой базой
engine.dispose()
@pytest.fixture
def db() -> Generator:
"""Фикстура для получения тестовой сессии БД."""
def db_session() -> Generator[Any, Any, None]:
"""Get database session"""
connection = engine.connect()
transaction = connection.begin()
session = TestingSessionLocal(bind=connection)
yield session
session.close()
transaction.rollback()
connection.close()
try:
yield session
finally:
session.close()
transaction.rollback()
connection.close()
@pytest.fixture
def client(db) -> TestClient:
"""Фикстура для получения тестового клиента."""
def override_get_db():
def client(db_session: Any) -> Generator[TestClient, Any, None]:
"""Get test client"""
def override_get_db() -> Generator[Any, Any, None]:
try:
yield db
yield db_session
finally:
pass
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
@pytest.fixture
def test_employee(db) -> Employee:
"""Фикстура для создания тестового сотрудника."""
employee = Employee(
first_name="Test",
last_name="Employee",
department="Test Department",
office="Test Office",
hashed_password=get_password_hash("testpassword"),
is_admin=False
)
db.add(employee)
db.commit()
db.refresh(employee)
return employee
@pytest.fixture
def test_admin(db) -> Employee:
"""Фикстура для создания тестового администратора."""
admin = Employee(
first_name="Admin",
last_name="User",
department="Admin Department",
office="Admin Office",
hashed_password=get_password_hash("adminpassword"),
is_admin=True
)
db.add(admin)
db.commit()
db.refresh(admin)
return admin
@pytest.fixture
def employee_token(client: TestClient, test_employee: Employee) -> str:
"""Фикстура для получения токена сотрудника."""
response = client.post(
"/api/auth/login",
data={"username": test_employee.last_name, "password": "testpassword"}
)
return response.json()["access_token"]
@pytest.fixture
def admin_token(client: TestClient, test_admin: Employee) -> str:
"""Фикстура для получения токена администратора."""
response = client.post(
"/api/auth/admin/login",
data={"username": test_admin.last_name, "password": "adminpassword"}
)
return response.json()["access_token"]
@pytest.fixture
def redis_mock():
"""Фикстура для мока Redis."""
return fakeredis.aioredis.FakeRedis()
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()

77
backend/tests/fixtures.py Normal file
View File

@@ -0,0 +1,77 @@
"""Test fixtures"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.crud import employees
from app.schemas.employee import EmployeeCreate
from app.utils.auth import get_password_hash
from app.models.employee import Employee
@pytest.fixture(scope="function")
def test_employee(db_session: Session) -> Employee:
"""Create test employee"""
# Удаляем существующего сотрудника, если есть
db_session.query(Employee).filter(
Employee.first_name == "Test",
Employee.last_name == "User"
).delete()
db_session.commit()
employee = EmployeeCreate(
first_name="Test",
last_name="User",
department="IT",
office="101",
password="testpass123",
is_admin=False
)
hashed_password = get_password_hash(employee.password)
db_employee = employees.create_employee(db_session, employee, hashed_password)
return db_employee
@pytest.fixture(scope="function")
def test_admin(db_session: Session) -> Employee:
"""Create test admin"""
# Удаляем существующего админа, если есть
db_session.query(Employee).filter(
Employee.first_name == "Admin",
Employee.last_name == "User"
).delete()
db_session.commit()
admin = EmployeeCreate(
first_name="Admin",
last_name="User",
department="IT",
office="102",
password="adminpass123",
is_admin=True
)
hashed_password = get_password_hash(admin.password)
db_admin = employees.create_employee(db_session, admin, hashed_password)
return db_admin
@pytest.fixture(scope="function")
def employee_token(client: TestClient, test_employee: Employee) -> str:
"""Get employee token"""
response = client.post(
"/api/auth/login",
data={
"username": f"{test_employee.first_name} {test_employee.last_name}",
"password": "testpass123"
}
)
return response.json()["access_token"]
@pytest.fixture(scope="function")
def admin_token(client: TestClient, test_admin: Employee) -> str:
"""Get admin token"""
response = client.post(
"/api/auth/admin/login",
data={
"username": f"{test_admin.first_name} {test_admin.last_name}",
"password": "adminpass123"
}
)
return response.json()["access_token"]

View File

@@ -1,77 +1,84 @@
"""Authentication tests."""
"""Authentication tests"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models.employee import Employee
def test_login_employee_success(client: TestClient, test_employee: Employee):
"""Тест успешной авторизации сотрудника."""
def test_login_success(client: TestClient, test_employee: dict):
"""Test successful login"""
response = client.post(
"/api/auth/login",
data={"username": test_employee.last_name, "password": "testpassword"}
data={
"username": f"{test_employee.first_name} {test_employee.last_name}",
"password": "testpass123"
}
)
assert response.status_code == 200
assert "access_token" in response.json()
assert "token_type" in response.json()
assert response.json()["token_type"] == "bearer"
def test_login_employee_wrong_password(client: TestClient, test_employee: Employee):
"""Тест авторизации сотрудника с неверным паролем."""
def test_login_wrong_password(client: TestClient, test_employee: dict):
"""Test login with wrong password"""
response = client.post(
"/api/auth/login",
data={"username": test_employee.last_name, "password": "wrongpassword"}
data={
"username": f"{test_employee.first_name} {test_employee.last_name}",
"password": "wrongpass"
}
)
assert response.status_code == 401
assert response.json()["detail"] == "Incorrect username or password"
def test_login_employee_wrong_username(client: TestClient):
"""Тест авторизации с несуществующим пользователем."""
def test_login_wrong_username(client: TestClient):
"""Test login with wrong username"""
response = client.post(
"/api/auth/login",
data={"username": "nonexistent", "password": "testpassword"}
data={
"username": "Wrong User",
"password": "testpass123"
}
)
assert response.status_code == 401
assert response.json()["detail"] == "Incorrect username or password"
def test_login_admin_success(client: TestClient, test_admin: Employee):
"""Тест успешной авторизации администратора."""
def test_login_invalid_username_format(client: TestClient):
"""Test login with invalid username format"""
response = client.post(
"/api/auth/login",
data={
"username": "InvalidFormat",
"password": "testpass123"
}
)
assert response.status_code == 401
assert response.json()["detail"] == "Username should be in format: 'First Last'"
def test_admin_login_success(client: TestClient, test_admin: dict):
"""Test successful admin login"""
response = client.post(
"/api/auth/admin/login",
data={"username": test_admin.last_name, "password": "adminpassword"}
data={
"username": f"{test_admin.first_name} {test_admin.last_name}",
"password": "adminpass123"
}
)
assert response.status_code == 200
assert "access_token" in response.json()
assert "token_type" in response.json()
assert response.json()["token_type"] == "bearer"
def test_login_admin_wrong_password(client: TestClient, test_admin: Employee):
"""Тест авторизации администратора с неверным паролем."""
def test_admin_login_not_admin(client: TestClient, test_employee: dict):
"""Test admin login with non-admin user"""
response = client.post(
"/api/auth/admin/login",
data={"username": test_admin.last_name, "password": "wrongpassword"}
data={
"username": f"{test_employee.first_name} {test_employee.last_name}",
"password": "testpass123"
}
)
assert response.status_code == 401
assert response.json()["detail"] == "Incorrect username or password"
def test_protected_route_with_valid_token(client: TestClient, employee_token: str, test_employee: Employee, db: Session):
"""Тест доступа к защищенному маршруту с валидным токеном."""
response = client.get(
"/api/employees/me",
headers={"Authorization": f"Bearer {employee_token}"}
)
assert response.status_code == 200
data = response.json()
assert data["first_name"] == test_employee.first_name
assert data["last_name"] == test_employee.last_name
def test_protected_route_without_token(client: TestClient):
"""Тест доступа к защищенному маршруту без токена."""
response = client.get("/api/employees/me")
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"
def test_protected_route_with_invalid_token(client: TestClient):
"""Тест доступа к защищенному маршруту с недействительным токеном."""
"""Test accessing protected route with invalid token"""
response = client.get(
"/api/employees/me",
headers={"Authorization": "Bearer invalid_token"}

View File

@@ -1,11 +1,10 @@
"""Employee tests."""
"""Employee tests"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models.employee import Employee
def test_create_employee(client: TestClient, admin_token: str, db: Session):
"""Тест создания сотрудника."""
def test_create_employee(client: TestClient, admin_token: str):
"""Test employee creation"""
response = client.post(
"/api/employees",
headers={"Authorization": f"Bearer {admin_token}"},
@@ -13,8 +12,9 @@ def test_create_employee(client: TestClient, admin_token: str, db: Session):
"first_name": "New",
"last_name": "Employee",
"department": "IT",
"office": "102",
"password": "newpassword"
"office": "103",
"password": "newpass123",
"is_admin": False
}
)
assert response.status_code == 201
@@ -22,26 +22,44 @@ def test_create_employee(client: TestClient, admin_token: str, db: Session):
assert data["first_name"] == "New"
assert data["last_name"] == "Employee"
assert data["department"] == "IT"
assert data["office"] == "102"
assert "id" in data
assert data["office"] == "103"
assert data["is_admin"] == False
def test_create_employee_unauthorized(client: TestClient):
"""Тест создания сотрудника без авторизации."""
"""Test employee creation without authorization"""
response = client.post(
"/api/employees",
json={
"first_name": "New",
"last_name": "Employee",
"department": "IT",
"office": "102",
"password": "newpassword"
"office": "103",
"password": "newpass123",
"is_admin": False
}
)
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"
def test_get_employees(client: TestClient, admin_token: str, test_employee: Employee, db: Session):
"""Тест получения списка сотрудников."""
def test_create_employee_not_admin(client: TestClient, employee_token: str):
"""Test employee creation by non-admin user"""
response = client.post(
"/api/employees",
headers={"Authorization": f"Bearer {employee_token}"},
json={
"first_name": "New",
"last_name": "Employee",
"department": "IT",
"office": "103",
"password": "newpass123",
"is_admin": False
}
)
assert response.status_code == 403
assert response.json()["detail"] == "Not enough permissions"
def test_get_employees(client: TestClient, admin_token: str):
"""Test getting all employees"""
response = client.get(
"/api/employees",
headers={"Authorization": f"Bearer {admin_token}"}
@@ -50,67 +68,24 @@ def test_get_employees(client: TestClient, admin_token: str, test_employee: Empl
data = response.json()
assert isinstance(data, list)
assert len(data) > 0
assert "first_name" in data[0]
assert "last_name" in data[0]
assert "department" in data[0]
assert "office" in data[0]
def test_get_employee_by_id(client: TestClient, admin_token: str, test_employee: Employee, db: Session):
"""Тест получения сотрудника по ID."""
def test_get_employees_unauthorized(client: TestClient):
"""Test getting employees without authorization"""
response = client.get("/api/employees")
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"
def test_get_employees_not_admin(client: TestClient, employee_token: str):
"""Test getting employees by non-admin user"""
response = client.get(
f"/api/employees/{test_employee.id}",
headers={"Authorization": f"Bearer {admin_token}"}
"/api/employees",
headers={"Authorization": f"Bearer {employee_token}"}
)
assert response.status_code == 200
data = response.json()
assert data["first_name"] == test_employee.first_name
assert data["last_name"] == test_employee.last_name
assert data["department"] == test_employee.department
assert data["office"] == test_employee.office
assert response.status_code == 403
assert response.json()["detail"] == "Not enough permissions"
def test_get_nonexistent_employee(client: TestClient, admin_token: str):
"""Тест получения несуществующего сотрудника."""
response = client.get(
"/api/employees/999",
headers={"Authorization": f"Bearer {admin_token}"}
)
assert response.status_code == 404
assert response.json()["detail"] == "Employee not found"
def test_update_employee(client: TestClient, admin_token: str, test_employee: Employee, db: Session):
"""Тест обновления данных сотрудника."""
response = client.put(
f"/api/employees/{test_employee.id}",
headers={"Authorization": f"Bearer {admin_token}"},
json={
"first_name": "Updated",
"last_name": "Name",
"department": "HR",
"office": "103"
}
)
assert response.status_code == 200
data = response.json()
assert data["first_name"] == "Updated"
assert data["last_name"] == "Name"
assert data["department"] == "HR"
assert data["office"] == "103"
def test_delete_employee(client: TestClient, admin_token: str, test_employee: Employee, db: Session):
"""Тест удаления сотрудника."""
response = client.delete(
f"/api/employees/{test_employee.id}",
headers={"Authorization": f"Bearer {admin_token}"}
)
assert response.status_code == 200
data = response.json()
assert data["first_name"] == test_employee.first_name
assert data["last_name"] == test_employee.last_name
assert data["department"] == test_employee.department
assert data["office"] == test_employee.office
def test_employee_me(client: TestClient, employee_token: str, test_employee: Employee, db: Session):
"""Тест получения информации о текущем сотруднике."""
def test_get_me(client: TestClient, employee_token: str, test_employee: dict):
"""Test getting current employee"""
response = client.get(
"/api/employees/me",
headers={"Authorization": f"Bearer {employee_token}"}
@@ -122,21 +97,35 @@ def test_employee_me(client: TestClient, employee_token: str, test_employee: Emp
assert data["department"] == test_employee.department
assert data["office"] == test_employee.office
def test_update_me(client: TestClient, employee_token: str, test_employee: Employee, db: Session):
"""Тест обновления информации о текущем сотруднике."""
def test_get_me_unauthorized(client: TestClient):
"""Test getting current employee without authorization"""
response = client.get("/api/employees/me")
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"
def test_update_me(client: TestClient, employee_token: str):
"""Test updating current employee"""
response = client.put(
"/api/employees/me",
headers={"Authorization": f"Bearer {employee_token}"},
json={
"first_name": "Updated",
"last_name": "Name",
"department": "Support",
"department": "HR",
"office": "104"
}
)
assert response.status_code == 200
data = response.json()
assert data["first_name"] == "Updated"
assert data["last_name"] == "Name"
assert data["department"] == "Support"
assert data["office"] == "104"
assert data["department"] == "HR"
assert data["office"] == "104"
def test_update_me_unauthorized(client: TestClient):
"""Test updating current employee without authorization"""
response = client.put(
"/api/employees/me",
json={
"department": "HR",
"office": "104"
}
)
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"

View File

@@ -1,55 +1,41 @@
"""Request tests."""
"""Request tests"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models.employee import Employee
from app.models.request import Request
def test_create_request(client: TestClient, employee_token: str, db: Session):
"""Тест создания заявки."""
def test_create_request(client: TestClient, employee_token: str):
"""Test request creation"""
response = client.post(
"/api/requests",
headers={"Authorization": f"Bearer {employee_token}"},
json={
"request_type": "equipment",
"description": "Test Description",
"description": "Need a new laptop",
"priority": "medium"
}
)
assert response.status_code == 201
data = response.json()
assert data["request_type"] == "equipment"
assert data["description"] == "Test Description"
assert data["description"] == "Need a new laptop"
assert data["priority"] == "medium"
assert data["status"] == "new"
assert "id" in data
def test_create_request_unauthorized(client: TestClient):
"""Тест создания заявки без авторизации."""
"""Test request creation without authorization"""
response = client.post(
"/api/requests",
json={
"request_type": "equipment",
"description": "Test Description",
"description": "Need a new laptop",
"priority": "medium"
}
)
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"
def test_get_employee_requests(client: TestClient, employee_token: str, test_employee: Employee, db: Session):
"""Тест получения списка заявок сотрудника."""
# Создаем тестовую заявку
request = Request(
request_type="equipment",
description="Test Description",
priority="medium",
status="new",
employee_id=test_employee.id
)
db.add(request)
db.commit()
def test_get_my_requests(client: TestClient, employee_token: str):
"""Test getting employee's requests"""
response = client.get(
"/api/requests/my",
headers={"Authorization": f"Bearer {employee_token}"}
@@ -57,23 +43,15 @@ def test_get_employee_requests(client: TestClient, employee_token: str, test_emp
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) > 0
assert data[0]["request_type"] == "equipment"
assert data[0]["description"] == "Test Description"
def test_admin_get_all_requests(client: TestClient, admin_token: str, test_employee: Employee, db: Session):
"""Тест получения всех заявок администратором."""
# Создаем тестовую заявку
request = Request(
request_type="equipment",
description="Test Description",
priority="medium",
status="new",
employee_id=test_employee.id
)
db.add(request)
db.commit()
def test_get_my_requests_unauthorized(client: TestClient):
"""Test getting employee's requests without authorization"""
response = client.get("/api/requests/my")
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"
def test_get_all_requests_admin(client: TestClient, admin_token: str):
"""Test getting all requests by admin"""
response = client.get(
"/api/requests/admin",
headers={"Authorization": f"Bearer {admin_token}"}
@@ -81,25 +59,39 @@ def test_admin_get_all_requests(client: TestClient, admin_token: str, test_emplo
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
assert len(data) > 0
assert data[0]["request_type"] == "equipment"
assert data[0]["description"] == "Test Description"
def test_update_request_status(client: TestClient, admin_token: str, test_employee: Employee, db: Session):
"""Тест обновления статуса заявки."""
# Создаем тестовую заявку
request = Request(
request_type="equipment",
description="Test Description",
priority="medium",
status="new",
employee_id=test_employee.id
def test_get_all_requests_unauthorized(client: TestClient):
"""Test getting all requests without authorization"""
response = client.get("/api/requests/admin")
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"
def test_get_all_requests_not_admin(client: TestClient, employee_token: str):
"""Test getting all requests by non-admin user"""
response = client.get(
"/api/requests/admin",
headers={"Authorization": f"Bearer {employee_token}"}
)
db.add(request)
db.commit()
assert response.status_code == 403
assert response.json()["detail"] == "Not enough permissions"
def test_update_request_status_admin(client: TestClient, admin_token: str):
"""Test updating request status by admin"""
# Сначала создаем запрос
response = client.post(
"/api/requests",
headers={"Authorization": f"Bearer {admin_token}"},
json={
"request_type": "equipment",
"description": "Need a new laptop",
"priority": "medium"
}
)
request_id = response.json()["id"]
# Обновляем статус
response = client.patch(
f"/api/requests/{request.id}/status",
f"/api/requests/{request_id}/status",
headers={"Authorization": f"Bearer {admin_token}"},
json={"status": "in_progress"}
)
@@ -107,42 +99,21 @@ def test_update_request_status(client: TestClient, admin_token: str, test_employ
data = response.json()
assert data["status"] == "in_progress"
def test_get_request_statistics(client: TestClient, admin_token: str, test_employee: Employee, db: Session):
"""Тест получения статистики по заявкам."""
# Создаем тестовые заявки с разными статусами
requests = [
Request(
request_type="equipment",
description="Test Description",
priority="medium",
status="new",
employee_id=test_employee.id
),
Request(
request_type="equipment",
description="Test Description",
priority="high",
status="in_progress",
employee_id=test_employee.id
),
Request(
request_type="equipment",
description="Test Description",
priority="low",
status="completed",
employee_id=test_employee.id
)
]
for req in requests:
db.add(req)
db.commit()
response = client.get(
"/api/statistics",
headers={"Authorization": f"Bearer {admin_token}"}
def test_update_request_status_not_admin(client: TestClient, employee_token: str):
"""Test updating request status by non-admin user"""
response = client.patch(
"/api/requests/1/status",
headers={"Authorization": f"Bearer {employee_token}"},
json={"status": "in_progress"}
)
assert response.status_code == 200
data = response.json()
assert "total" in data
assert "by_status" in data
assert data["total"] >= 3
assert response.status_code == 403
assert response.json()["detail"] == "Not enough permissions"
def test_update_request_status_unauthorized(client: TestClient):
"""Test updating request status without authorization"""
response = client.patch(
"/api/requests/1/status",
json={"status": "in_progress"}
)
assert response.status_code == 401
assert response.json()["detail"] == "Not authenticated"