Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions requirements.dev.txt
Original file line number Diff line number Diff line change
Expand Up @@ -5,3 +5,4 @@ isort
pytest
pytest-cov
pytest-mock
testcontainers[postgres]
167 changes: 167 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
from functools import lru_cache
from pathlib import Path

import pytest
from alembic import command
from alembic.config import Config as AlembicConfig
from fastapi.testclient import TestClient
from pytest import MonkeyPatch
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from testcontainers.postgres import PostgresContainer

from modal_backend.models.db import NoteType
from modal_backend.settings import Settings


class PostgresConfig:
"""Дата-класс со значениями для контейнера с тестовой БД и для alembic-миграции."""

container_name: str = "modal_backend-test"
username: str = "postgres"
host: str = "localhost"
external_port: int = 5433
image: str = "postgres:15"
host_auth_method: str = "trust"
alembic_ini: str = Path(__file__).resolve().parent.parent / "alembic.ini"

@classmethod
def get_url(cls) -> str:
"""Возвращает URI для подключения к БД."""
return f"postgresql://{cls.username}@{cls.host}:{cls.external_port}/postgres"


@pytest.fixture(scope="session")
def session_mp():
mp = MonkeyPatch()
yield mp
mp.undo()


@pytest.fixture(scope="session")
def get_settings_mock(session_mp):
"""Переопределение get_settings в modal_backend/settings.py."""

@lru_cache
def get_test_settings():
test_settings = Settings()
test_settings.DB_DSN = PostgresConfig.get_url()
return test_settings

dsn_mock = session_mp.setattr("modal_backend.settings.get_settings", get_test_settings)
return dsn_mock


@pytest.fixture(scope="session")
def get_app_with_test_settings(get_settings_mock):
"""Загрузка app с тестовыми настройками."""
from modal_backend.routes import app

return app


@pytest.fixture(scope="session")
def db_container(get_settings_mock):
"""Фикстура настройки БД для тестов в Docker-контейнере."""
container = (
PostgresContainer(
image=PostgresConfig.image, username=PostgresConfig.username, dbname=PostgresConfig.container_name
)
.with_bind_ports(5432, PostgresConfig.external_port)
.with_env("POSTGRES_HOST_AUTH_METHOD", PostgresConfig.host_auth_method)
.with_name(PostgresConfig.container_name)
)
container.start()
alembic_ini = PostgresConfig.alembic_ini
cfg = AlembicConfig(str(alembic_ini.resolve()))
cfg.set_main_option("script_location", "%(here)s/migrations")
command.upgrade(cfg, "head")
try:
yield PostgresConfig.get_url()
finally:
container.stop()


@pytest.fixture(scope="session")
def engine(db_container):
"""Фикстура настройки пула соединений к БД."""
engine = create_engine(str(db_container), pool_pre_ping=True)
yield engine
engine.dispose()


@pytest.fixture()
def dbsession(engine):
"""Фикстура настройки Session для работы с БД в тестах."""
TestingLocalSession = sessionmaker(bind=engine)
session = TestingLocalSession()
yield session
session.close()


@pytest.fixture()
def authlib_user_data():
"""
Данные о пользователе, возвращаемые сервисом auth.
"""
return {
"session_scopes": [{"id": 0, "name": "string", "comment": "string"}],
"user_scopes": [{"id": 0, "name": "string", "comment": "string"}],
"indirect_groups": [{"id": 0, "name": "string", "parent_id": 0}],
"groups": [{"id": 0, "name": "string", "parent_id": 0}],
"id": 0,
"email": "string",
"userdata": [
{"category": "Личная информация", "param": "Полное имя", "value": "Тестовый Тест"},
],
}


@pytest.fixture()
def authlib_mock(mocker):
auth_mock = mocker.patch("auth_lib.fastapi.UnionAuth.__call__", autospec=True)
return auth_mock


@pytest.fixture()
def user_mock(authlib_mock, authlib_user_data):
authlib_mock.return_value = authlib_user_data
return authlib_mock


@pytest.fixture()
def client(get_app_with_test_settings, user_mock):
app = get_app_with_test_settings
client = TestClient(app)
return client


def create_note_type(name: str, type_id: int):
"""Вспомогательная функция-мини-фабрика для создания разных типов модалок в фикстуре note_types."""
return NoteType(name=name, type_id=type_id)


@pytest.fixture()
def note_types(dbsession):
"""Создает три разных типа модалок."""
note_type_data = [
(
"Name 1",
1,
),
(
"Name 2",
2,
),
("Name 3", 3),
]

note_types = [create_note_type(*note_type) for note_type in note_type_data]

for note_type in note_types:
dbsession.add(note_type)
dbsession.commit()
yield note_types
for note_type in note_types:
dbsession.delete(note_type)
dbsession.commit()
68 changes: 68 additions & 0 deletions tests/test_routes/test_note_type.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
import pytest
from starlette import status

from modal_backend.models import NoteType
from modal_backend.schemas.models import NoteTypeGet
from modal_backend.settings import get_settings

url: str = "/notificationtype"
settings = get_settings()


@pytest.mark.parametrize(
"status_code",
Comment thread
petrCher marked this conversation as resolved.
[
(status.HTTP_200_OK),
],
)
def test_get_notification_type(client, note_types, status_code):
response = client.get(url)
assert response.status_code == status_code
type_ids_of_note_types = [note_type.type_id for note_type in note_types]
for note_type in response.json():
assert note_type.get("type_id") in type_ids_of_note_types


@pytest.mark.parametrize(
"status_code, body",
[
(
status.HTTP_200_OK,
{
"type_id": 4,
"name": "No_exist_type",
},
),
(
status.HTTP_409_CONFLICT,
{
"type_id": 1,
"name": "Already_exist_type",
},
),
(
status.HTTP_422_UNPROCESSABLE_ENTITY,
{
"type_id": 4,
"name": 123,
},
),
(
status.HTTP_422_UNPROCESSABLE_ENTITY,
{
"type_id": "string",
"name": "string",
},
),
],
)
def test_post_create_notification_type(client, dbsession, note_types, status_code, body):
response = client.post(url, json=body)
assert response.status_code == status_code

if status_code == status.HTTP_200_OK:
response_model = NoteTypeGet(**response.json())
exist_note_type = dbsession.query(NoteType).filter(NoteType.type_id == response_model.type_id).one_or_none()
assert exist_note_type
assert exist_note_type.name == body.get("name")
assert exist_note_type.type_id == body.get("type_id")
Loading