add list connections command

Author: alafanechere

octavia-cli/octavia_cli/list/commands.py

@@ -7,6 +7,7 @@
 import click
 
 from .listings import (
+    Connections,
     DestinationConnectorsDefinitions,
     Destinations,
     SourceConnectorsDefinitions,
@@ -60,7 +61,16 @@ def destinations(ctx: click.Context):
     click.echo(destinations)
 
 
-AVAILABLE_COMMANDS: List[click.Command] = [connectors, sources, destinations]
+@click.command(help="List existing connections.")
+@click.pass_context
+def connections(ctx: click.Context):
+    api_client = ctx.obj["API_CLIENT"]
+    workspace_id = ctx.obj["WORKSPACE_ID"]
+    connections = Connections(api_client, workspace_id)
+    click.echo(connections)
+
+
+AVAILABLE_COMMANDS: List[click.Command] = [connectors, sources, destinations, connections]
 
 
 def add_commands_to_list():

octavia-cli/octavia_cli/list/listings.py

@@ -8,6 +8,7 @@
 import airbyte_api_client
 import octavia_cli.list.formatting as formatting
 from airbyte_api_client.api import (
+    connection_api,
     destination_api,
     destination_definition_api,
     source_api,
@@ -85,7 +86,7 @@ class DestinationConnectorsDefinitions(BaseListing):
     list_function_name = "list_latest_destination_definitions"
 
 
-class SourcesAndDestinations(BaseListing, abc.ABC):
+class WorkspaceListing(BaseListing, abc.ABC):
     def __init__(self, api_client: airbyte_api_client.ApiClient, workspace_id: str):
         self.workspace_id = workspace_id
         super().__init__(api_client)
@@ -95,15 +96,22 @@ def list_function_kwargs(self) -> dict:
         return {"workspace_id_request_body": WorkspaceIdRequestBody(workspace_id=self.workspace_id)}
 
 
-class Sources(SourcesAndDestinations):
+class Sources(WorkspaceListing):
     api = source_api.SourceApi
     fields_to_display = ["name", "sourceName", "sourceId"]
     list_field_in_response = "sources"
     list_function_name = "list_sources_for_workspace"
 
 
-class Destinations(SourcesAndDestinations):
+class Destinations(WorkspaceListing):
     api = destination_api.DestinationApi
     fields_to_display = ["name", "destinationName", "destinationId"]
     list_field_in_response = "destinations"
     list_function_name = "list_destinations_for_workspace"
+
+
+class Connections(WorkspaceListing):
+    api = connection_api.ConnectionApi
+    fields_to_display = ["name", "connectionId", "status", "sourceId", "destinationId"]
+    list_field_in_response = "connections"
+    list_function_name = "list_connections_for_workspace"

octavia-cli/unit_tests/test_list/test_commands.py

@@ -7,7 +7,7 @@
 
 
 def test_available_commands():
-    assert commands.AVAILABLE_COMMANDS == [commands.connectors, commands.sources, commands.destinations]
+    assert commands.AVAILABLE_COMMANDS == [commands.connectors, commands.sources, commands.destinations, commands.connections]
 
 
 def test_commands_in_list_group():
@@ -50,3 +50,12 @@ def test_destinations(mocker):
     result = runner.invoke((commands.destinations), obj=context_object)
     commands.Destinations.assert_called_with(context_object["API_CLIENT"], context_object["WORKSPACE_ID"])
     assert result.output == "DestinationsRepr\n"
+
+
+def test_connections(mocker):
+    mocker.patch.object(commands, "Connections", mocker.Mock(return_value="ConnectionsRepr"))
+    context_object = {"API_CLIENT": mocker.Mock(), "WORKSPACE_ID": "my_workspace_id"}
+    runner = CliRunner()
+    result = runner.invoke((commands.connections), obj=context_object)
+    commands.Connections.assert_called_with(context_object["API_CLIENT"], context_object["WORKSPACE_ID"])
+    assert result.output == "ConnectionsRepr\n"

octavia-cli/unit_tests/test_list/test_listings.py

@@ -4,6 +4,7 @@
 
 import pytest
 from airbyte_api_client.api import (
+    connection_api,
     destination_api,
     destination_definition_api,
     source_api,
@@ -12,11 +13,12 @@
 from octavia_cli.list import listings
 from octavia_cli.list.listings import (
     BaseListing,
+    Connections,
     DestinationConnectorsDefinitions,
     Destinations,
     SourceConnectorsDefinitions,
     Sources,
-    SourcesAndDestinations,
+    WorkspaceListing,
 )
 
 
@@ -110,18 +112,18 @@ def test_init(self, mock_api_client):
         assert destination_connectors_definition.list_function_name == "list_latest_destination_definitions"
 
 
-class TestSourcesAndDestinations:
+class TestWorkspaceListing:
     @pytest.fixture
     def patch_base_class(self, mocker):
         # Mock abstract methods to enable instantiating abstract class
-        mocker.patch.object(SourcesAndDestinations, "__abstractmethods__", set())
-        mocker.patch.object(SourcesAndDestinations, "api", mocker.Mock())
+        mocker.patch.object(WorkspaceListing, "__abstractmethods__", set())
+        mocker.patch.object(WorkspaceListing, "api", mocker.Mock())
 
     def test_init(self, patch_base_class, mocker, mock_api_client):
         mocker.patch.object(listings, "WorkspaceIdRequestBody")
         mocker.patch.object(BaseListing, "__init__")
-        assert SourcesAndDestinations.__base__ == BaseListing
-        sources_and_destinations = SourcesAndDestinations(mock_api_client, "my_workspace_id")
+        assert WorkspaceListing.__base__ == BaseListing
+        sources_and_destinations = WorkspaceListing(mock_api_client, "my_workspace_id")
 
         assert sources_and_destinations.workspace_id == "my_workspace_id"
         assert sources_and_destinations.list_function_kwargs == {"workspace_id_request_body": listings.WorkspaceIdRequestBody.return_value}
@@ -130,12 +132,12 @@ def test_init(self, patch_base_class, mocker, mock_api_client):
 
     def test_abstract(self, mock_api_client):
         with pytest.raises(TypeError):
-            SourcesAndDestinations(mock_api_client)
+            WorkspaceListing(mock_api_client)
 
 
 class TestSources:
     def test_init(self, mock_api_client):
-        assert Sources.__base__ == SourcesAndDestinations
+        assert Sources.__base__ == WorkspaceListing
         sources = Sources(mock_api_client, "my_workspace_id")
         assert sources.api == source_api.SourceApi
         assert sources.fields_to_display == ["name", "sourceName", "sourceId"]
@@ -145,9 +147,19 @@ def test_init(self, mock_api_client):
 
 class TestDestinations:
     def test_init(self, mock_api_client):
-        assert Destinations.__base__ == SourcesAndDestinations
+        assert Destinations.__base__ == WorkspaceListing
         destinations = Destinations(mock_api_client, "my_workspace_id")
         assert destinations.api == destination_api.DestinationApi
         assert destinations.fields_to_display == ["name", "destinationName", "destinationId"]
         assert destinations.list_field_in_response == "destinations"
         assert destinations.list_function_name == "list_destinations_for_workspace"
+
+
+class TestConnections:
+    def test_init(self, mock_api_client):
+        assert Connections.__base__ == WorkspaceListing
+        connections = Connections(mock_api_client, "my_workspace_id")
+        assert connections.api == connection_api.ConnectionApi
+        assert connections.fields_to_display == ["name", "connectionId", "status", "sourceId", "destinationId"]
+        assert connections.list_field_in_response == "connections"
+        assert connections.list_function_name == "list_connections_for_workspace"