From 439d57554ca0f88a5d276d5ff835df04d7a67fdb Mon Sep 17 00:00:00 2001 From: "MH.Dmitrii" Date: Thu, 30 Jul 2026 15:49:08 +0300 Subject: [PATCH] integration tests --- src/database/auth/refresh_tokens.py | 20 ++++++ src/service/auth/auth.py | 4 +- src/service/auth/jwt.py | 4 ++ tests/integrated/test_auth.py | 94 +++++++++++++++++++++++++---- tests/unit/test_jwt.py | 48 +++++++-------- 5 files changed, 130 insertions(+), 40 deletions(-) diff --git a/src/database/auth/refresh_tokens.py b/src/database/auth/refresh_tokens.py index b0c4fe7..2bc156f 100644 --- a/src/database/auth/refresh_tokens.py +++ b/src/database/auth/refresh_tokens.py @@ -2,8 +2,10 @@ from uuid import UUID 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 @@ -11,6 +13,7 @@ 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: # noqa: SIM117 @@ -45,6 +48,23 @@ class JwtCrudActions: 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: # noqa: SIM117 diff --git a/src/service/auth/auth.py b/src/service/auth/auth.py index c547e89..586dd1c 100644 --- a/src/service/auth/auth.py +++ b/src/service/auth/auth.py @@ -146,10 +146,8 @@ class CurrentUser: ip_address=request.headers.get("x-forwarded-for", "").split(",")[0].strip() or (request.client.host if request.client else "unknown"), 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) diff --git a/src/service/auth/jwt.py b/src/service/auth/jwt.py index d1e9cb0..24c5bd8 100644 --- a/src/service/auth/jwt.py +++ b/src/service/auth/jwt.py @@ -34,6 +34,8 @@ class Jwt: def create_access_token(self, data:dict)->str: user_info=data.copy() + 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"}) return jwt.encode(user_info, env_settings.SECRET_KEY, env_settings.ALGORITHM) @@ -43,6 +45,8 @@ class Jwt: user_info=data.copy() jti=str(uuid4()) + 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 diff --git a/tests/integrated/test_auth.py b/tests/integrated/test_auth.py index 58ddc2a..072f22b 100644 --- a/tests/integrated/test_auth.py +++ b/tests/integrated/test_auth.py @@ -4,7 +4,7 @@ from uuid import uuid4 import allure import pytest -from fastapi import HTTPException +from fastapi import HTTPException, Request from jose import jwt from pydantic import ValidationError @@ -17,7 +17,7 @@ from src.service.auth.jwt import Hashes, Jwt class TestAuth: @pytest.mark.parametrize("user_data",[ - (SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True)) + 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: @@ -42,9 +42,9 @@ class TestAuth: @pytest.mark.parametrize("user_data, uuid, expected_exception",[ - (SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=False), uuid4(), HTTPException), - (SimpleNamespace(first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True),1234, HTTPException), - (SimpleNamespace(status=True),uuid4(), ValidationError) + 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: @@ -62,7 +62,7 @@ class TestAuth: @pytest.mark.parametrize("user_data, form_data_email,form_data_password",[ - (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"), + 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: @@ -77,15 +77,18 @@ class TestAuth: 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",[ - (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), - (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), - (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), + 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): @@ -117,8 +120,8 @@ class TestAuth: with allure.step("test logout with fake data"): - current_user_service.logout(token[0]) - + 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"), @@ -145,4 +148,69 @@ class TestAuth: with allure.step("test logout with fake data"), pytest.raises(expected_exception): current_user_service.logout(token) - \ No newline at end of file + + + @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) \ No newline at end of file diff --git a/tests/unit/test_jwt.py b/tests/unit/test_jwt.py index cc7e37b..d97c4a9 100644 --- a/tests/unit/test_jwt.py +++ b/tests/unit/test_jwt.py @@ -13,9 +13,7 @@ from src.service.auth.jwt import Hashes, Jwt class TestJwt: @pytest.mark.parametrize("data", [ - ({"sub": "123"}), - ({"sub":""}), - ({"":""}) + pytest.param({"sub": "123"}, id="full_sub") ]) def test_access_create_positive(self, jwt_service:Jwt, data:dict)->None: @@ -27,19 +25,19 @@ class TestJwt: - @pytest.mark.parametrize("data",[ - ("") + @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)->None: - with allure.step("create invalid access token"),pytest.raises(AttributeError): + 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", [ - ({"sub": "123"}), - ({"sub":""}), - ({"":""}) + pytest.param({"sub": "123"}, id="full_sub") ]) def test_refresh_create_positive(self, jwt_service:Jwt, data:dict)->None: @@ -51,16 +49,18 @@ class TestJwt: assert len(parts)==3 - @pytest.mark.parametrize("data",[ - ("") + @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)->None: - with allure.step("create invalid access token"), pytest.raises(AttributeError): + 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", [ - ({"sub": "123", "exp":datetime.now(UTC)+timedelta(minutes=15), "token_type":"access"}), + 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: @@ -77,11 +77,11 @@ class TestJwt: assert payload.get("token_type") @pytest.mark.parametrize("data, expected_exception", [ - ({"sub": "123", "exp":datetime.now(UTC)-timedelta(minutes=15), "token_type":"access"}, HTTPException), - ({"sub": "123", "exp":datetime.now(UTC)+timedelta(minutes=15)}, HTTPException), - ({"sub": "123", "token_type":"access"}, HTTPException), - ({}, HTTPException), - ("", AttributeError) + 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"): @@ -95,9 +95,9 @@ class TestJwt: @pytest.mark.parametrize("time, key, algorithm", [ - (15, "wrong_key", "HS256"), - (-15, "correct_key", "HS256"), - (15, "correct_key", "HS512"), + 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: @@ -113,7 +113,7 @@ class TestJwt: @pytest.mark.parametrize("password",[ - ("plain_password") + pytest.param("plain_password", id="correct_plain_password") ]) def test_hash_and_veryfy_positive(self, password:str, hash_service:Hashes)->None: