adds infrahubctl repository list command

Author: wvandeun

.github/workflows/ci.yml

@@ -116,6 +116,9 @@ jobs:
 
 
   unit-tests:
+    env:
+      # workaround for Rich table column width
+      COLUMNS: 140
     strategy:
       matrix:
         python-version:

changelog/+infrahubctl_repository_list.added.md

@@ -0,0 +1 @@
+adds `infrahubctl repository list` command

infrahub_sdk/ctl/repository.py

@@ -6,13 +6,14 @@
 import yaml
 from pydantic import ValidationError
 from rich.console import Console
+from rich.table import Table
 
 from infrahub_sdk.ctl.client import initialize_client
 
 from ..async_typer import AsyncTyper
 from ..ctl.exceptions import FileNotValidError
 from ..ctl.utils import init_logging
-from ..graphql import Mutation
+from ..graphql import Mutation, Query
 from ..schema.repository import InfrahubRepositoryConfig
 from ._file import read_file
 from .parameters import CONFIG_PARAM
@@ -102,3 +103,57 @@ async def add(
     )
 
     await client.execute_graphql(query=query.render(), branch_name=branch, tracker="mutation-repository-create")
+
+
+@app.command()
+async def list(
+    branch: str | None = None,
+    debug: bool = False,
+    _: str = CONFIG_PARAM,
+) -> None:
+    init_logging(debug=debug)
+
+    client = initialize_client(branch=branch)
+
+    repo_status_query = {
+        "CoreGenericRepository": {
+            "edges": {
+                "node": {
+                    "__typename": None,
+                    "name": {"value": None},
+                    "operational_status": {"value": None},
+                    "sync_status": {"value": None},
+                    "internal_status": {"value": None},
+                    "... on CoreReadOnlyRepository": {
+                        "ref": {"value": None},
+                    },
+                }
+            }
+        },
+    }
+
+    query = Query(name="GetRepositoryStatus", query=repo_status_query)
+    resp = await client.execute_graphql(query=query.render(), branch_name=branch, tracker="query-repository-list")
+
+    table = Table(title="List of all Repositories")
+
+    table.add_column("Name", justify="right", style="cyan", no_wrap=True)
+    table.add_column("Type")
+    table.add_column("Operational status")
+    table.add_column("Sync status")
+    table.add_column("Internal status")
+    table.add_column("Ref")
+
+    for repository_node in resp["CoreGenericRepository"]["edges"]:
+        repository = repository_node["node"]
+
+        table.add_row(
+            repository["name"]["value"],
+            repository["__typename"],
+            repository["operational_status"]["value"],
+            repository["sync_status"]["value"],
+            repository["internal_status"]["value"],
+            repository["ref"]["value"] if "ref" in repository else "",
+        )
+
+    console.print(table)

tests/fixtures/integration/test_infrahubctl/repository_list/output.txt

@@ -0,0 +1,7 @@
+                                                         List of all Repositories                                                          
+┏━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
+┃           Name ┃ Type                   ┃ Operational status ┃ Sync status ┃ Internal status ┃ Ref                                      ┃
+┡━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
+│ Demo Edge Repo │ CoreReadOnlyRepository │ unknown            │ in-sync     │ active          │ 5bffc938ba0d00dd111cb19331cdef6aab3729c2 │
+│    My Own Repo │ CoreRepository         │ in-sync            │ in-sync     │ active          │                                          │
+└────────────────┴────────────────────────┴────────────────────┴─────────────┴─────────────────┴──────────────────────────────────────────┘

tests/helpers/fixtures.py

@@ -5,3 +5,12 @@ def get_fixtures_dir() -> Path:
     """Get the directory which stores fixtures that are common to multiple unit/integration tests."""
     here = Path(__file__).parent.resolve()
     return here.parent / "fixtures"
+
+
+def read_fixture(file_name: str, fixture_subdir: str = ".") -> str:
+    """Read the contents of a fixture."""
+    file_path = get_fixtures_dir() / fixture_subdir / file_name
+    with file_path.open("r", encoding="utf-8") as fhd:
+        fixture_contents = fhd.read()
+
+    return fixture_contents

tests/unit/ctl/conftest.py

@@ -106,3 +106,43 @@ async def mock_repositories_query(httpx_mock: HTTPXMock) -> HTTPXMock:
     httpx_mock.add_response(method="POST", url="http://mock/graphql/main", json=response1)
     httpx_mock.add_response(method="POST", url="http://mock/graphql/cr1234", json=response2)
     return httpx_mock
+
+
+@pytest.fixture
+def mock_repositories_list(httpx_mock: HTTPXMock) -> HTTPXMock:
+    response = {
+        "data": {
+            "CoreGenericRepository": {
+                "edges": [
+                    {
+                        "node": {
+                            "__typename": "CoreReadOnlyRepository",
+                            "name": {"value": "Demo Edge Repo"},
+                            "operational_status": {"value": "unknown"},
+                            "sync_status": {"value": "in-sync"},
+                            "internal_status": {"value": "active"},
+                            "ref": {"value": "5bffc938ba0d00dd111cb19331cdef6aab3729c2"},
+                        }
+                    },
+                    {
+                        "node": {
+                            "__typename": "CoreRepository",
+                            "name": {"value": "My Own Repo"},
+                            "operational_status": {"value": "in-sync"},
+                            "sync_status": {"value": "in-sync"},
+                            "internal_status": {"value": "active"},
+                        }
+                    },
+                ]
+            }
+        }
+    }
+
+    httpx_mock.add_response(
+        method="POST",
+        status_code=200,
+        url="http://mock/graphql/main",
+        json=response,
+        match_headers={"X-Infrahub-Tracker": "query-repository-list"},
+    )
+    return httpx_mock

tests/unit/ctl/test_repository_app.py

@@ -8,6 +8,8 @@
 
 from infrahub_sdk.client import InfrahubClient
 from infrahub_sdk.ctl.cli_commands import app
+from tests.helpers.fixtures import read_fixture
+from tests.helpers.utils import strip_color
 
 runner = CliRunner()
 
@@ -24,11 +26,11 @@ def mock_client() -> mock.Mock:
 # ---------------------------------------------------------
 # infrahubctl  repository command tests
 # ---------------------------------------------------------
-@mock.patch("infrahub_sdk.ctl.repository.initialize_client")
 class TestInfrahubctlRepository:
     """Groups the 'infrahubctl repository' test cases."""
 
     @requires_python_310
+    @mock.patch("infrahub_sdk.ctl.repository.initialize_client")
     def test_repo_no_username(self, mock_init_client, mock_client) -> None:
         """Case allow no username to be passed in and set it as None rather than blank string that fails."""
         mock_cred = mock.AsyncMock()
@@ -89,6 +91,7 @@ def test_repo_no_username(self, mock_init_client, mock_client) -> None:
         )
 
     @requires_python_310
+    @mock.patch("infrahub_sdk.ctl.repository.initialize_client")
     def test_repo_username(self, mock_init_client, mock_client) -> None:
         """Case allow no username to be passed in and set it as None rather than blank string that fails."""
         mock_cred = mock.AsyncMock()
@@ -151,6 +154,7 @@ def test_repo_username(self, mock_init_client, mock_client) -> None:
         )
 
     @requires_python_310
+    @mock.patch("infrahub_sdk.ctl.repository.initialize_client")
     def test_repo_readonly_true(self, mock_init_client, mock_client) -> None:
         """Case allow no username to be passed in and set it as None rather than blank string that fails."""
         mock_cred = mock.AsyncMock()
@@ -212,6 +216,7 @@ def test_repo_readonly_true(self, mock_init_client, mock_client) -> None:
         )
 
     @requires_python_310
+    @mock.patch("infrahub_sdk.ctl.repository.initialize_client")
     def test_repo_description_commit_branch(self, mock_init_client, mock_client) -> None:
         """Case allow no username to be passed in and set it as None rather than blank string that fails."""
         mock_cred = mock.AsyncMock()
@@ -278,3 +283,8 @@ def test_repo_description_commit_branch(self, mock_init_client, mock_client) ->
             branch_name="develop",
             tracker="mutation-repository-create",
         )
+
+    def test_repo_list(self, mock_repositories_list) -> None:
+        result = runner.invoke(app, ["repository", "list", "--branch", "main"])
+        assert result.exit_code == 0
+        assert strip_color(result.stdout) == read_fixture("output.txt", "integration/test_infrahubctl/repository_list")

tests/unit/ctl/test_transform_app.py

@@ -13,6 +13,7 @@
 
 from infrahub_sdk.ctl.cli_commands import app
 from infrahub_sdk.repository import GitRepoManager
+from tests.helpers.fixtures import read_fixture
 from tests.helpers.utils import change_directory, strip_color
 
 runner = CliRunner()
@@ -25,14 +26,6 @@
 requires_python_310 = pytest.mark.skipif(sys.version_info < (3, 10), reason="Requires Python 3.10 or higher")
 
 
-def read_fixture(file_name: str, fixture_subdir: str = ".") -> str:
-    """Read the contents of a fixture."""
-    with Path(FIXTURE_BASE_DIR / fixture_subdir / file_name).open("r", encoding="utf-8") as fhd:
-        fixture_contents = fhd.read()
-
-    return fixture_contents
-
-
 @pytest.fixture
 def tags_transform_dir():
     temp_dir = tempfile.mkdtemp()
@@ -123,10 +116,12 @@ def test_infrahubctl_transform_cmd_success(httpx_mock: HTTPXMock, tags_transform
         httpx_mock.add_response(
             method="POST",
             url="http://mock/graphql/main",
-            json=json.loads(read_fixture("case_success_api_return.json", "transform_cmd")),
+            json=json.loads(read_fixture("case_success_api_return.json", "integration/test_infrahubctl/transform_cmd")),
         )
 
         with change_directory(tags_transform_dir):
             output = runner.invoke(app, ["transform", "tags_transform", "tag=red"])
-            assert strip_color(output.stdout) == read_fixture("case_success_output.txt", "transform_cmd")
+            assert strip_color(output.stdout) == read_fixture(
+                "case_success_output.txt", "integration/test_infrahubctl/transform_cmd"
+            )
             assert output.exit_code == 0