Move functions into SDK

Author: minitriga

infrahub_sdk/client.py

@@ -46,13 +46,13 @@
 )
 from .object_store import ObjectStore, ObjectStoreSync
 from .protocols_base import CoreNode, CoreNodeSync
-from .queries import get_commit_update_mutation
+from .queries import QUERY_USER, get_commit_update_mutation
 from .query_groups import InfrahubGroupContext, InfrahubGroupContextSync
 from .schema import InfrahubSchema, InfrahubSchemaSync, NodeSchemaAPI
 from .store import NodeStore, NodeStoreSync
 from .timestamp import Timestamp
 from .types import AsyncRequester, HTTPMethod, SyncRequester
-from .utils import decode_json, is_valid_uuid
+from .utils import decode_json, get_user_permissions, is_valid_uuid
 
 if TYPE_CHECKING:
     from types import TracebackType
@@ -274,6 +274,22 @@ def _initialize(self) -> None:
         self._request_method: AsyncRequester = self.config.requester or self._default_request_method
         self.group_context = InfrahubGroupContext(self)
 
+    async def get_version(self) -> str:
+        """Return the Infrahub version."""
+        response = await self.execute_graphql(query="query { InfrahubInfo { version }}")
+        version = response.get("InfrahubInfo", {}).get("version", "")
+        return version
+
+    async def get_user(self) -> dict:
+        """Return user information"""
+        user_info = await self.execute_graphql(query=QUERY_USER)
+        return user_info
+
+    async def get_user_permissions(self) -> dict:
+        """Return user permissions"""
+        user_info = await self.get_user()
+        return get_user_permissions(user_info["AccountProfile"]["member_of_groups"]["edges"])
+
     @overload
     async def create(
         self,
@@ -1427,6 +1443,22 @@ def _initialize(self) -> None:
         self._request_method: SyncRequester = self.config.sync_requester or self._default_request_method
         self.group_context = InfrahubGroupContextSync(self)
 
+    def get_version(self) -> str:
+        """Return the Infrahub version."""
+        response = self.execute_graphql(query="query { InfrahubInfo { version }}")
+        version = response.get("InfrahubInfo", {}).get("version", "")
+        return version
+
+    def get_user(self) -> dict:
+        """Return user information"""
+        user_info = self.execute_graphql(query=QUERY_USER)
+        return user_info
+
+    def get_user_permissions(self) -> dict:
+        """Return user permissions"""
+        user_info = self.get_user()
+        return get_user_permissions(user_info["AccountProfile"]["member_of_groups"]["edges"])
+
     @overload
     def create(
         self,

infrahub_sdk/ctl/cli_commands.py

@@ -39,7 +39,6 @@
 from ..ctl.utils import (
     catch_exception,
     execute_graphql_query,
-    get_user_permissions,
     load_yamlfile_from_disk_and_exit,
     parse_cli_vars,
 )
@@ -409,75 +408,6 @@ def version() -> None:
 def info(detail: bool = typer.Option(False, help="Display detailed information."), _: str = CONFIG_PARAM) -> None:  # noqa: PLR0915
     """Display the status of the Python SDK."""
 
-    query = """
-query GET_PROFILE_DETAILS {
-  AccountProfile {
-    id
-    display_label
-    account_type {
-      value
-      __typename
-      updated_at
-    }
-    status {
-      label
-      value
-      updated_at
-      __typename
-    }
-    description {
-      value
-      updated_at
-      __typename
-    }
-    label {
-      value
-      updated_at
-      __typename
-    }
-    member_of_groups {
-      count
-      edges {
-        node {
-          display_label
-          group_type {
-            value
-          }
-          ... on CoreAccountGroup {
-            id
-            roles {
-              count
-              edges {
-                node {
-                  permissions {
-                    count
-                    edges {
-                      node {
-                        display_label
-                        identifier {
-                          value
-                        }
-                      }
-                    }
-                  }
-                }
-              }
-            }
-            display_label
-          }
-        }
-      }
-    }
-    __typename
-    name {
-      value
-      updated_at
-      __typename
-    }
-  }
-}
-"""
-
     info: dict[str, Any] = {
         "error": None,
         "status": ":x:",
@@ -487,11 +417,10 @@ def info(detail: bool = typer.Option(False, help="Display detailed information."
     }
     try:
         client = initialize_client_sync()
-        version_response = client.execute_graphql(query="query { InfrahubInfo { version }}")
-        info["infrahub_version"] = version_response["InfrahubInfo"]["version"]
-        info["user_info"] = client.execute_graphql(query=query)
+        info["infrahub_version"] = client.get_version()
+        info["user_info"] = client.get_user()
         info["status"] = ":white_heavy_check_mark:"
-        info["groups"] = get_user_permissions(info["user_info"]["AccountProfile"]["member_of_groups"]["edges"])
+        info["groups"] = client.get_user_permissions()
     except Exception as e:
         info["error"] = f"{e!s} ({e.__class__.__name__})"
 
@@ -550,8 +479,8 @@ def info(detail: bool = typer.Option(False, help="Display detailed information."
 
             if groups := info["groups"]:
                 infrahub_info.add_row("Groups:", "")
-                for k, v in groups.items():
-                    infrahub_info.add_row("", k, v)
+                for group, roles in groups.items():
+                    infrahub_info.add_row("", group, ", ".join(roles))
 
         layout["infrahub_info"].update(Panel(infrahub_info, title="Infrahub Info"))
 

infrahub_sdk/ctl/utils.py

@@ -214,21 +214,3 @@ def load_yamlfile_from_disk_and_exit(
         raise typer.Exit(1)
 
     return data_files
-
-
-def get_user_permissions(data: list[dict]) -> dict:
-    groups = {}
-    for group in data:
-        group_name = group["node"]["display_label"]
-        permissions = []
-
-        roles = group["node"].get("roles", {}).get("edges", [])
-        for role in roles:
-            role_permissions = role["node"].get("permissions", {}).get("edges", [])
-            for permission in role_permissions:
-                permissions.append(permission["node"]["identifier"]["value"])
-
-        permissions_str = ", ".join(permissions)
-        groups[group_name] = permissions_str
-
-    return groups

infrahub_sdk/queries.py

@@ -42,3 +42,72 @@ def get_commit_update_mutation(is_read_only: bool = False) -> str:
   }
 }
 """
+
+QUERY_USER = """
+query GET_PROFILE_DETAILS {
+  AccountProfile {
+    id
+    display_label
+    account_type {
+      value
+      __typename
+      updated_at
+    }
+    status {
+      label
+      value
+      updated_at
+      __typename
+    }
+    description {
+      value
+      updated_at
+      __typename
+    }
+    label {
+      value
+      updated_at
+      __typename
+    }
+    member_of_groups {
+      count
+      edges {
+        node {
+          display_label
+          group_type {
+            value
+          }
+          ... on CoreAccountGroup {
+            id
+            roles {
+              count
+              edges {
+                node {
+                  permissions {
+                    count
+                    edges {
+                      node {
+                        display_label
+                        identifier {
+                          value
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+            display_label
+          }
+        }
+      }
+    }
+    __typename
+    name {
+      value
+      updated_at
+      __typename
+    }
+  }
+}
+"""

infrahub_sdk/utils.py

@@ -335,3 +335,20 @@ def write_to_file(path: Path, value: Any) -> bool:
     written = path.write_text(to_write)
 
     return written is not None
+
+
+def get_user_permissions(data: list[dict]) -> dict:
+    groups = {}
+    for group in data:
+        group_name = group["node"]["display_label"]
+        permissions = []
+
+        roles = group["node"].get("roles", {}).get("edges", [])
+        for role in roles:
+            role_permissions = role["node"].get("permissions", {}).get("edges", [])
+            for permission in role_permissions:
+                permissions.append(permission["node"]["identifier"]["value"])
+
+        groups[group_name] = permissions
+
+    return groups

tests/fixtures/integration/test_infrahubctl/info_cmd/account_profile.json renamed to tests/fixtures/account_profile.json

@@ -1,6 +1,8 @@
 import pytest
 from pytest_httpx import HTTPXMock
 
+from tests.unit.sdk.conftest import mock_query_infrahub_user, mock_query_infrahub_version  # noqa: F401
+
 
 @pytest.fixture
 async def mock_branches_list_query(httpx_mock: HTTPXMock) -> HTTPXMock:

tests/fixtures/integration/test_infrahubctl/info_cmd/infrahub_info.json

@@ -1,14 +1,10 @@
-import json
 import sys
 
 import pytest
-from pytest_httpx._httpx_mock import HTTPXMock
 from typer.testing import CliRunner
 
 from infrahub_sdk.ctl.cli import app
 
-from .test_transform_app import read_fixture
-
 runner = CliRunner()
 
 requires_python_310 = pytest.mark.skipif(sys.version_info < (3, 10), reason="Requires Python 3.10 or higher")
@@ -40,23 +36,8 @@ def test_version_command():
     assert "Python SDK: v" in result.stdout
 
 
-def mock_common_responses(httpx_mock: HTTPXMock):
-    httpx_mock.add_response(
-        method="POST",
-        url="http://mock/graphql/main",
-        json=json.loads(read_fixture("infrahub_info.json", "info_cmd")),
-    )
-    httpx_mock.add_response(
-        method="POST",
-        url="http://mock/graphql/main",
-        json=json.loads(read_fixture("account_profile.json", "info_cmd")),
-    )
-
-
 @requires_python_310
-def test_info_command_success(httpx_mock: HTTPXMock):
-    mock_common_responses(httpx_mock)
-
+def test_info_command_success(mock_query_infrahub_version, mock_query_infrahub_user):
     result = runner.invoke(app, ["info"])
     assert result.exit_code == 0
     for expected in ["Connection Status", "Python Version", "SDK Version", "Infrahub Version"]:
@@ -71,8 +52,7 @@ def test_info_command_failure():
 
 
 @requires_python_310
-def test_info_detail_command_success(httpx_mock: HTTPXMock):
-    mock_common_responses(httpx_mock)
+def test_info_detail_command_success(mock_query_infrahub_version, mock_query_infrahub_user):
     result = runner.invoke(app, ["info", "--detail"])
     assert result.exit_code == 0
     for expected in [

tests/unit/ctl/conftest.py

@@ -2031,6 +2031,19 @@ async def mock_query_mutation_location_create_failed(httpx_mock: HTTPXMock) -> H
     return httpx_mock
 
 
+@pytest.fixture
+async def mock_query_infrahub_version(httpx_mock: HTTPXMock) -> HTTPXMock:
+    httpx_mock.add_response(method="POST", json={"data": {"InfrahubInfo": {"version": "1.1.0"}}})
+    return httpx_mock
+
+
+@pytest.fixture
+async def mock_query_infrahub_user(httpx_mock: HTTPXMock) -> HTTPXMock:
+    response_text = (get_fixtures_dir() / "account_profile.json").read_text(encoding="UTF-8")
+    httpx_mock.add_response(method="POST", json=ujson.loads(response_text))
+    return httpx_mock
+
+
 @pytest.fixture
 def query_01() -> str:
     """Simple query with one document"""