From 2edb4f5322289cc6ee56538aea9dee5473108715 Mon Sep 17 00:00:00 2001 From: raf-nr Date: Sat, 9 Dec 2023 22:18:16 +0300 Subject: [PATCH] feat: add fixtures for tests --- tests/conftest.py | 54 ++++ tests/fixtures/fetch_data_from_db.py | 35 +++ tests/fixtures/insert_data_in_db.py | 363 +++++++++++++++++++++++++++ 3 files changed, 452 insertions(+) create mode 100644 tests/conftest.py create mode 100644 tests/fixtures/fetch_data_from_db.py create mode 100644 tests/fixtures/insert_data_in_db.py diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..5ffa4b28 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,54 @@ +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from app.db.base_model import Base +from app.settings import get_settings + +settings = get_settings() + +tables = [ + "Code", + "User", + "Role", + "Permission", + "Feedback", + "Device", + "Session", + "ColumnStats", + "FileFormat", + "FileInfo", + "Task", +] + +test_engine = create_engine(settings.postgres_test_dsn.unicode_string()) + + +@pytest.fixture(scope="session", autouse=True) +def session_test(): + Session = sessionmaker(test_engine, expire_on_commit=False) + yield Session + + +@pytest.fixture(scope="session", autouse=True) +def create_tables(): + Base.metadata.create_all(test_engine) + + +@pytest.fixture(scope="session", autouse=True) +def drop_tables_at_end(session_test): + yield + Base.metadata.drop_all(test_engine) + + +@pytest.fixture(scope="function", autouse=True) +def clean_tables(request, session_test): + if "fixture_name" in request.fixturenames: + yield + else: + with session_test() as session: + with session.connection(): + for table_name in tables: + table = Base.metadata.tables[table_name] + session.query(table).delete() + session.commit() + yield diff --git a/tests/fixtures/fetch_data_from_db.py b/tests/fixtures/fetch_data_from_db.py new file mode 100644 index 00000000..2523ecb6 --- /dev/null +++ b/tests/fixtures/fetch_data_from_db.py @@ -0,0 +1,35 @@ +import pytest +from app.db import tables + + +@pytest.fixture(scope="function") +def fetch_all_data_from_table(session_test): + def fetch_all_data(model_name: str): + with session_test() as session: + with session.connection(): + data = session.query(tables[model_name]).all() + return data + + return fetch_all_data + + +@pytest.fixture(scope="function") +def fetch_data_by_id_from_table(session_test): + def fetch_data_by_id(model_name: str, id: str | int): + with session_test() as session: + with session.connection(): + data = session.get(tables[model_name], id) + return data + + return fetch_data_by_id + + +@pytest.fixture(scope="function") +def fetch_first_from_table(session_test): + def fetch_first(model_name: str): + with session_test() as session: + with session.connection(): + data = session.query(tables[model_name]).first() + return data + + return fetch_first diff --git a/tests/fixtures/insert_data_in_db.py b/tests/fixtures/insert_data_in_db.py new file mode 100644 index 00000000..5386ac6c --- /dev/null +++ b/tests/fixtures/insert_data_in_db.py @@ -0,0 +1,363 @@ +import pytest +from app.db import tables +from uuid import UUID, uuid4 +from datetime import datetime + +# Here are fixtures that are not automatically executed and that return a function, +# that adds the arguments passed to it to the database, table that corresponds to the name of the fixture. + + +@pytest.fixture(scope="function") +def insert_permission_data(session_test): + def add_permission(id: int, permission: str): + with session_test() as session: + with session.connection(): + table = tables["permission"] + new_permission = table(id=id, permission=permission) + session.add(new_permission) + session.commit() + + return add_permission + + +@pytest.fixture(scope="function") +def insert_user_data(session_test): + def add_user( + email: str, + full_name: str, + country: str, + company_or_affiliation: str, + occupation: str, + account_status: str, + hashed_password: str, + created_at: datetime, + id: UUID = uuid4(), + deleted_at: datetime | None = None, + ): + with session_test() as session: + with session.connection(): + table = tables["user"] + new_user = table( + id=id, + email=email, + full_name=full_name, + country=country, + company_or_affiliation=company_or_affiliation, + occupation=occupation, + account_status=account_status, + hashed_password=hashed_password, + created_at=created_at, + deleted_at=deleted_at, + ) + session.add(new_user) + session.commit() + + return add_user + + +@pytest.fixture(scope="function") +def insert_role_data(session_test): + def add_role(user_id: UUID, type: str, permission_indices: str, id: UUID = uuid4()): + with session_test() as session: + with session.connection(): + table = tables["role"] + new_role = table( + id=id, + user_id=user_id, + type=type, + permission_indices=permission_indices, + ) + session.add(new_role) + session.commit() + + return add_role + + +@pytest.fixture(scope="function") +def insert_device_data(session_test): + def add_device( + id: str = uuid4(), + user_agent: str | None = None, + browser: str | None = None, + engine: str | None = None, + os: str | None = None, + os_version: str | None = None, + device: str | None = None, + cpu: str | None = None, + screen: str | None = None, + plugins: str | None = None, + time_zone: str | None = None, + language: str | None = None, + created_at: datetime = datetime.now(), + ): + with session_test() as session: + with session.connection(): + table = tables["device"] + new_device = table( + id=id, + user_agent=user_agent, + browser=browser, + engine=engine, + os=os, + os_version=os_version, + device=device, + cpu=cpu, + screen=screen, + plugins=plugins, + time_zone=time_zone, + language=language, + created_at=created_at, + ) + session.add(new_device) + session.commit() + + return add_device + + +@pytest.fixture(scope="function") +def insert_feedback_data(session_test): + def add_feedback( + user_id: UUID, + rating: int, + text: str, + id: UUID = uuid4(), + subject: str | None = None, + created_at: datetime = datetime.now(), + ): + with session_test() as session: + with session.connection(): + table = tables["feedback"] + new_feedback = table( + id=id, + user_id=user_id, + rating=rating, + subject=subject, + text=text, + created_at=created_at, + ) + session.add(new_feedback) + session.commit() + + return add_feedback + + +@pytest.fixture(scope="function") +def insert_session_data(session_test): + def add_session( + user_id: UUID, + device_id: str, + status: str, + id: UUID = uuid4(), + access_token_iat: int | None = None, + refresh_token_iat: int | None = None, + created_at: datetime = datetime.now(), + ): + with session_test() as session: + with session.connection(): + table = tables["session"] + new_session = table( + id=id, + user_id=user_id, + device_id=device_id, + status=status, + access_token_iat=access_token_iat, + refresh_token_iat=refresh_token_iat, + created_at=created_at, + ) + session.add(new_session) + session.commit() + + return add_session + + +@pytest.fixture(scope="function") +def insert_code_data(session_test): + def add_code( + type: str, + value: int, + user_id: UUID, + device_id: str, + id: UUID = uuid4(), + expiring_date: datetime = datetime.now(), + ): + with session_test() as session: + with session.connection(): + table = tables["code"] + new_code = table( + id=id, + type=type, + value=value, + expiring_date=expiring_date, + user_id=user_id, + device_id=device_id, + ) + session.add(new_code) + session.commit() + + return add_code + + +@pytest.fixture(scope="function") +def insert_file_info_data(session_test): + def add_file_info( + is_built_in: bool, + is_valid: bool, + file_name: str, + original_file_name: str, + has_header: bool, + delimiter: str, + renamed_header: str, + rows_count: int, + count_of_columns: int, + path: str, + id: UUID = uuid4(), + created_at: datetime = datetime.now(), + user_id: UUID | None = None, + mime_type: str | None = None, + encoding: str | None = None, + deleted_at: datetime | None = None, + ): + with session_test() as session: + with session.connection(): + table = tables["file_info"] + new_file_info = table( + id=id, + created_at=created_at, + user_id=user_id, + is_built_in=is_built_in, + is_valid=is_valid, + mime_type=mime_type, + encoding=encoding, + file_name=file_name, + original_file_name=original_file_name, + has_header=has_header, + delimiter=delimiter, + renamed_header=renamed_header, + rows_count=rows_count, + count_of_columns=count_of_columns, + path=path, + deleted_at=deleted_at, + ) + session.add(new_file_info) + session.commit() + + return add_file_info + + +@pytest.fixture(scope="function") +def insert_file_format_data(session_test): + def add_file_format( + id: UUID, + input_format: str, + tid_column_index: int | None = None, + item_column_index: int | None = None, + has_tid: bool | None = None, + ): + with session_test() as session: + with session.connection(): + table = tables["file_format"] + new_file_format = table( + id=id, + input_format=input_format, + tid_column_index=tid_column_index, + item_column_index=item_column_index, + has_tid=has_tid, + ) + session.add(new_file_format) + session.commit() + + return add_file_format + + +@pytest.fixture(scope="function") +def insert_column_stats_data(session_test): + def add_column_stats( + id: UUID, + column_index: int, + type: str, + distinct: int | None = None, + is_categorical: bool | None = None, + count: int | None = None, + avg: str | None = None, + std: str | None = None, + skewness: str | None = None, + kurtosis: str | None = None, + min: str | None = None, + max: str | None = None, + sum: str | None = None, + quantile25: str | None = None, + quantile50: str | None = None, + quantile75: str | None = None, + deleted_at: datetime | None = None, + ): + with session_test() as session: + with session.connection(): + table = tables["column_stats"] + new_column_stats = table( + id=id, + column_index=column_index, + type=type, + distinct=distinct, + is_categorical=is_categorical, + count=count, + avg=avg, + std=std, + skewness=skewness, + kurtosis=kurtosis, + min=min, + max=max, + sum=sum, + quantile25=quantile25, + quantile50=quantile50, + quantile75=quantile75, + deleted_at=deleted_at, + ) + session.add(new_column_stats) + session.commit() + + return add_column_stats + + +@pytest.fixture(scope="function") +def insert_task_data(session_test): + def add_task( + user_id: UUID, + file_id: UUID, + is_private: bool, + attempt_number: int, + status: str, + progres: float, + id_executed: bool, + id: UUID = uuid4(), + phase_name: str | None = None, + current_phase: int | None = None, + max_phase: int | None = None, + error_msg: int | None = None, + elapsed_time: float | None = None, + config: str | None = None, + result: str | None = None, + ): + with session_test() as session: + with session.connection(): + table = tables["task"] + new_task = table( + id=id, + user_id=user_id, + file_id=file_id, + is_private=is_private, + attempt_number=attempt_number, + status=status, + phase_name=phase_name, + current_phase=current_phase, + progres=progres, + max_phase=max_phase, + error_msg=error_msg, + id_executed=id_executed, + elapsed_time=elapsed_time, + config=config, + result=result, + ) + session.add(new_task) + session.commit() + + return add_task