226 lines
14 KiB
Python
226 lines
14 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
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
|
|
|
|
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
|
|
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,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),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, expected_status:int, 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) 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"),
|
|
])
|
|
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, 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, expected_status:int):
|
|
|
|
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) 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:
|
|
|
|
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, 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, expected_status:int)->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) 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")
|
|
])
|
|
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, 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, 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))
|
|
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) 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 |