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 CurrentUserService from src.service.auth.jwt import HashService, JwtService from unittest.mock import AsyncMock current_user_service=CurrentUserService() @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") ]) async def test_get_current_user_positive(self,current_user_service:CurrentUserService, jwt_service:JwtService, monkeypatch, user_data:SimpleNamespace)->None: with allure.step("create token"): token=await 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", AsyncMock(return_value=user_data)) with allure.step("test get_current_user_with_fake_token"): test_result= await 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") ]) async def test_get_current_user_negative(self,current_user_service:CurrentUserService, jwt_service:JwtService, monkeypatch, user_data:SimpleNamespace, expected_exception, uuid)->None: with allure.step("create token"): token=await 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", AsyncMock(return_value=user_data)) with allure.step("test get_current_user_with_fake_token"), pytest.raises(expected_exception): await 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"), ]) async def test_login_positive(self, jwt_service:JwtService,current_user_service:CurrentUserService, monkeypatch, user_data:SimpleNamespace, hash_service:HashService, 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", AsyncMock(return_value=user_data)) monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", AsyncMock(return_value=True)) monkeypatch.setattr(current_user_service.jwt_db_actions, "create_token", AsyncMock(return_value=True)) fake_request = requests with allure.step("test login_with_fake_data"): access, refresh=await 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"), ]) 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): 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", AsyncMock(return_value=user_data)) monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", AsyncMock(return_value=True)) monkeypatch.setattr(current_user_service.jwt_db_actions, "create_token", AsyncMock(return_value=True)) fake_request = requests with allure.step("test login_with_fake_data"), pytest.raises(expected_exception): await current_user_service.login(form_data_email, form_data_password,fake_request) async def test_logout_positive(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService)->None: with allure.step("create fake refresh token"): token=await jwt_service.create_refresh_token({"sub":str(uuid4())}) with allure.step("patching db call functions"): monkeypatch.setattr(current_user_service.jwt_db_actions, "logout", AsyncMock(return_value=True)) with allure.step("test logout with fake data"): 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"), ]) async def test_logout_negative(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService, expected_exception, jti, db_result)->None: with allure.step("patching db call functions"): monkeypatch.setattr(current_user_service.jwt_db_actions, "logout", AsyncMock(return_value=db_result) ) async 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=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): await 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") ]) async def test_refresh_token_positive(self, monkeypatch, current_user_service:CurrentUserService, db_result_token, requests:Request, jwt_service:JwtService,user_data_result_db )->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)) monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", AsyncMock(return_value=user_data_result_db)) monkeypatch.setattr(current_user_service.jwt_db_actions,"create_and_update_token", AsyncMock(return_value=True)) fake_request = requests async 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=await 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 = await 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, 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") ]) 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: 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)) monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", AsyncMock(return_value=user_data_result_db)) monkeypatch.setattr(current_user_service.jwt_db_actions, "create_and_update_token", AsyncMock(return_value=update_result)) fake_request = requests async 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=await fake_create_refresh_token(fake_token_data) with allure.step("test refresh token with fake data"), pytest.raises(expected_exception): await current_user_service.refresh_token(token, fake_request)