Merge pull request 'add tests for project' (#1) from test into dev1

Reviewed-on: #1

poetry.lock

@@ -129,12 +129,12 @@ version = "0.4.6"
 description = "Cross-platform colored terminal text."
 optional = false
 python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,!=3.6.*,>=2.7"
-groups = ["main"]
-markers = "platform_system == \"Windows\""
+groups = ["main", "test"]
 files = [
     {file = "colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6"},
     {file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"},
 ]
+markers = {main = "platform_system == \"Windows\"", test = "sys_platform == \"win32\""}
 
 [[package]]
 name = "dnspython"
@@ -290,6 +290,18 @@ files = [
 [package.extras]
 all = ["flake8 (>=7.1.1)", "mypy (>=1.11.2)", "pytest (>=8.3.2)", "ruff (>=0.6.2)"]
 
+[[package]]
+name = "iniconfig"
+version = "2.1.0"
+description = "brain-dead simple config-ini parsing"
+optional = false
+python-versions = ">=3.8"
+groups = ["test"]
+files = [
+    {file = "iniconfig-2.1.0-py3-none-any.whl", hash = "sha256:9deba5723312380e77435581c6bf4935c94cbfab9b1ed33ef8d238ea168eb760"},
+    {file = "iniconfig-2.1.0.tar.gz", hash = "sha256:3abbd2e30b36733fee78f9c7f7308f2d0050e88f0087fd25c2645f63c773e1c7"},
+]
+
 [[package]]
 name = "mako"
 version = "1.3.10"
@@ -381,6 +393,18 @@ files = [
     {file = "markupsafe-3.0.2.tar.gz", hash = "sha256:ee55d3edf80167e48ea11a923c7386f4669df67d7994554387f84e7d8b0a2bf0"},
 ]
 
+[[package]]
+name = "packaging"
+version = "25.0"
+description = "Core utilities for Python packages"
+optional = false
+python-versions = ">=3.8"
+groups = ["test"]
+files = [
+    {file = "packaging-25.0-py3-none-any.whl", hash = "sha256:29572ef2b1f17581046b3a2227d5c611fb25ec70ca1ba8554b24b0e69331a484"},
+    {file = "packaging-25.0.tar.gz", hash = "sha256:d443872c98d677bf60f6a1f2f8c1cb748e8fe762d2bf9d3148b5599295b0fc4f"},
+]
+
 [[package]]
 name = "passlib"
 version = "1.7.4"
@@ -399,6 +423,22 @@ bcrypt = ["bcrypt (>=3.1.0)"]
 build-docs = ["cloud-sptheme (>=1.10.1)", "sphinx (>=1.6)", "sphinxcontrib-fulltoc (>=1.2.0)"]
 totp = ["cryptography"]
 
+[[package]]
+name = "pluggy"
+version = "1.6.0"
+description = "plugin and hook calling mechanisms for python"
+optional = false
+python-versions = ">=3.9"
+groups = ["test"]
+files = [
+    {file = "pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746"},
+    {file = "pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3"},
+]
+
+[package.extras]
+dev = ["pre-commit", "tox"]
+testing = ["coverage", "pytest", "pytest-benchmark"]
+
 [[package]]
 name = "pydantic"
 version = "2.11.7"
@@ -557,6 +597,21 @@ gcp-secret-manager = ["google-cloud-secret-manager (>=2.23.1)"]
 toml = ["tomli (>=2.0.1)"]
 yaml = ["pyyaml (>=6.0.1)"]
 
+[[package]]
+name = "pygments"
+version = "2.19.2"
+description = "Pygments is a syntax highlighting package written in Python."
+optional = false
+python-versions = ">=3.8"
+groups = ["test"]
+files = [
+    {file = "pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b"},
+    {file = "pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887"},
+]
+
+[package.extras]
+windows-terminal = ["colorama (>=0.4.6)"]
+
 [[package]]
 name = "pyjwt"
 version = "2.10.1"
@@ -575,6 +630,47 @@ dev = ["coverage[toml] (==5.0.4)", "cryptography (>=3.4.0)", "pre-commit", "pyte
 docs = ["sphinx", "sphinx-rtd-theme", "zope.interface"]
 tests = ["coverage[toml] (==5.0.4)", "pytest (>=6.0.0,<7.0.0)"]
 
+[[package]]
+name = "pytest"
+version = "8.4.1"
+description = "pytest: simple powerful testing with Python"
+optional = false
+python-versions = ">=3.9"
+groups = ["test"]
+files = [
+    {file = "pytest-8.4.1-py3-none-any.whl", hash = "sha256:539c70ba6fcead8e78eebbf1115e8b589e7565830d7d006a8723f19ac8a0afb7"},
+    {file = "pytest-8.4.1.tar.gz", hash = "sha256:7c67fd69174877359ed9371ec3af8a3d2b04741818c51e5e99cc1742251fa93c"},
+]
+
+[package.dependencies]
+colorama = {version = ">=0.4", markers = "sys_platform == \"win32\""}
+iniconfig = ">=1"
+packaging = ">=20"
+pluggy = ">=1.5,<2"
+pygments = ">=2.7.2"
+
+[package.extras]
+dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "requests", "setuptools", "xmlschema"]
+
+[[package]]
+name = "pytest-asyncio"
+version = "1.1.0"
+description = "Pytest support for asyncio"
+optional = false
+python-versions = ">=3.9"
+groups = ["test"]
+files = [
+    {file = "pytest_asyncio-1.1.0-py3-none-any.whl", hash = "sha256:5fe2d69607b0bd75c656d1211f969cadba035030156745ee09e7d71740e58ecf"},
+    {file = "pytest_asyncio-1.1.0.tar.gz", hash = "sha256:796aa822981e01b68c12e4827b8697108f7205020f24b5793b3c41555dab68ea"},
+]
+
+[package.dependencies]
+pytest = ">=8.2,<9"
+
+[package.extras]
+docs = ["sphinx (>=5.3)", "sphinx-rtd-theme (>=1)"]
+testing = ["coverage (>=6.2)", "hypothesis (>=5.7.1)"]
+
 [[package]]
 name = "python-dotenv"
 version = "1.1.0"
@@ -805,4 +901,4 @@ standard = ["colorama (>=0.4) ; sys_platform == \"win32\"", "httptools (>=0.6.3)
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.12"
-content-hash = "7f9ca5ce7505707747e59087ccbb804dc0fef135c963fd2d2ebc8c91285ec188"
+content-hash = "cc2947613c2711ad32ccfa1b8f04a0d8bad6043c6e52bffaff12761ec76cc805"

pyproject.toml

@@ -91,3 +91,7 @@ ignore = [
 [tool.ruff.format]
 # Enable Ruff's formatter.
 docstring-code-format = true
+[tool.poetry.group.test.dependencies]
+pytest = "^8.4.1"
+pytest-asyncio = "^1.1.0"
+

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+pythonpath = . src
+asyncio_mode = auto

src/api/v1/tasks.py

@@ -3,8 +3,7 @@ from typing import Annotated
 from fastapi import APIRouter, Depends
 
 from src.api.dependacies.db_dep import sessionDep
-from src.api.dependacies.user_dep import ActiveUser, CurrentOrAdminTask, TaskOwnerDep
-from src.schemas.auth import TokenData
+from src.api.dependacies.user_dep import ActiveUser, TaskOwnerDep
 from src.schemas.tasks import TaskADDRequest
 from src.services.tasks import TaskService
 from src.services.users import UserService
@@ -19,7 +18,9 @@ async def get_tasks(session: sessionDep, user: ActiveUser):
 
 
 @router.get("/{id}")
-async def get_task_id(id: int): ...
+async def get_task_id(session: sessionDep, id: int, _: TaskOwnerDep):
+    task = await TaskService(session).get_task(id)
+    return task
 
 
 @router.post("/")

src/core/settings.py

@@ -34,7 +34,7 @@ class AccessToken(BaseSettings):
     model_config = SettingsConfigDict(
         env_file=".env", env_file_encoding="utf-8", env_prefix="ACCESS_TOKEN_"
     )
-    expire_minutes: int = 15
+    expire_minutes: int
     secret_key: str
     algorithm: str = "HS256"
     token_type: str = "bearer"  # noqa: S105
@@ -43,7 +43,7 @@ class AccessToken(BaseSettings):
 class Settings(BaseSettings):
     api: ApiPrefix = ApiPrefix()
     db: DbSettings = DbSettings()
-    access_token: AccessToken = AccessToken()
+    access_token: AccessToken = AccessToken()  # type: ignore
 
 
 settings = Settings()

src/repository/base.py

@@ -26,6 +26,7 @@ class BaseRepo(Generic[ModelType]):
         statement = insert(self.model).values(data).returning(self.model)
         result = await self.session.execute(statement)
         obj: ModelType = result.scalar_one()
+        print(obj)
         return obj
 
     async def get_one_or_none(self, **filter_by: Any) -> ModelType | None:

src/services/tasks.py

@@ -19,7 +19,8 @@ class TaskService(BaseService):
         return Task.model_validate(created_task_orm)
 
     async def get_task(self, task_id: int):
-        return await self.session.task.get_one_or_none(id=task_id)
+        task = await self.session.task.get_one_or_none(id=task_id)
+        return Task.model_validate(task)
 
     async def delete_task(self, task_id: int):
         await self.session.task.delete_one(id=task_id)

tests/conftest.py

@@ -0,0 +1,77 @@
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import NullPool, insert
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from src.api.dependacies.db_dep import get_db
+from src.core.auth_manager import AuthManager
+from src.core.database import Base
+from src.core.db_manager import DBManager
+from src.core.settings import settings
+from src.main import app
+from src.models import *  # noqa: F403
+
+engine_null_pool = create_async_engine(
+    "sqlite+aiosqlite:///tests/test_db.db", poolclass=NullPool
+)
+test_session_maker = async_sessionmaker(engine_null_pool, expire_on_commit=False)
+
+
+class TestDBManager(DBManager):
+    def __init__(self):
+        self.session_factory = test_session_maker
+
+
+async def get_test_db():
+    async with TestDBManager() as db:
+        yield db
+
+
+@pytest.fixture(scope="function")
+async def db():
+    async for db in get_test_db():
+        yield db
+
+
+@pytest.fixture(scope="function")
+async def ac():
+    async with AsyncClient(
+        transport=ASGITransport(app=app), base_url="http://test"
+    ) as ac:
+        yield ac
+
+
+app.dependency_overrides[get_db] = get_test_db
+
+
+@pytest.fixture(scope="session", autouse=True)
+async def setup_database():
+    async with engine_null_pool.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+        await conn.run_sync(Base.metadata.create_all)
+
+
+@pytest.fixture(scope="session", autouse=True)
+async def add_admin(setup_database):
+    hashed_pass = AuthManager.get_password_hash("admin")
+    user_admin = {
+        "username": "admin",
+        "hashed_password": hashed_pass,
+        "is_superuser": True,
+    }
+    async with test_session_maker() as conn:
+        result = await conn.execute(
+            insert(UsersORM).values(user_admin).returning(UsersORM)  # noqa: F405
+        )
+        await conn.commit()
+        admin = result.scalar_one()
+    assert admin.is_superuser is True
+
+
+@pytest.fixture
+def auth_token(ac, add_admin):
+    response = ac.post(
+        f"{settings.api.v1_login_url}/login",
+        data={"username": "admin", "password": "admin"},
+    )
+    return response.json()["access_token"]

tests/integration_tests/test_auth_api.py

@@ -0,0 +1,28 @@
+from httpx import AsyncClient
+
+from src.core.settings import settings
+from src.schemas.users import User
+
+
+async def test_registration(ac):
+    user = {"username": "kot", "email": "super@kot.ru", "password": "P@ssw0rd"}
+    result = await ac.post(
+        f"{settings.api.v1_login_url}/signup",
+        json=user,
+    )
+    assert result.status_code == 200
+    assert User.model_validate(result.json())
+    assert result.json()["is_active"]
+
+
+async def test_login(ac: AsyncClient):
+    result = await ac.post(
+        f"{settings.api.v1_login_url}/login",
+        data={
+            "grant_type": "password",
+            "username": "kot",
+            "password": "P@ssw0rd",
+        },
+    )
+    assert result.status_code == 200
+    assert result.json().get("access_token")

tests/unit_tests/test_auth_jwt.py

@@ -0,0 +1,10 @@
+from src.core.auth_manager import AuthManager
+
+
+async def test_jwt():
+    token = AuthManager.create_access_token(
+        data={"id": 1, "sub": "testuser", "is_active": "True"}
+    )
+    assert token
+    encode_token = AuthManager.decode_access_token(token=token)
+    assert encode_token["id"] == 1 and encode_token["sub"] == "testuser"

tests/unit_tests/test_repo_db.py

@@ -0,0 +1,31 @@
+from typing import TYPE_CHECKING
+
+from src.schemas.users import User
+
+if TYPE_CHECKING:
+    from tests.conftest import TestDBManager
+
+
+async def test_user_crud(db: "TestDBManager"):
+    data = {
+        "username": "test",
+        "hashed_password": "hashed_pass",
+        "email": "test@mail.ru",
+        "is_active": True,
+        "is_superuser": False,
+    }
+    user = await db.user.create_one(data=data)
+    assert user.username == data["username"]
+    filtered_user = await db.user.get_filtered(username=data["username"])
+    assert filtered_user[0] == user
+    new_user = User.model_validate(user)
+    new_user.username = "Test2"
+    new_user.email = None
+    await db.user.update_one(id=new_user.id, data=User.model_dump(new_user))
+    updated_user = await db.user.get_one_or_none(id=new_user.id)
+    assert updated_user
+    assert updated_user.username == new_user.username
+    assert not updated_user.email
+    await db.user.delete_one(id=updated_user.id)
+    delete_user = await db.user.get_one_or_none(id=new_user.id)
+    assert not delete_user