mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-08-27 12:30:13 +03:00
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/<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.
This commit is contained in:
parent
8d8b2c9692
commit
6aeb5dbda5
@ -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<br/>device/*.yaml<br/>feature/*.yaml]
|
||||
YAML[injection/*.yaml<br/>device/*.yaml + device/*/*.yaml<br/>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/<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:
|
||||
|
||||
@ -100,6 +100,12 @@ class SkillsLoader:
|
||||
"""
|
||||
Load all device skills from YAML files.
|
||||
|
||||
Supports two layouts:
|
||||
- Single file: device/<device>.yaml
|
||||
- Split directory: device/<device>/_base.yaml + device/<device>/<topic>.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.
|
||||
|
||||
@ -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()
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
348
tests/agent/test_skills_device_topics.py
Normal file
348
tests/agent/test_skills_device_topics.py
Normal file
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
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 <address>/<mask> <gateway>"
|
||||
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 <asn>"
|
||||
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 <prefix>/<len> <nexthop>"
|
||||
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 <asn>"
|
||||
# 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 <asn>"}},
|
||||
},
|
||||
},
|
||||
},
|
||||
"gns3_vpcs_telnet": {
|
||||
"name": "VPCS",
|
||||
"description": "VPCS test device",
|
||||
"category": "device",
|
||||
"config_commands": {"ip_config": {"syntax": "ip <address>/<mask>"}},
|
||||
},
|
||||
}
|
||||
)
|
||||
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 <asn>"
|
||||
|
||||
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 <address>/<mask>"
|
||||
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 <config_dir>/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)
|
||||
Loading…
x
Reference in New Issue
Block a user