Add get_api_info method

Author: HardNorth

reportportal_client/aio/client.py

@@ -683,6 +683,15 @@ async def get_project_settings(self) -> Optional[dict]:
         response = await AsyncHttpRequest((await self.session()).get, url=url, name="get_project_settings").make()
         return await response.json if response else None
 
+    async def get_api_info(self) -> Optional[dict]:
+        """Get server information, like version.
+
+        :return: server information.
+        """
+        url = root_uri_join("api/info")
+        response = await AsyncHttpRequest((await self.session()).get, url=url, name="get_api_info").make()
+        return await response.json if response else None
+
     async def log_batch(self, log_batch: Optional[list[AsyncRPRequestLog]]) -> Optional[tuple[str, ...]]:
         """Send batch logging message to the ReportPortal.
 
@@ -1109,6 +1118,13 @@ async def get_project_settings(self) -> Optional[dict]:
         """
         return await self.__client.get_project_settings()
 
+    async def get_api_info(self) -> Optional[dict]:
+        """Get server information, like version.
+
+        :return: server information.
+        """
+        return await self.__client.get_api_info()
+
     async def log(
         self,
         time: Union[str, datetime],
@@ -1565,6 +1581,15 @@ def get_project_settings(self) -> Task[Optional[str]]:
         result_task = self.create_task(result_coro)
         return result_task
 
+    def get_api_info(self) -> Task[Optional[dict]]:
+        """Get server information, like version.
+
+        :return: server information.
+        """
+        result_coro = self.__client.get_api_info()
+        result_task = self.create_task(result_coro)
+        return result_task
+
     async def _log_batch(self, log_rq: Optional[list[AsyncRPRequestLog]]) -> Optional[tuple[str, ...]]:
         return await self.__client.log_batch(log_rq)
 

reportportal_client/client.py

@@ -316,6 +316,14 @@ def get_project_settings(self) -> Optional[dict]:
         """
         raise NotImplementedError('"get_project_settings" method is not implemented!')
 
+    @abstractmethod
+    def get_api_info(self) -> Optional[dict]:
+        """Get server information, like version.
+
+        :return: server information.
+        """
+        raise NotImplementedError('"get_api_info" method is not implemented!')
+
     @abstractmethod
     def log(
         self,
@@ -1055,6 +1063,21 @@ def get_project_settings(self) -> Optional[dict]:
         ).make()
         return response.json if response else None
 
+    def get_api_info(self) -> Optional[dict]:
+        """Get server information, like version.
+
+        :return: server information.
+        """
+        url = uri_join(self.__endpoint, "api/info")
+        response = HttpRequest(
+            self.session.get,
+            url=url,
+            verify_ssl=self.verify_ssl,
+            http_timeout=self.http_timeout,
+            name="get_api_info",
+        ).make()
+        return response.json if response else None
+
     def _add_current_item(self, item: str) -> None:
         """Add the last item from the self._items queue."""
         self._item_stack.put(item)

tests/aio/test_aio_client.py

@@ -565,6 +565,7 @@ def request_error(*_, **__):
         ("get", "get_launch_ui_id", ["launch_uuid"]),
         ("get", "get_launch_ui_url", ["launch_uuid"]),
         ("get", "get_project_settings", []),
+        ("get", "get_api_info", []),
         (
             "post",
             "log_batch",
@@ -909,6 +910,21 @@ async def test_get_launch_ui_url(aio_client: Client):
     assert expected_uri == call_args[0][0]
 
 
+@pytest.mark.asyncio
+async def test_get_api_info(aio_client: Client):
+    # noinspection PyTypeChecker
+    session: mock.AsyncMock = await aio_client.session()
+    mock_basic_get_response(session)
+
+    expected_uri = "/api/info"
+
+    result = await aio_client.get_api_info()
+    assert result == RETURN_GET_JSON
+    session.get.assert_called_once()
+    call_args = session.get.call_args_list[0]
+    assert expected_uri == call_args[0][0]
+
+
 @pytest.mark.parametrize(
     "method, mock_method, call_method, arguments",
     [

tests/aio/test_async_client.py

@@ -193,3 +193,16 @@ async def test_logs_flush_on_close(async_client: AsyncRPClient):
     batcher.flush.assert_called_once()
     client.log_batch.assert_called_once()
     client.close.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_get_api_info(async_client: AsyncRPClient):
+    # noinspection PyTypeChecker
+    client: mock.AsyncMock = async_client.client
+    expected_info = {"version": "5.14.0"}
+    client.get_api_info.return_value = expected_info
+
+    result = await async_client.get_api_info()
+
+    assert result == expected_info
+    client.get_api_info.assert_called_once_with()

tests/aio/test_batched_client.py

@@ -167,3 +167,15 @@ def test_logs_flush_on_close(batched_client: BatchedRPClient):
     batcher.flush.assert_called_once()
     client.log_batch.assert_called_once()
     client.close.assert_called_once()
+
+
+def test_get_api_info():
+    aio_client = mock.AsyncMock()
+    expected_info = {"version": "5.14.0"}
+    aio_client.get_api_info.return_value = expected_info
+    client = BatchedRPClient("http://endpoint", "project", api_key="api_key", client=aio_client)
+
+    result = client.get_api_info().blocking_result()
+
+    assert result == expected_info
+    aio_client.get_api_info.assert_called_once_with()

tests/aio/test_threaded_client.py

@@ -165,3 +165,15 @@ def test_logs_flush_on_close(batched_client: ThreadedRPClient):
     batcher.flush.assert_called_once()
     client.log_batch.assert_called_once()
     client.close.assert_called_once()
+
+
+def test_get_api_info():
+    aio_client = mock.AsyncMock()
+    expected_info = {"version": "5.14.0"}
+    aio_client.get_api_info.return_value = expected_info
+    client = ThreadedRPClient("http://endpoint", "project", api_key="api_key", client=aio_client)
+
+    result = client.get_api_info().blocking_result()
+
+    assert result == expected_info
+    aio_client.get_api_info.assert_called_once_with()

tests/test_client.py

@@ -54,6 +54,7 @@ def invalid_response(*args, **kwargs):
         ("get", "get_launch_ui_id", []),
         ("get", "get_launch_ui_url", []),
         ("get", "get_project_settings", []),
+        ("get", "get_api_info", []),
         ("post", "start_launch", ["Test Launch", timestamp()]),
         ("post", "start_test_item", ["Test Item", timestamp(), "STEP"]),
         ("put", "update_test_item", ["test_item_id"]),
@@ -297,6 +298,7 @@ def test_attribute_sanitization_binary_and_number_limit(rp_client: RPClient):
         ("update_test_item", "put", ["test_item_uuid"]),
         ("get_launch_info", "get", []),
         ("get_project_settings", "get", []),
+        ("get_api_info", "get", []),
         ("get_item_id_by_uuid", "get", ["test_item_uuid"]),
         ("log", "post", [timestamp(), "Test Message"]),
     ],
@@ -344,6 +346,17 @@ def test_logs_flush_on_close(rp_client: RPClient):
     session.close.assert_called_once()
 
 
+def test_get_api_info_url(rp_client: RPClient):
+    # noinspection PyTypeChecker
+    session: mock.Mock = rp_client.session
+
+    rp_client.get_api_info()
+
+    session.get.assert_called_once()
+    request_args = session.get.call_args_list[0][0]
+    assert request_args[0] == "http://endpoint/api/info"
+
+
 def test_oauth_authentication_parameters():
     """Test that OAuth 2.0 authentication parameters work correctly."""
     client = RPClient(