feat: add ServerInfo's from_server_uri and to_dict

This commit introduces 2 new methods for ServerInfo class, 2 methods that are missing from splunktalib's implementation of ServerInfo.

After this commit is merged, splunktalib's implementation can be deprecated, because solnlib's implementation supports everything and even more.

Author: artemrys

solnlib/server_info.py

@@ -17,12 +17,12 @@
 """This module contains Splunk server info related functionalities."""
 
 import json
+from typing import Any, Dict, Optional
 
 from splunklib import binding
 
-from . import splunk_rest_client as rest_client
-from . import utils
-from .utils import retry
+from solnlib import splunk_rest_client as rest_client
+from solnlib import utils
 
 __all__ = ["ServerInfo", "ServerInfoException"]
 
@@ -42,10 +42,10 @@ class ServerInfo:
     def __init__(
         self,
         session_key: str,
-        scheme: str = None,
-        host: str = None,
-        port: int = None,
-        **context: dict
+        scheme: Optional[str] = None,
+        host: Optional[str] = None,
+        port: Optional[int] = None,
+        **context: Any
     ):
         """Initializes ServerInfo.
 
@@ -60,23 +60,75 @@ def __init__(
             session_key, "-", scheme=scheme, host=host, port=port, **context
         )
 
-    @retry(exceptions=[binding.HTTPError])
+    @classmethod
+    def from_server_uri(
+        cls, server_uri: str, session_key: str, **context: Any
+    ) -> "ServerInfo":
+        """Creates ServerInfo class using server_uri and session_key.
+
+        Note: splunktalib uses these parameters to create it's ServerInfo class,
+        so this method should ease the transition from splunktalib to solnlib.
+
+        Arguments:
+            server_uri: splunkd URI.
+            session_key: Splunk access token.
+            context: Other configurations for Splunk rest client.
+
+        Returns:
+            An instance of `ServerInfo`.
+
+        Raises:
+            ValueError: server_uri is in the wrong format.
+        """
+        scheme, host, port = utils.extract_http_scheme_host_port(server_uri)
+        return ServerInfo(
+            session_key,
+            scheme=scheme,
+            host=host,
+            port=port,
+            **context,
+        )
+
+    def to_dict(self) -> Dict:
+        """Returns server information in a form dictionary.
+
+        Note: This method is implemented here to have compatibility with splunktalib's
+        analogue.
+
+        Returns:
+            Server information in a dictionary format.
+        """
+        return self._server_info()
+
+    @utils.retry(exceptions=[binding.HTTPError])
     def _server_info(self):
         return self._rest_client.info
 
     @property
     def server_name(self) -> str:
-        """Get server name."""
+        """Get server name.
+
+        Returns:
+            Server name.
+        """
         return self._server_info()["serverName"]
 
     @property
     def guid(self) -> str:
-        """Get guid for the server."""
+        """Get guid for the server.
+
+        Returns:
+            Server GUID.
+        """
         return self._server_info()["guid"]
 
     @property
     def version(self) -> str:
-        """Get Splunk server version."""
+        """Get Splunk server version.
+
+        Returns:
+            Splunk version.
+        """
         return self._server_info()["version"]
 
     def is_captain(self) -> bool:
@@ -133,7 +185,7 @@ def is_shc_member(self) -> bool:
 
         return False
 
-    @retry(exceptions=[binding.HTTPError])
+    @utils.retry(exceptions=[binding.HTTPError])
     def get_shc_members(self) -> list:
         """Get SHC members.
 
@@ -162,7 +214,7 @@ def get_shc_members(self) -> list:
 
         return members
 
-    @retry(exceptions=[binding.HTTPError])
+    @utils.retry(exceptions=[binding.HTTPError])
     def is_captain_ready(self) -> bool:
         """Check if captain is ready.
 
@@ -173,7 +225,8 @@ def is_captain_ready(self) -> bool:
             True if captain is ready else False.
 
         Examples:
-            >>> serverinfo = solnlib.server_info.ServerInfo(session_key)
+            >>> from solnlib import server_info
+            >>> serverinfo = server_info.ServerInfo(session_key)
             >>> while 1:
             >>>    if serverinfo.is_captain_ready():
             >>>        break
@@ -189,7 +242,7 @@ def is_captain_ready(self) -> bool:
             cap_info["maintenance_mode"]
         )
 
-    @retry(exceptions=[binding.HTTPError])
+    @utils.retry(exceptions=[binding.HTTPError])
     def captain_info(self) -> dict:
         """Get captain information.
 

tests/integration/test_server_info.py

@@ -13,28 +13,33 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import unittest
-
 import context
+import pytest
 
 from solnlib import server_info
 
 
-class ServerInfoTest(unittest.TestCase):
-    def test_methods(self):
-        # This test does not check for guid, version and SHC related methods.
-        session_key = context.get_session_key()
-        si = server_info.ServerInfo(
-            session_key, context.scheme, context.host, context.port
-        )
-        self.assertEqual("custom-servername", si.server_name)
-        self.assertFalse(si.is_search_head())
-        self.assertFalse(si.is_shc_member())
-        with self.assertRaises(server_info.ServerInfoException):
-            si.get_shc_members()
-        with self.assertRaises(server_info.ServerInfoException):
-            si.captain_info()
-        with self.assertRaises(server_info.ServerInfoException):
-            si.is_captain_ready()
-        self.assertFalse(si.is_captain())
-        self.assertFalse(si.is_cloud_instance())
+def test_server_info_methods():
+    # This test does not check for guid, version and SHC related methods.
+    session_key = context.get_session_key()
+    si = server_info.ServerInfo(session_key, context.scheme, context.host, context.port)
+    assert "custom-servername" == si.server_name
+    assert si.is_search_head() is False
+    assert si.is_shc_member() is False
+    with pytest.raises(server_info.ServerInfoException):
+        si.get_shc_members()
+    with pytest.raises(server_info.ServerInfoException):
+        si.captain_info()
+    with pytest.raises(server_info.ServerInfoException):
+        si.is_captain_ready()
+    assert si.is_captain() is False
+    assert si.is_cloud_instance() is False
+
+
+def test_from_server_uri():
+    session_key = context.get_session_key()
+    si = server_info.ServerInfo.from_server_uri(
+        f"{context.scheme}://{context.host}:{context.port}", session_key
+    )
+    # Run 1 small test to check that .from_server_uri is working.
+    assert "custom-servername" == si.server_name

tests/unit/test_server_info.py

@@ -14,19 +14,24 @@
 # limitations under the License.
 #
 import common
+import pytest
 from splunklib import binding
 
 from solnlib import server_info
 
 
-class TestServerInfo:
-    def test_version(self, monkeypatch):
-        common.mock_splunkhome(monkeypatch)
-        common.mock_serverinfo(monkeypatch)
+def test_from_server_uri():
+    server_info.ServerInfo.from_server_uri("https://localhost:8089", common.SESSION_KEY)
 
-        si = server_info.ServerInfo(common.SESSION_KEY)
-        assert si.version == "6.3.1511.2"
 
+def test_from_server_uri_when_invalid_server_uri():
+    with pytest.raises(ValueError):
+        server_info.ServerInfo.from_server_uri(
+            "no-schema://localhost:99999", common.SESSION_KEY
+        )
+
+
+class TestServerInfo:
     def test_get_shc_members(self, monkeypatch):
         def _mock_get(self, path_segment, owner=None, app=None, sharing=None, **query):
             return common.make_response_record(