From 6aeb5dbda58db9e84e15e77c8296a69188bbe85c Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Fri, 21 Aug 2026 00:45:16 +0800 Subject: [PATCH] feat(copilot): device skills per-topic split layout and topic retrieval SkillsLoader.load_device_skills() now supports two device layouts: the existing single file and a split directory (device//_base.yaml + one YAML per protocol topic). Topic files are merged into the device skill under 'topics', keyed by their 'topic' field; mismatched device_type, missing _base.yaml and duplicate topics are handled with explicit log-and-skip. get_skill() and DeviceSkillsTool gain a 'topic' parameter following the injection list -> index -> issue pattern. Topic bodies are never returned without an explicit topic request - index/summary/full all serve a topic index instead - so growing a device with new protocol topics no longer grows the token cost of device-level lookups. Also fix reload_skills() to actually drop injection skills that fail validate_skill_format() instead of logging 'skipping' and merging them anyway. --- .../implemented/skills-repository.md | 30 +- .../agent/gns3_copilot/skills/loader.py | 117 +++++- .../agent/gns3_copilot/skills/manager.py | 13 +- .../agent/gns3_copilot/skills/registry.py | 76 +++- tests/agent/test_skills_device_topics.py | 348 ++++++++++++++++++ 5 files changed, 546 insertions(+), 38 deletions(-) create mode 100644 tests/agent/test_skills_device_topics.py diff --git a/docs/gns3-copilot/implemented/skills-repository.md b/docs/gns3-copilot/implemented/skills-repository.md index aa9c8f2db..167941465 100644 --- a/docs/gns3-copilot/implemented/skills-repository.md +++ b/docs/gns3-copilot/implemented/skills-repository.md @@ -14,7 +14,7 @@ GNS3 Copilot loads all skills, prompts, and security configurations from an exte The repository provides: - **Injection skills** (39 categories): Network fault scenarios for troubleshooting practice -- **Device skills**: Device-specific command knowledge (VPCS, etc.) +- **Device skills**: Device-specific command knowledge (VPCS, etc.) — large devices split into per-protocol **topics** - **Feature skills**: Topology planning, network design - **System prompts**: Agent personality and behavior definitions - **Forbidden commands**: Security rules for command filtering @@ -24,7 +24,7 @@ The repository provides: ```mermaid graph TD subgraph "GNS3-Skills Repository" - YAML[injection/*.yaml
device/*.yaml
feature/*.yaml] + YAML[injection/*.yaml
device/*.yaml + device/*/*.yaml
feature/*.yaml] MD[prompts/*.md] CFG[config/forbidden_commands.txt] end @@ -56,7 +56,11 @@ GNS3-Skills/ │ ├── vlan_issues.yaml │ └── ... ├── device/ # Device-specific skills -│ └── vpcs.yaml +│ ├── vpcs.yaml # small devices: one single file +│ └── frr/ # large devices: split per protocol topic +│ ├── _base.yaml # device-level skill (console model, notes, aliases) +│ ├── ospf.yaml # topic file (merged under "topics" at load time) +│ └── bgp.yaml ├── feature/ # Feature skills │ └── topology_planner.yaml ├── prompts/ # System prompts (Markdown) @@ -68,6 +72,26 @@ GNS3-Skills/ └── forbidden_commands.txt ``` +## Device Topics + +A device with knowledge for many protocols would grow one YAML file indefinitely. Such devices use a split layout instead: `device//_base.yaml` holds the device-level skill, and one file per protocol topic (`ospf.yaml`, `bgp.yaml`, ...) holds its commands and troubleshooting entries. The loader merges them into a single `SKILLS_REGISTRY` entry: + +``` +SKILLS_REGISTRY["frr_vtysh"] = { ..._base.yaml..., "topics": { "ospf": {...}, "bgp": {...} } } +``` + +Topic files must declare `device_type` (matching their `_base.yaml`), `topic` and `name`; the CI validator in the skills repository enforces this. + +The `device_skills` tool exposes a three-step drill-down (mirroring `injection_skills`'s list → index → issue pattern): + +```json +{"action": "list"} +{"device_type": "frr_vtysh", "detail": "index"} +{"device_type": "frr_vtysh", "topic": "bgp"} +``` + +Topic bodies are only served on an explicit `topic` request — every other detail level returns a topic index — so adding topics to a device does not grow the token cost of device-level lookups. + ## Configuration Skills repository settings are configured in `gns3_server.conf` under the `[Server]` section: diff --git a/gns3server/agent/gns3_copilot/skills/loader.py b/gns3server/agent/gns3_copilot/skills/loader.py index 4d585cac5..fee183ef6 100644 --- a/gns3server/agent/gns3_copilot/skills/loader.py +++ b/gns3server/agent/gns3_copilot/skills/loader.py @@ -100,6 +100,12 @@ class SkillsLoader: """ Load all device skills from YAML files. + Supports two layouts: + - Single file: device/.yaml + - Split directory: device//_base.yaml + device//.yaml + (topic files are merged into the base skill under "topics", + keyed by their "topic" field) + Returns: Dictionary mapping skill keys to skill definitions """ @@ -107,33 +113,110 @@ class SkillsLoader: logger.error("PyYAML is not installed. Cannot load skills from YAML.") return {} - skills = {} + skills: Dict[str, Dict[str, Any]] = {} device_dir = self.skills_dir / "device" if not device_dir.exists(): logger.warning(f"Device skills directory not found: {device_dir}") return {} - for yaml_file in device_dir.glob("*.yaml"): - try: - skill_data = self._load_yaml(yaml_file) - if not skill_data: - logger.warning(f"Skipping empty YAML file: {yaml_file}") - continue - # Use device_type from YAML content as the key - # Fallback to filename stem if device_type not present - skill_key = skill_data.get("device_type") if isinstance(skill_data, dict) else None - if not skill_key: - skill_key = yaml_file.stem - logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key") - skills[skill_key] = skill_data - logger.debug(f"Loaded device skill: {skill_key} from {yaml_file}") - except Exception as e: - logger.error(f"Failed to load skill from {yaml_file}: {e}") + for entry in sorted(device_dir.iterdir()): + if entry.is_file() and entry.suffix == ".yaml": + self._load_single_device_skill(skills, entry) + elif entry.is_dir(): + self._load_split_device_skill(skills, entry) logger.debug(f"Loaded {len(skills)} device skills from device directory") return skills + def _load_single_device_skill(self, skills: Dict[str, Dict[str, Any]], yaml_file: Path) -> None: + """ + Load a single-file device skill into the skills dictionary. + """ + try: + skill_data = self._load_yaml(yaml_file) + if not skill_data: + logger.warning(f"Skipping empty YAML file: {yaml_file}") + return + # Use device_type from YAML content as the key + # Fallback to filename stem if device_type not present + skill_key = skill_data.get("device_type") + if not skill_key: + skill_key = yaml_file.stem + logger.warning(f"No device_type in {yaml_file}, using filename '{skill_key}' as key") + skills[skill_key] = skill_data + logger.debug(f"Loaded device skill: {skill_key} from {yaml_file}") + except Exception as e: + logger.error(f"Failed to load skill from {yaml_file}: {e}") + + def _load_split_device_skill(self, skills: Dict[str, Dict[str, Any]], device_path: Path) -> None: + """ + Load a split device skill (directory with _base.yaml + topic files). + + The base file provides the device-level skill; every other YAML file + in the directory is a protocol topic merged under "topics". + """ + base_file = device_path / "_base.yaml" + if not base_file.exists(): + logger.error(f"No _base.yaml in device directory: {device_path}, skipping") + return + + try: + base_data = self._load_yaml(base_file) + except Exception as e: + logger.error(f"Failed to load skill from {base_file}: {e}") + return + if not base_data: + logger.warning(f"Skipping empty YAML file: {base_file}") + return + + skill_key = base_data.get("device_type") + if not skill_key: + skill_key = device_path.name + logger.warning(f"No device_type in {base_file}, using directory name '{skill_key}' as key") + + # Seed topics from the base file (if any), then merge topic files + base_topics = base_data.get("topics") + topics: Dict[str, Any] = dict(base_topics) if isinstance(base_topics, dict) else {} + + for yaml_file in sorted(device_path.glob("*.yaml")): + if yaml_file.name == "_base.yaml": + continue + try: + topic_data = self._load_yaml(yaml_file) + if not topic_data: + logger.warning(f"Skipping empty YAML file: {yaml_file}") + continue + + topic_device_type = topic_data.pop("device_type", None) + if topic_device_type is not None and topic_device_type != skill_key: + logger.error( + f"device_type mismatch in {yaml_file}: '{topic_device_type}' " + f"!= base '{skill_key}', skipping topic" + ) + continue + + topic_key = topic_data.pop("topic", None) + if not topic_key: + topic_key = yaml_file.stem + logger.warning(f"No 'topic' field in {yaml_file}, using filename '{topic_key}'") + if topic_key in topics: + logger.warning(f"Duplicate topic '{topic_key}' in {device_path.name} (from {yaml_file}), overwriting") + + # category/topics belong to the base skill only + topic_data.pop("category", None) + topic_data.pop("topics", None) + + topics[topic_key] = topic_data + logger.debug(f"Loaded device topic: {skill_key}/{topic_key} from {yaml_file}") + except Exception as e: + logger.error(f"Failed to load topic from {yaml_file}: {e}") + + if topics: + base_data["topics"] = topics + skills[skill_key] = base_data + logger.debug(f"Loaded device skill: {skill_key} from {device_path} ({len(topics)} topics)") + def load_feature_skills(self) -> Dict[str, Dict[str, Any]]: """ Load all feature skills from YAML files. diff --git a/gns3server/agent/gns3_copilot/skills/manager.py b/gns3server/agent/gns3_copilot/skills/manager.py index 6a663e0e8..e14b8a8dc 100644 --- a/gns3server/agent/gns3_copilot/skills/manager.py +++ b/gns3server/agent/gns3_copilot/skills/manager.py @@ -221,11 +221,18 @@ class SkillsManager: logger.warning("No injection skills loaded, keeping existing skills") return False - # Validate injection skills + # Validate injection skills (drop invalid ones before merging) + valid_injection_skills = {} for skill_key, skill_data in new_injection_skills.items(): - if not self.loader.validate_skill_format(skill_data): + if self.loader.validate_skill_format(skill_data): + valid_injection_skills[skill_key] = skill_data + else: logger.error(f"Invalid skill format for {skill_key}, skipping") - continue + + if not valid_injection_skills: + logger.warning("No valid injection skills loaded, keeping existing skills") + return False + new_injection_skills = valid_injection_skills # Load new device skills from YAML files new_device_skills = self.loader.load_device_skills() diff --git a/gns3server/agent/gns3_copilot/skills/registry.py b/gns3server/agent/gns3_copilot/skills/registry.py index 4d0cf1a53..a003413be 100644 --- a/gns3server/agent/gns3_copilot/skills/registry.py +++ b/gns3server/agent/gns3_copilot/skills/registry.py @@ -373,6 +373,7 @@ def get_skill( category: str | None = None, detail: str = "full", issue: str | None = None, + topic: str | None = None, ) -> dict[str, Any]: """ Get skill by device_type, with configurable detail level. @@ -382,6 +383,9 @@ def get_skill( category: Optional category filter detail: Detail level - "index" (names only), "summary" (+desc/sev/diff), "full" (all) issue: Optional specific issue key to retrieve + topic: Optional protocol topic to retrieve (split devices only). + Topic bodies are NEVER included without an explicit topic + request - all other detail levels return a topic index. Returns: Skill dictionary (detail varies by level), or error dict @@ -412,6 +416,28 @@ def get_skill( ], } + topics = skill.get("topics", {}) + + # Single topic lookup (topic bodies stay out of every other response) + if topic: + topic_data = topics.get(topic) + if not topic_data: + for key, data in topics.items(): + if key.lower() == topic.lower(): + topic_data = data + topic = key + break + if not topic_data: + return { + "error": f"Unknown topic '{topic}' in {device_type}", + "available_topics": list(topics.keys()), + } + return { + "device_type": device_type, + "skill_name": skill.get("name"), + "topic": {topic: topic_data}, + } + issues = skill.get("issues", {}) # Single issue lookup (most token-efficient) @@ -429,17 +455,20 @@ def get_skill( } if detail == "index": - # Minimal: only issue keys and names (90%+ token savings) - return { + # Minimal: only issue/topic keys and names (90%+ token savings) + result = { "device_type": device_type, "name": skill.get("name"), "description": skill.get("description"), "issues": {k: v["name"] for k, v in issues.items()}, } + if topics: + result["topics"] = {k: v.get("name", k) for k, v in topics.items()} + return result if detail == "summary": # Moderate: names + description + severity + difficulty - return { + result = { "device_type": device_type, "name": skill.get("name"), "description": skill.get("description"), @@ -453,14 +482,22 @@ def get_skill( for k, v in issues.items() }, } + if topics: + result["topics"] = { + k: {"name": v.get("name", k), "description": v.get("description", "")} + for k, v in topics.items() + } + return result - # Full detail (original behavior) - result = dict(skill) + # Full detail: topic bodies are replaced by the topic index + result = {k: v for k, v in skill.items() if k != "topics"} result["device_type"] = device_type + if topics: + result["topics"] = {k: v.get("name", k) for k, v in topics.items()} return result -def list_available_skills(category: str | None = None) -> list[dict[str, str]]: +def list_available_skills(category: str | None = None) -> list[dict[str, Any]]: """List all available device/feature skills, optionally filtered by category.""" skills = [] for did, skill in SKILLS_REGISTRY.items(): @@ -470,12 +507,14 @@ def list_available_skills(category: str | None = None) -> list[dict[str, str]]: "device_type": did, "name": skill.get("name", did), "category": skill.get("category"), + "topic_count": len(skill.get("topics", {})), }) else: skills.append({ "device_type": did, "name": skill.get("name", did), "category": skill.get("category"), + "topic_count": len(skill.get("topics", {})), }) return skills @@ -642,15 +681,21 @@ class DeviceSkillsTool(BaseTool): Provides access to device command knowledge (VPCS), topology planning, etc. For fault injection skills, use the injection_skills tool. - INPUT FORMAT (JSON string): - { - "action": "get", # "get" (default) or "list" - "device_type": "gns3_vpcs_telnet", # Required for action="get" - "detail": "full" # "full" (default) for complete skill information - } + TOKEN-EFFICIENT USAGE: + 1. List devices: {"action": "list"} + 2. List topics of a device: {"device_type": "frr_vtysh", "detail": "index"} + 3. Get ONE protocol topic (devices with topics): {"device_type": "frr_vtysh", "topic": "bgp"} + 4. Devices without topics: {"device_type": "gns3_vpcs_telnet"} - For action="list": - {"action": "list"} # Lists all available device/feature skills + Topic bodies are NEVER returned without an explicit "topic" - fetching a + device without one only returns its base skill plus the topic index, so + always request the specific protocol topic before configuring it. + + PARAMETERS: + - action: "list" or "get" (default "get") + - device_type: Required for action="get" (e.g., "frr_vtysh") + - topic: Protocol topic key from the topic index (e.g., "ospf", "bgp") + - detail: "index" | "summary" | "full" (default "full") """ def _run( @@ -693,8 +738,9 @@ class DeviceSkillsTool(BaseTool): category = params.get("category") detail = params.get("detail", "full") issue = params.get("issue") + topic = params.get("topic") - skill = get_skill(device_type, category, detail=detail, issue=issue) + skill = get_skill(device_type, category, detail=detail, issue=issue, topic=topic) return json.dumps(skill, ensure_ascii=False, indent=2) diff --git a/tests/agent/test_skills_device_topics.py b/tests/agent/test_skills_device_topics.py new file mode 100644 index 000000000..68d8371c5 --- /dev/null +++ b/tests/agent/test_skills_device_topics.py @@ -0,0 +1,348 @@ +#!/usr/bin/env python +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +""" +Tests for device skill topic splitting: directory layout loading +(_base.yaml + topic files) and topic-level retrieval via get_skill / +DeviceSkillsTool. +""" + +import pytest + +from gns3server.agent.gns3_copilot.skills.loader import SkillsLoader +from gns3server.agent.gns3_copilot.skills.registry import ( + DeviceSkillsTool, + get_skill, +) + + +@pytest.fixture +def skills_dir(tmp_path): + """ + Build a skills directory with both device layouts: + + - vpcs.yaml: single-file device (no topics) + - frr/: split device (_base.yaml + ospf/bgp topic files, + one mismatched topic file, one file without a topic field) + - orphan/: directory without _base.yaml (skipped) + """ + device_dir = tmp_path / "device" + device_dir.mkdir() + + (device_dir / "vpcs.yaml").write_text( + """ +name: "VPCS" +description: "VPCS test device" +device_type: "gns3_vpcs_telnet" +category: "device" +config_commands: + ip_config: + syntax: "ip
/ " + description: "Set PC address" +""", + encoding="utf-8", + ) + + frr_dir = device_dir / "frr" + frr_dir.mkdir() + (frr_dir / "_base.yaml").write_text( + """ +name: "FRR (Free Range Routing)" +description: "FRR test device" +device_type: "frr_vtysh" +category: "device" +config_commands: + write_memory: + syntax: "write memory" + description: "Save config" +""", + encoding="utf-8", + ) + (frr_dir / "ospf.yaml").write_text( + """ +device_type: "frr_vtysh" +topic: ospf +name: "OSPF (FRR 10.x)" +description: "OSPF topic" +config_commands: + ospfv2: + syntax: "router ospf" + description: "OSPFv2 process" +""", + encoding="utf-8", + ) + (frr_dir / "bgp.yaml").write_text( + """ +device_type: "frr_vtysh" +topic: bgp +name: "BGP (FRR 10.x)" +description: "BGP topic" +config_commands: + bgp_base: + syntax: "router bgp " + description: "BGP base" +""", + encoding="utf-8", + ) + # device_type mismatch with the base -> topic must be skipped + (frr_dir / "mpls.yaml").write_text( + """ +device_type: "other_device_type" +topic: mpls +name: "MPLS" +config_commands: + mpls_base: + syntax: "router mpls" + description: "..." +""", + encoding="utf-8", + ) + # no topic field -> falls back to the filename stem + (frr_dir / "static.yaml").write_text( + """ +device_type: "frr_vtysh" +name: "Static routing" +config_commands: + static_routes: + syntax: "ip route / " + description: "..." +""", + encoding="utf-8", + ) + + # directory without _base.yaml -> whole device skipped + orphan_dir = device_dir / "orphan" + orphan_dir.mkdir() + (orphan_dir / "some_topic.yaml").write_text( + """ +device_type: "orphan_device" +topic: anything +name: "Orphan" +""", + encoding="utf-8", + ) + + return tmp_path + + +class TestDeviceSkillsLoading: + """ + SkillsLoader.load_device_skills() with single-file and split layouts. + """ + + def test_both_layouts_are_loaded(self, skills_dir): + skills = SkillsLoader(str(skills_dir)).load_device_skills() + + assert "gns3_vpcs_telnet" in skills + assert "frr_vtysh" in skills + # the orphan directory (no _base.yaml) must not produce an entry + assert "orphan_device" not in skills + assert len(skills) == 2 + + def test_topics_are_merged_under_topics(self, skills_dir): + skills = SkillsLoader(str(skills_dir)).load_device_skills() + frr = skills["frr_vtysh"] + + # base-level content stays at the top level + assert frr["config_commands"]["write_memory"]["syntax"] == "write memory" + assert frr["category"] == "device" + + topics = frr["topics"] + assert topics["ospf"]["name"] == "OSPF (FRR 10.x)" + assert topics["bgp"]["config_commands"]["bgp_base"]["syntax"] == "router bgp " + # file without a topic field falls back to its filename stem + assert "static" in topics + # mismatched device_type is skipped + assert "mpls" not in topics + + def test_topic_metadata_is_stripped(self, skills_dir): + skills = SkillsLoader(str(skills_dir)).load_device_skills() + for topic_data in skills["frr_vtysh"]["topics"].values(): + assert "device_type" not in topic_data + assert "topic" not in topic_data + assert "category" not in topic_data + assert "topics" not in topic_data + + def test_single_file_device_has_no_topics_key(self, skills_dir): + skills = SkillsLoader(str(skills_dir)).load_device_skills() + assert "topics" not in skills["gns3_vpcs_telnet"] + + +@pytest.fixture +def device_registry(): + """ + Populate SKILLS_REGISTRY with a split device and a single-file device, + restoring the previous content afterwards. + """ + from gns3server.agent.gns3_copilot.skills import registry + + saved = dict(registry.SKILLS_REGISTRY) + registry.SKILLS_REGISTRY.clear() + registry.SKILLS_REGISTRY.update( + { + "frr_vtysh": { + "name": "FRR (Free Range Routing)", + "description": "FRR test device", + "category": "device", + "config_commands": {"write_memory": {"syntax": "write memory"}}, + "topics": { + "ospf": { + "name": "OSPF (FRR 10.x)", + "description": "OSPF topic", + "config_commands": {"ospfv2": {"syntax": "router ospf"}}, + }, + "bgp": { + "name": "BGP (FRR 10.x)", + "description": "BGP topic", + "config_commands": {"bgp_base": {"syntax": "router bgp "}}, + }, + }, + }, + "gns3_vpcs_telnet": { + "name": "VPCS", + "description": "VPCS test device", + "category": "device", + "config_commands": {"ip_config": {"syntax": "ip
/"}}, + }, + } + ) + yield registry + registry.SKILLS_REGISTRY.clear() + registry.SKILLS_REGISTRY.update(saved) + + +class TestGetSkillTopics: + """ + Topic-level retrieval and topic index behavior in get_skill(). + """ + + def test_topic_lookup_returns_topic_body(self, device_registry): + result = get_skill("frr_vtysh", topic="bgp") + assert result["device_type"] == "frr_vtysh" + assert result["skill_name"] == "FRR (Free Range Routing)" + assert result["topic"]["bgp"]["config_commands"]["bgp_base"]["syntax"] == "router bgp " + + def test_topic_lookup_is_case_insensitive(self, device_registry): + result = get_skill("frr_vtysh", topic="BGP") + assert "bgp" in result["topic"] + + def test_unknown_topic_lists_available_topics(self, device_registry): + result = get_skill("frr_vtysh", topic="mpls") + assert "error" in result + assert sorted(result["available_topics"]) == ["bgp", "ospf"] + + def test_full_without_topic_returns_index_not_bodies(self, device_registry): + result = get_skill("frr_vtysh", detail="full") + # base-level content is included... + assert result["config_commands"]["write_memory"]["syntax"] == "write memory" + # ...but topic bodies are never included without an explicit topic + assert result["topics"] == { + "ospf": "OSPF (FRR 10.x)", + "bgp": "BGP (FRR 10.x)", + } + assert "config_commands" not in result["topics"]["ospf"] + + def test_index_includes_topic_index(self, device_registry): + result = get_skill("frr_vtysh", detail="index") + assert result["topics"]["bgp"] == "BGP (FRR 10.x)" + + def test_summary_includes_topic_descriptions(self, device_registry): + result = get_skill("frr_vtysh", detail="summary") + assert result["topics"]["ospf"] == { + "name": "OSPF (FRR 10.x)", + "description": "OSPF topic", + } + + def test_single_file_device_still_works(self, device_registry): + result = get_skill("gns3_vpcs_telnet") + assert result["config_commands"]["ip_config"]["syntax"] == "ip
/" + assert "topics" not in result + + +class TestDeviceSkillsToolTopics: + """ + DeviceSkillsTool passes the topic parameter through to get_skill(). + """ + + def test_tool_topic_request(self, device_registry): + import json + + tool = DeviceSkillsTool() + result = json.loads(tool._run('{"device_type": "frr_vtysh", "topic": "ospf"}')) + assert result["topic"]["ospf"]["config_commands"]["ospfv2"]["syntax"] == "router ospf" + + def test_tool_list_shows_topic_counts(self, device_registry): + import json + + tool = DeviceSkillsTool() + result = json.loads(tool._run('{"action": "list"}')) + by_type = {s["device_type"]: s for s in result["skills"]} + assert by_type["frr_vtysh"]["topic_count"] == 2 + assert by_type["gns3_vpcs_telnet"]["topic_count"] == 0 + + +class TestReloadSkillsValidation: + """ + Invalid injection skills are dropped instead of merged into the registry. + """ + + def test_invalid_injection_skill_is_dropped(self, tmp_path, monkeypatch): + from gns3server.config import Config + from gns3server.agent.gns3_copilot.skills import registry + from gns3server.agent.gns3_copilot.skills.manager import SkillsManager + + # SkillsManager derives its local path from /skills + injection_dir = tmp_path / "skills" / "injection" + injection_dir.mkdir(parents=True) + (injection_dir / "valid.yaml").write_text( + """ +name: "OSPF Issues Injection" +description: "OSPF faults" +category: "injection" +issues: + ospf_area_mismatch: + name: "OSPF Area Mismatch" + description: "Areas differ" +""", + encoding="utf-8", + ) + # missing the required "issues" field -> invalid, must be dropped + (injection_dir / "broken.yaml").write_text( + """ +name: "Broken Injection" +description: "No issues field" +""", + encoding="utf-8", + ) + + monkeypatch.setattr(Config, "config_dir", property(lambda self: str(tmp_path))) + + saved_injection = dict(registry.INJECTION_SKILLS_REGISTRY) + saved_skills = dict(registry.SKILLS_REGISTRY) + registry.INJECTION_SKILLS_REGISTRY.clear() + registry.SKILLS_REGISTRY.clear() + try: + manager = SkillsManager(repo_url="https://example.invalid/gns3-skills.git") + manager._repo = None + assert manager.reload_skills() is True + assert "injection_valid" in registry.INJECTION_SKILLS_REGISTRY + assert "injection_broken" not in registry.INJECTION_SKILLS_REGISTRY + finally: + registry.INJECTION_SKILLS_REGISTRY.clear() + registry.INJECTION_SKILLS_REGISTRY.update(saved_injection) + registry.SKILLS_REGISTRY.clear() + registry.SKILLS_REGISTRY.update(saved_skills)