test: cover the kernel-datapath review fixes

Regression coverage added with the fixes:
- capture-slot semantics: second-port refusal, failed-start rollback, and
  a non-owner port's stop not reaching `capture stop_kernel`;
- QEMU stop/start teardown on a mid-body or pre-launch failure, IOU start
  cleanup and filter guards, Dynamips half-attach undo plus retry;
- switch VLAN reconcile skipping a deferred anchor and rolling the
  ports_mapping back on failure;
- Docker/QEMU occupied-port refusal, probe ctime invalidation, probe
  survival of a failing uBridge start;
- env-marker dialects (mid-line mentions, falsy values) and the agent
  merge rejecting entries without a device_name.

Test-body imports of product modules move to the shared header.
This commit is contained in:
YueGuobin committed 2026-10-06 19:55:31 +08:00
1 parent c3c025ab2a
commit 35cf964882
13 files changed
+438 -40

No files matched your search

+38 -6
View File
@@ -10,6 +10,8 @@ overwriting each other's commands and outputs.
import json
from unittest.mock import MagicMock, patch
import pytest
from gns3server.agent.gns3_copilot.utils.device_configs import (
merge_duplicate_device_configs,
)
@@ -334,6 +336,27 @@ class TestDuplicateDeviceMerging:
assert result[0]["status"] == "partial_success"
assert set(result[0]["blocked_commands"]) == {"debug ip routing", "debug ospf events"}
def test_entry_without_device_name_fails_the_whole_batch(self):
"""An unaddressable entry must abort with an error, not vanish."""
from gns3server.agent.gns3_copilot.tools_v2.display_tools_nornir import (
ExecuteMultipleDeviceCommands,
)
result = ExecuteMultipleDeviceCommands()._run(
json.dumps(
{
"project_id": PROJECT_ID,
"device_configs": [
{"device_name": "R1", "commands": ["show version"]},
{"commands": ["show clock"]},
],
}
)
)
assert result[0]["status"] == "failed"
assert "device_configs[1]" in result[0]["error"]
class TestMergeDuplicateDeviceConfigs:
"""Unit tests for the shared normalization helper."""
@@ -361,12 +384,21 @@ class TestMergeDuplicateDeviceConfigs:
assert merged == [{"device_name": "R1", "commands": ["a", "b"], "extra": 1}]
assert original[0]["commands"] == ["a"]
def test_entry_without_device_name_is_dropped(self):
merged = merge_duplicate_device_configs(
[{"commands": ["a"]}, {"device_name": "R1", "commands": ["b"]}],
commands_field="commands",
)
assert merged == [{"device_name": "R1", "commands": ["b"]}]
def test_entry_without_device_name_is_rejected(self):
# Silently dropping it would let the batch execute partially while
# every returned row reports success.
with pytest.raises(ValueError, match=r"device_configs\[0\]"):
merge_duplicate_device_configs(
[{"commands": ["a"]}, {"device_name": "R1", "commands": ["b"]}],
commands_field="commands",
)
def test_entry_that_is_not_an_object_is_rejected(self):
with pytest.raises(ValueError, match=r"device_configs\[1\]"):
merge_duplicate_device_configs(
[{"device_name": "R1", "commands": ["a"]}, "R2"],
commands_field="commands",
)
def test_missing_commands_field_defaults_to_empty(self):
merged = merge_duplicate_device_configs(