HTTP Client

overhave/transport/http/api_client/client.py

@@ -0,0 +1,184 @@
+import json
+import logging
+from typing import cast
+
+from overhave.transport.http import BaseHttpClient, BearerAuth
+from overhave.transport.http.api_client.models import (
+    ApiEmulationRunResponse,
+    ApiFeatureResponse,
+    ApiFeatureTypeResponse,
+    ApiTagResponse,
+    ApiTestRunResponse,
+    ApiTestUserResponse,
+)
+from overhave.transport.http.api_client.settings import OverhaveApiClientSettings
+from overhave.transport.http.base_client import HttpMethod
+
+logger = logging.getLogger(__name__)
+
+
+class OverhaveApiClient(BaseHttpClient[OverhaveApiClientSettings]):
+    """Client for overhave api."""
+
+    def __init__(self, settings: OverhaveApiClientSettings):
+        super().__init__(settings=settings)
+
+    def get_feature_tags_item(self, value: str) -> ApiTagResponse:
+        logger.debug("Start get feature tags item with [value: %s]", value)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_feature_tags_item_url,
+            params={"value": value},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+
+        logger.debug("Get tags item successfully")
+
+        return cast("ApiTagResponse", self._parse_or_raise(response, ApiTagResponse))
+
+    def get_feature_tags_list(self, value: str) -> list[ApiTagResponse]:
+        logger.debug("Start get feature tags list with [value: %s]", value)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_feature_tags_list_url,
+            params={"value": value},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get tags list successfully")
+
+        return [ApiTagResponse.model_validate(data) for data in response.json()]
+
+    def get_feature_types(self) -> list[ApiFeatureTypeResponse]:
+        logger.debug("Start get feature types list")
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_feature_types_list_url,
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get feature types successfully")
+
+        return [ApiFeatureTypeResponse.model_validate(data) for data in response.json()]
+
+    def get_features_by_tag_id(self, tag_id: int) -> list[ApiFeatureResponse]:
+        logger.debug("Start get feature with [tag_id: %s]", tag_id)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_feature_url,
+            params={"tag_id": tag_id},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get feature successfully")
+
+        return [ApiFeatureResponse.model_validate(data) for data in response.json()]
+
+    def get_features_by_tag_value(self, tag_value: str) -> list[ApiFeatureResponse]:
+        logger.debug("Start get feature with [tag_value: %s]", tag_value)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_feature_url,
+            params={"tag_value": tag_value},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get feature successfully")
+
+        return [ApiFeatureResponse.model_validate(data) for data in response.json()]
+
+    def get_test_run(self, test_run_id: int) -> ApiTestRunResponse:
+        logger.debug("Start get test run with [test_run_id: %s]", test_run_id)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_test_run_url,
+            params={"test_run_id": test_run_id},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get test run successfully")
+
+        return cast("ApiTestRunResponse", self._parse_or_raise(response, ApiTestRunResponse))
+
+    def create_test_run(self, tag_value: str) -> list[str]:
+        logger.debug("Start create test run with [tag_value: %s]", tag_value)
+        response = self._make_request(
+            method=HttpMethod.POST,
+            url=self._settings.get_test_run_create_url,
+            params={"tag_value": tag_value},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Create test run successfully")
+
+        return cast(list[str], response.json())
+
+    def get_emulation_runs(self, test_user_id: int) -> list[ApiEmulationRunResponse]:
+        logger.debug("Start get list of EmulationRun")
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_emulation_run_list_url,
+            params={"test_user_id": test_user_id},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get list of EmulationRun successfully")
+        return [ApiEmulationRunResponse.model_validate(data) for data in response.json()]
+
+    def get_test_user_by_user_id(self, user_id: int) -> ApiTestUserResponse:
+        logger.debug("Start get test user by user_id: %s", user_id)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_test_user_url,
+            params={"user_id": user_id},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get test user by user_id successfully")
+        return cast("ApiTestUserResponse", self._parse_or_raise(response, ApiTestUserResponse))
+
+    def get_test_user_by_user_key(self, user_key: str) -> ApiTestUserResponse:
+        logger.debug("Start get test user by user_key: %s", user_key)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_test_user_url,
+            params={"user_key": user_key},
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get test user by user_key successfully")
+        return ApiTestUserResponse.model_validate(response.json())
+
+    def get_test_users(self, feature_type: str, allow_update: bool) -> list[ApiTestUserResponse]:
+        logger.debug("Start get test users with feature_type: %s and allow_update: %s", feature_type, allow_update)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_test_user_list_url,
+            params={
+                "feature_type": feature_type,
+                "allow_update": allow_update,
+            },
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get tests users successfully")
+        return [ApiTestUserResponse.model_validate(data) for data in response.json()]
+
+    def delete_test_user(self, user_id: int) -> None:
+        logger.debug("Start delete user by user_id: %s", user_id)
+        self._make_request(
+            method=HttpMethod.DELETE,
+            url=self._settings.get_test_user_id_url(user_id),
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Delete test user successfully")
+
+    def get_test_user_specification(self, user_id: int) -> dict[str, str | None]:
+        logger.debug("Start get user specification by user_id: %s", user_id)
+        response = self._make_request(
+            method=HttpMethod.GET,
+            url=self._settings.get_test_user_id_spec_url(user_id),
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Get user specification successfully")
+        return cast(dict[str, str | None], response.json())
+
+    def update_test_user_specification(self, user_id: int, specification: dict[str, str | None]) -> None:
+        logger.debug("Start update user specification by user_id: %s", user_id)
+        self._make_request(
+            method=HttpMethod.PUT,
+            url=self._settings.get_test_user_id_spec_url(user_id),
+            data=json.dumps(specification),
+            auth=BearerAuth(self._settings.auth_token),
+        )
+        logger.debug("Update user specification successfully")

overhave/transport/http/api_client/models.py

@@ -1,5 +1,8 @@
+import enum
+from datetime import datetime
 from typing import Literal
 
+import allure
 from pydantic import BaseModel
 
 
@@ -9,3 +12,97 @@ class TokenRequestData(BaseModel):
     grant_type: Literal["password"] = "password"
     username: str
     password: str
+
+
+class ApiTagResponse(BaseModel):
+    """Model for Tag response data."""
+
+    id: int
+    value: str
+    created_by: str
+
+
+class ApiFeatureTypeResponse(BaseModel):
+    """Model for Feature Type response data."""
+
+    id: int
+    name: str
+
+
+class ApiFeatureResponse(BaseModel):
+    """Model for Feature response data."""
+
+    id: int
+    created_at: datetime
+    name: str
+    author: str
+    type_id: int
+    last_edited_by: str
+    last_edited_at: datetime
+    task: list[str]
+    file_path: str
+    released: bool
+    severity: allure.severity_level
+
+    feature_type: ApiFeatureTypeResponse
+    feature_tags: list[ApiTagResponse]
+
+
+class ApiTestRunResponse(BaseModel):
+    """Model for TestRun response data."""
+
+    id: int
+    created_at: datetime
+    name: str
+    executed_by: str
+    start: datetime | None
+    end: datetime | None
+    status: str
+    report_status: str
+    report: str | None
+    traceback: str | None
+    scenario_id: int
+
+
+class ApiEmulationStatus(enum.StrEnum):
+    """Enum for Emulation Status."""
+
+    CREATED = "CREATED"
+    REQUESTED = "REQUESTED"
+    READY = "READY"
+    ERROR = "ERROR"
+
+
+class ApiTestUserResponse(BaseModel):
+    """Model for TestUser response data."""
+
+    id: int
+    created_at: datetime
+    key: str
+    name: str
+    created_by: str
+    specification: dict[str, str | None]
+    feature_type_id: int
+    feature_type: ApiFeatureTypeResponse
+    allow_update: bool
+    changed_at: datetime
+
+
+class ApiEmulationResponse(BaseModel):
+    """Model for Emulation response data."""
+
+    id: int
+    command: str
+    test_user: ApiTestUserResponse
+
+
+class ApiEmulationRunResponse(BaseModel):
+    """Model for EmulationRun response data."""
+
+    id: int
+    emulation_id: int
+    changed_at: datetime
+    status: ApiEmulationStatus
+    port: int | None
+    initiated_by: str
+    emulation: ApiEmulationResponse

overhave/transport/http/api_client/settings.py

@@ -1,4 +1,5 @@
 import httpx
+from pydantic import StrictStr
 
 from overhave.base_settings import OVERHAVE_ENV_PREFIX
 from overhave.transport.http.base_client import BaseHttpClientSettings
@@ -15,3 +16,66 @@ class Config:
     @property
     def get_auth_token_url(self) -> httpx.URL:
         return httpx.URL(f"{self.url}/{self.auth_token_path}")
+
+
+class OverhaveApiClientSettings(BaseHttpClientSettings):
+    """Settings for :class:`OverhaveApiClient`."""
+
+    auth_token: str
+    feature_path: StrictStr = StrictStr("feature/")
+    feature_tags_item_path: StrictStr = StrictStr("feature/tags/item")
+    feature_tags_list_path: StrictStr = StrictStr("feature/tags/list")
+    feature_types_list_path: StrictStr = StrictStr("feature/types/list")
+
+    test_run_path: StrictStr = StrictStr("test_run")
+    test_run_create_path: StrictStr = StrictStr("test_run/create/")
+
+    emulation_run_list_path: StrictStr = StrictStr("emulation/run/list")
+
+    test_user_path: StrictStr = StrictStr("test_user/")
+    test_user_list_path: StrictStr = StrictStr("test_user/list")
+
+    @property
+    def get_feature_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.feature_path}")
+
+    @property
+    def get_feature_tags_item_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.feature_tags_item_path}")
+
+    @property
+    def get_feature_tags_list_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.feature_tags_list_path}")
+
+    @property
+    def get_feature_types_list_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.feature_types_list_path}")
+
+    @property
+    def get_test_run_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.test_run_path}")
+
+    @property
+    def get_test_run_create_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.test_run_create_path}")
+
+    @property
+    def get_emulation_run_list_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.emulation_run_list_path}")
+
+    @property
+    def get_test_user_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.test_user_path}")
+
+    @property
+    def get_test_user_list_url(self) -> httpx.URL:
+        return httpx.URL(f"{self.url}/{self.test_user_list_path}")
+
+    def get_test_user_id_url(self, user_id: int) -> httpx.URL:
+        return httpx.URL(f"{self.url}/test_user/{user_id}")
+
+    def get_test_user_id_spec_url(self, user_id: int) -> httpx.URL:
+        return httpx.URL(f"{self.url}/test_user/{user_id}/specification")
+
+    class Config:
+        env_prefix = "OVERHAVE_API_CLIENT_"

overhave/transport/http/base_client/objects.py

@@ -6,3 +6,5 @@ class HttpMethod(enum.StrEnum):
 
     GET = "get"
     POST = "post"
+    PUT = "put"
+    DELETE = "delete"

tests/integration/admin/views/test_login_view.py

@@ -23,9 +23,7 @@ def test_show_flash_with_chat_for_unregistered_user(
         mock_support_chat_url: None,
     ) -> None:
         test_app.config["WTF_CSRF_ENABLED"] = False
-
         response = test_client.post("/login", data={"username": "kek", "password": "12345"}, follow_redirects=True)
-
         assert (
             "Username 'kek' is not registered! Please contact the <a href='https://localhost'>support channel</a>!"
             in response.data.decode("utf-8")
@@ -37,7 +35,6 @@ def test_show_flash_without_chat_for_unregistered_user(
         test_client: FlaskClient,
     ) -> None:
         test_app.config["WTF_CSRF_ENABLED"] = False
-
         response = test_client.post("/login", data={"username": "kek", "password": "12345"}, follow_redirects=True)
 
         assert "Username 'kek' is not registered!" in response.data.decode(