Merge pull request 'clan-vm-manager: Fix regression' (#1944) from Qubasa/clan-core:Qubasa-new_clan_vm_manager into main

pkgs/clan-app/clan_app/__init__.py

@@ -1,6 +1,10 @@
 import logging
 import sys
 
+# Remove working directory from sys.path
+if "" in sys.path:
+    sys.path.remove("")
+
 from clan_cli.profiler import profile
 
 from clan_app.app import MainApplication

pkgs/clan-cli/clan_cli/__init__.py

@@ -44,14 +44,14 @@ except ImportError:
 def flake_path(arg: str) -> FlakeId:
     flake_dir = Path(arg).resolve()
     if flake_dir.exists() and flake_dir.is_dir():
-        return FlakeId(flake_dir)
+        return FlakeId(str(flake_dir))
     return FlakeId(arg)
 
 
 def default_flake() -> FlakeId | None:
     val = get_clan_flake_toplevel_or_env()
     if val:
-        return FlakeId(val)
+        return FlakeId(str(val))
     return None
 
 

pkgs/clan-cli/clan_cli/clan/inspect.py

@@ -28,6 +28,8 @@ class FlakeConfig:
     def __post_init__(self) -> None:
         if isinstance(self.vm, dict):
             self.vm = VmConfig(**self.vm)
+        if isinstance(self.flake_url, dict):
+            self.flake_url = FlakeId(**self.flake_url)
 
 
 def run_cmd(cmd: list[str]) -> str:
@@ -46,7 +48,7 @@ def inspect_flake(flake_url: str | Path, machine_name: str) -> FlakeConfig:
             f"Machine {machine_name} not found in {flake_url}. Available machines: {', '.join(machines)}"
         )
 
-    machine = Machine(machine_name, FlakeId(flake_url))
+    machine = Machine(machine_name, FlakeId(str(flake_url)))
     vm = inspect_vm(machine)
 
     # Make symlink to gcroots from vm.machine_icon
@@ -89,7 +91,7 @@ def inspect_flake(flake_url: str | Path, machine_name: str) -> FlakeConfig:
     meta = nix_metadata(flake_url)
     return FlakeConfig(
         vm=vm,
-        flake_url=FlakeId(flake_url),
+        flake_url=FlakeId(str(flake_url)),
         clan_name=clan_name,
         flake_attr=machine_name,
         nar_hash=meta["locked"]["narHash"],
@@ -109,7 +111,7 @@ class InspectOptions:
 def inspect_command(args: argparse.Namespace) -> None:
     inspect_options = InspectOptions(
         machine=args.machine,
-        flake=args.flake or FlakeId(Path.cwd()),
+        flake=args.flake or FlakeId(str(Path.cwd())),
     )
     res = inspect_flake(
         flake_url=str(inspect_options.flake), machine_name=inspect_options.machine

pkgs/clan-cli/clan_cli/clan_uri.py

@@ -9,16 +9,19 @@ from .errors import ClanError
 
 @dataclass
 class FlakeId:
-    loc: str | Path
+    loc: str
 
     def __post_init__(self) -> None:
         assert isinstance(
-            self.loc, str | Path
+            self.loc, str
         ), f"Flake {self.loc} has an invalid format: {type(self.loc)}"
 
     def __str__(self) -> str:
         return str(self.loc)
 
+    def __hash__(self) -> int:
+        return hash(str(self.loc))
+
     @property
     def path(self) -> Path:
         assert self.is_local(), f"Flake {self.loc} is not a local path"
@@ -87,20 +90,11 @@ class ClanURI:
             self.machine_name = components.fragment
 
     def _parse_url(self, comps: urllib.parse.ParseResult) -> FlakeId:
-        comb = (
-            comps.scheme,
-            comps.netloc,
-            comps.path,
-            comps.params,
-            comps.query,
-            comps.fragment,
-        )
-        match comb:
-            case ("file", "", path, "", "", _) | ("", "", path, "", "", _):  # type: ignore
-                flake_id = FlakeId(Path(path).expanduser().resolve())
-            case _:
-                flake_id = FlakeId(comps.geturl())
-
+        if comps.scheme == "" or "file" in comps.scheme:
+            res_p = Path(comps.path).expanduser().resolve()
+            flake_id = FlakeId(str(res_p))
+        else:
+            flake_id = FlakeId(comps.geturl())
         return flake_id
 
     def get_url(self) -> str:

pkgs/clan-cli/clan_cli/history/add.py

@@ -62,7 +62,6 @@ def list_history() -> list[HistoryEntry]:
 
 def new_history_entry(url: str, machine: str) -> HistoryEntry:
     flake = inspect_flake(url, machine)
-    flake.flake_url = flake.flake_url
     return HistoryEntry(
         flake=flake,
         last_used=datetime.datetime.now().isoformat(),

pkgs/clan-cli/clan_cli/vms/inspect.py

@@ -32,6 +32,8 @@ class VmConfig:
             self.flake_url = FlakeId(self.flake_url)
         if isinstance(self.waypipe, dict):
             self.waypipe = WaypipeConfig(**self.waypipe)
+        if isinstance(self.flake_url, dict):
+            self.flake_url = FlakeId(**self.flake_url)
 
 
 def inspect_vm(machine: Machine) -> VmConfig:
@@ -48,7 +50,7 @@ class InspectOptions:
 def inspect_command(args: argparse.Namespace) -> None:
     inspect_options = InspectOptions(
         machine=args.machine,
-        flake=args.flake or FlakeId(Path.cwd()),
+        flake=args.flake or FlakeId(str(Path.cwd())),
     )
 
     machine = Machine(inspect_options.machine, inspect_options.flake)

pkgs/clan-cli/tests/test_deserializers.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Literal
+from typing import Any, Literal
 
 import pytest
 
@@ -128,7 +128,7 @@ def test_simple_field_missing() -> None:
     class Person:
         name: str
 
-    person_dict = {}
+    person_dict: Any = {}
 
     with pytest.raises(ClanError):
         from_dict(Person, person_dict)
@@ -151,7 +151,7 @@ def test_nullable_non_exist() -> None:
     class Person:
         name: None
 
-    person_dict = {}
+    person_dict: Any = {}
 
     with pytest.raises(ClanError):
         from_dict(Person, person_dict)

pkgs/clan-cli/tests/test_history_cli.py

@@ -27,7 +27,7 @@ def test_history_add(
     history_file = user_history_file()
     assert history_file.exists()
     history = [HistoryEntry(**entry) for entry in json.loads(open(history_file).read())]
-    assert str(history[0].flake.flake_url["loc"]) == str(test_flake_with_core.path)
+    assert str(history[0].flake.flake_url) == str(test_flake_with_core.path)
 
 
 @pytest.mark.impure

pkgs/clan-cli/tests/test_modules.py

@@ -61,7 +61,7 @@ def test_add_module_to_inventory(
         ]
     )
     create_machine(
-        FlakeId(base_path),
+        FlakeId(str(base_path)),
         Machine(
             name="machine1", tags=[], system="x86_64-linux", deploy=MachineDeploy()
         ),

pkgs/clan-cli/tests/test_secrets_generate.py

@@ -48,7 +48,9 @@ def test_generate_secret(
     )
     cmd = ["facts", "generate", "--flake", str(test_flake_with_core.path), "vm1"]
     cli.run(cmd)
-    store1 = SecretStore(Machine(name="vm1", flake=FlakeId(test_flake_with_core.path)))
+    store1 = SecretStore(
+        Machine(name="vm1", flake=FlakeId(str(test_flake_with_core.path)))
+    )
 
     assert store1.exists("", "age.key")
     assert store1.exists("", "zerotier-identity-secret")
@@ -76,7 +78,9 @@ def test_generate_secret(
         secrets_folder / "vm1-zerotier-identity-secret" / "machines" / "vm1"
     ).exists()
 
-    store2 = SecretStore(Machine(name="vm2", flake=FlakeId(test_flake_with_core.path)))
+    store2 = SecretStore(
+        Machine(name="vm2", flake=FlakeId(str(test_flake_with_core.path)))
+    )
 
     assert store2.exists("", "password")
     assert store2.exists("", "password-hash")

pkgs/clan-cli/tests/test_secrets_password_store.py

@@ -49,7 +49,7 @@ def test_upload_secret(
     cli.run(["facts", "generate", "vm1"])
 
     store = SecretStore(
-        Machine(name="vm1", flake=FlakeId(test_flake_with_core_and_pass.path))
+        Machine(name="vm1", flake=FlakeId(str(test_flake_with_core_and_pass.path)))
     )
 
     network_id = machine_get_fact(

pkgs/clan-cli/tests/test_vars.py

@@ -4,17 +4,17 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 
 import pytest
+from age_keys import SopsSetup
+from fixtures_flakes import generate_flake
+from helpers import cli
+from helpers.nixos_config import nested_dict
+from root import CLAN_CORE
 
 from clan_cli.clan_uri import FlakeId
 from clan_cli.machines.machines import Machine
 from clan_cli.nix import nix_shell
 from clan_cli.vars.public_modules import in_repo
 from clan_cli.vars.secret_modules import password_store, sops
-from tests.age_keys import SopsSetup
-from tests.fixtures_flakes import generate_flake
-from tests.helpers import cli
-from tests.helpers.nixos_config import nested_dict
-from tests.root import CLAN_CORE
 
 
 def test_get_subgraph() -> None:

pkgs/clan-cli/tests/test_vars_deployment.py

@@ -1,13 +1,12 @@
 from pathlib import Path
 
 import pytest
-
-from tests.age_keys import SopsSetup
-from tests.fixtures_flakes import generate_flake
-from tests.helpers import cli
-from tests.helpers.nixos_config import nested_dict
-from tests.helpers.vms import qga_connect, run_vm_in_thread, wait_vm_down
-from tests.root import CLAN_CORE
+from age_keys import SopsSetup
+from fixtures_flakes import generate_flake
+from helpers import cli
+from helpers.nixos_config import nested_dict
+from helpers.vms import qga_connect, run_vm_in_thread, wait_vm_down
+from root import CLAN_CORE
 
 
 @pytest.mark.impure

pkgs/clan-cli/tests/test_vms_cli.py

@@ -3,16 +3,15 @@ from pathlib import Path
 from typing import TYPE_CHECKING
 
 import pytest
+from fixtures_flakes import FlakeForTest, generate_flake
+from helpers import cli
+from helpers.nixos_config import nested_dict
+from helpers.vms import qga_connect, qmp_connect, run_vm_in_thread, wait_vm_down
+from root import CLAN_CORE
 from stdout import CaptureOutput
 
-from tests.fixtures_flakes import FlakeForTest, generate_flake
-from tests.helpers import cli
-from tests.helpers.nixos_config import nested_dict
-from tests.helpers.vms import qga_connect, qmp_connect, run_vm_in_thread, wait_vm_down
-from tests.root import CLAN_CORE
-
 if TYPE_CHECKING:
-    from tests.age_keys import KeyPair
+    from age_keys import KeyPair
 
 no_kvm = not os.path.exists("/dev/kvm")
 

pkgs/clan-vm-manager/.vscode/lhebendanz.weaudit

@@ -2,30 +2,7 @@
   "clientRemote": "",
   "gitRemote": "",
   "gitSha": "",
-  "treeEntries": [
-    {
-      "label": "source of problem",
-      "entryType": 0,
-      "author": "lhebendanz",
-      "locations": [
-        {
-          "path": "../clan-cli/clan_cli/history/add.py",
-          "startLine": 45,
-          "endLine": 59,
-          "label": "",
-          "description": ""
-        }
-      ],
-      "details": {
-        "severity": "",
-        "difficulty": "",
-        "type": "",
-        "description": "",
-        "exploit": "",
-        "recommendation": "Short term, \nLong term, \n"
-      }
-    }
-  ],
+  "treeEntries": [],
   "auditedFiles": [],
   "resolvedEntries": []
 }

pkgs/clan-vm-manager/clan_vm_manager/components/vmobj.py

@@ -13,7 +13,7 @@ from typing import IO, ClassVar
 
 import gi
 from clan_cli import vms
-from clan_cli.clan_uri import ClanURI
+from clan_cli.clan_uri import ClanURI, FlakeId
 from clan_cli.dirs import vm_state_dir
 from clan_cli.history.add import HistoryEntry
 from clan_cli.machines.machines import Machine
@@ -51,7 +51,7 @@ class VMObject(GObject.Object):
 
         # Store the data from the history entry
         self.data: HistoryEntry = data
-
+        assert isinstance(self.data.flake.vm.flake_url, FlakeId)
         self.build_log_cb = build_log_cb
 
         # Create a process object to store the VM process
@@ -99,6 +99,7 @@ class VMObject(GObject.Object):
         return GLib.SOURCE_REMOVE
 
     def update(self, data: HistoryEntry) -> None:
+        assert isinstance(data.flake.flake_url, FlakeId)
         self.data = data
 
     def _on_vm_status_changed(self, source: "VMObject") -> None:
@@ -180,12 +181,14 @@ class VMObject(GObject.Object):
             return GLib.SOURCE_REMOVE
 
     def __start(self) -> None:
+        assert isinstance(self.data.flake.vm.flake_url, FlakeId)
         with self._create_machine() as machine:
             # Start building VM
             tstart = datetime.now()
             log.info(f"Building VM {self.get_id()}")
             log_dir = Path(str(self.log_dir.name))
 
+            assert isinstance(self.data.flake.vm.flake_url, FlakeId)
             # Start the build process
             self.build_process = spawn(
                 on_except=None,
@@ -272,6 +275,7 @@ class VMObject(GObject.Object):
             self.emit("vm_status_changed", self)
             return
         log.debug(f"VM state dir {self.log_dir.name}")
+        assert isinstance(self.data.flake.vm.flake_url, FlakeId)
         self._start_thread = threading.Thread(target=self.__start)
         self._start_thread.start()
 

pkgs/clan-vm-manager/clan_vm_manager/singletons/use_vms.py

@@ -4,7 +4,7 @@ from pathlib import Path
 from typing import Any, ClassVar
 
 import gi
-from clan_cli.clan_uri import ClanURI
+from clan_cli.clan_uri import ClanURI, FlakeId
 from clan_cli.history.add import HistoryEntry
 from clan_cli.machines.machines import Machine
 
@@ -34,7 +34,7 @@ class Emitter(GObject.GObject):
 
 class ClanStore:
     _instance: "None | ClanStore" = None
-    _clan_store: GKVStore[str, VMStore]
+    _clan_store: GKVStore[FlakeId, VMStore]
 
     _emitter: Emitter
 
@@ -65,6 +65,7 @@ class ClanStore:
 
     def set_logging_vm(self, ident: str) -> VMObject | None:
         vm = self.get_vm(ClanURI(f"clan://{ident}"))
+
         if vm is not None:
             self._logging_vm = vm
 
@@ -92,7 +93,7 @@ class ClanStore:
         self.clan_store.register_on_change(on_clanstore_change)
 
     @property
-    def clan_store(self) -> GKVStore[str, VMStore]:
+    def clan_store(self) -> GKVStore[FlakeId, VMStore]:
         return self._clan_store
 
     def create_vm_task(self, vm: HistoryEntry) -> bool:
@@ -109,8 +110,12 @@ class ClanStore:
         def log_details(gfile: Gio.File) -> None:
             self.log_details(vm, gfile)
 
+        assert isinstance(entry.flake.flake_url, FlakeId)
+
         vm = VMObject(icon=icon, data=entry, build_log_cb=log_details)
+        assert isinstance(vm.data.flake.flake_url, FlakeId)
         self.push(vm)
+        assert isinstance(vm.data.flake.flake_url, FlakeId)
 
     def log_details(self, vm: VMObject, gfile: Gio.File) -> None:
         views = ViewStack.use().view
@@ -136,7 +141,7 @@ class ClanStore:
         # we cannot check this type, python is not smart enough
 
     def push(self, vm: VMObject) -> None:
-        url = str(vm.data.flake.flake_url)
+        url = vm.data.flake.flake_url
 
         # Only write to the store if the Clan is not already in it
         # Every write to the KVStore rerenders bound widgets to the clan_store
@@ -160,11 +165,12 @@ class ClanStore:
                 vm_store.append(vm)
 
     def remove(self, vm: VMObject) -> None:
-        del self.clan_store[str(vm.data.flake.flake_url)][vm.data.flake.flake_attr]
+        del self.clan_store[vm.data.flake.flake_url][vm.data.flake.flake_attr]
 
     def get_vm(self, uri: ClanURI) -> None | VMObject:
         machine = Machine(uri.machine_name, uri.flake)
-        vm_store = self.clan_store.get(str(machine.flake))
+        vm_store = self.clan_store.get(machine.flake)
+
         if vm_store is None:
             return None
         vm = vm_store.get(str(machine.name), None)

pkgs/clan-vm-manager/clan_vm_manager/views/list.py

@@ -103,7 +103,7 @@ class ClanList(Gtk.Box):
 
         grp = Adw.PreferencesGroup()
         grp.set_title(vm.data.flake.clan_name)
-        grp.set_description(vm.data.flake.flake_url)
+        grp.set_description(str(vm.data.flake.flake_url))
 
         add_action = Gio.SimpleAction.new("add", GLib.VariantType.new("s"))
         add_action.connect("activate", self.on_add)

pkgs/clan-vm-manager/clan_vm_manager/windows/main_window.py

@@ -2,6 +2,7 @@ import logging
 import threading
 
 import gi
+from clan_cli.clan_uri import FlakeId
 from clan_cli.history.list import list_history
 
 from clan_vm_manager.components.interfaces import ClanConfig
@@ -73,6 +74,7 @@ class MainWindow(Adw.ApplicationWindow):
         # Execute `clan flakes add <path>` to democlan for this to work
         # TODO: Make list_history a generator function
         for entry in list_history():
+            assert isinstance(entry.flake.flake_url, FlakeId)
             GLib.idle_add(ClanStore.use().create_vm_task, entry)
 
         GLib.idle_add(self._set_clan_store_ready)