MH.Dmitrii/The_DisExcel_project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'dev' (#9) from dev into main

Browse Source 
Reviewed-on: #9
This commit was merged in pull request #9.
This commit is contained in:
MH.Dmitrii
2026-08-24 15:26:58 +00:00
parent 098461cd58 97e4b07c47
commit 862a1eeddc
29 changed files with 1227 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -21,6 +21,8 @@ Thumbs.db
*.env
#db
*.db
#logs
logs/
#Примеры документов
input/
configs/.e2e.env.example
+4
View File
@@ -0,0 +1,4 @@
HOST="change_me"
PORT="change_me"
TEST_USERNAME="change_me"
TEST_PASSWORD="change_me"
docker/compose-dev.yaml
+8 -14
View File
@@ -25,14 +25,11 @@ services:
source: ../DB
target: /home/excel-project/DB
- type: bind
source: ../upload
target: /home/excel-project/upload
source: ../uploads
target: /home/excel-project/uploads
- type: bind
source: ../upload_bad
target: /home/excel-project/upload_bad
- type: bind
source: ../upload_finished
target: /home/excel-project/upload_finished
source: ../logs
target: /home/excel-project/logs
networks:
- backend
ports:
@@ -57,14 +54,11 @@ services:
source: ../DB
target: /home/excel-project/DB
- type: bind
source: ../upload
target: /home/excel-project/upload
source: ../uploads
target: /home/excel-project/uploads
- type: bind
source: ../upload_bad
target: /home/excel-project/upload_bad
- type: bind
source: ../upload_finished
target: /home/excel-project/upload_finished
source: ../logs
target: /home/excel-project/logs
networks:
- backend
restart: unless-stopped
main.py
+7 -3
View File
@@ -4,6 +4,7 @@ from pathlib import Path
# import uvicorn
from fastapi import FastAPI
from src.logging.logger import LoggingMiddleware, ProcessingTimeMiddleware
from src.web.protected_routes.auth_routes import router as protected_router
from src.web.protected_routes.protected_user_action_routes import (
router as protected_user_action_routes,
@@ -18,6 +19,8 @@ async def lifespan(app: FastAPI):
app=FastAPI(root_path="/", lifespan=lifespan)
app.add_middleware(LoggingMiddleware)
app.add_middleware(ProcessingTimeMiddleware)
app.include_router(router=protected_router)
app.include_router(router=protected_user_action_routes)
@@ -30,9 +33,10 @@ async def root()->dict:
def create_dirs():
dirs_to_create=("./DB",
"./upload",
"./upload_bad",
"./upload_finished")
"./uploads/upload",
"./uploads/upload_bad",
"./uploads/upload_finished",
"./logs")
for x in dirs_to_create:
Path(x).mkdir(parents=True, exist_ok=True)
makefile
+4 -1
View File
@@ -13,6 +13,9 @@ help:
##
## Init section
##
.PHONY: run
run: ## Run dev local application
${VENV} uvicorn main:app --host 0.0.0.0 --port 8000 --reload
.PHONY: run-dev
run-dev: ## Run dev application
@@ -72,7 +75,7 @@ allure: ## Generate allure report
.PHONY: coverage
coverage: ## Run pytest coverage
${VENV} pytest --cov=src tests/
${VENV} pytest --cov=src tests/ --cov-report=term-missing
.PHONY: clear
clear: ## Delete old test results
poetry.lock
Generated
+93 -3
View File
@@ -1,5 +1,17 @@
# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand.
[[package]]
name = "aiofiles"
version = "25.1.0"
description = "File support for asyncio."
optional = false
python-versions = ">=3.9"
groups = ["main"]
files = [
{file = "aiofiles-25.1.0-py3-none-any.whl", hash = "sha256:abe311e527c862958650f9438e859c1fa7568a141b22abcd015e120e86a85695"},
{file = "aiofiles-25.1.0.tar.gz", hash = "sha256:a8d728f0a29de45dc521f18f07297428d56992a742f0cd2701ba86e44d23d5b2"},
]
[[package]]
name = "aiosqlite"
version = "0.22.1"
@@ -98,7 +110,7 @@ version = "4.14.2"
description = "High-level concurrency and networking framework on top of asyncio or Trio"
optional = false
python-versions = ">=3.10"
groups = ["main"]
groups = ["main", "dev"]
files = [
{file = "anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494"},
{file = "anyio-4.14.2.tar.gz", hash = "sha256:cfa139f3ed1a23ee8f88a145ddb5ac7605b8bbfd8592baacd7ce3d8bb4313c7f"},
@@ -703,12 +715,34 @@ version = "0.16.0"
description = "A pure-Python, bring-your-own-I/O implementation of HTTP/1.1"
optional = false
python-versions = ">=3.8"
groups = ["main"]
groups = ["main", "dev"]
files = [
{file = "h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86"},
{file = "h11-0.16.0.tar.gz", hash = "sha256:4e35b956cf45792e4caa5885e69fba00bdbc6ffafbfa020300e549b208ee5ff1"},
]
[[package]]
name = "httpcore"
version = "1.0.9"
description = "A minimal low-level HTTP client."
optional = false
python-versions = ">=3.8"
groups = ["dev"]
files = [
{file = "httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55"},
{file = "httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8"},
]
[package.dependencies]
certifi = "*"
h11 = ">=0.16"
[package.extras]
asyncio = ["anyio (>=4.0,<5.0)"]
http2 = ["h2 (>=3,<5)"]
socks = ["socksio (==1.*)"]
trio = ["trio (>=0.22.0,<1.0)"]
[[package]]
name = "httpie"
version = "3.2.4"
@@ -737,6 +771,32 @@ setuptools = "*"
dev = ["Jinja2", "flake8", "flake8-comprehensions", "flake8-deprecated", "flake8-mutable", "flake8-tuple", "pyopenssl", "pytest", "pytest-cov", "pytest-httpbin (>=0.0.6)", "pytest-mock", "pyyaml", "responses", "twine", "werkzeug (<2.1.0)", "wheel"]
test = ["pytest", "pytest-httpbin (>=0.0.6)", "pytest-mock", "responses", "werkzeug (<2.1.0)"]
[[package]]
name = "httpx"
version = "0.28.1"
description = "The next generation HTTP client."
optional = false
python-versions = ">=3.8"
groups = ["dev"]
files = [
{file = "httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad"},
{file = "httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc"},
]
[package.dependencies]
anyio = "*"
certifi = "*"
httpcore = "==1.*"
idna = "*"
socksio = {version = "==1.*", optional = true, markers = "extra == \"socks\""}
[package.extras]
brotli = ["brotli ; platform_python_implementation == \"CPython\"", "brotlicffi ; platform_python_implementation != \"CPython\""]
cli = ["click (==8.*)", "pygments (==2.*)", "rich (>=10,<14)"]
http2 = ["h2 (>=3,<5)"]
socks = ["socksio (==1.*)"]
zstd = ["zstandard (>=0.18.0)"]
[[package]]
name = "idna"
version = "3.18"
@@ -1844,6 +1904,24 @@ urllib3 = ">=1.26,<3"
socks = ["PySocks (>=1.5.6,!=1.5.7)"]
use-chardet-on-py3 = ["chardet (>=3.0.2,<8)"]
[[package]]
name = "requests-async"
version = "0.2.4"
description = "Simple async HTTP client with requests-like interface, powered by httpx"
optional = false
python-versions = ">=3.7"
groups = ["dev"]
files = [
{file = "requests_async-0.2.4-py3-none-any.whl", hash = "sha256:a967efb84eb03a2053a847b1bfbac5917df8100d101368d883e20ae9ab15f7a2"},
{file = "requests_async-0.2.4.tar.gz", hash = "sha256:7eaa42cbfe4d0f1a5f1ef78625c625248cac5c5323afcbd76e7a2a26b85b56b9"},
]
[package.dependencies]
httpx = {version = ">=0.23.0", extras = ["socks"]}
[package.extras]
dev = ["pytest (>=7.0.0)", "pytest-asyncio (>=0.21.0)", "pytest-cov (>=4.0.0)"]
[[package]]
name = "requests-toolbelt"
version = "1.0.0"
@@ -1942,6 +2020,18 @@ files = [
{file = "six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81"},
]
[[package]]
name = "socksio"
version = "1.0.0"
description = "Sans-I/O implementation of SOCKS4, SOCKS4A, and SOCKS5."
optional = false
python-versions = ">=3.6"
groups = ["dev"]
files = [
{file = "socksio-1.0.0-py3-none-any.whl", hash = "sha256:95dc1f15f9b34e8d7b16f06d74b8ccf48f609af32ab33c608d08761c5dcbb1f3"},
{file = "socksio-1.0.0.tar.gz", hash = "sha256:f88beb3da5b5c38b9890469de67d0cb0f9d494b78b106ca1845f96c10b91c4ac"},
]
[[package]]
name = "sqlalchemy"
version = "2.0.51"
@@ -2185,4 +2275,4 @@ files = [
[metadata]
lock-version = "2.1"
python-versions = ">=3.13"
content-hash = "6881b7672e5e0ed6a48d6416f117209ba1d0aefaf3b14db1ccec48360ebc22cb"
content-hash = "ab9787e2afb0885dd340270dcdff2b44a81cbf34bc03fc9ac1e9ffb1fdf476f4"
pyproject.toml
+4 -2
View File
@@ -22,7 +22,8 @@ dependencies = [
"python-jose (>=3.5.0,<4.0.0)",
"python-multipart (>=0.0.32,<0.0.33)",
"aiosqlite (>=0.22.1,<0.23.0)",
"greenlet (>=3.5.4,<4.0.0)"
"greenlet (>=3.5.4,<4.0.0)",
"aiofiles (>=25.1.0,<26.0.0)",
]
[tool.poetry.group.dev.dependencies]
@@ -33,6 +34,7 @@ allure-pytest = ">=2.16.0,<3.0.0"
ipython = ">=9.15.0,<10.0.0"
httpie = ">=3.2.4,<4.0.0"
pytest-asyncio = ">=1.4.0,<2.0.0"
requests-async = ">=0.2.4,<0.3.0"
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
@@ -40,7 +42,7 @@ build-backend = "poetry.core.masonry.api"
[tool.coverage.run]
omit = [
"*/models/*",
# "*/models/*",
"*/migrations/*",
"*/database/*",
"*/errors/*",
pytest.ini
+2
View File
@@ -2,6 +2,7 @@
addopts =
-l
-v
-s
--alluredir=tests/allure-results/reports/
testpaths =
tests
@@ -12,3 +13,4 @@ markers=
smoke: smoke tests
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
src/database/users/crud.py
+97 -2
View File
@@ -1,9 +1,14 @@
from uuid import UUID
from sqlalchemy import select
from sqlalchemy import delete, select, update
from sqlalchemy.ext.asyncio import async_sessionmaker
from src.models.database_models.model import User, engine
from src.models.database_models.model import (
Permissions,
PermissionsGroups,
User,
engine,
)
from src.models.pydantic_models.model import UserOutDB
@@ -13,17 +18,107 @@ class UsersCrudActions:
self.Session=async_sessionmaker(bind=engine)
async def get_user_by_email(self, email:str)->UserOutDB|None:
async with self.Session() as session, session.begin():
query=select(User).where(User.email==email)
response=(await session.scalars(query)).one_or_none()
if response is None:
return None
return UserOutDB.model_validate(response)
async def get_user_by_id(self, id:UUID)->UserOutDB|None:
async with self.Session() as session, session.begin():
query=select(User).where(User.id==id)
response=(await session.scalars(query)).one_or_none()
if response is None:
return None
return UserOutDB.model_validate(response)
async def create_user(self, data:dict)->UserOutDB|None:
async with self.Session() as session, session.begin():
groups_name=data.pop("group", None)
permissions_name=data.pop("direct_permissions", None)
new_user=User(**data)
if groups_name:
query=select(PermissionsGroups).where(PermissionsGroups.group.in_(groups_name))
response=(await session.scalars(query)).all()
if response is None:
new_user.group=[]
else:
new_user.group=list(response)
else:
new_user.group=[]
if permissions_name:
query=select(Permissions).where(Permissions.permission.in_(permissions_name))
response=(await session.scalars(query)).all()
if response is None:
new_user.direct_permissions=[]
else:
new_user.direct_permissions=list(response)
else:
new_user.direct_permissions=[]
session.add(new_user)
await session.flush()
return UserOutDB.model_validate(new_user)
async def delete_user_soft(self, user_email:str)->bool|None:
async with self.Session() as session, session.begin():
query=select(User).where(User.email == user_email)
response=(await session.scalars(query)).one_or_none()
if response is None:
return None
response.status=False
return True
async def delete_user_hard(self, user_email:str)->bool|None:
async with self.Session() as session, session.begin():
query=delete(User).where(User.email == user_email).returning(User)
response=(await session.scalars(query)).one_or_none()
if response is None:
return None
return True
async def update_user_partially(self, user_email:str, data:dict)->UserOutDB|None:
async with self.Session() as session, session.begin():
groups_name=data.pop("group", None)
permissions_name=data.pop("direct_permissions", None)
query = update(User).where(User.email == user_email).values(**data).returning(User)
user_edit=(await session.scalars(query)).one_or_none()
if user_edit is None:
return None
if groups_name is not None:
query=select(PermissionsGroups).where(PermissionsGroups.group.in_(permissions_name))
groups=(await session.scalars(query)).all()
user_edit.group=list(groups)
if permissions_name is not None:
query=select(Permissions).where(Permissions.permission.in_(permissions_name))
groups=(await session.scalars(query)).all()
user_edit.direct_permissions=list(groups)
await session.flush()
return UserOutDB.model_validate(user_edit)
src/logging/__init__.py
+9
View File
@@ -0,0 +1,9 @@
#logger decorators and middlewares
import logging
from .logger import LoggerDB
sql_logger = logging.getLogger("sqlalchemy.engine")
sql_logger.setLevel(logging.INFO)
sql_logger.addHandler(LoggerDB())
src/logging/logger.py
+66
View File
@@ -0,0 +1,66 @@
import asyncio
import json
import logging
from time import gmtime, perf_counter, strftime
from typing import cast
import aiofiles
from fastapi import Request
from starlette.concurrency import iterate_in_threadpool
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response, StreamingResponse
class ProcessingTimeMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next)->Response:
start_time = perf_counter()
response = await call_next(request)
process_time = perf_counter() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
file_time=strftime("%b_%Y", gmtime())
response = await call_next(request)
streaming_response = cast(StreamingResponse, response)
chunks = []
async for chunk in streaming_response.body_iterator:
if isinstance(chunk, str):
chunks.append(chunk.encode())
else:
chunks.append(bytes(chunk))
body_bytes = b"".join(chunks)
streaming_response.body_iterator = iterate_in_threadpool(iter([body_bytes]))
try:
body = json.loads(body_bytes)
body=body.get("detail", None)
except (json.JSONDecodeError, TypeError):
body = None
client_ip = request.headers.get('x-forwarded-for', '').split(',')[0].strip() or (request.client.host if request.client else 'unknown')
async with aiofiles.open(f"./logs/endpoints_log_{file_time}.txt", "a") as file:
await file.write(f"[{current_time}] [{response.status_code}] [{body}] [{client_ip}]\n")
return response
class LoggerDB(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
msg = self.format(record)
asyncio.create_task(self._write(msg))
async def _write(self, msg: str) -> None:
file_time = strftime("%b_%Y", gmtime())
current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
async with aiofiles.open(f"./logs/sql_log_{file_time}.txt", "a") as f:
await f.write(f"[{current_time}] {msg}\n")
src/migrations/env.py
+6 -11
View File
@@ -1,13 +1,13 @@
from logging.config import fileConfig
from alembic import context
from sqlalchemy import create_engine, pool
from src.models.database_models import Model
from src.models.database_models.model import engine
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
sync_url = engine.url.render_as_string(hide_password=False).replace("+aiosqlite", "")
sync_engine = create_engine(sync_url, poolclass=pool.NullPool)
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
@@ -23,7 +23,7 @@ if config.config_file_name is not None:
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Model.metadata
config.set_main_option("sqlalchemy.url", engine.url.render_as_string(hide_password=False))
config.set_main_option("sqlalchemy.url", sync_url)
# other values from the config, defined by the needs of env.py,
# can be acquired:
@@ -62,13 +62,8 @@ def run_migrations_online() -> None:
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
with sync_engine.connect() as connection:
context.configure(
connection=connection, target_metadata=target_metadata, render_as_batch=True
)
src/migrations/versions/ebe80cd28822_.py
+46
View File
@@ -0,0 +1,46 @@
"""empty message
Revision ID: ebe80cd28822
Revises: 74814eb1b7f8
Create Date: 2026-08-21 13:52:49.788041
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ebe80cd28822'
down_revision: Union[str, Sequence[str], None] = '74814eb1b7f8'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table('user_direct_permissions', schema=None) as batch_op:
batch_op.drop_constraint(batch_op.f('fk_user_direct_permissions_user_id_users'), type_='foreignkey')
batch_op.create_foreign_key(batch_op.f('fk_user_direct_permissions_user_id_users'), 'users', ['user_id'], ['id'], ondelete='CASCADE')
with op.batch_alter_table('user_group', schema=None) as batch_op:
batch_op.drop_constraint(batch_op.f('fk_user_group_user_id_users'), type_='foreignkey')
batch_op.create_foreign_key(batch_op.f('fk_user_group_user_id_users'), 'users', ['user_id'], ['id'], ondelete='CASCADE')
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table('user_group', schema=None) as batch_op:
batch_op.drop_constraint(batch_op.f('fk_user_group_user_id_users'), type_='foreignkey')
batch_op.create_foreign_key(batch_op.f('fk_user_group_user_id_users'), 'users', ['user_id'], ['id'])
with op.batch_alter_table('user_direct_permissions', schema=None) as batch_op:
batch_op.drop_constraint(batch_op.f('fk_user_direct_permissions_user_id_users'), type_='foreignkey')
batch_op.create_foreign_key(batch_op.f('fk_user_direct_permissions_user_id_users'), 'users', ['user_id'], ['id'])
# ### end Alembic commands ###
src/models/configs_read/env.py
+1 -1
View File
@@ -12,4 +12,4 @@ class Env(Base):
model_config=SettingsConfigDict(env_file="configs/.env", extra=None)
env_settings=Env()
env_settings=Env() # type: ignore[call-arg]
src/models/database_models/model.py
+3 -3
View File
@@ -15,7 +15,7 @@ from sqlalchemy import (
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
engine = create_async_engine("sqlite+aiosqlite:///DB/database.db", echo=True)
engine = create_async_engine("sqlite+aiosqlite:///DB/database.db")
'''remember as a boilerplate, or just cp/pst'''
class Model(DeclarativeBase):
@@ -75,14 +75,14 @@ class Permissions(Model):
user_group_of_permissions=Table(
"user_group",
Model.metadata,
Column("user_id", ForeignKey("users.id"), primary_key=True),
Column("user_id", ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
Column("permission_group_id", ForeignKey("groups_of_permissions.id"), primary_key=True)
)
user_permission=Table(
"user_direct_permissions",
Model.metadata,
Column("user_id",ForeignKey("users.id"), primary_key=True),
Column("user_id",ForeignKey("users.id", ondelete="CASCADE"), primary_key=True),
Column("permission_id", ForeignKey("permissions.id"), primary_key=True)
)
src/models/pydantic_models/model.py
+19 -4
View File
@@ -2,14 +2,28 @@ from datetime import datetime
from typing import Annotated
from uuid import UUID
from pydantic import BaseModel, EmailStr, Field
from pydantic import AfterValidator, BaseModel, EmailStr, Field
def validate_password(password: str) -> str:
PUNCTUATION: set[str] = {"$", "@", "#", "%", "!", "^", "&", "*", "(", ")", "-", "_", "+", "=", "{", "}", "[", "]"}
if len(password) < 8 or len(password) > 72:
raise ValueError("Password must be 8-72 characters")
if (
not any(c.isupper() for c in password)
or not any(c.islower() for c in password)
or not any(c.isdigit() for c in password)
or not any(c in PUNCTUATION for c in password)
):
raise ValueError("Password must contain uppercase, lowercase, digit and special char")
return password
PasswordStr = Annotated[str, AfterValidator(validate_password)]
class Base(BaseModel):
model_config = {"from_attributes": True}
class PermissionsCreate(Base):
permission:Annotated[str, Field(..., max_length=30, description="permission name")]
@@ -28,6 +42,7 @@ class PermissionsGroupsOut(Base):
id:Annotated[int, Field(..., description="id of the permission group")]
group:Annotated[str, Field(..., max_length=255, description="group name for the permissions")]
permissions: Annotated[list[PermissionsOut], Field(..., description="permissions granted by this group")]
class UserCreate(Base):
@@ -35,8 +50,7 @@ class UserCreate(Base):
last_name:Annotated[str, Field(...,max_length=64, description="last name of the user")]
middle_name:Annotated[str, Field(...,max_length=64, description="middle name of the user")]
email:Annotated[EmailStr, Field(...,min_length=5, max_length=255, description="email of the user")]
plain_password:Annotated[str, Field(...,min_length=8,max_length=72, description="plain password of the user")]
status:Annotated[bool, Field(..., description="status of the user")]
plain_password:Annotated[PasswordStr, Field(..., description="plain password of the user")]
direct_permissions:Annotated[list[str], Field(..., description="permissions of the user")]
group:Annotated[list[str], Field(..., description="permissions groups of the user")]
@@ -63,6 +77,7 @@ class UserUpdate(Base):
last_name:Annotated[str|None, Field(None, max_length=64,description="last name of the user")]
middle_name:Annotated[str|None, Field(None, max_length=64,description="middle name of the user")]
email:Annotated[EmailStr|None, Field(None, min_length=5, max_length=255, description="email of the user")]
plain_password:Annotated[PasswordStr|None, Field(None, description="plain password of the user")]
status:Annotated[bool|None, Field(None, description="status of the user")]
direct_permissions:Annotated[list[str]|None, Field(None, description="permissions of the user")]
group:Annotated[list[str]|None, Field(None, description="permissions groups of the user")]
src/service/auth/auth.py
+7 -1
View File
@@ -49,7 +49,7 @@ class CurrentUserService:
)
async def get_current_user(self, token:str)->UserOut:
async def get_current_user(self, token:str, *permissions: str)->UserOut:
payload= await self.jwt_service.jwt_decode(token)
sub=payload.get("sub")
@@ -69,6 +69,12 @@ class CurrentUserService:
if user.status is False:
raise self.error.credentials_error(detail="This user is deactivated")
effective = {p.permission for p in user.direct_permissions} | {p.permission for group in user.group for p in group.permissions}
missing = set(permissions) - effective
if missing:
raise self.error.forbidden_error(detail=f"Missing: {missing}")
return UserOut.model_validate(user)
src/service/users_crud/users_crud.py
+62 -1
View File
@@ -1,6 +1,9 @@
import asyncio
from src.database.users.crud import UsersCrudActions
from src.errors.http_errors.errors import Errors
from src.models.pydantic_models.model import UserOut
from src.models.pydantic_models.model import UserCreate, UserOut, UserUpdate
from src.service.auth.jwt import HashService
class CrudService:
@@ -8,12 +11,70 @@ class CrudService:
def __init__(self) -> None:
self.errors=Errors()
self.crud_db_actions=UsersCrudActions()
self.hash_service=HashService()
async def _plain_to_hash(self, user_data:dict)->dict:
user_data["hashed_password"]=user_data.pop("plain_password")
user_data["hashed_password"]= await asyncio.to_thread(self.hash_service.plain_to_hash, user_data["hashed_password"])
return user_data
async def get_user_by_email(self, email:str)->UserOut:
user_entity=await self.crud_db_actions.get_user_by_email(email)
if not user_entity:
raise self.errors.not_found_error(detail="User wasn't found")
return UserOut.model_validate(user_entity)
async def create_user(self, data:UserCreate)->UserOut:
user_data=UserCreate.model_dump(data)
user_data=await self._plain_to_hash(user_data)
user_entity=await self.crud_db_actions.create_user(user_data)
if not user_entity:
raise self.errors.validation_error(detail="User creation gone wrong")
return UserOut.model_validate(user_entity)
async def delete_user_soft(self, email:str)->bool:
user_entity=await self.crud_db_actions.delete_user_soft(email)
if not user_entity:
raise self.errors.not_found_error(detail="User not found")
return user_entity
async def delete_user_hard(self, email:str, current_user)->bool:
user_entity=await self.crud_db_actions.delete_user_hard(email)
if not user_entity:
raise self.errors.not_found_error(detail="User not found")
return user_entity
async def update_user(self, email:str, data:UserUpdate)->UserOut:
user_data=UserUpdate.model_dump(data, exclude_unset=True)
if not user_data:
raise self.errors.bad_request_error(detail="User info to update can not be empty")
if user_data.get("plain_password"):
user_data=await self._plain_to_hash(user_data)
user_entity=await self.crud_db_actions.update_user_partially(email, user_data)
if not user_entity:
raise self.errors.not_found_error(detail="User not found")
return UserOut.model_validate(user_entity)
async def crud_service()->CrudService:
return CrudService()
src/web/protected_routes/auth_routes.py
+12 -7
View File
@@ -8,6 +8,16 @@ from src.service.auth.auth import CurrentUserService, auth_service
router=APIRouter(prefix="/protected")
oauth2_schema=OAuth2PasswordBearer(tokenUrl="/protected/token", refreshUrl="/protected/refresh")
def require_permissions(*permissions: str): #permissions check dependency
async def checker(
token: str = Depends(oauth2_schema),
auth: CurrentUserService = Depends(auth_service), #noqa: B008
) -> UserOut:
return UserOut.model_validate(await auth.get_current_user(token, *permissions))
return checker
@router.post("/token")
async def get_access_token(request: Request,response:Response,auth:CurrentUserService=Depends(auth_service), form_data:OAuth2PasswordRequestForm=Depends())->dict: # noqa: B008
@@ -40,16 +50,11 @@ async def get_refresh_token(request:Request,response:Response, refresh_token: st
return {"access_token":access_token, "token_type": "bearer"}
async def get_current_user(token:str = Depends(oauth2_schema), auth:CurrentUserService=Depends(auth_service)) -> UserOut: # noqa: B008
return UserOut.model_validate(await auth.get_current_user(token))
@router.get("/logout")
async def logout(response:Response,refresh_token: str = Cookie(),auth:CurrentUserService=Depends(auth_service),current_user:UserOut=Depends(get_current_user))->bool: # noqa: B008
async def logout(response:Response,refresh_token: str = Cookie(),auth:CurrentUserService=Depends(auth_service),current_user:UserOut=Depends(require_permissions()))->bool: # noqa: B008
response.delete_cookie("refresh_token")
return await auth.logout(refresh_token)
@router.get("")
async def protected(current_user:UserOut=Depends(get_current_user))->dict: # noqa: B008
async def protected(current_user:UserOut=Depends(require_permissions()))->dict: # noqa: B008
return {"protected router": "Hello, this is a protected router"}
src/web/protected_routes/protected_user_action_routes.py
+20 -3
View File
@@ -1,11 +1,28 @@
from fastapi import APIRouter, Depends
from src.models.pydantic_models.model import UserOut
from src.models.pydantic_models.model import UserCreate, UserOut, UserUpdate
from src.service.users_crud.users_crud import CrudService, crud_service
from src.web.protected_routes.auth_routes import get_current_user
from src.web.protected_routes.auth_routes import require_permissions
router=APIRouter(prefix="/user")
@router.get("/get_by_email")
async def get_current_user_by_email(email:str, crud:CrudService=Depends(crud_service), current_user=Depends(get_current_user))->UserOut: # noqa: B008
async def get_current_user_by_email(email:str, crud:CrudService=Depends(crud_service), current_user=Depends(require_permissions("admin")))->UserOut: # noqa: B008
return await crud.get_user_by_email(email)
@router.post("/create_user")
async def create_user(data:UserCreate, crud:CrudService=Depends(crud_service), current_user=Depends(require_permissions("admin")))->UserOut: #noqa: B008
return await crud.create_user(data)
@router.post("/delete_user_soft")
async def delete_user_soft(email:str, crud:CrudService=Depends(crud_service), current_user=Depends(require_permissions("admin")))->bool: #noqa: B008
return await crud.delete_user_soft(email)
@router.post("/delete_user_hard")
async def delete_user_hard(email:str, crud:CrudService=Depends(crud_service), current_user=Depends(require_permissions("admin")))->bool: #noqa: B008
return await crud.delete_user_hard(email, current_user)
@router.patch("/patch_user")
async def patch_user(email:str, data:UserUpdate, crud:CrudService=Depends(crud_service), current_user=Depends(require_permissions("admin")))->UserOut: #noqa: B008
return await crud.update_user(email, data)
tests/e2e/conftest.py
+77
View File
@@ -0,0 +1,77 @@
from uuid import uuid4
import pytest_asyncio
import requests_async
from pydantic_settings import BaseSettings, SettingsConfigDict
class Env(BaseSettings):
HOST:str
PORT:str
TEST_USERNAME:str
TEST_PASSWORD:str
model_config=SettingsConfigDict(env_file="configs/.e2e.env", extra=None)
e2e_settings=Env() # type: ignore[call-arg]
class MySession(requests_async.AsyncSession):
def __init__(self):
super().__init__()
self.headers = {}
self.token = None
async def request(self, method:str, url:str, **kwargs):
if self.token:
self.headers['Authorization'] = f"Bearer {self.token}"
kwargs.setdefault('headers', self.headers)
return await super().request(method, url, **kwargs)
@pytest_asyncio.fixture(scope="function")
async def auth_fixture(target_url:str):
payload = {"username": e2e_settings.TEST_USERNAME, "password": e2e_settings.TEST_PASSWORD}
async with MySession() as session:
response = await session.post(target_url + "/protected/token", data=payload)
response.raise_for_status()
session.token = response.json()["access_token"]
yield session
@pytest_asyncio.fixture(scope="function")
async def test_user_fixture(request, auth_fixture: MySession, target_url:str):
test_id=uuid4()
direct_permission_param, group_param = request.param
new_user_record={
"first_name":f"TEST_{test_id}",
"last_name":f"TEST_{test_id}",
"middle_name":f"TEST_{test_id}",
"email":f"TEST_{test_id}@d.d",
"plain_password":"Test1234!",
"direct_permissions":direct_permission_param,
"group":group_param
}
response=await auth_fixture.post(f"{target_url}/user/create_user", json=new_user_record)
response.raise_for_status()
async with MySession() as session:
payload={"username": new_user_record.get("email"), "password": new_user_record.get("plain_password")}
response = await session.post(target_url + "/protected/token", data=payload)
response.raise_for_status()
session.token = response.json()["access_token"]
yield (session, new_user_record)
response=await auth_fixture.post(f"{target_url}/user/delete_user_hard", params={"email":new_user_record.get("email")})
response.raise_for_status()
@pytest_asyncio.fixture()
async def target_url()->str:
return f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
tests/e2e/test_auth.py
+46
View File
@@ -0,0 +1,46 @@
import allure
import pytest
import requests_async
from httpx import HTTPStatusError
class TestAuth:
async def test_get_access_token_positive(self, target_url:str)->None:
with allure.step("get_access_token"), pytest.raises(HTTPStatusError) as exc_info:
response = await requests_async.post(f"{target_url}/protected/token")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
assert exc_info.value.response.status_code != 401
async def test_get_refresh_token_positive(self, target_url:str)->None:
with allure.step("get_refresh_token"), pytest.raises(HTTPStatusError) as exc_info:
response = await requests_async.post(f"{target_url}/protected/refresh")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
assert exc_info.value.response.status_code != 401
async def test_get_root_positive(self, target_url:str)->None:
with allure.step("get_root"), pytest.raises(HTTPStatusError) as exc_info:
response = await requests_async.get(f"{target_url}/protected")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
assert exc_info.value.response.status_code == 401
@pytest.mark.parametrize("test_user_fixture", [([], [])], indirect=True)
async def test_get_logout_positive(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("get_root"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.get(f"{target_url}/protected/logout")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
assert exc_info.value.response.status_code != 401
tests/e2e/test_users_crud.py
+383
View File
@@ -0,0 +1,383 @@
from uuid import uuid4
import allure
import pytest
from httpx import HTTPStatusError
from tests.e2e.conftest import MySession
@pytest.mark.integra
class TestCrud:
@pytest.mark.parametrize("test_user_fixture", [(["admin"], ["admin_group"])], indirect=True)
async def test_get_user_by_email_positive(self,test_user_fixture, target_url:str)->None:
session, new_user_record=test_user_fixture
with allure.step("Get user by email"):
email = new_user_record.get("email") #get email from the fixture in yield sector
response = await session.get(f"{target_url}/user/get_by_email",params={"email":email})
response.raise_for_status()
response=response.json()
with allure.step("Validate response"):
assert response.get("email")==email
assert "TEST_" in response.get("first_name")
assert "TEST_" in response.get("last_name")
assert "TEST_" in response.get("middle_name")
assert response.get("direct_permissions") != []
assert response.get("group") != []
assert not response.get("hashed_password") or not response.get("plain_password") or not response.get("password")
@pytest.mark.parametrize("email, expected_status", [
pytest.param("test@test.test", 404, id="non_existed_email"),
pytest.param("test",404, id="wrong_email"),
pytest.param("@d", 404, id="wrong_email")
])
async def test_get_user_by_email_negative(self, email:str, expected_status:int, auth_fixture:MySession,target_url:str)->None:
with allure.step("Get user by email"):
with pytest.raises(HTTPStatusError) as exc_info:
response = await auth_fixture.get(f"{target_url}/user/get_by_email", params={"email": email})
response.raise_for_status()
assert exc_info.value.response.status_code == expected_status
@pytest.mark.parametrize("new_user_record",[
pytest.param({ "first_name":"TEST",
"last_name":"TEST",
"middle_name":"TEST",
"email":f"TEST_{uuid4()}@d.d",
"plain_password":"Test1234!",
"direct_permissions":[],
"group":[]}, id="Positive_user_creation_with_all_the_fields"),
pytest.param({ "first_name":"TEST",
"last_name":"TEST",
"middle_name":"TEST",
"email":f"TEST_{uuid4()}@d.d",
"plain_password":"Test1234!",
"direct_permissions":["WRONG_PERMISSIONS"],
"group":["WRONG_GROUP"]},id="Positive_wrong_permissions"),
])
async def test_create_delete_user_hard_positive(self, new_user_record:dict,auth_fixture:MySession, target_url:str)->None:
with allure.step("Create new test user and check for the new user"):
response=await auth_fixture.post(f"{target_url}/user/create_user", json=new_user_record)
response.raise_for_status()
try:
with allure.step("Check for the new user"):
response = await auth_fixture.get(f"{target_url}/user/get_by_email",params={"email":new_user_record.get("email")})
response.raise_for_status()
response=response.json()
assert response.get("email")==new_user_record["email"]
assert response.get("first_name")==new_user_record["first_name"]
assert response.get("last_name")==new_user_record["last_name"]
assert response.get("middle_name")==new_user_record["middle_name"]
assert response.get("direct_permissions") == new_user_record["direct_permissions"] or response.get("direct_permissions") == []
assert response.get("group") == new_user_record["group"] or response.get("group") == []
assert not response.get("hashed_password") or not response.get("plain_password") or not response.get("password")
finally:
with allure.step("delete new user"):
response=await auth_fixture.post(f"{target_url}/user/delete_user_hard", params={"email":new_user_record["email"]})
response.raise_for_status()
@pytest.mark.parametrize("new_user_record, expected_status", [
pytest.param({ "first_name":"TEST",
"last_name":"TEST",
"middle_name":"TEST",
"email":"WRONGEMAIL",
"plain_password":"Test1234!",
"direct_permissions":[],
"group":[]},422,id="Non_existed_email"),
pytest.param({ "first_name":"TEST",
"last_name":"TEST",
"middle_name":"TEST",
"email":"TEST1@d.d",
"plain_password":"1234",
"direct_permissions":[],
"group":[]},422,id="Wrong_password"),
pytest.param({ "first_name":"TEST",
"email":"TEST1@d.d",
"plain_password":"Test1234!",
},422,id="Not_all_the_fields"),
])
async def test_create_user_negative(self, new_user_record:dict, auth_fixture:MySession, expected_status:int, target_url:str):
with allure.step("Preparing data to create new user negative"):
user_created=False
try:
with allure.step("Create new test user and check for the new user"):
response=await auth_fixture.post(f"{target_url}/user/create_user", json=new_user_record)
if response.status_code < 400:
user_created = True
with pytest.raises(HTTPStatusError) as exc_info:
response.raise_for_status()
assert exc_info.value.response.status_code == expected_status
finally:
if user_created:
with allure.step("delete new user"):
response = await auth_fixture.post(
f"{target_url}/user/delete_user_hard",
params={"email": new_user_record["email"]}
)
response.raise_for_status()
@pytest.mark.parametrize("new_user_record",[
pytest.param({ "first_name":"TEST",
"last_name":"TEST",
"middle_name":"TEST",
"email":f"TEST_{uuid4()}@d.d",
"plain_password":"Test1234!",
"direct_permissions":[],
"group":[]}, id="Positive_user_delete_soft"),
])
async def test_user_create_delete_soft_positive(self, new_user_record:dict, auth_fixture:MySession, target_url:str)->None:
with allure.step("Create new test user and check for the new user"):
response=await auth_fixture.post(f"{target_url}/user/create_user", json=new_user_record)
response.raise_for_status()
try:
with allure.step("Check for the new user"):
response = await auth_fixture.get(f"{target_url}/user/get_by_email",params={"email":new_user_record.get("email")})
response.raise_for_status()
with allure.step("Delete user soft"):
response=await auth_fixture.post(f"{target_url}/user/delete_user_soft", params={"email":new_user_record["email"]})
response.raise_for_status()
finally:
with allure.step("delete new user"):
response=await auth_fixture.post(f"{target_url}/user/delete_user_hard", params={"email":new_user_record["email"]})
response.raise_for_status()
@pytest.mark.parametrize("email, expected_status, ",[
pytest.param("Test", 404,id="Wrong_email")
])
async def test_user_delete_soft_negative(self, email:str,expected_status:int, auth_fixture:MySession, target_url:str)->None:
with allure.step("Delete user soft"):
with pytest.raises(HTTPStatusError) as exc_info:
response=await auth_fixture.post(f"{target_url}/user/delete_user_soft", params={"email":email})
response.raise_for_status()
assert exc_info.value.response.status_code == expected_status
@pytest.mark.parametrize("email, expected_status, ",[
pytest.param("Test", 404,id="Wrong_email")
])
async def test_user_delete_hard_negative(self, email:str,expected_status:int, auth_fixture:MySession, target_url:str)->None:
with allure.step("Delete user hard"):
with pytest.raises(HTTPStatusError) as exc_info:
response=await auth_fixture.post(f"{target_url}/user/delete_user_soft", params={"email":email})
response.raise_for_status()
assert exc_info.value.response.status_code == expected_status
@pytest.mark.parametrize("test_user_fixture", [
(["admin"], ["admin_group"])
], indirect=True)
@pytest.mark.parametrize("user_record_to_update",[
pytest.param({"first_name": "Test_New"},id="Positive_user_update_partially")
])
async def test_user_update_partially_positive(self, test_user_fixture, user_record_to_update:dict, target_url:str)->None:
session, new_user_record=test_user_fixture
with allure.step("Update user"):
response= await session.patch(f"{target_url}/user/patch_user", json=user_record_to_update, params={"email":new_user_record["email"]})
response.raise_for_status()
with allure.step("Check for the updated user"):
response=await session.get(f"{target_url}/user/get_by_email", params={"email":new_user_record["email"]})
response.raise_for_status()
response=response.json()
actual_permissions = [item.get("permission") for item in response.get("direct_permissions")] #unpacking json like {group:[{},{}]}
actual_groups =[item.get("group") for item in response.get("group")]
assert response.get("email")==new_user_record["email"]
assert response.get("first_name")==user_record_to_update["first_name"]
assert response.get("last_name")==new_user_record["last_name"]
assert response.get("middle_name")==new_user_record["middle_name"]
assert actual_permissions == new_user_record["direct_permissions"]
assert actual_groups == new_user_record["group"]
assert not response.get("hashed_password") or not response.get("plain_password") or not response.get("password")
@pytest.mark.parametrize("test_user_fixture", [
(["admin"], ["admin_group"])
], indirect=True)
@pytest.mark.parametrize("user_record_to_update, expected_exception, expected_status",[
pytest.param({"plain_password": "Wrong_pass"},HTTPStatusError,422,id="Wrong_password"),
pytest.param({"email": "Wrong_email"},HTTPStatusError,422,id="Wrong_email"),
pytest.param({},HTTPStatusError, 400,id="Negative_user_update_nothing")
])
async def test_user_update_partially_negative(self, test_user_fixture, user_record_to_update:dict, expected_exception, expected_status:int, target_url:str)->None:
session, new_user_record=test_user_fixture
with allure.step("Update user"), pytest.raises(expected_exception) as exc_info:
response= await session.patch(f"{target_url}/user/patch_user", json=user_record_to_update, params={"email":new_user_record["email"]})
response.raise_for_status()
assert exc_info.value.response.status_code == expected_status
@pytest.mark.parametrize("test_user_fixture", [(["admin"], ["admin_group"])], indirect=True)
async def test_get_user_by_email_permissions_positive(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.get(f"{target_url}/user/get_by_email")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
@pytest.mark.parametrize("test_user_fixture", [([], [])], indirect=True)
async def test_get_user_by_email_permissions_negative(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.get(f"{target_url}/user/get_by_email")
response.raise_for_status()
assert exc_info.value.response.status_code == 403
@pytest.mark.parametrize("test_user_fixture", [(["admin"], ["admin_group"])], indirect=True)
async def test_create_user_permissions_positive(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.post(f"{target_url}/user/create_user")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
@pytest.mark.parametrize("test_user_fixture", [([], [])], indirect=True)
async def test_create_user_permissions_negative(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.post(f"{target_url}/user/create_user")
response.raise_for_status()
assert exc_info.value.response.status_code == 403
@pytest.mark.parametrize("test_user_fixture", [(["admin"], ["admin_group"])], indirect=True)
async def test_patch_user_permissions_positive(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.patch(f"{target_url}/user/patch_user")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
@pytest.mark.parametrize("test_user_fixture", [([], [])], indirect=True)
async def test_patch_user_permissions_negative(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.patch(f"{target_url}/user/patch_user")
response.raise_for_status()
assert exc_info.value.response.status_code == 403
@pytest.mark.parametrize("test_user_fixture", [(["admin"], ["admin_group"])], indirect=True)
async def test_delete_user_soft_permissions_positive(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.post(f"{target_url}/user/delete_user_soft")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
@pytest.mark.parametrize("test_user_fixture", [([], [])], indirect=True)
async def test_delete_user_soft_permissions_negative(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.post(f"{target_url}/user/delete_user_soft")
response.raise_for_status()
assert exc_info.value.response.status_code == 403
@pytest.mark.parametrize("test_user_fixture", [(["admin"], ["admin_group"])], indirect=True)
async def test_delete_user_hard_permissions_positive(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.post(f"{target_url}/user/delete_user_hard")
response.raise_for_status()
assert exc_info.value.response.status_code != 403
@pytest.mark.parametrize("test_user_fixture", [([], [])], indirect=True)
async def test_delete_user_hard_permissions_negative(self,test_user_fixture, target_url:str)->None:
session=test_user_fixture[0]
with allure.step("Get user by email"), pytest.raises(HTTPStatusError) as exc_info:
response = await session.post(f"{target_url}/user/delete_user_hard")
response.raise_for_status()
assert exc_info.value.response.status_code == 403
tests/integrated/conftest.py
+6
View File
@@ -2,6 +2,7 @@ import pytest_asyncio
from fastapi import Request
from src.service.auth.auth import CurrentUserService
from src.service.users_crud.users_crud import CrudService
@pytest_asyncio.fixture
@@ -14,3 +15,8 @@ async def requests(mocker):
fake_request = mocker.MagicMock(spec=Request)
fake_request.headers = {"user-agent": "pytest-agent", "x-forwarded-for":"127.0.0.1"}
return fake_request
@pytest_asyncio.fixture
async def crud_service()->CrudService:
crud_service=CrudService()
return crud_service
tests/integrated/test_auth.py
+39 -28
View File
@@ -41,14 +41,16 @@ class TestAuth:
assert test_result.email==user_data.email
assert test_result.direct_permissions==user_data.direct_permissions
assert test_result.group==user_data.group
assert not hasattr(test_result, "password") or not hasattr(test_result, "plain_password") or not hasattr(test_result, "hashed_password")
assert not hasattr(test_result, "status")
@pytest.mark.parametrize("user_data, uuid, expected_exception",[
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=False), uuid4(), HTTPException, id="false_status"),
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True),1234, HTTPException, id="wrong_id"),
pytest.param(SimpleNamespace(status=True),uuid4(), ValidationError,id="empty_model_data")
@pytest.mark.parametrize("user_data, uuid, expected_exception,expected_status",[
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=False), uuid4(), HTTPException,401, id="false_status"),
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True),1234, HTTPException,401, id="wrong_id"),
pytest.param(SimpleNamespace(status=True,direct_permissions=[],group=[]),uuid4(), ValidationError,None,id="empty_model_data")
])
async def test_get_current_user_negative(self,current_user_service:CurrentUserService, jwt_service:JwtService, monkeypatch, user_data:SimpleNamespace, expected_exception, uuid)->None:
async def test_get_current_user_negative(self,current_user_service:CurrentUserService, jwt_service:JwtService, monkeypatch, user_data:SimpleNamespace,expected_exception, expected_status:int, uuid)->None:
with allure.step("create token"):
@@ -58,10 +60,11 @@ class TestAuth:
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", AsyncMock(return_value=user_data))
with allure.step("test get_current_user_with_fake_token"), pytest.raises(expected_exception):
with allure.step("test get_current_user_with_fake_token"), pytest.raises(expected_exception) as exc_info:
await current_user_service.get_current_user(token)
if expected_exception is HTTPException:
assert exc_info.value.status_code == expected_status
@pytest.mark.parametrize("user_data, form_data_email,form_data_password",[
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "1234", id="correct_data"),
@@ -87,12 +90,12 @@ class TestAuth:
assert len(parts_b)==3
@pytest.mark.parametrize("user_data, form_data_email,form_data_password, expected_exception",[
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "wrong_password", HTTPException, id="wrong_password"),
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=False), "d@d.d", "1234",HTTPException, id="false_status"),
pytest.param(SimpleNamespace(id=1234,hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "1234",ValidationError, id="wrong_id"),
@pytest.mark.parametrize("user_data, form_data_email,form_data_password, expected_exception, expected_status",[
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "wrong_password", HTTPException,401, id="wrong_password"),
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=False), "d@d.d", "1234",HTTPException,401, id="false_status"),
pytest.param(SimpleNamespace(id=1234,hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "1234",ValidationError,None, id="wrong_id"),
])
async def test_login_negative(self, current_user_service:CurrentUserService, user_data:SimpleNamespace, jwt_service:JwtService, monkeypatch, requests, hash_service:HashService, form_data_email:str, form_data_password:str, expected_exception):
async def test_login_negative(self, current_user_service:CurrentUserService, user_data:SimpleNamespace, jwt_service:JwtService, monkeypatch, requests, hash_service:HashService, form_data_email:str, form_data_password:str, expected_exception, expected_status:int):
with allure.step("patching db call functions"):
@@ -103,10 +106,12 @@ class TestAuth:
fake_request = requests
with allure.step("test login_with_fake_data"), pytest.raises(expected_exception):
with allure.step("test login_with_fake_data"), pytest.raises(expected_exception) as exc_info:
await current_user_service.login(form_data_email, form_data_password,fake_request)
if expected_exception is HTTPException:
assert exc_info.value.status_code == expected_status
async def test_logout_positive(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService)->None:
@@ -124,12 +129,12 @@ class TestAuth:
status=await current_user_service.logout(token[0])
assert status is True
@pytest.mark.parametrize("jti,db_result, expected_exception",[
pytest.param(None, True, HTTPException, id="jti_none"),
pytest.param(1234, True, HTTPException, id="jti_int"),
pytest.param(str(uuid4()), False, HTTPException, id="db_result_none"),
@pytest.mark.parametrize("jti,db_result, expected_exception, expected_status",[
pytest.param(None, True, HTTPException,401, id="jti_none"),
pytest.param(1234, True, HTTPException,401, id="jti_int"),
pytest.param(str(uuid4()), False, HTTPException,404,id="db_result_none"),
])
async def test_logout_negative(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService, expected_exception, jti, db_result)->None:
async def test_logout_negative(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService, expected_exception, jti, db_result, expected_status:int)->None:
with allure.step("patching db call functions"):
@@ -145,10 +150,13 @@ class TestAuth:
token=await fake_create_refresh_token({"sub":str(uuid4()), "jti":jti, "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)})
with allure.step("test logout with fake data"), pytest.raises(expected_exception):
with allure.step("test logout with fake data"), pytest.raises(expected_exception) as exc_info:
await current_user_service.logout(token)
if expected_exception is HTTPException:
assert exc_info.value.status_code==expected_status
@pytest.mark.parametrize("db_result_token, user_data_result_db", [
pytest.param(SimpleNamespace(is_revoked=False, expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True), id="correct_data")
@@ -186,15 +194,15 @@ class TestAuth:
assert new_access_token!=token
assert new_refresh_token!=token
@pytest.mark.parametrize("db_result_token, user_data_result_db, update_result, fake_token_data,expected_exception", [
pytest.param(SimpleNamespace(is_revoked=True,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),False,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="false_revoke_status"),
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException, id="jti_missing"),
pytest.param(None,SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()),"token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException, id="token_missing"),
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=False),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="false_user_status"),
pytest.param(SimpleNamespace(is_revoked=False, user_id="123",expires_at=datetime.now(UTC)+timedelta(days=15)),None,True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="user_missing"),
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)-timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="wrong_exp")
@pytest.mark.parametrize("db_result_token, user_data_result_db, update_result, fake_token_data,expected_exception, expected_status", [
pytest.param(SimpleNamespace(is_revoked=True,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),False,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,404,id="false_revoke_status"),
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,401, id="jti_missing"),
pytest.param(None,SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()),"token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException, 404,id="token_missing"),
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=False),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,401,id="false_user_status"),
pytest.param(SimpleNamespace(is_revoked=False, user_id="123",expires_at=datetime.now(UTC)+timedelta(days=15)),None,True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,404,id="user_missing"),
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)-timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,401,id="wrong_exp")
])
async def test_refresh_token_negative(self, monkeypatch, current_user_service:CurrentUserService, db_result_token, requests, jwt_service:JwtService,user_data_result_db, expected_exception, fake_token_data, update_result)->None:
async def test_refresh_token_negative(self, monkeypatch, current_user_service:CurrentUserService, db_result_token, requests, jwt_service:JwtService,user_data_result_db, expected_exception, fake_token_data, update_result, expected_status:int)->None:
with allure.step("patching db call functions"):
monkeypatch.setattr(current_user_service.jwt_db_actions,"get_token_by_id", AsyncMock(return_value=db_result_token))
monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", AsyncMock(return_value=True))
@@ -211,5 +219,8 @@ class TestAuth:
with allure.step("create fake refresh token"):
token=await fake_create_refresh_token(fake_token_data)
with allure.step("test refresh token with fake data"), pytest.raises(expected_exception):
with allure.step("test refresh token with fake data"), pytest.raises(expected_exception) as exc_info:
await current_user_service.refresh_token(token, fake_request)
if expected_exception is HTTPException:
assert exc_info.value.status_code==expected_status
tests/integrated/test_users_crud.py
+68
View File
@@ -0,0 +1,68 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock
import allure
import pytest
from fastapi import HTTPException
from src.models.pydantic_models.model import UserCreate, UserUpdate
from src.service.users_crud.users_crud import CrudService
@pytest.mark.integra
class TestCrud:
@pytest.mark.parametrize("user_data", [
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[], plain_password="Test1234!"),id="Create_user_positive")
])
async def test_create_user_positive(self, crud_service:CrudService, monkeypatch, user_data:SimpleNamespace)->None:
with allure.step("Patching functions"):
monkeypatch.setattr(crud_service.crud_db_actions, "create_user", AsyncMock(return_value=user_data))
with allure.step("Test Create User"):
await crud_service.create_user(UserCreate.model_validate(user_data))
@pytest.mark.parametrize("user_data,expected_exception,expected_status ", [
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[], plain_password="Test1234!"),HTTPException,422,id="Create_user_None")
])
async def test_create_user_negative(self, crud_service:CrudService, monkeypatch, user_data:SimpleNamespace, expected_exception, expected_status:int)->None:
with allure.step("Patching functions"):
monkeypatch.setattr(crud_service.crud_db_actions, "create_user", AsyncMock(return_value=None))
with allure.step("Test Create User"), pytest.raises(expected_exception) as exc_info:
await crud_service.create_user(UserCreate.model_validate(user_data))
if expected_exception is HTTPException:
assert exc_info.value.status_code == expected_status
@pytest.mark.parametrize("user_data", [
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[], plain_password="Test1234!"),id="Create_user_positive")
])
async def test_update_user_positive(self, crud_service:CrudService, monkeypatch, user_data:SimpleNamespace)->None:
with allure.step("Patching functions"):
monkeypatch.setattr(crud_service.crud_db_actions, "update_user_partially", AsyncMock(return_value=user_data))
with allure.step("Test Update User"):
await crud_service.update_user(user_data.email,UserUpdate.model_validate(user_data))
@pytest.mark.parametrize("user_data,expected_exception,expected_status ", [
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[], plain_password="Test1234!"),HTTPException,404,id="Update_user_None"),
pytest.param(SimpleNamespace(),HTTPException,400,id="Empty_user_data")
])
async def test_update_user_negative(self, crud_service:CrudService, monkeypatch, user_data:SimpleNamespace, expected_exception, expected_status:int)->None:
with allure.step("Patching functions"):
monkeypatch.setattr(crud_service.crud_db_actions, "update_user_partially", AsyncMock(return_value=None))
with allure.step("Test Update User"), pytest.raises(expected_exception) as exc_info:
await crud_service.update_user("d@d.d",UserUpdate.model_validate(user_data))
if expected_exception is HTTPException:
assert exc_info.value.status_code == expected_status
tests/unit/conftest.py
+7
View File
@@ -1,2 +1,9 @@
import pytest_asyncio
from src.service.users_crud.users_crud import CrudService
@pytest_asyncio.fixture
async def crud_service()->CrudService:
crud_service=CrudService()
return crud_service
tests/unit/test_crud.py
+103
View File
@@ -0,0 +1,103 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock
import allure
import pytest
from fastapi import HTTPException
from src.service.users_crud.users_crud import CrudService
@pytest.mark.unit
class TestCrud:
@pytest.mark.parametrize("user_data",[
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True),id="Get_user_by_email_positive")
])
async def test_get_user_by_email_positive(self, monkeypatch, user_data:SimpleNamespace, crud_service:CrudService)->None:
with allure.step("Patching functions"):
monkeypatch.setattr(crud_service.crud_db_actions, "get_user_by_email",AsyncMock(return_value=user_data))
with allure.step("Test get_by_email"):
test_result = await crud_service.get_user_by_email(user_data.email)
assert test_result.email==user_data.email
assert test_result.first_name==user_data.first_name
assert test_result.last_name==user_data.last_name
assert test_result.middle_name==user_data.middle_name
assert test_result.email==user_data.email
assert test_result.email==user_data.email
assert not hasattr(test_result, "password") or not hasattr(test_result, "plain_password") or not hasattr(test_result, "hashed_password")
assert test_result.direct_permissions==user_data.direct_permissions
assert test_result.group==user_data.group
assert not hasattr(test_result, "status")
@pytest.mark.parametrize("email, expected_exception, expected_status",[
pytest.param("Wrong_email", HTTPException, 404,id="Wrong_email"),
pytest.param("",HTTPException, 404,id="Empty_email"),
])
async def test_get_user_by_email_negative(self, email, crud_service:CrudService, expected_exception, expected_status:int)->None:
with allure.step("Test get_by_email"), pytest.raises(expected_exception) as exc_info:
await crud_service.get_user_by_email(email)
if expected_exception is HTTPException:
assert exc_info.value.status_code==expected_status
@pytest.mark.parametrize("user_data",[
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True),id="Get_user_by_email_positive")
])
async def test_delete_user_soft_positive(self, monkeypatch, user_data:SimpleNamespace, crud_service:CrudService)->None:
with allure.step("Patching functions"):
monkeypatch.setattr(crud_service.crud_db_actions, "delete_user_soft",AsyncMock(return_value=True))
with allure.step("Test get_by_email"):
test_result = await crud_service.delete_user_soft(user_data.email)
assert test_result == True
@pytest.mark.parametrize("email, expected_exception, expected_status",[
pytest.param("Wrong_email", HTTPException, 404,id="Wrong_email"),
pytest.param("",HTTPException, 404,id="Empty_email"),
])
async def test_delete_user_soft_negative(self, email, crud_service:CrudService, expected_exception, expected_status:int)->None:
with allure.step("Test get_by_email"), pytest.raises(expected_exception) as exc_info:
await crud_service.delete_user_soft(email)
if expected_exception is HTTPException:
assert exc_info.value.status_code==expected_status
@pytest.mark.parametrize("user_data",[
pytest.param(SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True),id="Get_user_by_email_positive")
])
async def test_delete_user_hard_positive(self, monkeypatch, user_data:SimpleNamespace, crud_service:CrudService)->None:
with allure.step("Patching functions"):
monkeypatch.setattr(crud_service.crud_db_actions, "delete_user_hard",AsyncMock(return_value=True))
with allure.step("Test get_by_email"):
test_result = await crud_service.delete_user_hard(user_data.email, "current_user")
assert test_result==True
@pytest.mark.parametrize("email, expected_exception, expected_status",[
pytest.param("Wrong_email", HTTPException, 404,id="Wrong_email"),
pytest.param("",HTTPException, 404,id="Empty_email"),
])
async def test_delete_user_hard_negative(self, email, crud_service:CrudService, expected_exception, expected_status:int)->None:
with allure.step("Test get_by_email"), pytest.raises(expected_exception) as exc_info:
await crud_service.delete_user_hard(email, "current_user")
if expected_exception is HTTPException:
assert exc_info.value.status_code==expected_status
tests/unit/test_jwt.py
+19 -15
View File
@@ -25,15 +25,16 @@ class TestJwt:
@pytest.mark.parametrize("data, expected_exception",[
pytest.param("", AttributeError,id="not_dict_value"),
pytest.param({"sub":""},HTTPException, id="empty_value"),
pytest.param({"":""},HTTPException, id="empty_key_value")
@pytest.mark.parametrize("data, expected_exception, expected_status",[
pytest.param("", AttributeError,None,id="not_dict_value"),
pytest.param({"sub":""},HTTPException,401,id="empty_value"),
pytest.param({"":""},HTTPException,401,id="empty_key_value")
])
async def test_access_create_negative(self, jwt_service:JwtService, data:dict, expected_exception)->None:
with allure.step("create invalid access token"),pytest.raises(expected_exception):
async def test_access_create_negative(self, jwt_service:JwtService, data:dict, expected_exception, expected_status:int)->None:
with allure.step("create invalid access token"),pytest.raises(expected_exception) as exc_info:
await jwt_service.create_access_token(data)
if expected_exception is HTTPException:
assert expected_status == exc_info.value.status_code
@pytest.mark.parametrize("data", [
@@ -76,23 +77,26 @@ class TestJwt:
assert payload.get("exp")
assert payload.get("token_type")
@pytest.mark.parametrize("data, expected_exception", [
pytest.param({"sub": "123", "exp":datetime.now(UTC)-timedelta(minutes=15), "token_type":"access"}, HTTPException, id="wrong_exp"),
pytest.param({"sub": "123", "exp":datetime.now(UTC)+timedelta(minutes=15)}, HTTPException, id="no_token_type"),
pytest.param({"sub": "123", "token_type":"access"}, HTTPException,id="no_exp"),
pytest.param({}, HTTPException, id="empty_data"),
pytest.param("", AttributeError, id="not_dict_data")
@pytest.mark.parametrize("data, expected_exception, expected_status", [
pytest.param({"sub": "123", "exp":datetime.now(UTC)-timedelta(minutes=15), "token_type":"access"}, HTTPException,401, id="wrong_exp"),
pytest.param({"sub": "123", "exp":datetime.now(UTC)+timedelta(minutes=15)}, HTTPException, 401,id="no_token_type"),
pytest.param({"sub": "123", "token_type":"access"}, HTTPException,401,id="no_exp"),
pytest.param({}, HTTPException, 401,id="empty_data"),
pytest.param("", AttributeError, None,id="not_dict_data")
])
async def test_jwt_decode_invalid(self,jwt_service:JwtService, expected_exception, data, monkeypatch)->None:
async def test_jwt_decode_invalid(self,jwt_service:JwtService, expected_exception, data, monkeypatch, expected_status)->None:
with allure.step("patch a create token function"):
async def fake_create_access_token(data:dict)->str:
return jwt.encode(data, env_settings.SECRET_KEY, env_settings.ALGORITHM)
monkeypatch.setattr(jwt_service, "create_access_token", fake_create_access_token)
with allure.step("create and decode invalid token"), pytest.raises(expected_exception):
with allure.step("create and decode invalid token"), pytest.raises(expected_exception) as exc_info:
fake_token=await jwt_service.create_access_token(data)
await jwt_service.jwt_decode(fake_token)
if expected_exception is HTTPException:
assert expected_status == exc_info.value.status_code
@pytest.mark.parametrize("time, key, algorithm", [
pytest.param(15, "wrong_key", "HS256",id="wrong_key"),