From fab955ee29fd77e8daff8843fa31275b3717caa5 Mon Sep 17 00:00:00 2001 From: Florent Carli Date: Mon, 7 Sep 2026 16:47:21 +0200 Subject: [PATCH 1/5] tests: cover the vm_manager_cluster VM lifecycle disable_vm, start, stop, status and is_enabled are thin Pacemaker wrappers that no test reached. They now run against the fake Pacemaker, which grows show(), delete(), start() and stop(), plus a resource state and an undeletable-resource set the tests drive. list_resources() moves from the fake to the stub the cluster fixture builds, so it answers on the class as well as on an instance. The real Pacemaker declares it static, and is_enabled() calls it that way. Signed-off-by: Florent Carli --- tests/test_vm_manager_cluster_unit.py | 175 +++++++++++++++++++++++++- 1 file changed, 171 insertions(+), 4 deletions(-) diff --git a/tests/test_vm_manager_cluster_unit.py b/tests/test_vm_manager_cluster_unit.py index 3ec1bf8..5208234 100644 --- a/tests/test_vm_manager_cluster_unit.py +++ b/tests/test_vm_manager_cluster_unit.py @@ -50,6 +50,8 @@ REAL_REMOVE = vmc.remove REAL_CREATE_XML = vmc._create_xml REAL_GET_REMOTE_NODES = vmc._get_remote_nodes +REAL_DISABLE_VM = vmc.disable_vm +REAL_IS_ENABLED = vmc.is_enabled RBD_SECRET = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" @@ -262,10 +264,6 @@ def __exit__(self, *exc_info): def call_names(self): return [call[0] for call in self.calls] - def list_resources(self): - self.calls.append(("list_resources",)) - return list(self.collaborators.resources) - def add_vm(self, vm_options, nostart=False): self.calls.append(("add_vm", dict(vm_options), nostart)) if not self.collaborators.add_vm_fails: @@ -289,6 +287,21 @@ def manage(self): def wait_for(self, state): self.calls.append(("wait_for", state)) + def show(self): + self.calls.append(("show",)) + return self.collaborators.state + + def delete(self, force=False, clean=False): + self.calls.append(("delete", force, clean)) + if self.vm_name not in self.collaborators.undeletable_resources: + self.collaborators.resources.discard(self.vm_name) + + def start(self): + self.calls.append(("start",)) + + def stop(self): + self.calls.append(("stop",)) + class FakeDomain: """The subset of a libvirt domain that the cluster backend uses.""" @@ -372,6 +385,11 @@ def __init__(self, rbd, libvirt, ceph_conf): self.pacemakers = [] self.resources = set() self.add_vm_fails = False + # What show() reports for the resource under test. + self.state = "Stopped (disabled)" + # Resources a delete must not remove, to simulate a resource + # that survives the call that was supposed to delete it. + self.undeletable_resources = set() self.observer = None self.remote_nodes = [] self.enabled = [] @@ -459,6 +477,17 @@ class PacemakerStub(FakePacemaker): def __init__(self, vm_name): super().__init__(collaborators, vm_name) + def list_resources(self=None): + """Answer both call styles. + + The real Pacemaker declares this static, and list_vms() and + is_enabled() do call it on the class. Every other caller has + an instance, whose call sequence the tests assert on. + """ + if self is not None: + self.calls.append(("list_resources",)) + return list(collaborators.resources) + @staticmethod def is_valid_host(host): return host in collaborators.valid_hosts @@ -526,6 +555,20 @@ def real_remote_nodes(monkeypatch, cluster): return cluster +@pytest.fixture +def real_disable_vm(monkeypatch, cluster): + """Put the real disable_vm() back, to test it rather than mock it.""" + monkeypatch.setattr(vmc, "disable_vm", REAL_DISABLE_VM) + return cluster + + +@pytest.fixture +def real_is_enabled(monkeypatch, cluster): + """Put the real is_enabled() back, to test it rather than mock it.""" + monkeypatch.setattr(vmc, "is_enabled", REAL_IS_ENABLED) + return cluster + + @pytest.fixture def real_create_xml(monkeypatch, cluster): """Put the real _create_xml() back, with a Ceph secret to find.""" @@ -1723,6 +1766,130 @@ def test_success_is_logged(self, real_remove, caplog): assert "removed" in caplog.text +class TestDisableVm: + """disable_vm(): take the resource out of Pacemaker, keep the disks.""" + + def test_absent_vm_is_only_warned_about(self, real_disable_vm, caplog): + with caplog.at_level("WARNING", logger=vmc.logger.name): + vmc.disable_vm(SRC) + assert "is not on the cluster" in caplog.text + assert "delete" not in real_disable_vm.pacemaker.call_names + + def test_failed_vm_is_cleaned_and_force_deleted(self, real_disable_vm): + real_disable_vm.resources.add(SRC) + real_disable_vm.state = "FAILED (blocked)" + vmc.disable_vm(SRC) + assert ("delete", True, True) in real_disable_vm.pacemaker.calls + + def test_running_vm_is_force_deleted(self, real_disable_vm): + real_disable_vm.resources.add(SRC) + real_disable_vm.state = "Started" + vmc.disable_vm(SRC) + assert ("delete", True, False) in real_disable_vm.pacemaker.calls + + def test_stopped_vm_is_deleted_without_force(self, real_disable_vm): + real_disable_vm.resources.add(SRC) + real_disable_vm.state = "Stopped (disabled)" + vmc.disable_vm(SRC) + assert ("delete", False, False) in real_disable_vm.pacemaker.calls + + def test_surviving_resource_is_reported(self, real_disable_vm): + real_disable_vm.resources.add(SRC) + real_disable_vm.undeletable_resources.add(SRC) + with pytest.raises(Exception, match="Could not remove VM"): + vmc.disable_vm(SRC) + + def test_success_is_logged(self, real_disable_vm, caplog): + real_disable_vm.resources.add(SRC) + with caplog.at_level("INFO", logger=vmc.logger.name): + vmc.disable_vm(SRC) + assert "disabled from the cluster" in caplog.text + + +class TestStart: + """start(): resume a resource Pacemaker already knows.""" + + def test_absent_vm_is_reported(self, cluster): + with pytest.raises(Exception, match="is not on the cluster"): + vmc.start(SRC) + + def test_stopped_vm_is_started_and_waited_for(self, cluster): + cluster.resources.add(SRC) + cluster.state = "Stopped (disabled)" + vmc.start(SRC) + assert ("start",) in cluster.pacemaker.calls + assert ("wait_for", "Started") in cluster.pacemaker.calls + + def test_started_vm_is_left_alone(self, cluster, caplog): + cluster.resources.add(SRC) + cluster.state = "Started" + with caplog.at_level("INFO", logger=vmc.logger.name): + vmc.start(SRC) + assert "start" not in cluster.pacemaker.call_names + assert "already started" in caplog.text + + +class TestIsEnabled: + """is_enabled(): whether Pacemaker carries the resource.""" + + def test_resource_on_the_cluster(self, real_is_enabled): + real_is_enabled.resources.add(SRC) + assert vmc.is_enabled(SRC) is True + + def test_resource_outside_the_cluster(self, real_is_enabled): + assert vmc.is_enabled(SRC) is False + + +class TestStatus: + """status(): Ceph decides Undefined, Pacemaker decides the rest.""" + + def test_vm_without_a_group_is_undefined(self, cluster): + assert vmc.status("ghostvm") == "Undefined" + + def test_vm_outside_the_cluster_is_disabled(self, cluster): + assert vmc.status(SRC) == "Disabled" + + def test_enabled_vm_reports_the_pacemaker_state(self, cluster): + cluster.resources.add(SRC) + cluster.state = "Started" + assert vmc.status(SRC) == "Started" + + def test_undefined_vm_does_not_reach_pacemaker(self, cluster): + vmc.status("ghostvm") + assert cluster.pacemakers == [] + + +class TestStop: + """stop(): ask Pacemaker to stop the resource and wait for it.""" + + def test_absent_vm_is_reported(self, cluster): + with pytest.raises(Exception, match="is not on the cluster"): + vmc.stop(SRC) + + def test_running_vm_is_stopped_and_waited_for(self, cluster): + cluster.resources.add(SRC) + cluster.state = "Started" + vmc.stop(SRC) + assert ("stop",) in cluster.pacemaker.calls + assert ("wait_for", "Stopped (disabled)") in cluster.pacemaker.calls + + def test_stopped_vm_is_left_alone(self, cluster, caplog): + cluster.resources.add(SRC) + cluster.state = "Stopped (disabled)" + with caplog.at_level("INFO", logger=vmc.logger.name): + vmc.stop(SRC) + assert "stop" not in cluster.pacemaker.call_names + assert "already stopped" in caplog.text + + def test_force_is_announced_as_unimplemented(self, cluster, caplog): + cluster.resources.add(SRC) + cluster.state = "Started" + with caplog.at_level("INFO", logger=vmc.logger.name): + vmc.stop(SRC, force=True) + assert "isn't implemented yet" in caplog.text + assert ("stop",) in cluster.pacemaker.calls + + class TestListAllUuids: """list_all_uuids(): the UUID index built from the group XMLs.""" From c9127a8ba949c1838a1f63742e26b39698d700c8 Mon Sep 17 00:00:00 2001 From: Florent Carli Date: Mon, 7 Sep 2026 16:47:21 +0200 Subject: [PATCH 2/5] tests: cover vm_manager_cluster _create_vm_group and list_vms _create_vm_group owns the force path that removes an existing VM and the check that the group Ceph reported creating is really there. The fake gets create_group() and a set of group names it must accept without creating, to reach that check. list_vms picks its source from the enabled flag: the Ceph groups, or the Pacemaker resources without touching Ceph at all. Signed-off-by: Florent Carli --- tests/test_vm_manager_cluster_unit.py | 63 +++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/tests/test_vm_manager_cluster_unit.py b/tests/test_vm_manager_cluster_unit.py index 5208234..c3022e4 100644 --- a/tests/test_vm_manager_cluster_unit.py +++ b/tests/test_vm_manager_cluster_unit.py @@ -50,6 +50,7 @@ REAL_REMOVE = vmc.remove REAL_CREATE_XML = vmc._create_xml REAL_GET_REMOTE_NODES = vmc._get_remote_nodes +REAL_CREATE_VM_GROUP = vmc._create_vm_group REAL_DISABLE_VM = vmc.disable_vm REAL_IS_ENABLED = vmc.is_enabled @@ -98,6 +99,9 @@ def __init__(self, images=(), metadata=None, groups=None): # survive the call that was supposed to delete them. self.undeletable_groups = set() self.undeletable_images = set() + # Group names a creation must not create, to simulate Ceph + # accepting the call and producing nothing. + self.uncreatable_groups = set() # Image name -> list of {"id", "name", "timestamp"} self.snapshots = {} self.calls = [] @@ -174,6 +178,11 @@ def list_group_images(self, name): self._record("list_group_images", name) return sorted(self.groups[name]) + def create_group(self, name): + self._record("create_group", name) + if name not in self.uncreatable_groups: + self.groups.setdefault(name, set()) + def remove_group(self, name): self._record("remove_group", name) if name not in self.undeletable_groups: @@ -555,6 +564,13 @@ def real_remote_nodes(monkeypatch, cluster): return cluster +@pytest.fixture +def real_create_vm_group(monkeypatch, cluster): + """Put the real _create_vm_group() back, to test it for real.""" + monkeypatch.setattr(vmc, "_create_vm_group", REAL_CREATE_VM_GROUP) + return cluster + + @pytest.fixture def real_disable_vm(monkeypatch, cluster): """Put the real disable_vm() back, to test it rather than mock it.""" @@ -1948,6 +1964,53 @@ def test_without_a_group_only_the_system_disk(self, cluster): ] +class TestCreateVmGroup: + """_create_vm_group(): the Ceph group holding a VM's disks.""" + + def test_group_is_created(self, real_create_vm_group): + vmc._create_vm_group(DST) + assert ("create_group", DST) in real_create_vm_group.rbd.calls + assert DST in real_create_vm_group.rbd.groups + + def test_existing_group_is_refused(self, real_create_vm_group): + with pytest.raises(Exception, match="already exists"): + vmc._create_vm_group(SRC) + assert "create_group" not in real_create_vm_group.rbd.call_names + + def test_force_removes_the_existing_vm_first(self, real_create_vm_group): + vmc._create_vm_group(SRC, force=True) + assert real_create_vm_group.removed == [SRC] + assert ("create_group", SRC) in real_create_vm_group.rbd.calls + + def test_group_that_does_not_appear_is_reported( + self, real_create_vm_group + ): + real_create_vm_group.rbd.uncreatable_groups.add(DST) + with pytest.raises(Exception, match="Could not create group"): + vmc._create_vm_group(DST) + + def test_success_is_logged(self, real_create_vm_group, caplog): + with caplog.at_level("INFO", logger=vmc.logger.name): + vmc._create_vm_group(DST) + assert "created successfully" in caplog.text + + +class TestListVms: + """list_vms(): Ceph groups, or the Pacemaker resources.""" + + def test_all_vms_come_from_the_ceph_groups(self, cluster): + cluster.rbd.groups[DST] = set() + assert sorted(vmc.list_vms()) == sorted([SRC, DST]) + + def test_enabled_vms_come_from_pacemaker(self, cluster): + cluster.resources.add(DST) + assert vmc.list_vms(enabled=True) == [DST] + + def test_enabled_does_not_reach_ceph(self, cluster): + vmc.list_vms(enabled=True) + assert cluster.rbd.calls == [] + + def snapshot(index, name, timestamp): """Build a snapshot as the Ceph bindings report it.""" return {"id": index, "name": name, "timestamp": timestamp} From a877391d281ed184bdeaae0fc9110aa3bd888482 Mon Sep 17 00:00:00 2001 From: Florent Carli Date: Mon, 7 Sep 2026 16:47:21 +0200 Subject: [PATCH 3/5] tests: cover the vm_manager_cluster metadata and remote helpers list_snapshots, list_metadata, get_metadata, set_metadata, add_colocation, add_pacemaker_remote and remove_pacemaker_remote are the last uncovered functions of the module. The fakes grow list_image_metadata() on the Ceph side and add_colocation(), add_meta() and remove_meta() on the Pacemaker side. The tests pin down what the wrappers actually decide: the name validation set_metadata, add_colocation and add_pacemaker_remote run before touching anything, the optional port and timeout that stay out of both stores when they are not given, and the KeyError remove_pacemaker_remote swallows on a VM that carries no remote configuration. With them vm_manager_cluster.py reaches 100% statement and branch coverage. Signed-off-by: Florent Carli --- tests/test_vm_manager_cluster_unit.py | 178 ++++++++++++++++++++++++++ 1 file changed, 178 insertions(+) diff --git a/tests/test_vm_manager_cluster_unit.py b/tests/test_vm_manager_cluster_unit.py index c3022e4..24fa883 100644 --- a/tests/test_vm_manager_cluster_unit.py +++ b/tests/test_vm_manager_cluster_unit.py @@ -131,6 +131,10 @@ def remove_image_metadata(self, disk, key): self._record("remove_image_metadata", disk, key) del self.metadata[disk][key] + def list_image_metadata(self, disk): + self._record("list_image_metadata", disk) + return sorted(self.metadata.get(disk, {})) + def image_exists(self, name): self._record("image_exists", name) return name in self.images @@ -311,6 +315,15 @@ def start(self): def stop(self): self.calls.append(("stop",)) + def add_colocation(self, *resources, strong=False): + self.calls.append(("add_colocation", resources, strong)) + + def add_meta(self, key, value): + self.calls.append(("add_meta", key, value)) + + def remove_meta(self, key): + self.calls.append(("remove_meta", key)) + class FakeDomain: """The subset of a libvirt domain that the cluster backend uses.""" @@ -2254,6 +2267,171 @@ def test_rollback_is_logged(self, cluster, caplog): assert "successfully rollbacked" in caplog.text +class TestListSnapshots: + """list_snapshots(): the snapshot names of the system disk.""" + + def test_snapshot_names_are_returned(self, cluster): + cluster.rbd.snapshots[SRC_DISK] = [ + snapshot(0, "snap1", None), + snapshot(1, "snap2", None), + ] + assert vmc.list_snapshots(SRC) == ["snap1", "snap2"] + + def test_no_snapshot_gives_an_empty_list(self, cluster): + assert vmc.list_snapshots(SRC) == [] + + +class TestMetadata: + """list_metadata(), get_metadata() and set_metadata().""" + + def test_metadata_names_are_listed(self, cluster): + cluster.rbd.metadata[SRC_DISK]["name"] = SRC + assert vmc.list_metadata(SRC) == ["_base_xml", "name"] + + def test_a_metadata_value_is_read(self, cluster): + assert vmc.get_metadata(SRC, "_base_xml") == BASE_XML + + def test_an_unknown_metadata_raises(self, cluster): + with pytest.raises(KeyError): + vmc.get_metadata(SRC, "absent") + + def test_a_metadata_value_is_written(self, cluster): + vmc.set_metadata(SRC, "colour", "blue") + assert cluster.rbd.metadata[SRC_DISK]["colour"] == "blue" + + def test_a_reserved_metadata_name_is_refused(self, cluster): + with pytest.raises(ValueError, match="reserved word"): + vmc.set_metadata(SRC, vmc.RESERVED_NAMES[0], "value") + assert "set_image_metadata" not in cluster.rbd.call_names + + def test_a_metadata_name_with_special_chars_is_refused(self, cluster): + with pytest.raises(ValueError, match="special chars"): + vmc.set_metadata(SRC, "not a name", "value") + + def test_writing_a_metadata_is_logged(self, cluster, caplog): + with caplog.at_level("INFO", logger=vmc.logger.name): + vmc.set_metadata(SRC, "colour", "blue") + assert "metadata set: (colour:blue)" in caplog.text + + +class TestAddColocation: + """add_colocation(): the constraint handed over to Pacemaker.""" + + def test_resources_reach_pacemaker(self, cluster): + vmc.add_colocation(SRC, "res1", "res2") + assert ( + "add_colocation", + ("res1", "res2"), + False, + ) in cluster.pacemaker.calls + + def test_strong_is_passed_through(self, cluster): + vmc.add_colocation(SRC, "res1", strong=True) + assert ("add_colocation", ("res1",), True) in cluster.pacemaker.calls + + def test_a_bad_vm_name_is_refused(self, cluster): + with pytest.raises(ValueError, match="special chars"): + vmc.add_colocation("not a name", "res1") + assert cluster.pacemakers == [] + + +REMOTE_METADATA = ( + "_remote_node", + "_remote_node_address", + "_remote_node_port", + "_remote_node_timeout", +) + +REMOTE_META = ( + "remote-node", + "remote-addr", + "remote-port", + "remote-connect-timeout", +) + + +class TestRemovePacemakerRemote: + """remove_pacemaker_remote(): drop the remote config on both sides.""" + + @pytest.fixture + def configured(self, cluster): + """A VM already carrying a full pacemaker-remote config.""" + cluster.rbd.metadata[SRC_DISK].update( + {name: "value" for name in REMOTE_METADATA} + ) + return cluster + + def test_the_metadata_are_removed(self, configured): + vmc.remove_pacemaker_remote(SRC) + assert not set(REMOTE_METADATA) & set( + configured.rbd.metadata[SRC_DISK] + ) + + def test_the_crm_meta_attributes_are_removed(self, configured): + vmc.remove_pacemaker_remote(SRC) + for key in REMOTE_META: + assert ("remove_meta", key) in configured.pacemaker.calls + + def test_a_vm_without_the_metadata_is_still_cleaned(self, cluster): + vmc.remove_pacemaker_remote(SRC) + for key in REMOTE_META: + assert ("remove_meta", key) in cluster.pacemaker.calls + + def test_a_partial_config_stops_at_the_first_gap(self, cluster): + cluster.rbd.metadata[SRC_DISK]["_remote_node"] = "remote1" + vmc.remove_pacemaker_remote(SRC) + assert "_remote_node" not in cluster.rbd.metadata[SRC_DISK] + + +class TestAddPacemakerRemote: + """add_pacemaker_remote(): store the remote config on both sides.""" + + def test_the_node_and_address_are_stored(self, cluster): + vmc.add_pacemaker_remote(SRC, "remote1", "10.0.0.1") + assert cluster.rbd.metadata[SRC_DISK]["_remote_node"] == "remote1" + assert ( + cluster.rbd.metadata[SRC_DISK]["_remote_node_address"] + == "10.0.0.1" + ) + assert ("add_meta", "remote-node", "remote1") in ( + cluster.pacemaker.calls + ) + assert ("add_meta", "remote-addr", "10.0.0.1") in ( + cluster.pacemaker.calls + ) + + def test_the_port_and_timeout_are_optional(self, cluster): + vmc.add_pacemaker_remote(SRC, "remote1", "10.0.0.1") + assert "_remote_node_port" not in cluster.rbd.metadata[SRC_DISK] + assert "_remote_node_timeout" not in cluster.rbd.metadata[SRC_DISK] + assert [ + call for call in cluster.pacemaker.calls if call[0] == "add_meta" + ] == [ + ("add_meta", "remote-addr", "10.0.0.1"), + ("add_meta", "remote-node", "remote1"), + ] + + def test_the_port_and_timeout_are_stored_when_given(self, cluster): + vmc.add_pacemaker_remote( + SRC, + "remote1", + "10.0.0.1", + remote_node_port="3121", + remote_node_timeout="60s", + ) + assert cluster.rbd.metadata[SRC_DISK]["_remote_node_port"] == "3121" + assert cluster.rbd.metadata[SRC_DISK]["_remote_node_timeout"] == "60s" + assert ("add_meta", "remote-port", "3121") in cluster.pacemaker.calls + assert ("add_meta", "remote-connect-timeout", "60s") in ( + cluster.pacemaker.calls + ) + + def test_a_bad_remote_node_name_is_refused(self, cluster): + with pytest.raises(ValueError, match="special chars"): + vmc.add_pacemaker_remote(SRC, "not a name", "10.0.0.1") + assert cluster.rbd.calls == [] + + class TestConsole: """console(): find the hypervisor, then open virsh on it.""" From e8d6012bfa1813fd38ead2fbec1b90bc5d991c78 Mon Sep 17 00:00:00 2001 From: Florent Carli Date: Mon, 7 Sep 2026 17:25:18 +0200 Subject: [PATCH 4/5] create: apply --disable and --enable-live-migration Both assignments sat behind "enable" in args and "live_migration" in args. Neither is an argparse dest, so both guards were always false and the two flags did nothing. Assign unconditionally, the way add-to-cluster already does. Signed-off-by: Florent Carli --- tests/test_vm_manager_cmd.py | 14 -------------- vm_manager/vm_manager_cmd.py | 12 ++++-------- 2 files changed, 4 insertions(+), 22 deletions(-) diff --git a/tests/test_vm_manager_cmd.py b/tests/test_vm_manager_cmd.py index 1420f66..fe1e764 100644 --- a/tests/test_vm_manager_cmd.py +++ b/tests/test_vm_manager_cmd.py @@ -614,26 +614,12 @@ def test_pinned_host_is_forwarded(self, run_cli, api, xml_file): options = self._create(run_cli, api, xml_file, "--pinned-host", "hyp1") assert options["pinned_host"] == "hyp1" - @pytest.mark.xfail( - strict=True, - reason="main() guards the assignment with 'if \"enable\" in args', " - "but 'enable' is never an argparse dest for create, so args.enable " - "is never set. _configure_vm() treats a missing 'enable' key as " - "True, so 'create --disable' enables the VM anyway.", - ) def test_disable_is_forwarded_as_enable_false( self, run_cli, api, xml_file ): options = self._create(run_cli, api, xml_file, "--disable") assert options["enable"] is False - @pytest.mark.xfail( - strict=True, - reason='main() guards the assignment with \'if "live_migration" in ' - "args', but 'live_migration' is never an argparse dest, so the key " - "is never set and _configure_vm() never writes the _live_migration " - "metadata. clone and add-to-cluster assign it unconditionally.", - ) def test_enable_live_migration_is_renamed(self, run_cli, api, xml_file): options = self._create( run_cli, api, xml_file, "--enable-live-migration" diff --git a/vm_manager/vm_manager_cmd.py b/vm_manager/vm_manager_cmd.py index 4ec63fe..552e0e4 100755 --- a/vm_manager/vm_manager_cmd.py +++ b/vm_manager/vm_manager_cmd.py @@ -538,16 +538,12 @@ def main(): elif args.command == "create": with open(args.xml, "r") as xml: args.base_xml = xml.read() - if "live_migration" in args: - args.live_migration = args.enable_live_migration - if "add_crm_config_cmd" in args: - args.crm_config_cmd = args.add_crm_config_cmd + args.live_migration = args.enable_live_migration + args.crm_config_cmd = args.add_crm_config_cmd if "disable" in args and args.disable: - if "enable" in args: - args.enable = not args.disable + args.enable = not args.disable else: - if "enable" in args: - args.enable = True + args.enable = True vm_manager.create(vars(args)) elif args.command == "clone": args.base_xml = None From cef23a901104514d4f7f2e79ac9361fc461e4582 Mon Sep 17 00:00:00 2001 From: Florent Carli Date: Mon, 7 Sep 2026 17:25:18 +0200 Subject: [PATCH 5/5] clone: apply --disable The clone branch never set args.enable, and _configure_vm() reads a missing key as True, so --disable enabled the clone anyway. Signed-off-by: Florent Carli --- tests/test_vm_manager_cmd.py | 7 ------- vm_manager/vm_manager_cmd.py | 4 ++++ 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/tests/test_vm_manager_cmd.py b/tests/test_vm_manager_cmd.py index fe1e764..75b0562 100644 --- a/tests/test_vm_manager_cmd.py +++ b/tests/test_vm_manager_cmd.py @@ -654,13 +654,6 @@ def test_live_migration_is_renamed(self, run_cli, api): _, args, _ = api.only assert args[0]["live_migration"] is True - @pytest.mark.xfail( - strict=True, - reason="the clone branch of main() never sets args.enable, and " - "_configure_vm() treats a missing 'enable' key as True, so " - "'clone --disable' enables the clone anyway. Same root cause as " - "TestMainCreate.test_disable_is_forwarded_as_enable_false.", - ) def test_disable_is_forwarded_as_enable_false(self, run_cli, api): run_cli("clone", "-n", "vm1", "--dst_name", "vm2", "--disable") _, args, _ = api.only diff --git a/vm_manager/vm_manager_cmd.py b/vm_manager/vm_manager_cmd.py index 552e0e4..e92ac9e 100755 --- a/vm_manager/vm_manager_cmd.py +++ b/vm_manager/vm_manager_cmd.py @@ -552,6 +552,10 @@ def main(): args.base_xml = xml.read() args.live_migration = args.enable_live_migration args.crm_config_cmd = args.add_crm_config_cmd + if "disable" in args and args.disable: + args.enable = not args.disable + else: + args.enable = True vm_manager.clone(vars(args)) elif args.command == "disable": vm_manager.disable_vm(args.name)