11 Commits

Author SHA1 Message Date
3705ac4f0c conftest fix 2026-07-30 15:54:43 +03:00
439d57554c integration tests 2026-07-30 15:49:08 +03:00
f87f54de55 tests logout 2026-07-29 18:32:22 +03:00
00403191f5 integrated tests 2026-07-29 00:44:14 +03:00
d24b99b8b0 integrated tests + omit unnecessary 2026-07-28 15:16:52 +03:00
ef1e39d506 fix gitignore 2026-07-28 13:35:38 +03:00
2781317797 rm coverage 2026-07-28 13:34:14 +03:00
f81ba19da4 hash tests 2026-07-28 13:31:58 +03:00
7a4df2933d ruff fix 2026-07-28 01:43:35 +03:00
5522447b07 auth unit tests & ruff visual fix 2026-07-28 01:42:29 +03:00
1d25b0edc3 refactoring some code by removing unnecessary checks and moving tests block to root / 2026-07-24 17:13:37 +03:00
30 changed files with 776 additions and 72 deletions

2
.gitignore vendored
View File

@@ -25,3 +25,5 @@ Thumbs.db
#Примеры документов
input/
output/
allure-results/
.coverage

13
main.py
View File

@@ -1,5 +1,6 @@
from fastapi import FastAPI
from src.web.protected_routes.routes import router as protected_router
from pathlib import Path
import uvicorn
app=FastAPI(root_path="/")
@@ -9,7 +10,19 @@ app.include_router(router=protected_router)
def root()->dict:
return {"root":"hello, this is root"}
def create_dirs():
dirs_to_create=("./DB",
"./upload",
"./upload_bad",
"./upload_finished")
for x in dirs_to_create:
Path(x).mkdir(parents=True, exist_ok=True)
def main():
create_dirs()
uvicorn.run("main:app", reload=True)
if __name__=="__main__":

View File

@@ -1,7 +1,9 @@
VENV=source .venv/bin/activate;
ALLURE=.venv/allure-2.44.0/bin/allure #linux&macos
#ALLURE=.venv\allure-2.44.0\bin\allure #Windows
.PHONY:
run, m_gen, m_up
run, m_gen, m_up, allure, coverage, pytest
run:
${VENV} python3 main.py
@@ -9,3 +11,9 @@ m_gen:
${VENV} alembic revision --autogenerate
m_up:
${VENV} alembic upgrade head
pytest:
${VENV} pytest
allure:
${VENV} ${ALLURE} generate tests/allure-results/reports --single-file -o tests/allure-results/html --clean
coverage:
${VENV} pytest --cov=src tests/

188
poetry.lock generated
View File

@@ -20,6 +20,38 @@ typing-extensions = ">=4.12"
[package.extras]
tz = ["tzdata"]
[[package]]
name = "allure-pytest"
version = "2.16.0"
description = "Allure pytest integration"
optional = false
python-versions = "*"
groups = ["main"]
files = [
{file = "allure_pytest-2.16.0-py3-none-any.whl", hash = "sha256:e5035a3b1f541b0c2ee566822df6fec20e0628b8536780e5bdfe39060ac0c71b"},
{file = "allure_pytest-2.16.0.tar.gz", hash = "sha256:3cc883595b1ce4280b0b9a5fdaa0ce3bb5cdbaa1b43c7269d8b82e825e6e107c"},
]
[package.dependencies]
allure-python-commons = "2.16.0"
pytest = ">=4.5.0"
[[package]]
name = "allure-python-commons"
version = "2.16.0"
description = "Contains the API for end users as well as helper functions and classes to build Allure adapters for Python test frameworks"
optional = false
python-versions = ">=3.6"
groups = ["main"]
files = [
{file = "allure_python_commons-2.16.0-py3-none-any.whl", hash = "sha256:6d42a500078aca8a2e68823075c1ffc2396987bb268d62b19af82390b205ce88"},
{file = "allure_python_commons-2.16.0.tar.gz", hash = "sha256:ecdc92bafea074bab96b5f2c4eb3100825340188f5aece608ae80eced709b36f"},
]
[package.dependencies]
attrs = ">=16.0.0"
pluggy = ">=0.4.0"
[[package]]
name = "annotated-doc"
version = "0.0.4"
@@ -78,6 +110,18 @@ files = [
astroid = ["astroid (>=2,<5)"]
test = ["astroid (>=2,<5)", "pytest (<9.0)", "pytest-cov", "pytest-xdist"]
[[package]]
name = "attrs"
version = "26.1.0"
description = "Classes Without Boilerplate"
optional = false
python-versions = ">=3.9"
groups = ["main"]
files = [
{file = "attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309"},
{file = "attrs-26.1.0.tar.gz", hash = "sha256:d03ceb89cb322a8fd706d4fb91940737b6642aa36998fe130a9bc96c985eff32"},
]
[[package]]
name = "bcrypt"
version = "5.0.0"
@@ -298,6 +342,110 @@ files = [
{file = "colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44"},
]
[[package]]
name = "coverage"
version = "7.15.2"
description = "Code coverage measurement for Python"
optional = false
python-versions = ">=3.10"
groups = ["main"]
files = [
{file = "coverage-7.15.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:9b5bd92ff1ec22e535eab0de75fa6db021992791f461a2aceb7822c625a1187d"},
{file = "coverage-7.15.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:44826758cfe73fcd0e6af5deb4ba6d5417cc1d13df3acb35c93484a11160f846"},
{file = "coverage-7.15.2-cp310-cp310-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:09f5c6ec5901f667bd97dd140b5b9a2586b10efec66f46fb1e6d8135f8b95bdf"},
{file = "coverage-7.15.2-cp310-cp310-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:1d16e3a7104ea84f03e614611b3edbf6fb6892554b3ab0fe7fbb3f2b2ef04376"},
{file = "coverage-7.15.2-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d46e62cb35d91e6e2589fda6d28074426b0e276422b5d2ebef2c6b11dc60dbfd"},
{file = "coverage-7.15.2-cp310-cp310-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:dfd3db045e95960ae3683059571e597fda7cc610106a8916f77c5839048c1deb"},
{file = "coverage-7.15.2-cp310-cp310-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:affd532502d34c0472d0cdb181325c89f1d2c44992fef0c17e88e7b1576259a1"},
{file = "coverage-7.15.2-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:d17d7512151fedfcc64c1821a8977fc9be0dbf495754669afcab7b57abc98ae9"},
{file = "coverage-7.15.2-cp310-cp310-musllinux_1_2_i686.whl", hash = "sha256:e26ff680768b8095e8874aabe0e9d3a47a2a9f176a8340d05f8604c56457c23a"},
{file = "coverage-7.15.2-cp310-cp310-musllinux_1_2_ppc64le.whl", hash = "sha256:7e8f27131dc7cd53de2c137dd207b3720919320b3c20d499dc30aa9ee6173287"},
{file = "coverage-7.15.2-cp310-cp310-musllinux_1_2_riscv64.whl", hash = "sha256:728a33676d4c3f0db977990a4bd421dcaa3be3e53b5b6273036fff6666008e89"},
{file = "coverage-7.15.2-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:29c052f7c83ccfcc5c577eaae025d2e4a9bb80daf03c0ac31c996e83b000ce88"},
{file = "coverage-7.15.2-cp310-cp310-win32.whl", hash = "sha256:1268ac8fb9ddcd783d3948dbabaf80a5d53bfdaa0575e873e2139a692f797443"},
{file = "coverage-7.15.2-cp310-cp310-win_amd64.whl", hash = "sha256:9f4432898c4bf2fba0435bbe35dd4437d7264565e5a88a21f5b49d8662a6b629"},
{file = "coverage-7.15.2-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:2f1ec6f304b156669cfde653b4e9a953f5de87e247ea02ac599bce0ab2744036"},
{file = "coverage-7.15.2-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4d3361879d736f469f45723c11ea1a5bbdaf1f6928f0e632c940378b5aa9b660"},
{file = "coverage-7.15.2-cp311-cp311-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:c6a98d698f9e2c8008d0370ec7fc452ebfcc530002ae2d0061170d768b992589"},
{file = "coverage-7.15.2-cp311-cp311-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:d50dd325e18ec25bfcc10cd7f99b04df1ab9ec76b0918c260e60817ad0643dee"},
{file = "coverage-7.15.2-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:67d7602480a47bdf5b675635403625553ebaa70d5a62a657c035149fd401cea0"},
{file = "coverage-7.15.2-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:cee0f89f4767a6057c8fbf168f8135f18be651300496086bd873e3189fed0487"},
{file = "coverage-7.15.2-cp311-cp311-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:a29ec5305a7335aacee2d799e3422e91e1c8a12474986e2b3b07e315c91be82f"},
{file = "coverage-7.15.2-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:48ccc6395958eda89093ecdc35644c86f23a8b23a7f4d44958812b721aad67c1"},
{file = "coverage-7.15.2-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:81f382c5a94b434ec1f6da607edb904c76d7212e618cd4d1bc9f97bed4120ef5"},
{file = "coverage-7.15.2-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:bbc808daf4f5cd567af8075ecc72d21c6dfef9a254709a621a84c217c935ebc0"},
{file = "coverage-7.15.2-cp311-cp311-musllinux_1_2_riscv64.whl", hash = "sha256:a4c46b247b5d4b78f613bd89fea926d32b25c6cc61a50bd1e99ba310348f3dad"},
{file = "coverage-7.15.2-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:094dd37f3ef7b2da8b068b583d1f4c40f91c65197e16c52a71962d5d537fc5db"},
{file = "coverage-7.15.2-cp311-cp311-win32.whl", hash = "sha256:a63b9e190711134d581c4d703df5df09851b1acf99792c7aacbbe9f41f0283c9"},
{file = "coverage-7.15.2-cp311-cp311-win_amd64.whl", hash = "sha256:8bb9f4b4279187560796a4cdaca3b0a93dd97e48ee667df005f4ed9a97403688"},
{file = "coverage-7.15.2-cp311-cp311-win_arm64.whl", hash = "sha256:8c726b232659cbd2ae57ade46509eb068c9bd7a06df9fcbff6fe484870006934"},
{file = "coverage-7.15.2-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:1adac78e5abc7c5438f7a209c9ca69d06542f0bf481d728b6989ea80b813fdf9"},
{file = "coverage-7.15.2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:b868acc62aa5de3be7a9d05c2333bf8359ca987e43f9cb30ff8fbda6a024ab73"},
{file = "coverage-7.15.2-cp312-cp312-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:6f6966fc30e6f06ca8f98fb0ce51eda6b111b3ee8d066a8b1ec9e77fa06ab55d"},
{file = "coverage-7.15.2-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:68af907f595ab01a78f794932ff3bdf929c316d3000810d38dbc247129e26f8b"},
{file = "coverage-7.15.2-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:afa29e2eff3d5729267e2cb2fd4ce9d61c952932fb2694e34ccb5d9540c6a296"},
{file = "coverage-7.15.2-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:bbf44513ceb1589e31948e20eafbde9deaface90e1a1afa5f5f77b4423d17ce6"},
{file = "coverage-7.15.2-cp312-cp312-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:9deddf09eecb717b7f980414b43d90a5b22ff3967d2949ab29cb0aa83d9e9098"},
{file = "coverage-7.15.2-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:ae901f7e55ba405c84ee1cab3d3e962e4e871e4a2bcb9c90911adbd69b42ac5a"},
{file = "coverage-7.15.2-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:a0f47002c6eeb7c280228467a4cb0cc15ca2103a8421b986b2d3ec04a0f9bd8b"},
{file = "coverage-7.15.2-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:1cd7a5beb7af3e864a13b1f0fb26efd3695da43ef0daf71e586adfffaf34d5b2"},
{file = "coverage-7.15.2-cp312-cp312-musllinux_1_2_riscv64.whl", hash = "sha256:97a5c5457a9fb1d6c4e06cfb5dc835871fbfb6a6a51addc9e925bdeff5ef7440"},
{file = "coverage-7.15.2-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:0901cfe6c13bcd2302da4f83e884555d2a22bda6e4c476f09ef204ba20ca536e"},
{file = "coverage-7.15.2-cp312-cp312-win32.whl", hash = "sha256:b171bdd71cb7ff792bf32e376173b0ace7e7963e7e57c58dfc42063a6a7174cd"},
{file = "coverage-7.15.2-cp312-cp312-win_amd64.whl", hash = "sha256:582edc45c2040543fef83341be23c43024a3ab3ae0c2d8bc498a06282905ad40"},
{file = "coverage-7.15.2-cp312-cp312-win_arm64.whl", hash = "sha256:a638db90c61cd219aeee65e83a24fdaa57269a741ae0cf773309208ac862cee3"},
{file = "coverage-7.15.2-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:1121caa19159a38b5463eaae4b1e1fde81e525b15ecc5e000cd5b1a108f743a8"},
{file = "coverage-7.15.2-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:a300c6934e0989c327b9e8a1e110329da4641149f872bbe9f70168be66da76c1"},
{file = "coverage-7.15.2-cp313-cp313-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:2617f8799d268fabdeef42a7e89ac3a23e1deee9025427db2df970f99a89a578"},
{file = "coverage-7.15.2-cp313-cp313-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:7dc2950a2992cd676d35c20ae63522836deeb034f08874699d14068710af3dc1"},
{file = "coverage-7.15.2-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9e36686f7a442185db2400b3df171aac520869faf9deb59df687d28659eda2a6"},
{file = "coverage-7.15.2-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:7d29ca7bd67af6e12e74632d65f026eabc1364da5c254494cd914446a28a3ef7"},
{file = "coverage-7.15.2-cp313-cp313-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:db9c8438057e5b0f6a22a0af99c0c1d26b57fbbdbd1be5861ddb8f897fcc3a2d"},
{file = "coverage-7.15.2-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:63022c4c8dec1d0342f05c3ede99842fe3d007689acc45e86f123a1746e4a026"},
{file = "coverage-7.15.2-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:6c0be82b4d4aa5b2704e08518e2252f3e3d110164bcca826816801052e48a7aa"},
{file = "coverage-7.15.2-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:4510fb9cdf6bb02dfa6af0be4a534b8102d086e22e4a33f8836df663da3d660d"},
{file = "coverage-7.15.2-cp313-cp313-musllinux_1_2_riscv64.whl", hash = "sha256:42ec3d989421b174a2ab607c1539f24127ad362757b7f1c0c0d7a2993f7eb37b"},
{file = "coverage-7.15.2-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:e8f91bce78e32343af184c3b7fa28fcf5a9e2641f4b6623d392038f804939188"},
{file = "coverage-7.15.2-cp313-cp313-win32.whl", hash = "sha256:434e68d531858205895eb0d74b73d20b84260de426387d53c422a5acda2cf050"},
{file = "coverage-7.15.2-cp313-cp313-win_amd64.whl", hash = "sha256:26c3b04a6377fd7c09800921fa934e3a17c0020439cd59df73e73ae1d4b6a78c"},
{file = "coverage-7.15.2-cp313-cp313-win_arm64.whl", hash = "sha256:3ed010aa1b69cda8e827aabfca9866216c980e2dca82ab9a78c5f83689964c8b"},
{file = "coverage-7.15.2-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:40f633c5c5fc783732f6312280122e859538fa24461235597c13d803ea9a108a"},
{file = "coverage-7.15.2-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:075560438765b7a2ef43bf7aa7758661b53d889df47f062a31bda6c1ade553a2"},
{file = "coverage-7.15.2-cp314-cp314-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:25fd15dd40a0a2c51a500d664ca29053c09c3259d998407bf982b6e114696138"},
{file = "coverage-7.15.2-cp314-cp314-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:b9a6367e4aff723e8ee8190836836124284e8fcd4265e307c844010cfa074f3f"},
{file = "coverage-7.15.2-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9854ca62c152874b2060772503535be2e8f53f70b8aaa7686b094888d872f984"},
{file = "coverage-7.15.2-cp314-cp314-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:913b6c56e110da40e035bbd168353bf7aaa2544a5eaccea5d98a4629aac156c7"},
{file = "coverage-7.15.2-cp314-cp314-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:aaccad4129d735a8a4d526f26929894c9a4e8ef7034566f210b176749d6906e3"},
{file = "coverage-7.15.2-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:a164b50081fc7357331c4024ef4d17b78ba325f8380d05f5a69599a7e05257ee"},
{file = "coverage-7.15.2-cp314-cp314-musllinux_1_2_i686.whl", hash = "sha256:bfd341ccf78128e72c094bc70cc25b3ef309c33c7c2c66ba3ed4309549e02de1"},
{file = "coverage-7.15.2-cp314-cp314-musllinux_1_2_ppc64le.whl", hash = "sha256:1473b3ba8e7ee0f076117b1a72c23f579a2b9e2bb742f48a8d86ea27ca93f91a"},
{file = "coverage-7.15.2-cp314-cp314-musllinux_1_2_riscv64.whl", hash = "sha256:17c432b5f73ad52ef46fb06019f6fa7c66ce381961cf0f7dfd1d3a4bd3a98145"},
{file = "coverage-7.15.2-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:77f0ef5011df53a4bd1b35211ab122287f8d9b8d7aa1c4553e5c2deb24b1d446"},
{file = "coverage-7.15.2-cp314-cp314-win32.whl", hash = "sha256:f653e5d7248c1191ec988a85c72edeab46c3ff44f90639a4ed4874ec0be90243"},
{file = "coverage-7.15.2-cp314-cp314-win_amd64.whl", hash = "sha256:9911f31aad8906abe337c271343485cf20df5e70df5d2f57f9f136e7b55f26bc"},
{file = "coverage-7.15.2-cp314-cp314-win_arm64.whl", hash = "sha256:e38def96ad59853824c97953fdcd2c320a84ba3ce99b417db78af8bb6c3db635"},
{file = "coverage-7.15.2-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:835ec4e20b45f0a7f63ed78f94065aca00de033403df8377bfe8b9c6abc0a7be"},
{file = "coverage-7.15.2-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:7466cc7ab6dc0db871d264bf99e8779f0917ee63d40730af0552f71535a6e072"},
{file = "coverage-7.15.2-cp314-cp314t-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:e370c12133095ff18432de8c044962be85a5a96d90c6fcbce8e17e76236d2328"},
{file = "coverage-7.15.2-cp314-cp314t-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:fe41909c9515c3bfdb5f02c4d1f857dba322d9a9a1178069b91eea77889df63a"},
{file = "coverage-7.15.2-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:6aa28cfb6488e5453b5b762d65f73aa586380f6693a04d58078ce228a29b06c0"},
{file = "coverage-7.15.2-cp314-cp314t-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:bcc0aae933921d03096f53b0b03eeb702129fd406dee59f08d2efacc68681fa5"},
{file = "coverage-7.15.2-cp314-cp314t-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:7c63387e21ab21f512c69c9756a8c7dadd322c7275edb064064433c9a09c3743"},
{file = "coverage-7.15.2-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:0e55510bc98ae943cece9e667a6c0fe94c6a92913720dea34243657a17993d0c"},
{file = "coverage-7.15.2-cp314-cp314t-musllinux_1_2_i686.whl", hash = "sha256:2ff08701be2d1556fc78b326c80a3e8042da09352ecb3819105f8e386c8a3071"},
{file = "coverage-7.15.2-cp314-cp314t-musllinux_1_2_ppc64le.whl", hash = "sha256:38c9518b7103826c403a461544e3c2e77151e8676d06eaed85911a97e962584a"},
{file = "coverage-7.15.2-cp314-cp314t-musllinux_1_2_riscv64.whl", hash = "sha256:dee88b1ed88587abd8c0269a1fc1f4cc77f7750d1dfde2869e2a123af420e67d"},
{file = "coverage-7.15.2-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:2fbeeeecea279727f8ac16c8e1133ddfeee793e985c86ae343d6a5ce744eef8c"},
{file = "coverage-7.15.2-cp314-cp314t-win32.whl", hash = "sha256:cb0fddaa6884be6aae36ced9544b5e90f7d5f03845a2853bf47a14953a4e8688"},
{file = "coverage-7.15.2-cp314-cp314t-win_amd64.whl", hash = "sha256:77f091ea3a9cc611cd29f433565476bc1936c084ac8eee00ea0e7e70c27e4199"},
{file = "coverage-7.15.2-cp314-cp314t-win_arm64.whl", hash = "sha256:6fc448c377d6eeb00a47c673494bd9bae29280ca53987e1869e67ebedfe20658"},
{file = "coverage-7.15.2-py3-none-any.whl", hash = "sha256:eb6bcae8d1a9d305351ecb108232441d11c5cfe9de840a04388ba5d2db8d735c"},
{file = "coverage-7.15.2.tar.gz", hash = "sha256:3df60dc267f0a2ca23cb7a9ab1109c62b9335ffbf519fcfe167157c28c09b81d"},
]
[package.extras]
toml = ["tomli ; python_full_version <= \"3.11.0a6\""]
[[package]]
name = "decorator"
version = "5.3.1"
@@ -1536,6 +1684,44 @@ pygments = ">=2.7.2"
[package.extras]
dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "requests", "setuptools", "xmlschema"]
[[package]]
name = "pytest-cov"
version = "7.1.0"
description = "Pytest plugin for measuring coverage."
optional = false
python-versions = ">=3.9"
groups = ["main"]
files = [
{file = "pytest_cov-7.1.0-py3-none-any.whl", hash = "sha256:a0461110b7865f9a271aa1b51e516c9a95de9d696734a2f71e3e78f46e1d4678"},
{file = "pytest_cov-7.1.0.tar.gz", hash = "sha256:30674f2b5f6351aa09702a9c8c364f6a01c27aae0c1366ae8016160d1efc56b2"},
]
[package.dependencies]
coverage = {version = ">=7.10.6", extras = ["toml"]}
pluggy = ">=1.2"
pytest = ">=7"
[package.extras]
testing = ["process-tests", "pytest-xdist", "virtualenv"]
[[package]]
name = "pytest-mock"
version = "3.15.1"
description = "Thin-wrapper around the mock package for easier use with pytest"
optional = false
python-versions = ">=3.9"
groups = ["main"]
files = [
{file = "pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d"},
{file = "pytest_mock-3.15.1.tar.gz", hash = "sha256:1849a238f6f396da19762269de72cb1814ab44416fa73a8686deac10b0d87a0f"},
]
[package.dependencies]
pytest = ">=6.2.5"
[package.extras]
dev = ["pre-commit", "pytest-asyncio", "tox"]
[[package]]
name = "python-dateutil"
version = "2.9.0.post0"
@@ -1965,4 +2151,4 @@ files = [
[metadata]
lock-version = "2.1"
python-versions = ">=3.13"
content-hash = "22aa39f5ef7ad5d2415b44f3908af10f00430ba0b9be5071910d691fa9a9cb34"
content-hash = "ebbba0a60f3f121f79117d05b5a6dbd8914ca1d971776808c7d1cfd146b562bc"

View File

@@ -22,10 +22,28 @@ dependencies = [
"python-jose (>=3.5.0,<4.0.0)",
"python-multipart (>=0.0.32,<0.0.33)",
"ipython (>=9.15.0,<10.0.0)",
"httpie (>=3.2.4,<4.0.0)"
"httpie (>=3.2.4,<4.0.0)",
"pytest-cov (>=7.1.0,<8.0.0)",
"allure-pytest (>=2.16.0,<3.0.0)",
"pytest-mock (>=3.15.1,<4.0.0)"
]
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.coverage.run]
omit = [
"*/models/*",
"*/migrations/*",
"*/database/*",
"*/errors/*",
"__init__.py",
"*/docker/*"
]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
]

12
pytest.ini Normal file
View File

@@ -0,0 +1,12 @@
[pytest]
addopts =
-l
-v
--alluredir=tests/allure-results/reports/
testpaths =
tests
markers=
unit: unit tests
integra: integrations test
e2e: e2e tests
smoke: smoke tests

View File

@@ -1,16 +1,22 @@
from uuid import UUID
from src.models.database_models.model import engine, RefreshTokens
from sqlalchemy import and_, not_, select
from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm import sessionmaker
from src.errors.http_errors.errors import Errors
from src.models.database_models.model import RefreshTokens, engine
from src.models.pydantic_models.model import RefreshTokensOut
class JwtCrudActions:
def __init__(self) -> None:
self.Session=sessionmaker(bind=engine)
self.error=Errors()
def get_token_by_user_id(self, user_id:UUID)->RefreshTokensOut|None:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
query=select(RefreshTokens).where(and_(RefreshTokens.user_id==user_id, not_(RefreshTokens.is_revoked)))
response=session.scalars(query).one_or_none()
@@ -19,7 +25,7 @@ class JwtCrudActions:
return RefreshTokensOut.model_validate(response)
def get_token_by_id(self, id:UUID)->RefreshTokensOut|None:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
query=select(RefreshTokens).where(RefreshTokens.id==id)
response=session.scalars(query).one_or_none()
@@ -28,23 +34,40 @@ class JwtCrudActions:
return RefreshTokensOut.model_validate(response)
def create_token(self, data:dict)->None:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
new_token=RefreshTokens(**data)
response=session.add(new_token)
return response
def update_token(self, old_jti:UUID, new_jti:UUID)->bool:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
query=select(RefreshTokens).where(RefreshTokens.id==old_jti)
response=session.scalars(query).one()
response.is_revoked=True
response.replaced_by=new_jti
return True
def create_and_update_token(self,data:dict, old_jti:UUID, new_jti:UUID)->bool:
with self.Session() as session: #noqa:SIM117
with session.begin():
new_token=RefreshTokens(**data)
session.add(new_token)
query=select(RefreshTokens).where(RefreshTokens.id==old_jti)
try:
response = session.scalars(query).one()
except NoResultFound as e:
raise self.error.not_found_error(detail="Token not found") from e
response.is_revoked=True
response.replaced_by=new_jti
return True
def revoke_all(self, user_id:UUID)->bool:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
query=select(RefreshTokens).where(RefreshTokens.user_id==user_id)
response=session.scalars(query).all()
@@ -53,7 +76,7 @@ class JwtCrudActions:
return True
def logout(self,id:UUID)->bool:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
query=select(RefreshTokens).where(RefreshTokens.id == id)
response=session.scalars(query).one_or_none()

View File

@@ -1,14 +1,18 @@
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.orm import sessionmaker
from src.models.database_models.model import User, engine
from src.models.pydantic_models.model import UserOutDB
from sqlalchemy.orm import sessionmaker
from uuid import UUID
class UsersCrudActions:
def __init__(self) -> None:
self.Session=sessionmaker(bind=engine)
def get_user_by_email(self, email:str)->UserOutDB|None:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
query=select(User).where(User.email==email)
response=session.scalars(query).one_or_none()
@@ -17,7 +21,7 @@ class UsersCrudActions:
return UserOutDB.model_validate(response)
def get_user_by_id(self, id:UUID)->UserOutDB|None:
with self.Session() as session:
with self.Session() as session: # noqa: SIM117
with session.begin():
query=select(User).where(User.id==id)
response=session.scalars(query).one_or_none()

View File

@@ -1,5 +1,6 @@
from pydantic_settings import BaseSettings, SettingsConfigDict
class Base(BaseSettings):
pass

View File

@@ -1,7 +1,10 @@
from src.models.database_models.model import Model
from uuid import UUID, uuid1
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from uuid import UUID, uuid1
from src.models.database_models.model import Model
class AccountantSettings(Model):
__tablename__="accountant_settings"

View File

@@ -1,9 +1,11 @@
from datetime import datetime
from sqlalchemy import TIMESTAMP, String, func, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.models.database_models.model import Model
from uuid import UUID, uuid1
from sqlalchemy import TIMESTAMP, ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.models.database_models.model import Model
class Stored(Model):
__tablename__="reports"

View File

@@ -1,8 +1,11 @@
from decimal import Decimal
from src.models.database_models.model import Model
from sqlalchemy import Boolean, ForeignKey, Integer, Numeric, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.models.database_models.model import Model
class Nomenclature(Model):
__tablename__="goods"

View File

@@ -1,7 +1,20 @@
from sqlalchemy import TIMESTAMP, Table, create_engine, String, Boolean, MetaData, Column, ForeignKey, func, Uuid
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase, relationship
from uuid import UUID, uuid4
from datetime import datetime
from uuid import UUID, uuid4
from sqlalchemy import (
TIMESTAMP,
Boolean,
Column,
ForeignKey,
MetaData,
String,
Table,
Uuid,
create_engine,
func,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
engine = create_engine("sqlite:///DB/database.db", echo=True)
'''remember as a boilerplate, or just cp/pst'''

View File

@@ -1,7 +1,11 @@
from typing import Annotated
from src.models.pydantic_models.model import Base
from pydantic import Field
from uuid import UUID
from pydantic import Field
from src.models.pydantic_models.model import Base
class AccountantCreate(Base):
name:Annotated[str, Field(...,max_length=64,description="name of the accountant setting")]

View File

@@ -1,9 +1,12 @@
from datetime import datetime
from typing import Annotated
from uuid import UUID
from src.models.pydantic_models.model import Base
from pydantic import Field
from src.models.pydantic_models.model import Base
class ReportCreate(Base):
filename:Annotated[str,Field(..., min_length=2, max_length=255, description="name of the report")]
doc_date:Annotated[datetime, Field(..., description="ts of the report")]

View File

@@ -1,8 +1,11 @@
from decimal import Decimal
from src.models.pydantic_models.model import Base
from pydantic import Field
from typing import Annotated
from pydantic import Field
from src.models.pydantic_models.model import Base
class NomenclatureCreate(Base):
article:Annotated[str, Field(...,min_length=5,max_length=16, description="name of the article")]

View File

@@ -1,8 +1,9 @@
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, EmailStr, Field
from uuid import UUID
from pydantic import BaseModel, EmailStr, Field
class Base(BaseModel):
model_config = {"from_attributes": True}

View File

@@ -1,13 +1,15 @@
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from uuid import UUID
from fastapi import Request
from .jwt import Jwt, Hashes
from src.database.users.crud import UsersCrudActions
from src.database.auth.refresh_tokens import JwtCrudActions
from src.database.users.crud import UsersCrudActions
from src.errors.http_errors.errors import Errors
from src.models.pydantic_models.model import RefreshTokensCreate, UserOut
from src.models.configs_read.env import env_settings
from src.models.pydantic_models.model import RefreshTokensCreate, UserOut
from .jwt import Hashes, Jwt
class CurrentUser:
@@ -22,6 +24,7 @@ class CurrentUser:
def _check(self, form_data_email:str, form_data_password:str,):
'''check user by email'''
user=self.crud_db_actions.get_user_by_email(form_data_email)
if user is None:
raise self.error.credentials_error(detail="Wrong credentials")
@@ -36,20 +39,21 @@ class CurrentUser:
def get_current_user(self, token:str)->UserOut:
payload=self.jwt_service.jwt_decode(token)
sub=payload.get("sub")
if (sub:=payload.get("sub")) is None:
raise self.error.credentials_error(detail="Jwt token is incorrect")
try:
sub=UUID(sub)
except (ValueError, TypeError):
raise self.error.credentials_error(detail="Jwt token is incorrect")
except (ValueError, TypeError) as e:
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
user=self.crud_db_actions.get_user_by_id(sub)
if user is None:
raise self.error.not_found_error(detail="User with this email address not found")
if user.status is False:
raise self.error.credentials_error(detail="This user is deactivated")
return UserOut.model_validate(user)
@@ -61,12 +65,13 @@ class CurrentUser:
def create_refresh_token(self,user_id:UUID, request:Request)->str:
token, jti=self.jwt_service.create_refresh_token({"sub":str(user_id)})
try:
jti=UUID(jti)
except (ValueError, TypeError):
raise self.error.credentials_error(detail="Jwt token is incorrect")
except (ValueError, TypeError) as e:
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
'''create new refresh token if all the checks are successful'''
token_record=RefreshTokensCreate(
@@ -75,7 +80,7 @@ class CurrentUser:
token_hash=self.hash.token_to_hash(token),
device_info=request.headers.get("user-agent", "unknown"),
ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"),
expires_at=datetime.now(timezone.utc)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS)
expires_at=datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS)
)
self.jwt_db_actions.create_token(RefreshTokensCreate.model_dump(token_record))
@@ -83,16 +88,19 @@ class CurrentUser:
def refresh_token(self, refresh_token:str, request:Request)->tuple[str, str]:
'''decode old refresh token'''
old_refresh_token=self.jwt_service.jwt_decode(refresh_token)
if (sub:=old_refresh_token.get("sub")) is None or (old_jti:=old_refresh_token.get("jti")) is None:
sub=old_refresh_token.get("sub")
if (old_jti:=old_refresh_token.get("jti")) is None:
raise self.error.credentials_error(detail="Jwt token is incorrect")
try:
old_jti=UUID(old_jti)
sub=UUID(sub)
except (ValueError, TypeError):
raise self.error.credentials_error(detail="Jwt token is incorrect")
except (ValueError, TypeError) as e:
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
'''old refresh token check'''
@@ -107,8 +115,8 @@ class CurrentUser:
'''sqlite constraints about timezone'''
expires_at=old_record.expires_at
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
if expires_at<datetime.now(timezone.utc):
expires_at = expires_at.replace(tzinfo=UTC)
if expires_at<datetime.now(UTC):
raise self.error.credentials_error(detail="Token expired")
'''user check'''
@@ -122,6 +130,7 @@ class CurrentUser:
'''create new refresh token if all the checks are successful'''
new_refresh_token, new_jti=self.jwt_service.create_refresh_token({"sub":str(sub)})
new_access_token=self.create_access_token(user_id=sub)
try:
new_jti=UUID(new_jti)
except (ValueError, TypeError):
@@ -135,12 +144,10 @@ class CurrentUser:
token_hash=self.hash.token_to_hash(new_refresh_token),
device_info=request.headers.get("user-agent", "unknown"),
ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"),
expires_at=datetime.now(timezone.utc)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
expires_at=datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
)
self.jwt_db_actions.create_token(RefreshTokensCreate.model_dump(new_token_record))
'''update old token to deactivate it and assign replaced_by'''
self.jwt_db_actions.update_token(old_jti, new_jti)
self.jwt_db_actions.create_and_update_token(RefreshTokensCreate.model_dump(new_token_record), old_jti, new_jti)
return (new_access_token,new_refresh_token)
@@ -152,11 +159,12 @@ class CurrentUser:
payload=self.jwt_service.jwt_decode(refresh_token)
if (jti:=payload.get("jti")) is None:
raise self.error.credentials_error(detail="Invalid Refresh Token")
raise self.error.credentials_error(detail="Jwt token is incorrect")
try:
jti=UUID(jti)
except (ValueError, TypeError):
raise self.error.credentials_error(detail="Jwt token is incorrect")
except (ValueError, TypeError) as e:
raise self.error.credentials_error(detail="Jwt token is incorrect") from e
current_token = self.jwt_db_actions.get_token_by_id(jti)
if current_token is None:

View File

@@ -1,10 +1,12 @@
from jose import JWTError, jwt
import bcrypt
from src.errors.http_errors.errors import Errors
from datetime import datetime, timedelta, timezone
from src.models.configs_read.env import env_settings
from uuid import uuid4
import hashlib
from datetime import UTC, datetime, timedelta
from uuid import uuid4
import bcrypt
from jose import JWTError, jwt
from src.errors.http_errors.errors import Errors
from src.models.configs_read.env import env_settings
'''Hash/Check hash'''
class Hashes:
@@ -32,9 +34,10 @@ class Jwt:
def create_access_token(self, data:dict)->str:
user_info=data.copy()
user_info.update({"exp": datetime.now(timezone.utc)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES),
if not (user_info.get("sub")) or user_info.get("sub") == "":
raise self.error.credentials_error(detail="Jwt token is incorrect")
user_info.update({"exp": datetime.now(UTC)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES),
"token_type":"access"})
print(f"DEBUG: expires at {datetime.now(timezone.utc)+timedelta(minutes=env_settings.ACCESS_TOKEN_EXPIRE_MINUTES)}, minutes={env_settings.ACCESS_TOKEN_EXPIRE_MINUTES}")
return jwt.encode(user_info, env_settings.SECRET_KEY, env_settings.ALGORITHM)
@@ -42,7 +45,9 @@ class Jwt:
user_info=data.copy()
jti=str(uuid4())
user_info.update({"exp":datetime.now(timezone.utc)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
if not (user_info.get("sub")) or user_info.get("sub") == "":
raise self.error.credentials_error(detail="Jwt token is incorrect")
user_info.update({"exp":datetime.now(UTC)+timedelta(days=env_settings.REFRESH_TOKEN_EXPIRE_DAYS),
"token_type":"refresh",
"jti":jti
})
@@ -54,10 +59,10 @@ class Jwt:
def jwt_decode(self, token:str)->dict:
try:
payload=jwt.decode(token, env_settings.SECRET_KEY, algorithms=[env_settings.ALGORITHM])
payload=jwt.decode(token, env_settings.SECRET_KEY, algorithms=[env_settings.ALGORITHM], options={"require_exp": True} )
if (payload.get("sub")) is None:
raise self.error.credentials_error(detail="Sub block is missing")
if not (payload.get("sub")) or not (payload.get("token_type")):
raise self.error.credentials_error(detail="Jwt token is incorrect")
except JWTError as e:
raise self.error.credentials_error(detail="JWTerror") from e

View File

@@ -1 +0,0 @@
'''unit, integr, e2e tests'''

View File

@@ -1,15 +1,15 @@
from fastapi import APIRouter, Depends, Request, Response, Cookie
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from fastapi import APIRouter, Cookie, Depends, Request, Response
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from src.models.configs_read.env import env_settings
from src.models.pydantic_models.model import UserOut
from src.service.auth.auth import auth
router=APIRouter(prefix="/protected")
oauth2_schema=OAuth2PasswordBearer(tokenUrl="/protected/token", refreshUrl="/protected/refresh")
@router.post("/token")
async def get_access_token(request: Request,response:Response, form_data:OAuth2PasswordRequestForm=Depends())->dict:
async def get_access_token(request: Request,response:Response, form_data:OAuth2PasswordRequestForm=Depends())->dict: # noqa: B008
access_token, refresh_token=auth.login(form_data_email=form_data.username, form_data_password=form_data.password, request=request)
@@ -46,10 +46,10 @@ async def get_current_user(token:str = Depends(oauth2_schema)) -> UserOut:
@router.get("/logout")
async def logout(response:Response,refresh_token: str = Cookie(),current_user:UserOut=Depends(get_current_user))->bool:
async def logout(response:Response,refresh_token: str = Cookie(),current_user:UserOut=Depends(get_current_user))->bool: # noqa: B008
response.delete_cookie("refresh_token")
return auth.logout(refresh_token)
@router.get("")
async def protected(current_user:UserOut=Depends(get_current_user))->dict:
async def protected(current_user:UserOut=Depends(get_current_user))->dict: # noqa: B008
return {"protected router": "Hello, this is a protected router"}

1
tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
'''unit, integr, e2e tests'''

14
tests/conftest.py Normal file
View File

@@ -0,0 +1,14 @@
import pytest
from src.service.auth.jwt import Hashes, Jwt
@pytest.fixture
def jwt_service()->Jwt:
jwt_service=Jwt()
return jwt_service
@pytest.fixture
def hash_service()->Hashes:
hash_service=Hashes()
return hash_service

0
tests/e2e/conftest.py Normal file
View File

View File

View File

@@ -0,0 +1,16 @@
import pytest
from fastapi import Request
from src.service.auth.auth import CurrentUser
@pytest.fixture
def current_user_service()->CurrentUser:
current_user_service=CurrentUser()
return current_user_service
@pytest.fixture
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

View File

@@ -0,0 +1,216 @@
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from uuid import uuid4
import allure
import pytest
from fastapi import HTTPException, Request
from jose import jwt
from pydantic import ValidationError
from src.models.configs_read.env import env_settings
from src.service.auth.auth import CurrentUser
from src.service.auth.jwt import Hashes, Jwt
@pytest.mark.integra
class TestAuth:
@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="correct_data")
])
def test_get_current_user_positive(self,current_user_service:CurrentUser, jwt_service:Jwt, monkeypatch, user_data:SimpleNamespace)->None:
with allure.step("create token"):
token=jwt_service.create_access_token({"sub":str(uuid4())})
with allure.step("patching db call functions"):
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", lambda user:user_data)
with allure.step("test get_current_user_with_fake_token"):
test_result= current_user_service.get_current_user(token)
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.direct_permissions==user_data.direct_permissions
assert test_result.group==user_data.group
@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")
])
def test_get_current_user_negative(self,current_user_service:CurrentUser, jwt_service:Jwt, monkeypatch, user_data:SimpleNamespace, expected_exception, uuid)->None:
with allure.step("create token"):
token=jwt_service.create_access_token({"sub":str(uuid)})
with allure.step("patching db call functions"):
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", lambda user:user_data)
with allure.step("test get_current_user_with_fake_token"), pytest.raises(expected_exception):
current_user_service.get_current_user(token)
@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"),
])
def test_login_positive(self, jwt_service:Jwt,current_user_service:CurrentUser, monkeypatch, user_data:SimpleNamespace, hash_service:Hashes, form_data_email:str, form_data_password:str, requests)->None:
with allure.step("patching db call functions"):
user_data.hashed_password=hash_service.plain_to_hash(user_data.hashed_password)
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_email", lambda user:user_data)
monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", lambda user_id: True)
monkeypatch.setattr(current_user_service.jwt_db_actions, "create_token", lambda token: True )
fake_request = requests
with allure.step("test login_with_fake_data"):
access, refresh=current_user_service.login(form_data_email, form_data_password,fake_request)
parts_a=access.split(".")
parts_b=refresh.split(".")
assert isinstance(access, str)
assert len(parts_a)==3
assert isinstance(refresh, str)
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"),
])
def test_login_negative(self, current_user_service:CurrentUser, user_data:SimpleNamespace, jwt_service:Jwt, monkeypatch, requests, hash_service:Hashes, form_data_email:str, form_data_password:str, expected_exception):
with allure.step("patching db call functions"):
user_data.hashed_password=hash_service.plain_to_hash(user_data.hashed_password)
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_email", lambda user:user_data)
monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", lambda user_id: True)
monkeypatch.setattr(current_user_service.jwt_db_actions, "create_token", lambda token: True )
fake_request = requests
with allure.step("test login_with_fake_data"), pytest.raises(expected_exception):
current_user_service.login(form_data_email, form_data_password,fake_request)
def test_logout_positive(self, jwt_service:Jwt, monkeypatch, current_user_service:CurrentUser)->None:
with allure.step("create fake refresh token"):
token=jwt_service.create_refresh_token({"sub":str(uuid4())})
with allure.step("patching db call functions"):
monkeypatch.setattr(current_user_service.jwt_db_actions, "get_token_by_id", lambda jti: "test")
monkeypatch.setattr(current_user_service.jwt_db_actions, "logout", lambda jti: True )
with allure.step("test logout with fake data"):
status=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()), None, HTTPException, id="db_result_none"),
])
def test_logout_negative(self, jwt_service:Jwt, monkeypatch, current_user_service:CurrentUser, expected_exception, jti, db_result)->None:
with allure.step("patching db call functions"):
monkeypatch.setattr(current_user_service.jwt_db_actions, "get_token_by_id", lambda jti: db_result)
monkeypatch.setattr(current_user_service.jwt_db_actions, "logout", lambda jti: True )
def fake_create_refresh_token(data:dict)->str:
return jwt.encode(data, env_settings.SECRET_KEY, env_settings.ALGORITHM)
monkeypatch.setattr(jwt_service, "create_refresh_token", fake_create_refresh_token)
with allure.step("create fake refresh token"):
token=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):
current_user_service.logout(token)
@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")
])
def test_refresh_token_positive(self, monkeypatch, current_user_service:CurrentUser, db_result_token, requests:Request, jwt_service:Jwt,user_data_result_db )->None:
with allure.step("patching db call functions"):
monkeypatch.setattr(current_user_service.jwt_db_actions,"get_token_by_id", lambda jti: db_result_token)
monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", lambda data:True)
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", lambda user:user_data_result_db)
monkeypatch.setattr(current_user_service.jwt_db_actions,"create_and_update_token", lambda data0, data1, data2: True)
fake_request = requests
def fake_create_refresh_token(data:dict)->str:
return jwt.encode(data, env_settings.SECRET_KEY, env_settings.ALGORITHM)
monkeypatch.setattr(jwt_service, "create_refresh_token", fake_create_refresh_token)
with allure.step("create fake refresh token"):
token=fake_create_refresh_token({"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)})
with allure.step("test refresh token with fake data"):
new_access_token, new_refresh_token = current_user_service.refresh_token(token, fake_request)
parts_a=new_access_token.split(".")
parts_b=new_refresh_token.split(".")
assert isinstance(new_access_token, str)
assert len(parts_a)==3
assert isinstance(new_refresh_token, str)
assert len(parts_b)==3
assert new_access_token!=new_refresh_token
assert new_access_token!=token
assert new_refresh_token!=token
@pytest.mark.parametrize("db_result_token, user_data_result_db, 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),{"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),{"sub":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException, id="jti_missing"),
pytest.param(None,SimpleNamespace(status=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),{"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,{"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),{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="wrong_exp")
])
def test_refresh_token_negative(self, monkeypatch, current_user_service:CurrentUser, db_result_token, requests, jwt_service:Jwt,user_data_result_db, expected_exception, fake_token_data)->None:
with allure.step("patching db call functions"):
monkeypatch.setattr(current_user_service.jwt_db_actions,"get_token_by_id", lambda jti: db_result_token)
monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", lambda user_id:True)
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", lambda user:user_data_result_db)
monkeypatch.setattr(current_user_service.jwt_db_actions, "create_token", lambda new_token_record:True)
monkeypatch.setattr(current_user_service.jwt_db_actions,"update_token", lambda old_jti, new_jti: True )
fake_request = requests
def fake_create_refresh_token(data:dict)->str:
return jwt.encode(data, env_settings.SECRET_KEY, env_settings.ALGORITHM)
monkeypatch.setattr(jwt_service, "create_refresh_token", fake_create_refresh_token)
with allure.step("create fake refresh token"):
token=fake_create_refresh_token(fake_token_data)
with allure.step("test refresh token with fake data"), pytest.raises(expected_exception):
current_user_service.refresh_token(token, fake_request)

0
tests/unit/__init__.py Normal file
View File

2
tests/unit/conftest.py Normal file
View File

@@ -0,0 +1,2 @@

144
tests/unit/test_jwt.py Normal file
View File

@@ -0,0 +1,144 @@
from datetime import UTC, datetime, timedelta
import allure
import pytest
from fastapi import HTTPException
from jose import jwt
from src.models.configs_read.env import env_settings
from src.service.auth.jwt import Hashes, Jwt
@pytest.mark.unit
class TestJwt:
@pytest.mark.parametrize("data", [
pytest.param({"sub": "123"}, id="full_sub")
])
def test_access_create_positive(self, jwt_service:Jwt, data:dict)->None:
with allure.step("create correct access token"):
token = jwt_service.create_access_token(data)
parts=token.split(".")
assert isinstance(token, str)
assert len(parts)==3
@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")
])
def test_access_create_negative(self, jwt_service:Jwt, data:dict, expected_exception)->None:
with allure.step("create invalid access token"),pytest.raises(expected_exception):
jwt_service.create_access_token(data)
@pytest.mark.parametrize("data", [
pytest.param({"sub": "123"}, id="full_sub")
])
def test_refresh_create_positive(self, jwt_service:Jwt, data:dict)->None:
with allure.step("create correct access token"):
token = jwt_service.create_refresh_token(data)
parts=token[0].split(".")
assert isinstance(token[0], str)
assert isinstance(token[1], str)
assert len(parts)==3
@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")
])
def test_refresh_create_negative(self, jwt_service:Jwt, data, expected_exception)->None:
with allure.step("create invalid access token"), pytest.raises(expected_exception):
jwt_service.create_refresh_token(data)
@pytest.mark.parametrize("data", [
pytest.param({"sub": "123", "exp":datetime.now(UTC)+timedelta(minutes=15), "token_type":"access"}, id="correct_data"),
])
def test_jwt_decode_positive(self, data:dict, monkeypatch, jwt_service:Jwt)->None:
with allure.step("patch a create token function"):
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 correct token"):
fake_token = jwt_service.create_access_token(data)
payload=jwt_service.jwt_decode(fake_token)
assert payload.get("sub")
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")
])
def test_jwt_decode_invalid(self,jwt_service:Jwt, expected_exception, data, monkeypatch)->None:
with allure.step("patch a create token function"):
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):
fake_token=jwt_service.create_access_token(data)
jwt_service.jwt_decode(fake_token)
@pytest.mark.parametrize("time, key, algorithm", [
pytest.param(15, "wrong_key", "HS256",id="wrong_key"),
pytest.param(-15, "correct_key", "HS256", id="wrong_time"),
pytest.param(15, "correct_key", "HS512", id="wrong_algorithm"),
])
def test_jwt_decode_wrong_env(self, monkeypatch, time:int, key:str, algorithm:str, jwt_service)->None:
with allure.step("patch a create token function"):
def fake_create_access_token(data:dict, key:str, algorithm:str)->str:
data.update({"exp":datetime.now(UTC)+timedelta(minutes=time)})
return jwt.encode(data, key, algorithm)
monkeypatch.setattr(jwt_service, "create_access_token", fake_create_access_token)
with allure.step("create and decode token with wrong data inside"), pytest.raises(HTTPException):
fake_token=jwt_service.create_access_token({"sub": "123", "token_type":"access"}, key, algorithm)
jwt_service.jwt_decode(fake_token)
@pytest.mark.parametrize("password",[
pytest.param("plain_password", id="correct_plain_password")
])
def test_hash_and_veryfy_positive(self, password:str, hash_service:Hashes)->None:
with allure.step("encode password"):
encoded_password=hash_service.plain_to_hash(password)
assert isinstance(encoded_password, str)
assert encoded_password!=password
with allure.step("decode password"):
assert hash_service.verify_password(password, encoded_password) is True
def test_verify_wrong_password(self, hash_service:Hashes)->None:
with allure.step("encode password"):
encoded_password=hash_service.plain_to_hash("plain_password")
with allure.step("decode password"):
assert hash_service.verify_password("wrong_password", encoded_password) is False
def test_token_to_hash_determistic(self, hash_service:Hashes)->None:
assert hash_service.token_to_hash("abc")==hash_service.token_to_hash("abc")
def test_token_to_hash_different_input(self, hash_service:Hashes)->None:
assert hash_service.token_to_hash("abc")!=hash_service.token_to_hash("xyz")