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:32:31 +06:00
parent d667221deb
commit 24f969425f
4 changed files with 102 additions and 23 deletions

View File

@@ -1,5 +1,6 @@
"""Application configuration"""
import os
from typing import Any
from pydantic_settings import BaseSettings
from functools import lru_cache
@@ -12,7 +13,6 @@ class Settings(BaseSettings):
POSTGRES_PORT: str = "5432"
POSTGRES_DB: str = "app"
POSTGRES_TEST_DB: str = "test_app"
DATABASE_URL: str | None = None
# JWT
SECRET_KEY: str = "your-secret-key"
@@ -20,7 +20,7 @@ class Settings(BaseSettings):
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# Режим тестирования
TESTING: bool = bool(os.getenv("TESTING"))
TESTING: bool = bool(os.getenv("TESTING", ""))
# Redis
REDIS_HOST: str = "redis"
@@ -28,11 +28,18 @@ class Settings(BaseSettings):
REDIS_DB: int = 0
REDIS_TEST_DB: int = 1
# Telegram
TELEGRAM_BOT_TOKEN: str = ""
TELEGRAM_CHAT_ID: str = ""
model_config = {
"case_sensitive": True,
"env_file": ".env",
"extra": "allow"
}
def get_database_url(self) -> str:
"""Get database URL"""
if self.DATABASE_URL:
return self.DATABASE_URL
if self.TESTING:
return f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@localhost:5432/{self.POSTGRES_TEST_DB}"
return f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
@@ -43,15 +50,6 @@ class Settings(BaseSettings):
host = "localhost" if self.TESTING else self.REDIS_HOST
return f"redis://{host}:{self.REDIS_PORT}/{db}"
# Telegram
TELEGRAM_BOT_TOKEN: str = os.getenv("TELEGRAM_BOT_TOKEN", "")
TELEGRAM_CHAT_ID: str = os.getenv("TELEGRAM_CHAT_ID", "")
class Config:
"""Pydantic config"""
env_file = ".env"
case_sensitive = True
@lru_cache()
def get_settings() -> Settings:
"""Get cached settings"""

View File

@@ -11,7 +11,6 @@ class TestSettings(BaseSettings):
POSTGRES_HOST: str = "localhost"
POSTGRES_PORT: str = "5432"
POSTGRES_DB: str = "test_app"
DATABASE_URL: str = "postgresql://postgres:postgres@localhost:5432/test_app"
# JWT
SECRET_KEY: str = "test_secret_key"
@@ -26,9 +25,14 @@ class TestSettings(BaseSettings):
# Testing
TESTING: bool = True
class Config:
"""Pydantic config"""
case_sensitive = True
env_file = ".env.test"
model_config = {
"case_sensitive": True,
"env_file": ".env.test",
"extra": "allow"
}
def get_database_url(self) -> str:
"""Get database URL"""
return f"postgresql://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
test_settings = TestSettings()

View File

@@ -7,16 +7,16 @@ from fastapi.testclient import TestClient
from typing import Generator, Any
# Устанавливаем переменную окружения для тестов
os.environ["TESTING"] = "True"
os.environ["TESTING"] = "1"
from app.database import Base
from app.main import app
from app.core.test_config import test_settings
from app.dependencies import get_db
from .fixtures import * # импортируем все фикстуsры
from .test_fixtures import * # импортируем все фикстуры
# Создаем тестовый движок базы данных
engine = create_engine(test_settings.DATABASE_URL)
engine = create_engine(test_settings.get_database_url())
# Создаем тестовую фабрику сессий
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@@ -25,7 +25,7 @@ TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engin
def setup_test_db() -> Generator[None, Any, None]:
"""Setup test database"""
# Пробуем создать базу данных test_app
default_engine = create_engine("postgresql://postgres:postgres@localhost:5432/postgres")
default_engine = create_engine(f"postgresql://{test_settings.POSTGRES_USER}:{test_settings.POSTGRES_PASSWORD}@{test_settings.POSTGRES_HOST}:{test_settings.POSTGRES_PORT}/postgres")
with default_engine.connect() as conn:
conn.execute(text("COMMIT")) # Завершаем текущую транзакцию
try:

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"]