implement init command

Author: alafanechere

octavia-cli/octavia_cli/entrypoint.py

@@ -8,9 +8,10 @@
 import click
 from airbyte_api_client.api import workspace_api
 
+from .init import commands as init_commands
 from .list import commands as list_commands
 
-AVAILABLE_COMMANDS: List[click.Command] = [list_commands._list]
+AVAILABLE_COMMANDS: List[click.Command] = [list_commands._list, init_commands.init]
 
 
 @click.group()
@@ -38,11 +39,6 @@ def add_commands_to_octavia():
         octavia.add_command(command)
 
 
-@octavia.command(help="Scaffolds a local project directories.")
-def init():
-    raise click.ClickException("The init command is not yet implemented.")
-
-
 @octavia.command(name="import", help="Import an existing resources from the Airbyte instance.")
 def _import() -> None:
     raise click.ClickException("The import command is not yet implemented.")

octavia-cli/octavia_cli/init/__init__.py

@@ -0,0 +1,3 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#

octavia-cli/octavia_cli/init/commands.py

@@ -0,0 +1,33 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+import os
+from typing import Iterable, Tuple
+
+import click
+
+DIRECTORIES_TO_CREATE = {"connections", "destinations", "sources"}
+
+
+def create_directories(directories_to_create: Iterable[str]) -> Tuple[Iterable[str], Iterable[str]]:
+    created_directories = []
+    not_created_directories = []
+    for directory in directories_to_create:
+        try:
+            os.mkdir(directory)
+            created_directories.append(directory)
+        except FileExistsError:
+            not_created_directories.append(directory)
+    return created_directories, not_created_directories
+
+
+@click.command(help="Initialize required directories for the project.")
+def init():
+    created_directories, not_created_directories = create_directories(DIRECTORIES_TO_CREATE)
+    if created_directories:
+        message = f"✅ Created the following directories: {', '.join(created_directories)}"
+        click.echo(click.style(message, fg="green"))
+    if not_created_directories:
+        message = f"❌ Did not create the following already existing directories: {', '.join(not_created_directories) }"
+        click.echo(click.style(message, fg="red"))

octavia-cli/unit_tests/test_entrypoint.py

@@ -44,7 +44,7 @@ def test_commands_in_octavia_group():
 
 @pytest.mark.parametrize(
     "command",
-    [entrypoint.init, entrypoint.apply, entrypoint.create, entrypoint.delete, entrypoint._import],
+    [entrypoint.apply, entrypoint.create, entrypoint.delete, entrypoint._import],
 )
 def test_not_implemented_commands(command):
     runner = CliRunner()
@@ -54,4 +54,4 @@ def test_not_implemented_commands(command):
 
 
 def test_available_commands():
-    assert entrypoint.AVAILABLE_COMMANDS == [entrypoint.list_commands._list]
+    assert entrypoint.AVAILABLE_COMMANDS == [entrypoint.list_commands._list, entrypoint.init_commands.init]

octavia-cli/unit_tests/test_init/__init__.py

@@ -0,0 +1,3 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#

octavia-cli/unit_tests/test_init/test_commands.py

@@ -0,0 +1,55 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+import pytest
+from click.testing import CliRunner
+from octavia_cli.init import commands
+
+
+def test_directories_to_create():
+    assert commands.DIRECTORIES_TO_CREATE == {"connections", "destinations", "sources"}
+
+
+@pytest.mark.parametrize(
+    "directories_to_create,mkdir_side_effects,expected_created_directories,expected_not_created_directories",
+    [
+        (["dir_a", "dir_b"], None, ["dir_a", "dir_b"], []),
+        (["dir_a", "dir_b"], FileExistsError(), [], ["dir_a", "dir_b"]),
+        (["dir_a", "dir_b"], [None, FileExistsError()], ["dir_a"], ["dir_b"]),
+    ],
+)
+def test_create_directories(
+    mocker, directories_to_create, mkdir_side_effects, expected_created_directories, expected_not_created_directories
+):
+    mocker.patch.object(commands, "os", mocker.Mock(mkdir=mocker.Mock(side_effect=mkdir_side_effects)))
+    created_directories, not_created_directories = commands.create_directories(directories_to_create)
+    assert created_directories == expected_created_directories
+    assert not_created_directories == expected_not_created_directories
+    commands.os.mkdir.assert_has_calls([mocker.call(d) for d in directories_to_create])
+
+
+def test_init(mocker):
+    runner = CliRunner()
+    mocker.patch.object(commands, "create_directories", mocker.Mock(return_value=(["dir_a", "dir_b"], [])))
+    result = runner.invoke(commands.init)
+    assert result.exit_code == 0
+    assert result.output == "✅ Created the following directories: dir_a, dir_b\n"
+
+
+def test_init_some_existing_directories(mocker):
+    runner = CliRunner()
+    mocker.patch.object(commands, "create_directories", mocker.Mock(return_value=(["dir_a"], ["dir_b"])))
+    result = runner.invoke(commands.init)
+    assert result.exit_code == 0
+    assert (
+        result.output == "✅ Created the following directories: dir_a\n❌ Did not create the following already existing directories: dir_b\n"
+    )
+
+
+def test_init_all_existing_directories(mocker):
+    runner = CliRunner()
+    mocker.patch.object(commands, "create_directories", mocker.Mock(return_value=([], ["dir_a", "dir_b"])))
+    result = runner.invoke(commands.init)
+    assert result.exit_code == 0
+    assert result.output == "❌ Did not create the following already existing directories: dir_a, dir_b\n"

octavia-cli/unit_tests/test_list/__init__.py

@@ -0,0 +1,3 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#