e2e crud 0.2
This commit is contained in:
@@ -75,7 +75,7 @@ allure: ## Generate allure report
|
||||
|
||||
.PHONY: coverage
|
||||
coverage: ## Run pytest coverage
|
||||
${VENV} pytest --cov=src tests/
|
||||
${VENV} pytest --cov=src tests/ --cov-report=term-missing
|
||||
|
||||
.PHONY: clear
|
||||
clear: ## Delete old test results
|
||||
|
||||
+1
-1
@@ -41,7 +41,7 @@ build-backend = "poetry.core.masonry.api"
|
||||
|
||||
[tool.coverage.run]
|
||||
omit = [
|
||||
"*/models/*",
|
||||
# "*/models/*",
|
||||
"*/migrations/*",
|
||||
"*/database/*",
|
||||
"*/errors/*",
|
||||
|
||||
+2
-1
@@ -1,7 +1,8 @@
|
||||
[pytest]
|
||||
addopts =
|
||||
-l
|
||||
-vv
|
||||
-v
|
||||
-s
|
||||
--alluredir=tests/allure-results/reports/
|
||||
testpaths =
|
||||
tests
|
||||
|
||||
@@ -22,7 +22,7 @@ async def delete_user_soft(email:str, crud:CrudService=Depends(crud_service), cu
|
||||
async def delete_user_hard(email:str, crud:CrudService=Depends(crud_service), current_user=Depends(get_current_user))->bool: #noqa: B008
|
||||
return await crud.delete_user_hard(email, current_user)
|
||||
|
||||
@router.post("/patch_user")
|
||||
@router.patch("/patch_user")
|
||||
async def patch_user(email:str, data:UserUpdate, crud:CrudService=Depends(crud_service), current_user=Depends(get_current_user))->UserOut: #noqa: B008
|
||||
return await crud.update_user(email, data)
|
||||
|
||||
+17
-6
@@ -40,25 +40,36 @@ async def auth_fixture():
|
||||
yield session
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(scope="function", autouse=True)
|
||||
async def test_user_fixture(auth_fixture: MySession):
|
||||
@pytest_asyncio.fixture(scope="function")
|
||||
async def test_user_fixture(request, auth_fixture: MySession):
|
||||
|
||||
url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
test_id=uuid4()
|
||||
|
||||
direct_permission_param, group_param = request.param
|
||||
|
||||
new_user_record={
|
||||
"first_name":f"TEST_{test_id}",
|
||||
"last_name":f"TEST_{test_id}",
|
||||
"middle_name":f"TEST_{test_id}",
|
||||
"email":"test@d.d",
|
||||
"email":f"TEST_{test_id}@d.d",
|
||||
"plain_password":"Test1234!",
|
||||
"direct_permissions":[],
|
||||
"group":[]
|
||||
"direct_permissions":direct_permission_param,
|
||||
"group":group_param
|
||||
}
|
||||
response=await auth_fixture.post(f"{url}/user/create_user", json=new_user_record)
|
||||
response.raise_for_status()
|
||||
|
||||
yield auth_fixture
|
||||
|
||||
async with MySession() as session:
|
||||
|
||||
payload={"username": new_user_record.get("email"), "password": new_user_record.get("plain_password")}
|
||||
|
||||
response = await session.post(url + "/protected/token", data=payload)
|
||||
response.raise_for_status()
|
||||
session.token = response.json()["access_token"]
|
||||
|
||||
yield (session, new_user_record)
|
||||
|
||||
response=await auth_fixture.post(f"{url}/user/delete_user_hard", params={"email":new_user_record.get("email")})
|
||||
response.raise_for_status()
|
||||
+250
-14
@@ -1,3 +1,6 @@
|
||||
|
||||
from uuid import uuid4
|
||||
|
||||
import allure
|
||||
import pytest
|
||||
from httpx import HTTPStatusError
|
||||
@@ -8,15 +11,17 @@ from tests.e2e.conftest import MySession, e2e_settings
|
||||
@pytest.mark.integra
|
||||
class TestCrud:
|
||||
|
||||
@pytest.mark.parametrize("email",[
|
||||
("test@d.d")
|
||||
])
|
||||
async def test_get_user_by_email_positive(self, email:str, auth_fixture:MySession)->None:
|
||||
@pytest.mark.parametrize("test_user_fixture", [(["admin"], ["admin_group"])], indirect=True)
|
||||
async def test_get_user_by_email_positive(self,test_user_fixture)->None:
|
||||
|
||||
session, new_user_record=test_user_fixture
|
||||
|
||||
with allure.step("Get user by email"):
|
||||
|
||||
email = new_user_record.get("email") #get email from the fixture in yield sector
|
||||
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
response = await auth_fixture.get(f"{target_url}/user/get_by_email",params={"email":email})
|
||||
response = await session.get(f"{target_url}/user/get_by_email",params={"email":email})
|
||||
response.raise_for_status()
|
||||
response=response.json()
|
||||
|
||||
@@ -26,23 +31,254 @@ class TestCrud:
|
||||
assert "TEST_" in response.get("first_name")
|
||||
assert "TEST_" in response.get("last_name")
|
||||
assert "TEST_" in response.get("middle_name")
|
||||
assert response.get("direct_permissions")==[]
|
||||
assert response.get("group")==[]
|
||||
assert response.get("direct_permissions") != []
|
||||
assert response.get("group") != []
|
||||
assert not response.get("hashed_password") or not response.get("plain_password") or not response.get("password")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("email, expected_exception", [
|
||||
("test@test.test", HTTPStatusError),
|
||||
("test", HTTPStatusError),
|
||||
("@d", HTTPStatusError)
|
||||
@pytest.mark.parametrize("email, expected_status", [
|
||||
pytest.param("test@test.test", 404, id="non_existed_email"),
|
||||
pytest.param("test",404, id="wrong_email"),
|
||||
pytest.param("@d", 404, id="wrong_email")
|
||||
])
|
||||
async def test_get_user_by_email_negative(self, email:str, expected_exception, auth_fixture:MySession)->None:
|
||||
|
||||
with allure.step("Get user by email"), pytest.raises(expected_exception):
|
||||
async def test_get_user_by_email_negative(self, email:str, expected_status:int, auth_fixture:MySession)->None:
|
||||
|
||||
with allure.step("Get user by email"):
|
||||
target_url = f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
with pytest.raises(HTTPStatusError) as exc_info:
|
||||
response = await auth_fixture.get(f"{target_url}/user/get_by_email", params={"email": email})
|
||||
response.raise_for_status()
|
||||
|
||||
assert exc_info.value.response.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.parametrize("new_user_record",[
|
||||
pytest.param({ "first_name":"TEST",
|
||||
"last_name":"TEST",
|
||||
"middle_name":"TEST",
|
||||
"email":f"TEST_{uuid4()}@d.d",
|
||||
"plain_password":"Test1234!",
|
||||
"direct_permissions":[],
|
||||
"group":[]}, id="Positive_user_creation_with_all_the_fields"),
|
||||
|
||||
pytest.param({ "first_name":"TEST",
|
||||
"last_name":"TEST",
|
||||
"middle_name":"TEST",
|
||||
"email":f"TEST_{uuid4()}@d.d",
|
||||
"plain_password":"Test1234!",
|
||||
"direct_permissions":["WRONG_PERMISSIONS"],
|
||||
"group":["WRONG_GROUP"]},id="Positive_wrong_permissions"),
|
||||
])
|
||||
async def test_create_delete_user_hard_positive(self, new_user_record:dict,auth_fixture:MySession)->None:
|
||||
|
||||
with allure.step("Setting target_url"):
|
||||
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
with allure.step("Create new test user and check for the new user"):
|
||||
|
||||
response=await auth_fixture.post(f"{target_url}/user/create_user", json=new_user_record)
|
||||
response.raise_for_status()
|
||||
|
||||
try:
|
||||
with allure.step("Check for the new user"):
|
||||
|
||||
response = await auth_fixture.get(f"{target_url}/user/get_by_email",params={"email":new_user_record.get("email")})
|
||||
response.raise_for_status()
|
||||
response=response.json()
|
||||
|
||||
assert response.get("email")==new_user_record["email"]
|
||||
assert response.get("first_name")==new_user_record["first_name"]
|
||||
assert response.get("last_name")==new_user_record["last_name"]
|
||||
assert response.get("middle_name")==new_user_record["middle_name"]
|
||||
assert response.get("direct_permissions") == new_user_record["direct_permissions"] or response.get("direct_permissions") == []
|
||||
assert response.get("group") == new_user_record["group"] or response.get("group") == []
|
||||
assert not response.get("hashed_password") or not response.get("plain_password") or not response.get("password")
|
||||
|
||||
finally:
|
||||
with allure.step("delete new user"):
|
||||
response=await auth_fixture.post(f"{target_url}/user/delete_user_hard", params={"email":new_user_record["email"]})
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("new_user_record, expected_status", [
|
||||
pytest.param({ "first_name":"TEST",
|
||||
"last_name":"TEST",
|
||||
"middle_name":"TEST",
|
||||
"email":"WRONGEMAIL",
|
||||
"plain_password":"Test1234!",
|
||||
"direct_permissions":[],
|
||||
"group":[]},422,id="Non_existed_email"),
|
||||
|
||||
pytest.param({ "first_name":"TEST",
|
||||
"last_name":"TEST",
|
||||
"middle_name":"TEST",
|
||||
"email":"TEST1@d.d",
|
||||
"plain_password":"1234",
|
||||
"direct_permissions":[],
|
||||
"group":[]},422,id="Wrong_password"),
|
||||
|
||||
pytest.param({ "first_name":"TEST",
|
||||
"email":"TEST1@d.d",
|
||||
"plain_password":"Test1234!",
|
||||
},422,id="Not_all_the_fields"),
|
||||
])
|
||||
async def test_create_user_negative(self, new_user_record:dict, auth_fixture:MySession, expected_status:int):
|
||||
|
||||
with allure.step("Preparing data to create new user negative"):
|
||||
|
||||
user_created=False
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
try:
|
||||
with allure.step("Create new test user and check for the new user"):
|
||||
|
||||
response=await auth_fixture.post(f"{target_url}/user/create_user", json=new_user_record)
|
||||
if response.status_code < 400:
|
||||
user_created = True
|
||||
|
||||
with pytest.raises(HTTPStatusError) as exc_info:
|
||||
response.raise_for_status()
|
||||
|
||||
assert exc_info.value.response.status_code == expected_status
|
||||
|
||||
finally:
|
||||
if user_created:
|
||||
with allure.step("delete new user"):
|
||||
|
||||
response = await auth_fixture.post(
|
||||
f"{target_url}/user/delete_user_hard",
|
||||
params={"email": new_user_record["email"]}
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("new_user_record",[
|
||||
pytest.param({ "first_name":"TEST",
|
||||
"last_name":"TEST",
|
||||
"middle_name":"TEST",
|
||||
"email":f"TEST_{uuid4()}@d.d",
|
||||
"plain_password":"Test1234!",
|
||||
"direct_permissions":[],
|
||||
"group":[]}, id="Positive_user_delete_soft"),
|
||||
])
|
||||
async def test_user_create_delete_soft_positive(self, new_user_record:dict, auth_fixture:MySession)->None:
|
||||
with allure.step("Setting target_url"):
|
||||
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
with allure.step("Create new test user and check for the new user"):
|
||||
|
||||
response=await auth_fixture.post(f"{target_url}/user/create_user", json=new_user_record)
|
||||
response.raise_for_status()
|
||||
|
||||
try:
|
||||
with allure.step("Check for the new user"):
|
||||
|
||||
response = await auth_fixture.get(f"{target_url}/user/get_by_email",params={"email":new_user_record.get("email")})
|
||||
response.raise_for_status()
|
||||
|
||||
with allure.step("Delete user soft"):
|
||||
response=await auth_fixture.post(f"{target_url}/user/delete_user_soft", params={"email":new_user_record["email"]})
|
||||
response.raise_for_status()
|
||||
|
||||
finally:
|
||||
with allure.step("delete new user"):
|
||||
response=await auth_fixture.post(f"{target_url}/user/delete_user_hard", params={"email":new_user_record["email"]})
|
||||
response.raise_for_status()
|
||||
|
||||
@pytest.mark.parametrize("email, expected_status, ",[
|
||||
pytest.param("Test", 404,id="Wrong_email")
|
||||
])
|
||||
async def test_user_delete_soft_negative(self, email:str,expected_status:int, auth_fixture:MySession)->None:
|
||||
|
||||
with allure.step("Delete user soft"):
|
||||
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
with pytest.raises(HTTPStatusError) as exc_info:
|
||||
|
||||
response=await auth_fixture.post(f"{target_url}/user/delete_user_soft", params={"email":email})
|
||||
response.raise_for_status()
|
||||
|
||||
assert exc_info.value.response.status_code == expected_status
|
||||
|
||||
|
||||
@pytest.mark.parametrize("email, expected_status, ",[
|
||||
pytest.param("Test", 404,id="Wrong_email")
|
||||
])
|
||||
async def test_user_delete_hard_negative(self, email:str,expected_status:int, auth_fixture:MySession)->None:
|
||||
|
||||
with allure.step("Delete user hard"):
|
||||
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
with pytest.raises(HTTPStatusError) as exc_info:
|
||||
|
||||
response=await auth_fixture.post(f"{target_url}/user/delete_user_soft", params={"email":email})
|
||||
response.raise_for_status()
|
||||
|
||||
assert exc_info.value.response.status_code == expected_status
|
||||
|
||||
|
||||
|
||||
@pytest.mark.parametrize("test_user_fixture", [
|
||||
([], [])
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize("user_record_to_update",[
|
||||
pytest.param({"first_name": "Test_New"},id="Positive_user_update_partially")
|
||||
])
|
||||
async def test_user_update_partially_positive(self, test_user_fixture, user_record_to_update:dict)->None:
|
||||
|
||||
session, new_user_record=test_user_fixture
|
||||
|
||||
with allure.step("Setting target_url"):
|
||||
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
with allure.step("Update user"):
|
||||
response= await session.patch(f"{target_url}/user/patch_user", json=user_record_to_update, params={"email":new_user_record["email"]})
|
||||
response.raise_for_status()
|
||||
|
||||
with allure.step("Check for the updated user"):
|
||||
response=await session.get(f"{target_url}/user/get_by_email", params={"email":new_user_record["email"]})
|
||||
response.raise_for_status()
|
||||
response=response.json()
|
||||
|
||||
actual_permissions = [item.get("direct_permissions") for item in response.get("direct_permissions")] #unpacking json like {group:[{},{}]}
|
||||
actual_groups =[item.get("group") for item in response.get("group")]
|
||||
|
||||
|
||||
assert response.get("email")==new_user_record["email"]
|
||||
assert response.get("first_name")==user_record_to_update["first_name"]
|
||||
assert response.get("last_name")==new_user_record["last_name"]
|
||||
assert response.get("middle_name")==new_user_record["middle_name"]
|
||||
assert actual_permissions == new_user_record["direct_permissions"] or actual_permissions == [None]
|
||||
assert actual_groups == new_user_record["group"] or actual_groups == [None]
|
||||
assert not response.get("hashed_password") or not response.get("plain_password") or not response.get("password")
|
||||
|
||||
|
||||
|
||||
@pytest.mark.parametrize("test_user_fixture", [
|
||||
([], [])
|
||||
], indirect=True)
|
||||
@pytest.mark.parametrize("user_record_to_update, expected_exception, expected_status",[
|
||||
pytest.param({"plain_password": "Wrong_pass"},HTTPStatusError,422,id="Wrong_password"),
|
||||
pytest.param({"email": "Wrong_email"},HTTPStatusError,422,id="Wrong_email"),
|
||||
pytest.param({},HTTPStatusError, 400,id="Positive_user_update_nothing")
|
||||
])
|
||||
async def test_user_update_partially_negative(self, test_user_fixture, user_record_to_update:dict, expected_exception, expected_status:int)->None:
|
||||
|
||||
session, new_user_record=test_user_fixture
|
||||
|
||||
with allure.step("Setting target_url"):
|
||||
|
||||
target_url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
|
||||
|
||||
with allure.step("Update user"), pytest.raises(expected_exception) as exc_info:
|
||||
|
||||
response= await session.patch(f"{target_url}/user/patch_user", json=user_record_to_update, params={"email":new_user_record["email"]})
|
||||
response.raise_for_status()
|
||||
|
||||
assert exc_info.value.response.status_code == expected_status
|
||||
|
||||
@@ -43,12 +43,12 @@ class TestAuth:
|
||||
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")
|
||||
@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, uuid)->None:
|
||||
async def test_get_current_user_negative(self,current_user_service:CurrentUserService, jwt_service:JwtService, monkeypatch, user_data:SimpleNamespace,expected_exception, expected_status:int, uuid)->None:
|
||||
|
||||
with allure.step("create token"):
|
||||
|
||||
@@ -58,10 +58,11 @@ class TestAuth:
|
||||
|
||||
monkeypatch.setattr(current_user_service.crud_db_actions, "get_user_by_id", AsyncMock(return_value=user_data))
|
||||
|
||||
with allure.step("test get_current_user_with_fake_token"), pytest.raises(expected_exception):
|
||||
|
||||
with allure.step("test get_current_user_with_fake_token"), pytest.raises(expected_exception) as exc_info:
|
||||
await current_user_service.get_current_user(token)
|
||||
|
||||
if expected_exception is HTTPException:
|
||||
assert exc_info.value.status_code == expected_status
|
||||
|
||||
@pytest.mark.parametrize("user_data, form_data_email,form_data_password",[
|
||||
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "1234", id="correct_data"),
|
||||
@@ -87,12 +88,12 @@ class TestAuth:
|
||||
assert len(parts_b)==3
|
||||
|
||||
|
||||
@pytest.mark.parametrize("user_data, form_data_email,form_data_password, expected_exception",[
|
||||
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "wrong_password", HTTPException, id="wrong_password"),
|
||||
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=False), "d@d.d", "1234",HTTPException, id="false_status"),
|
||||
pytest.param(SimpleNamespace(id=1234,hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "1234",ValidationError, id="wrong_id"),
|
||||
@pytest.mark.parametrize("user_data, form_data_email,form_data_password, expected_exception, expected_status",[
|
||||
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "wrong_password", HTTPException,401, id="wrong_password"),
|
||||
pytest.param(SimpleNamespace(id=uuid4(),hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=False), "d@d.d", "1234",HTTPException,401, id="false_status"),
|
||||
pytest.param(SimpleNamespace(id=1234,hashed_password="1234", first_name="test",last_name="test",middle_name="test",email="d@d.d",direct_permissions=[],group=[],status=True), "d@d.d", "1234",ValidationError,None, id="wrong_id"),
|
||||
])
|
||||
async def test_login_negative(self, current_user_service:CurrentUserService, user_data:SimpleNamespace, jwt_service:JwtService, monkeypatch, requests, hash_service:HashService, form_data_email:str, form_data_password:str, expected_exception):
|
||||
async def test_login_negative(self, current_user_service:CurrentUserService, user_data:SimpleNamespace, jwt_service:JwtService, monkeypatch, requests, hash_service:HashService, form_data_email:str, form_data_password:str, expected_exception, expected_status:int):
|
||||
|
||||
with allure.step("patching db call functions"):
|
||||
|
||||
@@ -103,10 +104,12 @@ class TestAuth:
|
||||
|
||||
fake_request = requests
|
||||
|
||||
with allure.step("test login_with_fake_data"), pytest.raises(expected_exception):
|
||||
with allure.step("test login_with_fake_data"), pytest.raises(expected_exception) as exc_info:
|
||||
|
||||
await current_user_service.login(form_data_email, form_data_password,fake_request)
|
||||
|
||||
if expected_exception is HTTPException:
|
||||
assert exc_info.value.status_code == expected_status
|
||||
|
||||
|
||||
async def test_logout_positive(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService)->None:
|
||||
@@ -124,12 +127,12 @@ class TestAuth:
|
||||
status=await current_user_service.logout(token[0])
|
||||
assert status is True
|
||||
|
||||
@pytest.mark.parametrize("jti,db_result, expected_exception",[
|
||||
pytest.param(None, True, HTTPException, id="jti_none"),
|
||||
pytest.param(1234, True, HTTPException, id="jti_int"),
|
||||
pytest.param(str(uuid4()), False, HTTPException, id="db_result_none"),
|
||||
@pytest.mark.parametrize("jti,db_result, expected_exception, expected_status",[
|
||||
pytest.param(None, True, HTTPException,401, id="jti_none"),
|
||||
pytest.param(1234, True, HTTPException,401, id="jti_int"),
|
||||
pytest.param(str(uuid4()), False, HTTPException,404,id="db_result_none"),
|
||||
])
|
||||
async def test_logout_negative(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService, expected_exception, jti, db_result)->None:
|
||||
async def test_logout_negative(self, jwt_service:JwtService, monkeypatch, current_user_service:CurrentUserService, expected_exception, jti, db_result, expected_status:int)->None:
|
||||
|
||||
with allure.step("patching db call functions"):
|
||||
|
||||
@@ -145,10 +148,13 @@ class TestAuth:
|
||||
|
||||
token=await fake_create_refresh_token({"sub":str(uuid4()), "jti":jti, "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)})
|
||||
|
||||
with allure.step("test logout with fake data"), pytest.raises(expected_exception):
|
||||
with allure.step("test logout with fake data"), pytest.raises(expected_exception) as exc_info:
|
||||
|
||||
await current_user_service.logout(token)
|
||||
|
||||
if expected_exception is HTTPException:
|
||||
assert exc_info.value.status_code==expected_status
|
||||
|
||||
|
||||
@pytest.mark.parametrize("db_result_token, user_data_result_db", [
|
||||
pytest.param(SimpleNamespace(is_revoked=False, expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True), id="correct_data")
|
||||
@@ -186,15 +192,15 @@ class TestAuth:
|
||||
assert new_access_token!=token
|
||||
assert new_refresh_token!=token
|
||||
|
||||
@pytest.mark.parametrize("db_result_token, user_data_result_db, update_result, fake_token_data,expected_exception", [
|
||||
pytest.param(SimpleNamespace(is_revoked=True,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),False,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="false_revoke_status"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException, id="jti_missing"),
|
||||
pytest.param(None,SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()),"token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException, id="token_missing"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=False),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="false_user_status"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False, user_id="123",expires_at=datetime.now(UTC)+timedelta(days=15)),None,True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="user_missing"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)-timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,id="wrong_exp")
|
||||
@pytest.mark.parametrize("db_result_token, user_data_result_db, update_result, fake_token_data,expected_exception, expected_status", [
|
||||
pytest.param(SimpleNamespace(is_revoked=True,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),False,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,404,id="false_revoke_status"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,401, id="jti_missing"),
|
||||
pytest.param(None,SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()),"token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException, 404,id="token_missing"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)+timedelta(days=15)),SimpleNamespace(status=False),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,401,id="false_user_status"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False, user_id="123",expires_at=datetime.now(UTC)+timedelta(days=15)),None,True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,404,id="user_missing"),
|
||||
pytest.param(SimpleNamespace(is_revoked=False,user_id="123", expires_at=datetime.now(UTC)-timedelta(days=15)),SimpleNamespace(status=True),True,{"sub":str(uuid4()), "jti":str(uuid4()), "token_type":"refresh", "exp":datetime.now(UTC)+timedelta(days=45)}, HTTPException,401,id="wrong_exp")
|
||||
])
|
||||
async def test_refresh_token_negative(self, monkeypatch, current_user_service:CurrentUserService, db_result_token, requests, jwt_service:JwtService,user_data_result_db, expected_exception, fake_token_data, update_result)->None:
|
||||
async def test_refresh_token_negative(self, monkeypatch, current_user_service:CurrentUserService, db_result_token, requests, jwt_service:JwtService,user_data_result_db, expected_exception, fake_token_data, update_result, expected_status:int)->None:
|
||||
with allure.step("patching db call functions"):
|
||||
monkeypatch.setattr(current_user_service.jwt_db_actions,"get_token_by_id", AsyncMock(return_value=db_result_token))
|
||||
monkeypatch.setattr(current_user_service.jwt_db_actions, "revoke_all", AsyncMock(return_value=True))
|
||||
@@ -211,5 +217,8 @@ class TestAuth:
|
||||
with allure.step("create fake refresh token"):
|
||||
token=await fake_create_refresh_token(fake_token_data)
|
||||
|
||||
with allure.step("test refresh token with fake data"), pytest.raises(expected_exception):
|
||||
with allure.step("test refresh token with fake data"), pytest.raises(expected_exception) as exc_info:
|
||||
await current_user_service.refresh_token(token, fake_request)
|
||||
|
||||
if expected_exception is HTTPException:
|
||||
assert exc_info.value.status_code==expected_status
|
||||
+19
-15
@@ -25,15 +25,16 @@ class TestJwt:
|
||||
|
||||
|
||||
|
||||
@pytest.mark.parametrize("data, expected_exception",[
|
||||
pytest.param("", AttributeError,id="not_dict_value"),
|
||||
pytest.param({"sub":""},HTTPException, id="empty_value"),
|
||||
pytest.param({"":""},HTTPException, id="empty_key_value")
|
||||
@pytest.mark.parametrize("data, expected_exception, expected_status",[
|
||||
pytest.param("", AttributeError,None,id="not_dict_value"),
|
||||
pytest.param({"sub":""},HTTPException,401,id="empty_value"),
|
||||
pytest.param({"":""},HTTPException,401,id="empty_key_value")
|
||||
])
|
||||
async def test_access_create_negative(self, jwt_service:JwtService, data:dict, expected_exception)->None:
|
||||
with allure.step("create invalid access token"),pytest.raises(expected_exception):
|
||||
async def test_access_create_negative(self, jwt_service:JwtService, data:dict, expected_exception, expected_status:int)->None:
|
||||
with allure.step("create invalid access token"),pytest.raises(expected_exception) as exc_info:
|
||||
await jwt_service.create_access_token(data)
|
||||
|
||||
if expected_exception is HTTPException:
|
||||
assert expected_status == exc_info.value.status_code
|
||||
|
||||
|
||||
@pytest.mark.parametrize("data", [
|
||||
@@ -76,23 +77,26 @@ class TestJwt:
|
||||
assert payload.get("exp")
|
||||
assert payload.get("token_type")
|
||||
|
||||
@pytest.mark.parametrize("data, expected_exception", [
|
||||
pytest.param({"sub": "123", "exp":datetime.now(UTC)-timedelta(minutes=15), "token_type":"access"}, HTTPException, id="wrong_exp"),
|
||||
pytest.param({"sub": "123", "exp":datetime.now(UTC)+timedelta(minutes=15)}, HTTPException, id="no_token_type"),
|
||||
pytest.param({"sub": "123", "token_type":"access"}, HTTPException,id="no_exp"),
|
||||
pytest.param({}, HTTPException, id="empty_data"),
|
||||
pytest.param("", AttributeError, id="not_dict_data")
|
||||
@pytest.mark.parametrize("data, expected_exception, expected_status", [
|
||||
pytest.param({"sub": "123", "exp":datetime.now(UTC)-timedelta(minutes=15), "token_type":"access"}, HTTPException,401, id="wrong_exp"),
|
||||
pytest.param({"sub": "123", "exp":datetime.now(UTC)+timedelta(minutes=15)}, HTTPException, 401,id="no_token_type"),
|
||||
pytest.param({"sub": "123", "token_type":"access"}, HTTPException,401,id="no_exp"),
|
||||
pytest.param({}, HTTPException, 401,id="empty_data"),
|
||||
pytest.param("", AttributeError, None,id="not_dict_data")
|
||||
])
|
||||
async def test_jwt_decode_invalid(self,jwt_service:JwtService, expected_exception, data, monkeypatch)->None:
|
||||
async def test_jwt_decode_invalid(self,jwt_service:JwtService, expected_exception, data, monkeypatch, expected_status)->None:
|
||||
with allure.step("patch a create token function"):
|
||||
async def fake_create_access_token(data:dict)->str:
|
||||
return jwt.encode(data, env_settings.SECRET_KEY, env_settings.ALGORITHM)
|
||||
monkeypatch.setattr(jwt_service, "create_access_token", fake_create_access_token)
|
||||
|
||||
with allure.step("create and decode invalid token"), pytest.raises(expected_exception):
|
||||
with allure.step("create and decode invalid token"), pytest.raises(expected_exception) as exc_info:
|
||||
fake_token=await jwt_service.create_access_token(data)
|
||||
await jwt_service.jwt_decode(fake_token)
|
||||
|
||||
if expected_exception is HTTPException:
|
||||
assert expected_status == exc_info.value.status_code
|
||||
|
||||
|
||||
@pytest.mark.parametrize("time, key, algorithm", [
|
||||
pytest.param(15, "wrong_key", "HS256",id="wrong_key"),
|
||||
|
||||
Reference in New Issue
Block a user