"kiwi init" as in previous version

kiwi_scp/commands/cmd_init.py

@@ -1,20 +1,58 @@
+import logging
+import os
+from ipaddress import IPv4Network
+from pathlib import Path
+
 import click
 
 from .cli import KiwiCTX, pass_kiwi_ctx
+from .._constants import KIWI_CONF_NAME
+from ..config import Config
+from ..misc import user_query
 
 
 @click.command(
     "init",
     short_help="Initializes kiwi-scp"
 )
-@click.argument(
-    "path",
-    required=False,
-    type=click.Path(resolve_path=True)
+@click.option(
+    "-f/-F",
+    "--force/--no-force",
+    help=f"use default values even if {KIWI_CONF_NAME} is present",
+)
+@click.option(
+    "-s/-S",
+    "--show/--no-show",
+    help=f"show effective {KIWI_CONF_NAME} contents instead",
 )
 @pass_kiwi_ctx
-def cmd(ctx: KiwiCTX, path):
+def cmd(ctx: KiwiCTX, force: bool, show: bool):
     """Initialize or reconfigure a kiwi-scp instance"""
 
-    click.echo(f"Hello init, kiwi version {ctx.config.version}")
-    pass
+    current_config = Config() if force else ctx.config
+
+    if show:
+        # just show the currently effective kiwi.yml
+        click.echo_via_pager(current_config.kiwi_yml)
+        return
+
+    # check force switch
+    if force and os.path.isfile(KIWI_CONF_NAME):
+        logging.warning(f"Overwriting an existing '{KIWI_CONF_NAME}'!")
+
+    # build new kiwi dict
+    kiwi_dict = current_config.kiwi_dict
+    kiwi_dict.update({
+        "version": user_query("kiwi-scp version to use in this instance", current_config.version),
+        "storage": {
+            "directory": user_query("local directory for service data", current_config.storage.directory, Path),
+        },
+        "network": {
+            "name": user_query("name for local network hub", current_config.network.name),
+            "cidr": user_query("CIDRv4 block for local network hub", current_config.network.cidr, IPv4Network),
+        },
+    })
+
+    # write out as new kiwi.yml
+    with open(ctx.instance.joinpath(KIWI_CONF_NAME), "w") as file:
+        file.write(Config.parse_obj(kiwi_dict).kiwi_yml)

kiwi_scp/config.py

@@ -142,7 +142,7 @@ class Config(BaseModel):
     @classmethod
     @functools.lru_cache(maxsize=5)
     def from_instance(cls, instance: Path):
-        """parses an actual kiwi.yml from disk"""
+        """parses an actual kiwi.yml from disk (cached)"""
 
         try:
             with open(instance.joinpath(KIWI_CONF_NAME)) as kc:
@@ -151,7 +151,14 @@ class Config(BaseModel):
 
         except FileNotFoundError:
             # return the defaults if no kiwi.yml found
-            return cls()
+            return cls.from_default()
+
+    @classmethod
+    @functools.lru_cache(maxsize=1)
+    def from_default(cls):
+        """returns the default config (cached)"""
+
+        return cls()
 
     @property
     def kiwi_dict(self) -> Dict[str, Any]:
@@ -352,4 +359,3 @@ class Config(BaseModel):
         else:
             # undefined format
             raise ValueError("Invalid Network Format")
-

kiwi_scp/misc.py

@@ -1,3 +1,26 @@
+from typing import Any, Type
+
+import click
+
+
+def user_query(description: str, default: Any, cast_to: Type[Any] = str):
+    # prompt user as per argument
+    while True:
+        try:
+            str_value = input(f"Enter {description} [{default}] ").strip()
+            if str_value:
+                return cast_to(str_value)
+            else:
+                return default
+
+        except EOFError:
+            click.echo("Input aborted.")
+            return default
+
+        except Exception as e:
+            click.echo(f"Invalid input: {e}")
+
+
 def _surround(string, bang):
     midlane = f"{bang * 3} {string} {bang * 3}"
     sidelane = bang * len(midlane)

kiwi_scp/subcommands/init.py

@@ -1,63 +0,0 @@
-# system
-import logging
-import os
-
-# local
-from .._constants import KIWI_CONF_NAME
-from ..config import DefaultConfig, LoadedConfig
-from ..subcommand import SubCommand
-
-
-class InitCommand(SubCommand):
-    """kiwi init"""
-
-    def __init__(self):
-        super().__init__(
-            'init',
-            action=f"Initializing '{KIWI_CONF_NAME}' in",
-            description="Initialize or reconfigure kiwi-scp instance"
-        )
-
-        # -f switch: Initialize with default config
-        self._sub_parser.add_argument(
-            '-f', '--force',
-            action='store_true',
-            help=f"use default values even if {KIWI_CONF_NAME} is present"
-        )
-
-        # -s switch: Show current config instead
-        self._sub_parser.add_argument(
-            '-s', '--show',
-            action='store_true',
-            help=f"show effective {KIWI_CONF_NAME} contents instead"
-        )
-
-    def _run_instance(self, runner, args):
-        config = LoadedConfig.get()
-
-        # check show switch
-        if args.show:
-            print(config)
-            return True
-
-        # check force switch
-        if args.force and os.path.isfile(KIWI_CONF_NAME):
-            logging.warning(f"Overwriting existing '{KIWI_CONF_NAME}'!")
-            config = DefaultConfig.get()
-
-        # version
-        config.user_query('version')
-
-        # runtime
-        config.user_query('runtime:storage')
-
-        # markers
-        config.user_query('markers:project')
-        config.user_query('markers:disabled')
-
-        # network
-        config.user_query('network:name')
-        config.user_query('network:cidr')
-
-        config.save()
-        return True

tests/test_config.py

@@ -19,6 +19,8 @@ def test_default():
     c = Config()
     version = toml.load("./pyproject.toml")["tool"]["poetry"]["version"]
 
+    assert c == Config.from_default()
+
     assert c.version == version
     assert len(c.shells) == 1
     assert c.shells[0] == Path("/bin/bash")