conftest e2e auth/test user create

pytest.ini

@@ -1,7 +1,7 @@
 [pytest]
 addopts = 
     -l
-    -v
+    -vv
     --alluredir=tests/allure-results/reports/
 testpaths =
     tests
@@ -12,3 +12,4 @@ markers=
     smoke: smoke tests
 
 asyncio_mode = auto
+asyncio_default_fixture_loop_scope = function

src/models/pydantic_models/model.py

@@ -5,7 +5,7 @@ from uuid import UUID
 from pydantic import AfterValidator, BaseModel, EmailStr, Field
 
 
-async def validate_password(password: str) -> str:
+def validate_password(password: str) -> str:
     PUNCTUATION: set[str] = {"$", "@", "#", "%", "!", "^", "&", "*", "(", ")", "-", "_", "+", "=", "{", "}", "[", "]"}
     if len(password) < 8 or len(password) > 72:
         raise ValueError("Password must be 8-72 characters")

src/service/users_crud/users_crud.py

@@ -1,3 +1,5 @@
+import asyncio
+
 from src.database.users.crud import UsersCrudActions
 from src.errors.http_errors.errors import Errors
 from src.models.pydantic_models.model import UserCreate, UserOut, UserUpdate
@@ -15,7 +17,7 @@ class CrudService:
     async def _plain_to_hash(self, user_data:dict)->dict:
         
         user_data["hashed_password"]=user_data.pop("plain_password")
-        user_data["hashed_password"]=self.hash_service.plain_to_hash(user_data["hashed_password"])
+        user_data["hashed_password"]= await asyncio.to_thread(self.hash_service.plain_to_hash, user_data["hashed_password"])
         
         return user_data
     

tests/e2e/conftest.py

@@ -21,14 +21,14 @@ class MySession(requests_async.AsyncSession):
         self.headers = {}
         self.token = None
 
-    async def get(self, url:str, **kwargs):
+    async def request(self, method:str, url:str, **kwargs):
         if self.token:
             self.headers['Authorization'] = f"Bearer {self.token}"
         kwargs.setdefault('headers', self.headers)
-        return await super().get(url, **kwargs)
+        return await super().request(method, url, **kwargs)
 
-@pytest_asyncio.fixture(scope="session", autouse=True)
+@pytest_asyncio.fixture(scope="function", autouse=True)
-async def auth_fixture()->None:
+async def auth_fixture():
     
     payload = {"username": e2e_settings.TEST_USERNAME, "password": e2e_settings.TEST_PASSWORD}
     url = f'http://{e2e_settings.HOST}:{e2e_settings.PORT}'
@@ -37,25 +37,28 @@ async def auth_fixture()->None:
         response = await session.post(url + "/protected/token", data=payload)
         response.raise_for_status()
         session.token = response.json()["access_token"]
+        yield session
+    
 
 @pytest_asyncio.fixture(scope="function", autouse=True)
-async def test_user_fixture():
+async def test_user_fixture(auth_fixture: MySession):
     
     url=f"http://{e2e_settings.HOST}:{e2e_settings.PORT}"
-    test_id=uuid4
+    test_id=uuid4()
     
     new_user_record={
         "first_name":f"TEST_{test_id}",
         "last_name":f"TEST_{test_id}",
         "middle_name":f"TEST_{test_id}",
-        "email":"TEST@TEST.TEST",
+        "email":"test@d.d",
-        "plain_password":"1234",
+        "plain_password":"Test1234!",
         "direct_permissions":[],
         "group":[]
     }
-    response=await requests_async.post(f"{url}/create_user", data=new_user_record)
+    response=await auth_fixture.post(f"{url}/user/create_user", json=new_user_record)
-    print(response.json())
+    response.raise_for_status()
-    yield
+    
-    new_user_to_delete={"email":"TEST@TEST.TEST"}
+    yield auth_fixture
-    response=await requests_async.post(f"{url}/delete_user_hard", data=new_user_to_delete)
+    
-    print(response.json())
+    response=await auth_fixture.post(f"{url}/user/delete_user_hard", params={"email":new_user_record.get("email")})
+    response.raise_for_status()

tests/e2e/test_users_crud.py

@@ -1,14 +1,48 @@
 import allure
 import pytest
+from httpx import HTTPStatusError
+
+from tests.e2e.conftest import MySession, e2e_settings
 
 
 @pytest.mark.integra
 class TestCrud:
     
     @pytest.mark.parametrize("email",[
-        ("Test_Email@test.com")
+        ("test@d.d")
     ])
-    async def test_get_user_by_email_positive(self):
+    async def test_get_user_by_email_positive(self, email:str, auth_fixture:MySession)->None:
+        
+        with allure.step("Get user by email"):
+            
+            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.raise_for_status()
+            response=response.json()
+            
+        with allure.step("Validate response"):
+            
+            assert response.get("email")==email
+            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 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)
+    ])
+    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):
+            
+            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.raise_for_status()
+            
+    
             
-        with allure.step(""):
-            pass