Feature/auth tests api (CORE-13484)

utils/api_client.py → clients/api_client.py

@@ -6,11 +6,13 @@
 
 class APIClient:
 
-    def __init__(self, base_url: str, token: Optional[str] = None, api_key: Optional[str] = None):
+    def __init__(self, base_url: str, token: Optional[str] = None, refresh_token: Optional[str] = None, api_key: Optional[str] = None):
         self.base_url = base_url.rstrip('/')
         self.token = token
+        self.refresh_token = refresh_token
         self.api_key = api_key
         self.client = httpx.Client(timeout=30.0)
+        self.headers = self._get_headers()
 
     def _get_headers(self, additional_headers: Optional[Dict[str, str]] = None) -> Dict[str, str]:
         headers = {"Content-Type": "application/json"}
@@ -80,6 +82,48 @@ def delete(self, endpoint: str, headers: Optional[Dict[str, str]] = None) -> htt
 
         return response
 
+    @allure.step("PATCH {endpoint}")
+    def patch(self, endpoint: str, json: Optional[Dict[str, Any]] = None,
+            headers: Optional[Dict[str, str]] = None) -> httpx.Response:
+        url = f"{self.base_url}{endpoint}"
+        request_headers = self._get_headers(headers)
+
+        with allure.step(f"Request: PATCH {url}"):
+            allure.attach(str(json), "Request Body", allure.attachment_type.JSON)
+            response = self.client.patch(url, json=json, headers=request_headers)
+            allure.attach(response.text, "Response Body", allure.attachment_type.JSON)
+            allure.attach(str(response.status_code), "Status Code", allure.attachment_type.TEXT)
+
+        return response
+
+    @allure.step("Custom Request {method} {endpoint}")
+    def send_custom_request(self, method: str, endpoint: str, 
+                           json: Optional[Dict[str, Any]] = None,
+                           params: Optional[Dict[str, Any]] = None,
+                           headers: Optional[Dict[str, str]] = None) -> httpx.Response:
+        if method.upper() == "GET":
+            return self.get(endpoint, params=params, headers=headers)
+        elif method.upper() == "POST":
+            return self.post(endpoint, json=json, headers=headers)
+        elif method.upper() == "PUT":
+            return self.put(endpoint, json=json, headers=headers)
+        elif method.upper() == "PATCH":
+            return self.patch(endpoint, json=json, headers=headers)
+        elif method.upper() == "DELETE":
+            return self.delete(endpoint, headers=headers)
+        else:
+            raise ValueError(f"Unsupported HTTP method: {method}")
+
+    def set_cookies(self, cookies: Dict[str, str]):
+        for name, value in cookies.items():
+            self.client.cookies.set(name, value)
+
+    def get_cookies(self) -> Dict[str, str]:
+        return dict(self.client.cookies)
+
+    def clear_cookies(self):
+        self.client.cookies.clear()
+
     def close(self):
         self.client.close()
 
@@ -126,3 +170,36 @@ def register_worker(self, worker_id: str, worker_data: Dict[str, Any]) -> httpx.
 
     def confirm_deletion(self, node_id: str) -> httpx.Response:
         return self.post(f"/internal/nodes/{node_id}/confirm-delete")
+
+
+class AuthAPIClient(APIClient):
+
+    def __init__(self, settings: Settings, token: Optional[str] = None, refresh_token: Optional[str] = None):
+        super().__init__(
+            base_url=settings.cp_nodes_api_url,
+            token=token,
+            refresh_token=refresh_token
+        )
+
+    def login(self, username: Optional[str] = None, password: Optional[str] = None) -> httpx.Response:        
+        return self.post("/v1/auth/login", json={"username": username, "password": password})
+
+    def post_refresh(self, refresh_token: str) -> httpx.Response:
+        return self.post("/v1/auth/refresh", json={"refresh_token": refresh_token})
+
+    def logout(self, refresh_token: Optional[str] = None) -> httpx.Response:
+        payload = {"refresh_token": refresh_token} if refresh_token else {}
+        return self.post("/v1/auth/logout", json=payload)
+
+    def get_profile(self) -> httpx.Response:
+        return self.get("/v1/auth/profile")
+
+    def change_password(self, old_password: Optional[str] = None, new_password: Optional[str] = None) -> httpx.Response:
+        return self.put("/v1/auth/password", json={"old_password": old_password, "new_password": new_password})
+
+    def change_username(self, new_username: Optional[str] = None) -> httpx.Response:
+        return self.put("/v1/auth/username", json={"new_username": new_username})
+
+    def get_audit_log(self, page: Optional[int] = None, page_size: Optional[int] = None) -> httpx.Response:
+        params = {"page": page, "page_size": page_size}
+        return self.get("/v1/auth/audit-log", params=params)

fixtures/api_fixtures.py

@@ -1,9 +1,11 @@
 import pytest
-from utils.api_client import NodesAPIClient, InternalAPIClient
+from clients.api_client import NodesAPIClient, InternalAPIClient, AuthAPIClient
 from config.settings import Settings
 from faker import Faker
 
-fake = Faker()
+@pytest.fixture(scope="session")
+def faker():
+    return Faker()
 
 @pytest.fixture(scope="session")
 def nodes_api_client(config: Settings):
@@ -19,16 +21,65 @@ def internal_api_client(config: Settings):
     client.close()
 
 
-@pytest.fixture
-def authenticated_client(config: Settings):
+@pytest.fixture(scope="session")
+def authenticated_nodes_client(config: Settings):
     client = NodesAPIClient(config)
     yield client
     client.close()
 
-@pytest.fixture
-def invalid_username():
-    return fake.email()
+
+@pytest.fixture(scope="function")
+def auth_client(config: Settings):
+    client = AuthAPIClient(config)
+    yield client
+    client.close()
+
+
+@pytest.fixture(scope="function")
+def authenticated_auth_client(config: Settings):
+    client = AuthAPIClient(config)
+    if config.user_log and config.user_pass:
+        response = client.login(config.user_log, config.user_pass)
+        if response.status_code == 200:
+            token = response.json().get("access_token")
+            refresh_token = response.json().get("refresh_token")
+            client.token = token
+            client.refresh_token = refresh_token
+        else:
+            raise Exception("Login failed")
+    yield client
+    client.close()
+
 
 @pytest.fixture
-def invalid_password():
-    return fake.password()
+def valid_credentials(config: Settings):
+    return {
+        "username": config.user_log,
+        "password": config.user_pass
+    }
+
+
+@pytest.fixture(scope="function")
+def invalid_username(faker):
+    return faker.email()
+
+@pytest.fixture(scope="function")
+def valid_username(faker):
+    return faker.name()
+
+@pytest.fixture(scope="function")
+def valid_password(faker):
+    return faker.password()
+
+
+@pytest.fixture(scope="function")
+def invalid_password(faker):
+    return faker.password()
+
+
+@pytest.fixture(scope="function")
+def invalid_credentials(invalid_username, invalid_password):
+    return {
+        "username": invalid_username,
+        "password": invalid_password
+    }

fixtures/eth_fixtures.py

@@ -1,5 +1,5 @@
 import pytest
-from utils.eth_client import EthereumClient
+from clients.eth_client import EthereumClient
 from config.settings import Settings