diff --git a/changelog/109.added.md b/changelog/109.added.md new file mode 100644 index 0000000..31b8137 --- /dev/null +++ b/changelog/109.added.md @@ -0,0 +1 @@ +Adds `infrahubctl info` command to display information of the connectivity status of the SDK. \ No newline at end of file diff --git a/infrahub_sdk/client.py b/infrahub_sdk/client.py index f717893..96500bd 100644 --- a/infrahub_sdk/client.py +++ b/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 @@ -272,6 +272,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, @@ -1479,6 +1495,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, diff --git a/infrahub_sdk/ctl/cli_commands.py b/infrahub_sdk/ctl/cli_commands.py index 2471f72..09a73ec 100644 --- a/infrahub_sdk/ctl/cli_commands.py +++ b/infrahub_sdk/ctl/cli_commands.py @@ -4,6 +4,7 @@ import functools import importlib import logging +import platform import sys from pathlib import Path from typing import TYPE_CHECKING, Any, Callable @@ -12,7 +13,11 @@ import typer import ujson from rich.console import Console +from rich.layout import Layout from rich.logging import RichHandler +from rich.panel import Panel +from rich.pretty import Pretty +from rich.table import Table from rich.traceback import Traceback from .. import __version__ as sdk_version @@ -392,11 +397,106 @@ def protocols( @app.command(name="version") @catch_exception(console=console) -def version(_: str = CONFIG_PARAM) -> None: - """Display the version of Infrahub and the version of the Python SDK in use.""" +def version() -> None: + """Display the version of Python and the version of the Python SDK in use.""" - client = initialize_client_sync() - response = client.execute_graphql(query="query { InfrahubInfo { version }}") + console.print(f"Python: {platform.python_version()}\nPython SDK: v{sdk_version}") - infrahub_version = response["InfrahubInfo"]["version"] - console.print(f"Infrahub: v{infrahub_version}\nPython SDK: v{sdk_version}") + +@app.command(name="info") +@catch_exception(console=console) +def info(detail: bool = typer.Option(False, help="Display detailed information."), _: str = CONFIG_PARAM) -> None: # noqa: PLR0915 + """Display the status of the Python SDK.""" + + info: dict[str, Any] = { + "error": None, + "status": ":x:", + "infrahub_version": "N/A", + "user_info": {}, + "groups": {}, + } + try: + client = initialize_client_sync() + info["infrahub_version"] = client.get_version() + info["user_info"] = client.get_user() + info["status"] = ":white_heavy_check_mark:" + info["groups"] = client.get_user_permissions() + except Exception as e: + info["error"] = f"{e!s} ({e.__class__.__name__})" + + if detail: + layout = Layout() + + # Layout structure + new_console = Console(height=45) + layout = Layout() + layout.split_column( + Layout(name="body", ratio=1), + ) + layout["body"].split_row( + Layout(name="left"), + Layout(name="right"), + ) + + layout["left"].split_column( + Layout(name="connection_status", size=7), + Layout(name="client_info", ratio=1), + ) + + layout["right"].split_column( + Layout(name="version_info", size=7), + Layout(name="infrahub_info", ratio=1), + ) + + # Connection status panel + connection_status = Table(show_header=False, box=None) + connection_status.add_row("Server Address:", client.config.address) + connection_status.add_row("Status:", info["status"]) + if info["error"]: + connection_status.add_row("Error Reason:", info["error"]) + layout["connection_status"].update(Panel(connection_status, title="Connection Status")) + + # Version information panel + version_info = Table(show_header=False, box=None) + version_info.add_row("Python Version:", platform.python_version()) + version_info.add_row("Infrahub Version", info["infrahub_version"]) + version_info.add_row("Infrahub SDK:", sdk_version) + layout["version_info"].update(Panel(version_info, title="Version Information")) + + # SDK client configuration panel + pretty_model = Pretty(client.config.model_dump(), expand_all=True) + layout["client_info"].update(Panel(pretty_model, title="Client Info")) + + # Infrahub information planel + infrahub_info = Table(show_header=False, box=None) + if info["user_info"]: + infrahub_info.add_row("User:", info["user_info"]["AccountProfile"]["display_label"]) + infrahub_info.add_row("Description:", info["user_info"]["AccountProfile"]["description"]["value"]) + infrahub_info.add_row("Status:", info["user_info"]["AccountProfile"]["status"]["label"]) + infrahub_info.add_row( + "Number of Groups:", str(info["user_info"]["AccountProfile"]["member_of_groups"]["count"]) + ) + + if groups := info["groups"]: + infrahub_info.add_row("Groups:", "") + for group, roles in groups.items(): + infrahub_info.add_row("", group, ", ".join(roles)) + + layout["infrahub_info"].update(Panel(infrahub_info, title="Infrahub Info")) + + new_console.print(layout) + else: + # Simple output + table = Table(show_header=False, box=None) + table.add_row("Address:", client.config.address) + table.add_row("Connection Status:", info["status"]) + if info["error"]: + table.add_row("Connection Error:", info["error"]) + + table.add_row("Python Version:", platform.python_version()) + table.add_row("SDK Version:", sdk_version) + table.add_row("Infrahub Version:", info["infrahub_version"]) + if account := info["user_info"].get("AccountProfile"): + table.add_row("User:", account["display_label"]) + + console.print(table) diff --git a/infrahub_sdk/queries.py b/infrahub_sdk/queries.py index 75bc593..9f24f73 100644 --- a/infrahub_sdk/queries.py +++ b/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 + } + } +} +""" diff --git a/infrahub_sdk/utils.py b/infrahub_sdk/utils.py index 339c66e..1231dae 100644 --- a/infrahub_sdk/utils.py +++ b/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 diff --git a/tests/fixtures/account_profile.json b/tests/fixtures/account_profile.json new file mode 100644 index 0000000..b3dfa7b --- /dev/null +++ b/tests/fixtures/account_profile.json @@ -0,0 +1,79 @@ +{ + "data": { + "AccountProfile": { + "id": "1816ebcd-cea7-3bf7-3fc9-c51282f03fe7", + "display_label": "Admin", + "account_type": { + "value": "User", + "__typename": "TextAttribute", + "updated_at": "2025-01-02T16:06:15.565985+00:00" + }, + "status": { + "label": "Active", + "value": "active", + "updated_at": "2025-01-02T16:06:15.565985+00:00", + "__typename": "Dropdown" + }, + "description": { + "value": null, + "updated_at": "2025-01-02T16:06:15.565985+00:00", + "__typename": "TextAttribute" + }, + "label": { + "value": "Admin", + "updated_at": "2025-01-02T16:06:15.565985+00:00", + "__typename": "TextAttribute" + }, + "member_of_groups": { + "count": 1, + "edges": [ + { + "node": { + "display_label": "Super Administrators", + "group_type": { + "value": "default" + }, + "id": "1816ebce-1cbe-2e96-3fc3-c5124c324bac", + "roles": { + "count": 1, + "edges": [ + { + "node": { + "permissions": { + "count": 2, + "edges": [ + { + "node": { + "display_label": "super_admin 6", + "identifier": { + "value": "global:super_admin:allow_all" + } + } + }, + { + "node": { + "display_label": "* * any 6", + "identifier": { + "value": "object:*:*:any:allow_all" + } + } + } + ] + } + } + } + ] + } + } + } + ] + }, + "__typename": "CoreAccount", + "name": { + "value": "admin", + "updated_at": "2025-01-02T16:06:15.565985+00:00", + "__typename": "TextAttribute" + } + } + } +} \ No newline at end of file diff --git a/tests/unit/ctl/conftest.py b/tests/unit/ctl/conftest.py index 17c0ecc..7e00b02 100644 --- a/tests/unit/ctl/conftest.py +++ b/tests/unit/ctl/conftest.py @@ -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: diff --git a/tests/unit/ctl/test_cli.py b/tests/unit/ctl/test_cli.py index a6df7f9..d73a710 100644 --- a/tests/unit/ctl/test_cli.py +++ b/tests/unit/ctl/test_cli.py @@ -27,3 +27,46 @@ def test_validate_all_groups_have_names(): assert app.registered_groups for group in app.registered_groups: assert group.name + + +@requires_python_310 +def test_version_command(): + result = runner.invoke(app, ["version"]) + assert result.exit_code == 0 + assert "Python SDK: v" in result.stdout + + +@requires_python_310 +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"]: + assert expected in result.stdout, f"'{expected}' not found in info command output" + + +@requires_python_310 +def test_info_command_failure(): + result = runner.invoke(app, ["info"]) + assert result.exit_code == 0 + assert "Connection Error" in result.stdout + + +@requires_python_310 +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 [ + "Connection Status", + "Version Information", + "Client Info", + "Infrahub Info", + "Groups:", + ]: + assert expected in result.stdout, f"'{expected}' not found in detailed info command output" + + +@requires_python_310 +def test_info_detail_command_failure(): + result = runner.invoke(app, ["info", "--detail"]) + assert result.exit_code == 0 + assert "Error Reason" in result.stdout diff --git a/tests/unit/sdk/conftest.py b/tests/unit/sdk/conftest.py index 0be09a1..7f1d494 100644 --- a/tests/unit/sdk/conftest.py +++ b/tests/unit/sdk/conftest.py @@ -2141,6 +2141,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""" diff --git a/tests/unit/sdk/test_client.py b/tests/unit/sdk/test_client.py index cef5480..3f08780 100644 --- a/tests/unit/sdk/test_client.py +++ b/tests/unit/sdk/test_client.py @@ -83,6 +83,38 @@ async def test_method_count(clients, mock_query_repository_count, client_type): assert count == 5 +@pytest.mark.parametrize("client_type", client_types) +async def test_method_get_version(clients, mock_query_infrahub_version, client_type): # pylint: disable=unused-argument + if client_type == "standard": + version = await clients.standard.get_version() + else: + version = clients.sync.get_version() + + assert version == "1.1.0" + + +@pytest.mark.parametrize("client_type", client_types) +async def test_method_get_user(clients, mock_query_infrahub_user, client_type): # pylint: disable=unused-argument + if client_type == "standard": + user = await clients.standard.get_user() + else: + user = clients.sync.get_user() + + assert isinstance(user, dict) + assert user["AccountProfile"]["display_label"] == "Admin" + + +@pytest.mark.parametrize("client_type", client_types) +async def test_method_get_user_permissions(clients, mock_query_infrahub_user, client_type): # pylint: disable=unused-argument + if client_type == "standard": + groups = await clients.standard.get_user_permissions() + else: + groups = clients.sync.get_user_permissions() + + assert isinstance(groups, dict) + assert groups["Super Administrators"] == ["global:super_admin:allow_all", "object:*:*:any:allow_all"] + + @pytest.mark.parametrize("client_type", client_types) async def test_method_all_with_limit(clients, mock_query_repository_page1_2, client_type): # pylint: disable=unused-argument if client_type == "standard":