Merge pull request 'Feat(machine/templates): simplify template args for machines command' (#3937) from hsjobeki/template-ux into main

Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/3937

lib/build-clan/interface.nix

@@ -213,6 +213,8 @@ in
 
           secrets = lib.mkOption { type = lib.types.raw; };
 
+          templates = lib.mkOption { type = lib.types.raw; };
+
           machines = lib.mkOption { type = lib.types.raw; };
         };
       };

lib/build-clan/module.nix

@@ -250,6 +250,9 @@ in
       # We should have only clan.modules. (consistent with clan.templates)
       inherit (clan-core) clanModules;
 
+      # Statically export the predefined clan modules
+      templates = clan-core.clan.templates;
+
       secrets = config.secrets;
 
       # machine specifics

pkgs/clan-cli/clan_cli/machines/create.py

@@ -2,7 +2,6 @@ import argparse
 import logging
 import re
 from dataclasses import dataclass
-from pathlib import Path
 
 from clan_lib.api import API
 from clan_lib.dirs import get_clan_flake_toplevel_or_env
@@ -13,15 +12,9 @@ from clan_lib.nix_models.clan import InventoryMachine
 from clan_lib.nix_models.clan import InventoryMachineDeploy as MachineDeploy
 from clan_lib.persist.inventory_store import InventoryStore
 from clan_lib.persist.util import set_value_by_path
-from clan_lib.templates import (
+from clan_lib.templates.handler import machine_template
-    InputPrio,
-    TemplateName,
-    get_template,
-)
-from clan_lib.templates.filesystem import copy_from_nixstore
 
 from clan_cli.completions import add_dynamic_completer, complete_tags
-from clan_cli.machines.list import list_full_machines
 
 log = logging.getLogger(__name__)
 
@@ -30,14 +23,14 @@ log = logging.getLogger(__name__)
 class CreateOptions:
     clan_dir: Flake
     machine: InventoryMachine
+    template: str = "new-machine"
     target_host: str | None = None
-    input_prio: InputPrio | None = None
-    template_name: str | None = None
 
 
 @API.register
 def create_machine(
-    opts: CreateOptions, commit: bool = True, _persist: bool = True
+    opts: CreateOptions,
+    commit: bool = True,
 ) -> None:
     """
     Create a new machine in the clan directory.
@@ -54,42 +47,12 @@ def create_machine(
         description = "Import machine only works on local clans"
         raise ClanError(msg, description=description)
 
-    if not opts.template_name:
-        opts.template_name = "new-machine"
-
     clan_dir = opts.clan_dir.path
 
-    # TODO(@Qubasa): make this a proper template handler
-    # i.e. with_template (use context manager)
-    # And move the checks and template handling into the template handler
-    template = get_template(
-        TemplateName(opts.template_name),
-        "machine",
-        input_prio=opts.input_prio,
-        clan_dir=opts.clan_dir,
-    )
-    log.info(f"Found template '{template.name}' in '{template.input_variant}'")
-
     machine_name = opts.machine.get("name")
-    if opts.template_name in list_full_machines(
+    if not machine_name:
-        Flake(str(clan_dir))
+        msg = "Machine name is required"
-    ) and not opts.machine.get("name"):
+        raise ClanError(msg, location="Create Machine")
-        msg = f"{opts.template_name} is already defined in {clan_dir}"
-        raise ClanError(msg)
-
-    machine_name = machine_name if machine_name else opts.template_name
-    src = Path(template.src["path"])
-
-    if not src.exists():
-        msg = f"Template {template} does not exist"
-        raise ClanError(msg)
-    if not src.is_dir():
-        msg = f"Template {template} is not a directory"
-        raise ClanError(msg)
-
-    dst = clan_dir / "machines"
-    dst.mkdir(exist_ok=True)
-    dst /= machine_name
 
     # TODO: Move this into nix code
     hostname_regex = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$"
@@ -97,21 +60,12 @@ def create_machine(
         msg = "Machine name must be a valid hostname"
         raise ClanError(msg, location="Create Machine")
 
-    if dst.exists():
+    with machine_template(
-        msg = f"Machine {machine_name} already exists in {clan_dir}"
+        flake=opts.clan_dir,
-        description = "Please remove the existing machine folder"
+        template_ident=opts.template,
-        raise ClanError(msg, description=description)
+        dst_machine_name=machine_name,
-
+    ) as _machine_dir:
-    # TODO(@Qubasa): move this into the template handler
+        # Write to the inventory if persist is true
-    if not (src / "configuration.nix").exists():
-        msg = f"Template machine '{opts.template_name}' does not contain a configuration.nix"
-        description = "Template machine must contain a configuration.nix"
-        raise ClanError(msg, description=description)
-
-    # TODO(@Qubasa): move this into the template handler
-    copy_from_nixstore(src, dst)
-
-    if _persist:
         target_host = opts.target_host
         new_machine = opts.machine
         new_machine["deploy"] = {"targetHost": target_host}  # type: ignore
@@ -133,16 +87,14 @@ def create_machine(
         )
         inventory_store.write(inventory, message=f"machine '{machine_name}'")
 
+        if commit:
+            commit_file(
+                clan_dir / "machines" / machine_name,
+                repo_dir=clan_dir,
+                commit_message=f"Add machine {machine_name}",
+            )
+    # Invalidate the cache since this modified the flake
     opts.clan_dir.invalidate_cache()
-    # Commit at the end in that order to avoid committing halve-baked machines
-    # TODO: automatic rollbacks if something goes wrong
-
-    if commit:
-        commit_file(
-            clan_dir / "machines" / machine_name,
-            repo_dir=clan_dir,
-            commit_message=f"Add machine {machine_name}",
-        )
 
 
 def create_command(args: argparse.Namespace) -> None:
@@ -159,25 +111,15 @@ def create_command(args: argparse.Namespace) -> None:
         )
         raise ClanError(msg, description=description)
 
-    if len(args.input) == 0:
-        args.input = ["clan", "clan-core"]
-
-    if args.no_self:
-        input_prio = InputPrio.try_inputs(tuple(args.input))
-    else:
-        input_prio = InputPrio.try_self_then_inputs(tuple(args.input))
-
     machine = InventoryMachine(
         name=args.machine_name,
         tags=args.tags,
         deploy=MachineDeploy(targetHost=args.target_host),
     )
     opts = CreateOptions(
-        input_prio=input_prio,
         clan_dir=clan_dir,
         machine=machine,
-        template_name=args.template_name,
+        template=args.template,
-        target_host=args.target_host,
     )
     create_machine(opts)
 
@@ -196,29 +138,17 @@ def register_create_parser(parser: argparse.ArgumentParser) -> None:
         help="Tags to associate with the machine. Can be used to assign multiple machines to services.",
     )
     add_dynamic_completer(tag_parser, complete_tags)
-    parser.add_argument(
-        "--template-name",
-        type=str,
-        help="The name of the template machine to import",
-    )
     parser.add_argument(
         "--target-host",
         type=str,
         help="Address of the machine to install and update, in the format of user@host:1234",
     )
     parser.add_argument(
-        "--input",
+        "-t",
+        "--template",
         type=str,
-        help="""Flake input name to use as template source
+        help="""Reference to the template to use for the machine. default="new-machine". In the format '<flake_ref>#template_name' Where <flake_ref> is a flake reference (e.g. github:org/repo) or a local path (e.g. '.' ).
-        can be specified multiple times, inputs are tried in order of definition
+        Omitting '<flake_ref>#' will use the builtin templates (e.g. just 'new-machine' from clan-core ).
-        Example: --input clan --input clan-core
         """,
-        action="append",
+        default="new-machine",
-        default=[],
-    )
-    parser.add_argument(
-        "--no-self",
-        help="Do not look into own flake for templates",
-        action="store_true",
-        default=False,
     )

pkgs/clan-cli/clan_cli/machines/morph.py

@@ -8,9 +8,10 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 
 from clan_lib.cmd import Log, RunOpts, run
-from clan_lib.dirs import get_clan_flake_toplevel_or_env
+from clan_lib.dirs import get_clan_flake_toplevel_or_env, specific_machine_dir
 from clan_lib.errors import ClanError
 from clan_lib.flake import Flake
+from clan_lib.machines.actions import list_machines
 from clan_lib.machines.machines import Machine
 from clan_lib.nix import nix_build, nix_command
 from clan_lib.nix_models.clan import InventoryMachine
@@ -41,7 +42,7 @@ def random_hostname() -> str:
 
 
 def morph_machine(
-    flake: Flake, template_name: str, ask_confirmation: bool, name: str | None = None
+    flake: Flake, template: str, ask_confirmation: bool, name: str | None = None
 ) -> None:
     cmd = nix_command(
         [
@@ -69,12 +70,13 @@ def morph_machine(
         if name is None:
             name = random_hostname()
 
-        create_opts = CreateOptions(
+        if name not in list_machines(flake):
-            template_name=template_name,
+            create_opts = CreateOptions(
-            machine=InventoryMachine(name=name),
+                template=template,
-            clan_dir=Flake(str(flakedir)),
+                machine=InventoryMachine(name=name),
-        )
+                clan_dir=Flake(str(flakedir)),
-        create_machine(create_opts, commit=False, _persist=False)
+            )
+            create_machine(create_opts, commit=False)
 
         machine = Machine(name=name, flake=Flake(str(flakedir)))
 
@@ -89,9 +91,9 @@ def morph_machine(
         # facter_json = run(["nixos-facter"]).stdout
         # run(["cp", "facter.json", f"{flakedir}/machines/{name}/facter.json"]).stdout
 
-        Path(f"{flakedir}/machines/{name}/facter.json").write_text(
+        machine_dir = specific_machine_dir(machine)
-            '{"system": "x86_64-linux"}'
+        machine_dir.mkdir(parents=True, exist_ok=True)
-        )
+        Path(f"{machine_dir}/facter.json").write_text('{"system": "x86_64-linux"}')
         result_path = run(
             nix_build(
                 [f"{flakedir}#nixosConfigurations.{name}.config.system.build.toplevel"]
@@ -149,7 +151,7 @@ def morph_command(args: argparse.Namespace) -> None:
 
     morph_machine(
         flake=Flake(str(args.flake)),
-        template_name=args.template_name,
+        template=args.template,
         ask_confirmation=args.confirm_firing,
         name=args.name,
     )
@@ -159,7 +161,7 @@ def register_morph_parser(parser: argparse.ArgumentParser) -> None:
     parser.set_defaults(func=morph_command)
 
     parser.add_argument(
-        "template_name",
+        "template",
         default="new-machine",
         type=str,
         help="The name of the template to use",

pkgs/clan-cli/clan_lib/flake/flake.py

@@ -598,6 +598,9 @@ class Flake:
         assert isinstance(self._is_local, bool)
         return self._is_local
 
+    def get_input_names(self) -> list[str]:
+        return self.select("inputs", apply="builtins.attrNames")
+
     @property
     def path(self) -> Path:
         if self._path is None:
@@ -680,6 +683,7 @@ class Flake:
         self,
         selectors: list[str],
         nix_options: list[str] | None = None,
+        apply: str = "v: v",
     ) -> None:
         """
         Retrieves specific attributes from a Nix flake using the provided selectors.
@@ -754,7 +758,7 @@ class Flake:
                 result = builtins.toJSON [
                     {" ".join(
                         [
-                            f"(selectLib.applySelectors (builtins.fromJSON ''{attr}'') flake)"
+                            f"(({apply}) (selectLib.applySelectors (builtins.fromJSON ''{attr}'') flake))"
                             for attr in str_selectors
                         ]
                     )}
@@ -823,6 +827,7 @@ class Flake:
         self,
         selector: str,
         nix_options: list[str] | None = None,
+        apply: str = "v: v",
     ) -> Any:
         """
         Selects a value from the cache based on the provided selector string.
@@ -839,6 +844,6 @@ class Flake:
 
         if not self._cache.is_cached(selector):
             log.debug(f"Cache miss for {selector}")
-            self.get_from_nix([selector], nix_options)
+            self.get_from_nix([selector], nix_options, apply=apply)
         value = self._cache.select(selector)
         return value

pkgs/clan-cli/clan_lib/templates/handler.py

@@ -0,0 +1,108 @@
+import logging
+import shutil
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+
+from clan_lib.dirs import specific_machine_dir
+from clan_lib.errors import ClanError
+from clan_lib.flake import Flake
+from clan_lib.machines.actions import MachineID, list_machines
+from clan_lib.templates.filesystem import copy_from_nixstore, realize_nix_path
+from clan_lib.templates.template_url import transform_url
+
+log = logging.getLogger(__name__)
+
+
+@contextmanager
+def machine_template(
+    flake: Flake, template_ident: str, dst_machine_name: str
+) -> Iterator[Path]:
+    """
+    Create a machine from a template.
+    This function will copy the template files to the machine specific directory of the specified flake.
+
+    :param flake: The flake to create the machine in.
+    :param template_ident: The identifier of the template to use. Example ".#template_name"
+    :param dst_machine_name: The name of the machine to create.
+
+    Example usage:
+
+    >>> with machine_template(
+    ...     Flake("/home/johannes/git/clan-core"), ".#new-machine", "my-machine"
+    ... ) as machine_path:
+    ...     # Use `machine_path` here if you want to access the created machine directory
+
+    ... The machine directory is removed if the context raised any errors.
+    ... Only if the context is exited without errors, the machine directory is kept.
+    """
+
+    # Check for duplicates
+    if dst_machine_name in list_machines(flake):
+        msg = f"Machine '{dst_machine_name}' already exists"
+        raise ClanError(
+            msg,
+            description="Please remove the existing machine or choose a different name",
+        )
+
+    # Get the clan template from the specifier
+    [flake_ref, template_selector] = transform_url(
+        "machine", template_ident, flake=flake
+    )
+    # For pretty error messages
+    printable_template_ref = f"{flake_ref}#{template_selector}"
+
+    template_flake = Flake(flake_ref)
+    try:
+        template = template_flake.select(template_selector)
+    except ClanError as e:
+        msg = f"Failed to select template '{template_ident}' from flake '{flake_ref}' (via attribute path: {printable_template_ref})"
+        raise ClanError(msg) from e
+
+    src = template.get("path")
+    if not src:
+        msg = f"Malformed template: {printable_template_ref} does not have a 'path' attribute"
+        raise ClanError(msg)
+
+    src_path = Path(src).resolve()
+
+    realize_nix_path(template_flake, str(src_path))
+
+    if not src_path.exists():
+        msg = f"Template {printable_template_ref} does not exist at {src_path}"
+        raise ClanError(msg)
+
+    if not src_path.is_dir():
+        msg = f"Template {printable_template_ref} is not a directory at {src_path}"
+        raise ClanError(msg)
+
+    # TODO: Do we really need to check for a specific file in the template?
+    if not (src_path / "configuration.nix").exists():
+        msg = f"Template {printable_template_ref} does not contain a configuration.nix"
+        raise ClanError(
+            msg,
+            description="Template machine must contain a configuration.nix",
+        )
+
+    tmp_machine = MachineID(flake=flake, name=dst_machine_name)
+
+    dst_machine_dir = specific_machine_dir(tmp_machine)
+
+    dst_machine_dir.parent.mkdir(exist_ok=True, parents=True)
+
+    copy_from_nixstore(src_path, dst_machine_dir)
+
+    try:
+        yield dst_machine_dir
+    except Exception as e:
+        log.error(f"An error occurred inside the 'machine_template' context: {e}")
+
+        # Ensure that the directory is removed to avoid half-created machines
+        # Everything in the with block is considered part of the context
+        # So if the context raises an error, we clean up the machine directory
+        log.info(f"Removing left-over machine directory: {dst_machine_dir}")
+        shutil.rmtree(dst_machine_dir, ignore_errors=True)
+        raise
+    finally:
+        # If no error occurred, the machine directory is kept
+        pass

pkgs/clan-cli/clan_lib/templates/template_url.py

@@ -0,0 +1,108 @@
+import logging
+from pathlib import Path
+from typing import Protocol
+
+from clan_lib.errors import ClanError
+
+log = logging.getLogger(__name__)
+
+
+class Flake(Protocol):
+    """
+    Protocol for a local flake, which has a path attribute.
+    Pass clan_lib.flake.Flake or any other object that implements this protocol.
+    """
+
+    @property
+    def path(self) -> Path: ...
+
+    def get_input_names(self) -> list[str]: ...
+
+
+def transform_url(template_type: str, identifier: str, flake: Flake) -> tuple[str, str]:
+    """
+    Transform a template flake ref by injecting the context (clan|machine|disko) into the url.
+    We do this for shorthand notation of URLs.
+    If the attribute selector path is explicitly selecting an attribute, we don't transform it.
+
+    :param template_type: The type of the template (clan, machine, disko)
+    :param identifier: The identifier of the template, which can be a flake reference with a fragment.
+    :param local_path: The local flake path, which is used to resolve to a local flake reference, i.e. ".#" shorthand.
+
+    Examples:
+
+    1. injects "machine" as context
+    clan machines create --template .#new-machine
+    or
+    clan machines create --template #new-machine
+    -> .#clan.templates.machine.new-machine
+
+    2. injects "clan" as context
+    clan create --template .#default
+    -> .#clan.templates.clan.default
+
+    3. Dont transform explicit paths (e.g. when more than one attribute selector is present)
+    clan machines create --template .#clan.templates.machine.new-machine
+    -> .#clan.templates.machine.new-machine
+
+    clan machines create --template .#"new.machine"
+    -> .#clan.templates.machine."new.machine"
+
+    4. Builtin templates
+    clan machines create --template new.machine
+    -> clanInternals.templates.machine."new.machine"
+
+    5. Remote templates
+    clan machines create --template github:/org/repo#new.machine
+    -> clanInternals.templates.machine."new.machine"
+
+    6. Templates locked via inputs:
+    clan machines create --template clan-core#new-machine
+    path: clan-core matches one of the input attributes.
+    -> <local_flake>#inputs.clan-core.clan.templates.machine."new-machine"
+
+    As of URL specification (RFC 3986).
+    scheme:[//[user:password@]host[:port]][/path][?query][#fragment]
+
+    We can safely split the URL into a front part and the fragment
+    We can then analyze the fragment and inject the context into the path.
+
+    Of there is no fragment and no URL its a builtin template path.
+    new-machine -> #clanInternals.templates.machine."new-machine"
+
+    """
+    if identifier.count("#") > 1:
+        msg = "Invalid template identifier: More than one '#' found. Please use a single '#'"
+        raise ClanError(msg)
+
+    [flake_ref, selector] = (
+        identifier.split("#", 1) if "#" in identifier else ["", identifier]
+    )
+    # Substitute the flake reference with the local flake path if it is empty or just a dot.
+    # This is required if the command will be executed from a different place, than the local flake root.
+    if not flake_ref or flake_ref == ".":
+        flake_ref = str(flake.path)
+
+    if "#" not in identifier:
+        # No fragment, so we assume its a builtin template
+        return (flake_ref, f'clanInternals.templates.{template_type}."{selector}"')
+
+    input_prefix = ""
+    if flake_ref in flake.get_input_names():
+        # Interpret the flake reference as an input of the local flake.
+        input_prefix = f"inputs.{flake_ref}."
+        flake_ref = str(flake.path)
+
+    # TODO: implement support for quotes in the tail "a.b".c
+    # If the tail contains a dot, or is quoted we assume its a path and don't transform it.
+    if '"' in selector or "'" in selector:
+        log.warning(
+            "Quotes in template paths are not yet supported. Please use unquoted paths."
+        )
+        return (flake_ref, input_prefix + selector)
+
+    if "." in selector:
+        return (flake_ref, input_prefix + selector)
+
+    # Tail doesn't contain a dot at this point, so we can inject the context.
+    return (flake_ref, input_prefix + f'clan.templates.{template_type}."{selector}"')

pkgs/clan-cli/clan_lib/templates/template_url_test.py

@@ -0,0 +1,151 @@
+from pathlib import Path
+
+import pytest
+
+from clan_lib.errors import ClanError
+from clan_lib.templates.template_url import transform_url
+
+template_type = "machine"
+
+
+class DummyFlake:
+    def __init__(self, path: str) -> None:
+        self.path: Path = Path(path)
+
+    def get_input_names(self) -> list[str]:
+        return ["locked-input"]
+
+
+local_path = DummyFlake(".")
+
+
+def test_transform_url_self_explizit_dot() -> None:
+    user_input = ".#new-machine"
+    expected_selector = 'clan.templates.machine."new-machine"'
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+def test_transform_url_self_no_dot() -> None:
+    user_input = "#new-machine"
+    expected_selector = 'clan.templates.machine."new-machine"'
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+def test_transform_url_builtin_template() -> None:
+    user_input = "new-machine"
+    expected_selector = 'clanInternals.templates.machine."new-machine"'
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+def test_transform_url_remote_template() -> None:
+    user_input = "github:/org/repo#new-machine"
+    expected_selector = 'clan.templates.machine."new-machine"'
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+
+    assert flake_ref == "github:/org/repo"
+    assert selector == expected_selector
+
+
+def test_transform_url_explicit_path() -> None:
+    user_input = ".#clan.templates.machine.new-machine"
+    expected_selector = "clan.templates.machine.new-machine"
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+# Currently quoted selectors are treated as explicit paths.
+def test_transform_url_quoted_selector() -> None:
+    user_input = '.#"new.machine"'
+    expected_selector = '"new.machine"'
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+def test_single_quote_selector() -> None:
+    user_input = ".#'new.machine'"
+    expected_selector = "'new.machine'"
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+def test_custom_template_path() -> None:
+    user_input = "github:/org/repo#my.templates.custom.machine"
+    expected_selector = "my.templates.custom.machine"
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == "github:/org/repo"
+    assert selector == expected_selector
+
+
+def test_full_url_query_and_fragment() -> None:
+    user_input = "github:/org/repo?query=param#my.templates.custom.machine"
+    expected_flake_ref = "github:/org/repo?query=param"
+    expected_selector = "my.templates.custom.machine"
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == expected_flake_ref
+    assert selector == expected_selector
+
+
+def test_custom_template_type() -> None:
+    user_input = "#my.templates.custom.machine"
+    expected_selector = "my.templates.custom.machine"
+
+    flake_ref, selector = transform_url("custom", user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+def test_malformed_identifier() -> None:
+    user_input = "github:/org/repo#my.templates.custom.machine#extra"
+    with pytest.raises(ClanError) as exc_info:
+        _flake_ref, _selector = transform_url(
+            template_type, user_input, flake=local_path
+        )
+
+    assert isinstance(exc_info.value, ClanError)
+    assert (
+        str(exc_info.value)
+        == "Invalid template identifier: More than one '#' found. Please use a single '#'"
+    )
+
+
+def test_locked_input_template() -> None:
+    user_input = "locked-input#new-machine"
+    expected_selector = 'inputs.locked-input.clan.templates.machine."new-machine"'
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert flake_ref == str(local_path.path)
+    assert selector == expected_selector
+
+
+def test_locked_input_template_no_quotes() -> None:
+    user_input = 'locked-input#"new.machine"'
+    expected_selector = 'inputs.locked-input."new.machine"'
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert selector == expected_selector
+    assert flake_ref == str(local_path.path)
+
+
+def test_locked_input_template_no_dot() -> None:
+    user_input = "locked-input#new.machine"
+    expected_selector = "inputs.locked-input.new.machine"
+
+    flake_ref, selector = transform_url(template_type, user_input, flake=local_path)
+    assert selector == expected_selector
+    assert flake_ref == str(local_path.path)