diff --git a/README.md b/README.md index d39e324..4868faf 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ - **🎯 Dry Run Mode** - Preview changes before applying them - **✅ Validation & Schema** - Built-in YAML validation and IDE support - **🤖 MCP Integration** - Model Context Protocol support for AI-assisted development workflows +- **📚 Named Sources** - Save, inspect, validate, and reuse custom structure directories ## 🤔 Why structkit? @@ -79,6 +80,11 @@ structkit generate terraform-module ./my-terraform-module # List available structures structkit list +# Save and use a custom structures source +structkit sources add company ~/templates +structkit list --source company +structkit generate company/project/python ./app + # Validate a configuration structkit validate my-config.yaml diff --git a/docs/custom-structures.md b/docs/custom-structures.md index e9a3ecd..9c2424e 100644 --- a/docs/custom-structures.md +++ b/docs/custom-structures.md @@ -35,3 +35,55 @@ For this to work, you will need to set the path to the custom structures reposit ```sh structkit generate -s ~/path/to/custom-structures/structures file://.struct.yaml ./output ``` + +## Named Sources + +You can save frequently used custom structure directories as named sources with the `structkit sources` command group. Sources are stored in a user-level StructKit config file at `~/.config/structkit/sources.json` by default. Automation and tests can override that location with `STRUCTKIT_SOURCES_CONFIG`. + +```sh +structkit sources add company ~/path/to/custom-structures/structures +structkit sources list +structkit sources show company +structkit sources validate company +structkit sources remove company +``` + +Named sources currently support local filesystem directories. Remote URLs are reserved for future support. + +After a source is configured, select it explicitly with `--source`: + +```sh +structkit list --source company +structkit search --source company python +structkit info --source company project/python +structkit generate --source company project/python ./output +``` + +For `generate` and `info`, you can also prefix the structure name with a configured source name: + +```sh +structkit generate company/project/python ./output +``` + +### Precedence and compatibility + +Existing custom-structure behavior remains compatible: + +1. `--structures-path` and `STRUCTKIT_STRUCTURES_PATH` point commands at a custom directory. +2. If no structures path is set, `--source ` selects a named source. +3. If no structures path or `--source` is set, `generate` and `info` can resolve source-prefixed names such as `company/project/python`. +4. If none of the above are used, StructKit uses its bundled contrib structures. + +When a custom path or named source is selected, StructKit still falls back to bundled contrib structures for `list`, `search`, and `generate` in the same way as the existing `STRUCTKIT_STRUCTURES_PATH` workflow. + +### MCP tools + +The MCP server exposes the same source-management workflow for AI clients: + +- `list_sources` +- `add_source` +- `remove_source` +- `show_source` +- `validate_source` + +The structure tools also accept named sources. Pass `source` to `list_structures`, `get_structure_info`, or `generate_structure`, or use a source-prefixed structure name such as `company/project/python` with `get_structure_info` and `generate_structure`. diff --git a/structkit/commands/generate.py b/structkit/commands/generate.py index 37bee42..e57258c 100644 --- a/structkit/commands/generate.py +++ b/structkit/commands/generate.py @@ -5,6 +5,7 @@ from structkit.file_item import FileItem from structkit.completers import file_strategy_completer, structures_completer from structkit.template_renderer import TemplateRenderer +from structkit.sources import SourceConfigError, resolve_structures_path import subprocess @@ -24,6 +25,7 @@ def __init__(self, parser): help='Path to structure definitions (env: STRUCTKIT_STRUCTURES_PATH)', default=os.getenv('STRUCTKIT_STRUCTURES_PATH', None) ) + parser.add_argument('--source', type=str, help='Named source containing the structure definition') parser.add_argument('-n', '--input-store', type=str, help='Path to the input store (env: STRUCTKIT_INPUT_STORE)', default=os.getenv('STRUCTKIT_INPUT_STORE', '/tmp/structkit/input.json')) parser.add_argument('-d', '--dry-run', action='store_true', help='Perform a dry run without creating any files or directories') parser.add_argument('--diff', action='store_true', help='Show unified diffs for files that would change during dry-run or console output') @@ -161,6 +163,13 @@ def execute(self, args): # Load config to check for hooks config = None + try: + effective_structures_path, structure_definition = resolve_structures_path(args, args.structure_definition) + except SourceConfigError as e: + self.logger.error(f"❗ {e}") + return + args.structures_path = effective_structures_path + args.structure_definition = structure_definition config = self._load_yaml_config(args.structure_definition, args.structures_path) if config is None: return diff --git a/structkit/commands/info.py b/structkit/commands/info.py index db7dfe4..714c31a 100644 --- a/structkit/commands/info.py +++ b/structkit/commands/info.py @@ -3,6 +3,7 @@ import asyncio from structkit.commands import Command +from structkit.sources import SourceConfigError, resolve_structures_path # Info command class for exposing information about the structure @@ -16,6 +17,7 @@ def __init__(self, parser): '-s', '--structures-path', type=str, help='Path to structure definitions (env: STRUCTKIT_STRUCTURES_PATH)', default=os.getenv('STRUCTKIT_STRUCTURES_PATH', None) ) + parser.add_argument('--source', type=str, help='Named source containing the structure definition') parser.add_argument('--mcp', action='store_true', help='Enable MCP (Model Context Protocol) integration') parser.set_defaults(func=self.execute) @@ -33,11 +35,16 @@ def _get_info(self, args): with open(args.structure_definition[7:], 'r') as f: config = yaml.safe_load(f) else: - if args.structures_path is None: + try: + effective_structures_path, structure_definition = resolve_structures_path(args, args.structure_definition) + except SourceConfigError as e: + self.logger.error(f"❗ {e}") + return + if effective_structures_path is None: this_file = os.path.dirname(os.path.realpath(__file__)) - file_path = os.path.join(this_file, "..", "contribs", f"{args.structure_definition}.yaml") + file_path = os.path.join(this_file, "..", "contribs", f"{structure_definition}.yaml") else: - file_path = os.path.join(args.structures_path, f"{args.structure_definition}.yaml") + file_path = os.path.join(effective_structures_path, f"{structure_definition}.yaml") # show error if file is not found if not os.path.exists(file_path): self.logger.error(f"❗ File not found: {file_path}") diff --git a/structkit/commands/list.py b/structkit/commands/list.py index 41130dc..96f2d13 100644 --- a/structkit/commands/list.py +++ b/structkit/commands/list.py @@ -1,4 +1,5 @@ from structkit.commands import Command +from structkit.sources import SourceConfigError, resolve_structures_path import os import asyncio @@ -12,6 +13,7 @@ def __init__(self, parser): '-s', '--structures-path', type=str, help='Path to structure definitions (env: STRUCTKIT_STRUCTURES_PATH)', default=os.getenv('STRUCTKIT_STRUCTURES_PATH', None) ) + parser.add_argument('--source', type=str, help='Named source to list') parser.add_argument('--names-only', action='store_true', help='Print only structure names, one per line (for shell completion)') parser.add_argument('--mcp', action='store_true', help='Enable MCP (Model Context Protocol) integration') parser.set_defaults(func=self.execute) @@ -27,8 +29,14 @@ def _list_structures(self, args): this_file = os.path.dirname(os.path.realpath(__file__)) contribs_path = os.path.join(this_file, "..", "contribs") - if args.structures_path: - final_path = args.structures_path + try: + effective_structures_path, _ = resolve_structures_path(args) + except SourceConfigError as e: + self.logger.error(f"❗ {e}") + return + + if effective_structures_path: + final_path = effective_structures_path paths_to_list = [final_path, contribs_path] else: paths_to_list = [contribs_path] diff --git a/structkit/commands/mcp.py b/structkit/commands/mcp.py index dc603c1..702cb9e 100644 --- a/structkit/commands/mcp.py +++ b/structkit/commands/mcp.py @@ -50,6 +50,11 @@ def execute(self, args): print(" - get_structure_info: Get detailed information about a structure") print(" - generate_structure: Generate structures with various options") print(" - validate_structure: Validate structure configuration files") + print(" - list_sources: List configured custom structure sources") + print(" - add_source: Add or replace a local custom structure source") + print(" - remove_source: Remove a configured custom structure source") + print(" - show_source: Show a configured custom structure source") + print(" - validate_source: Validate a configured custom structure source") print("\nExamples:") print(" structkit mcp --server --transport stdio --debug") print(" structkit mcp --server --transport http --host 127.0.0.1 --port 9000 --path /mcp --uvicorn-log-level debug") diff --git a/structkit/commands/search.py b/structkit/commands/search.py index 137a901..3e22e2e 100644 --- a/structkit/commands/search.py +++ b/structkit/commands/search.py @@ -2,6 +2,7 @@ import yaml from structkit.commands import Command +from structkit.sources import SourceConfigError, resolve_structures_path class SearchCommand(Command): @@ -21,6 +22,11 @@ def __init__(self, parser): help='Path to structure definitions (env: STRUCTKIT_STRUCTURES_PATH)', default=os.getenv('STRUCTKIT_STRUCTURES_PATH', None), ) + parser.add_argument( + '--source', + type=str, + help='Named source to search', + ) parser.add_argument( '--names-only', action='store_true', @@ -36,8 +42,14 @@ def _search_structures(self, args): this_file = os.path.dirname(os.path.realpath(__file__)) contribs_path = os.path.join(this_file, '..', 'contribs') - if args.structures_path: - paths_to_search = [(args.structures_path, False), (contribs_path, True)] + try: + effective_structures_path, _ = resolve_structures_path(args) + except SourceConfigError as e: + self.logger.error(f"❗ {e}") + return + + if effective_structures_path: + paths_to_search = [(effective_structures_path, False), (contribs_path, True)] else: paths_to_search = [(contribs_path, True)] diff --git a/structkit/commands/sources.py b/structkit/commands/sources.py new file mode 100644 index 0000000..305ab1a --- /dev/null +++ b/structkit/commands/sources.py @@ -0,0 +1,85 @@ +from structkit.commands import Command +from structkit.sources import ( + SourceConfigError, + add_source, + get_source_path, + read_sources, + remove_source, + validate_source, +) + + +class SourcesCommand(Command): + """Manage named custom structure sources.""" + + def __init__(self, parser): + super().__init__(parser) + parser.description = "Manage named custom structure sources" + subparsers = parser.add_subparsers(dest="sources_command") + + list_parser = subparsers.add_parser("list", help="List configured sources") + list_parser.set_defaults(func=self.execute) + + add_parser = subparsers.add_parser("add", help="Add or replace a local source") + add_parser.add_argument("name", type=str, help="Source name") + add_parser.add_argument("path_or_url", type=str, help="Local path to structure definitions") + add_parser.set_defaults(func=self.execute) + + remove_parser = subparsers.add_parser("remove", help="Remove a configured source") + remove_parser.add_argument("name", type=str, help="Source name") + remove_parser.set_defaults(func=self.execute) + + show_parser = subparsers.add_parser("show", help="Show one configured source") + show_parser.add_argument("name", type=str, help="Source name") + show_parser.set_defaults(func=self.execute) + + validate_parser = subparsers.add_parser("validate", help="Validate one configured source") + validate_parser.add_argument("name", type=str, help="Source name") + validate_parser.set_defaults(func=self.execute) + + parser.set_defaults(func=self.execute) + + def execute(self, args): + command = getattr(args, "sources_command", None) + if command is None: + self.parser.print_help() + return + + try: + if command == "list": + self._list_sources() + elif command == "add": + self._add_source(args.name, args.path_or_url) + elif command == "remove": + self._remove_source(args.name) + elif command == "show": + self._show_source(args.name) + elif command == "validate": + self._validate_source(args.name) + except SourceConfigError as e: + self.logger.error(f"❗ {e}") + + def _list_sources(self): + sources = read_sources() + if not sources: + print("No sources configured") + return + print("📚 Configured structure sources\n") + for name in sorted(sources): + print(f" - {name}: {sources[name]}") + + def _add_source(self, name, path_or_url): + source_path = add_source(name, path_or_url) + print(f"Added source '{name}': {source_path}") + + def _remove_source(self, name): + source_path = remove_source(name) + print(f"Removed source '{name}': {source_path}") + + def _show_source(self, name): + source_path = get_source_path(name) + print(f"{name}: {source_path}") + + def _validate_source(self, name): + source_path = validate_source(name) + print(f"Source '{name}' is valid: {source_path}") diff --git a/structkit/main.py b/structkit/main.py index 73e499c..7f6167b 100644 --- a/structkit/main.py +++ b/structkit/main.py @@ -10,6 +10,7 @@ from structkit.commands.search import SearchCommand from structkit.commands.generate_schema import GenerateSchemaCommand from structkit.commands.mcp import MCPCommand +from structkit.commands.sources import SourcesCommand from structkit.logging_config import configure_logging # Optional dependency: shtab for static shell completion generation @@ -38,6 +39,7 @@ def get_parser(): SearchCommand(subparsers.add_parser('search', help='Search available structures by keyword')) GenerateSchemaCommand(subparsers.add_parser('generate-schema', help='Generate JSON schema for available structures')) MCPCommand(subparsers.add_parser('mcp', help='MCP (Model Context Protocol) support')) + SourcesCommand(subparsers.add_parser('sources', help='Manage named custom structure sources')) # init to create a basic .struct.yaml from structkit.commands.init import InitCommand diff --git a/structkit/mcp_server.py b/structkit/mcp_server.py index e18eba4..8adae9f 100644 --- a/structkit/mcp_server.py +++ b/structkit/mcp_server.py @@ -6,18 +6,29 @@ 2. Getting detailed information about structures 3. Generating structures with various options 4. Validating structure configurations +5. Managing named custom structure sources """ import asyncio import logging import os import sys import yaml +from types import SimpleNamespace from typing import Any, Dict, Optional from fastmcp import FastMCP from structkit.commands.generate import GenerateCommand from structkit.commands.validate import ValidateCommand +from structkit.sources import ( + SourceConfigError, + add_source as add_config_source, + get_source_path, + read_sources, + remove_source as remove_config_source, + resolve_structures_path, + validate_source as validate_config_source, +) from structkit import __version__ @@ -32,13 +43,20 @@ def __init__(self): # ===================== # Tool logic (transport-agnostic) # ===================== - def _list_structures_logic(self, structures_path: Optional[str] = None) -> str: + def _list_structures_logic(self, structures_path: Optional[str] = None, source: Optional[str] = None) -> str: this_file = os.path.dirname(os.path.realpath(__file__)) contribs_path = os.path.join(this_file, "contribs") + try: + effective_structures_path, _ = resolve_structures_path( + SimpleNamespace(structures_path=structures_path, source=source) + ) + except SourceConfigError as e: + return f"❗ {e}" + paths_to_list = [contribs_path] - if structures_path: - paths_to_list = [structures_path, contribs_path] + if effective_structures_path: + paths_to_list = [effective_structures_path, contribs_path] all_structures = set() for path in paths_to_list: @@ -56,7 +74,12 @@ def _list_structures_logic(self, structures_path: Optional[str] = None) -> str: result_text += "\n\nNote: Structures with '+' sign are custom structures" return result_text - def _get_structure_info_logic(self, structure_name: Optional[str], structures_path: Optional[str] = None) -> str: + def _get_structure_info_logic( + self, + structure_name: Optional[str], + structures_path: Optional[str] = None, + source: Optional[str] = None, + ) -> str: if not structure_name: return "Error: structure_name is required" @@ -64,9 +87,15 @@ def _get_structure_info_logic(self, structure_name: Optional[str], structures_pa if structure_name.startswith("file://") and structure_name.endswith(".yaml"): file_path = structure_name[7:] else: + try: + effective_structures_path, resolved_structure_name = resolve_structures_path( + SimpleNamespace(structures_path=structures_path, source=source), structure_name + ) + except SourceConfigError as e: + return f"❗ {e}" this_file = os.path.dirname(os.path.realpath(__file__)) - base = structures_path or os.path.join(this_file, "contribs") - file_path = os.path.join(base, f"{structure_name}.yaml") + base = effective_structures_path or os.path.join(this_file, "contribs") + file_path = os.path.join(base, f"{resolved_structure_name}.yaml") if not os.path.exists(file_path): return f"❗ Structure not found: {file_path}" @@ -119,6 +148,7 @@ def _generate_structure_logic( dry_run: bool = False, mappings: Optional[Dict[str, str]] = None, structures_path: Optional[str] = None, + source: Optional[str] = None, ) -> str: class Args: pass @@ -128,6 +158,7 @@ class Args: args.output = "console" if output == "console" else "file" args.dry_run = dry_run args.structures_path = structures_path + args.source = source args.vars = None args.mappings_file = None args.backup = None @@ -167,6 +198,54 @@ class Args: return f"Dry run completed for structure '{structure_definition}' at '{base_path}'" return f"Structure '{structure_definition}' generated successfully at '{base_path}'" + def _list_sources_logic(self) -> str: + sources = read_sources() + if not sources: + return "No sources configured" + + lines = ["📚 Configured structure sources\n"] + for name in sorted(sources): + lines.append(f" - {name}: {sources[name]}\n") + return "".join(lines).rstrip() + + def _add_source_logic(self, name: Optional[str], path_or_url: Optional[str]) -> str: + if not name: + return "Error: name is required" + if not path_or_url: + return "Error: path_or_url is required" + try: + source_path = add_config_source(name, path_or_url) + except SourceConfigError as e: + return f"❗ {e}" + return f"Added source '{name}': {source_path}" + + def _remove_source_logic(self, name: Optional[str]) -> str: + if not name: + return "Error: name is required" + try: + source_path = remove_config_source(name) + except SourceConfigError as e: + return f"❗ {e}" + return f"Removed source '{name}': {source_path}" + + def _show_source_logic(self, name: Optional[str]) -> str: + if not name: + return "Error: name is required" + try: + source_path = get_source_path(name) + except SourceConfigError as e: + return f"❗ {e}" + return f"{name}: {source_path}" + + def _validate_source_logic(self, name: Optional[str]) -> str: + if not name: + return "Error: name is required" + try: + source_path = validate_config_source(name) + except SourceConfigError as e: + return f"❗ {e}" + return f"Source '{name}' is valid: {source_path}" + def _validate_structure_logic(self, yaml_file: Optional[str]) -> str: if not yaml_file: return "Error: yaml_file is required" @@ -198,19 +277,23 @@ class Args: # ===================== def _register_tools(self): @self.app.tool(name="list_structures", description="List all available structure definitions") - async def list_structures(structures_path: Optional[str] = None) -> str: - self.logger.debug(f"MCP request: list_structures args={{'structures_path': {structures_path!r}}}") - result = self._list_structures_logic(structures_path) + async def list_structures(structures_path: Optional[str] = None, source: Optional[str] = None) -> str: + self.logger.debug( + f"MCP request: list_structures args={{'structures_path': {structures_path!r}, 'source': {source!r}}}" + ) + result = self._list_structures_logic(structures_path, source) preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" self.logger.debug(f"MCP response: list_structures len={len(result)} preview=\n{preview}") return result @self.app.tool(name="get_structure_info", description="Get detailed information about a specific structure") - async def get_structure_info(structure_name: str, structures_path: Optional[str] = None) -> str: + async def get_structure_info( + structure_name: str, structures_path: Optional[str] = None, source: Optional[str] = None + ) -> str: self.logger.debug( - f"MCP request: get_structure_info args={{'structure_name': {structure_name!r}, 'structures_path': {structures_path!r}}}" + f"MCP request: get_structure_info args={{'structure_name': {structure_name!r}, 'structures_path': {structures_path!r}, 'source': {source!r}}}" ) - result = self._get_structure_info_logic(structure_name, structures_path) + result = self._get_structure_info_logic(structure_name, structures_path, source) preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" self.logger.debug(f"MCP response: get_structure_info len={len(result)} preview=\n{preview}") return result @@ -223,6 +306,7 @@ async def generate_structure( dry_run: bool = False, mappings: Optional[Dict[str, str]] = None, structures_path: Optional[str] = None, + source: Optional[str] = None, ) -> str: self.logger.debug( "MCP request: generate_structure args=%s", @@ -233,6 +317,7 @@ async def generate_structure( "dry_run": dry_run, "mappings": mappings, "structures_path": structures_path, + "source": source, }, ) result = self._generate_structure_logic( @@ -242,11 +327,52 @@ async def generate_structure( dry_run, mappings, structures_path, + source, ) preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" self.logger.debug(f"MCP response: generate_structure len={len(result)} preview=\n{preview}") return result + @self.app.tool(name="list_sources", description="List configured custom structure sources") + async def list_sources() -> str: + self.logger.debug("MCP request: list_sources args={}") + result = self._list_sources_logic() + preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" + self.logger.debug(f"MCP response: list_sources len={len(result)} preview=\n{preview}") + return result + + @self.app.tool(name="add_source", description="Add or replace a local custom structure source") + async def add_source(name: str, path_or_url: str) -> str: + self.logger.debug(f"MCP request: add_source args={{'name': {name!r}, 'path_or_url': {path_or_url!r}}}") + result = self._add_source_logic(name, path_or_url) + preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" + self.logger.debug(f"MCP response: add_source len={len(result)} preview=\n{preview}") + return result + + @self.app.tool(name="remove_source", description="Remove a configured custom structure source") + async def remove_source(name: str) -> str: + self.logger.debug(f"MCP request: remove_source args={{'name': {name!r}}}") + result = self._remove_source_logic(name) + preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" + self.logger.debug(f"MCP response: remove_source len={len(result)} preview=\n{preview}") + return result + + @self.app.tool(name="show_source", description="Show a configured custom structure source") + async def show_source(name: str) -> str: + self.logger.debug(f"MCP request: show_source args={{'name': {name!r}}}") + result = self._show_source_logic(name) + preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" + self.logger.debug(f"MCP response: show_source len={len(result)} preview=\n{preview}") + return result + + @self.app.tool(name="validate_source", description="Validate a configured custom structure source") + async def validate_source(name: str) -> str: + self.logger.debug(f"MCP request: validate_source args={{'name': {name!r}}}") + result = self._validate_source_logic(name) + preview = result if len(result) <= 1000 else result[:1000] + f"... [truncated {len(result)-1000} chars]" + self.logger.debug(f"MCP response: validate_source len={len(result)} preview=\n{preview}") + return result + @self.app.tool(name="validate_structure", description="Validate a structure configuration YAML file") async def validate_structure(yaml_file: str) -> str: self.logger.debug(f"MCP request: validate_structure args={{'yaml_file': {yaml_file!r}}}") @@ -323,8 +449,9 @@ async def _handle_get_structure_info(self, params: Dict[str, Any]): """Compatibility method for tests that expect MCP-style responses.""" structure_name = params.get('structure_name') structures_path = params.get('structures_path') + source = params.get('source') - result_text = self._get_structure_info_logic(structure_name, structures_path) + result_text = self._get_structure_info_logic(structure_name, structures_path, source) # Mock MCP response structure class MockContent: diff --git a/structkit/sources.py b/structkit/sources.py new file mode 100644 index 0000000..abb6587 --- /dev/null +++ b/structkit/sources.py @@ -0,0 +1,146 @@ +"""Utilities for managing named StructKit structure sources.""" + +import json +import os +from pathlib import Path + + +CONFIG_ENV_VAR = "STRUCTKIT_SOURCES_CONFIG" +DEFAULT_CONFIG_PATH = Path.home() / ".config" / "structkit" / "sources.json" + + +class SourceConfigError(ValueError): + """Raised when source configuration is invalid.""" + + +def get_sources_config_path(): + """Return the user-level source config path. + + STRUCTKIT_SOURCES_CONFIG is primarily useful for tests and automation. Normal + users get a platform-neutral file under ~/.config/structkit/sources.json. + """ + configured_path = os.getenv(CONFIG_ENV_VAR) + if configured_path: + return Path(configured_path).expanduser() + return DEFAULT_CONFIG_PATH + + +def normalize_source_path(path_or_url): + """Normalize a local source path for storage.""" + if "://" in path_or_url: + raise SourceConfigError("Only local filesystem source paths are supported right now") + return str(Path(path_or_url).expanduser().resolve()) + + +def read_sources(config_path=None): + """Read source mappings from disk. + + Missing config files are treated as an empty mapping for first-run usage. + """ + path = Path(config_path).expanduser() if config_path else get_sources_config_path() + if not path.exists(): + return {} + with open(path, "r") as f: + data = json.load(f) + if isinstance(data, dict) and isinstance(data.get("sources"), dict): + return data["sources"] + if isinstance(data, dict): + return data + raise SourceConfigError("Source config must contain a JSON object") + + +def write_sources(sources, config_path=None): + """Write source mappings to disk.""" + path = Path(config_path).expanduser() if config_path else get_sources_config_path() + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as f: + json.dump({"sources": sources}, f, indent=2, sort_keys=True) + f.write("\n") + + +def add_source(name, path_or_url, config_path=None): + """Add or replace a named local source.""" + validate_source_name(name) + normalized_path = normalize_source_path(path_or_url) + if not os.path.isdir(normalized_path): + raise SourceConfigError(f"Source path does not exist or is not a directory: {normalized_path}") + sources = read_sources(config_path) + sources[name] = normalized_path + write_sources(sources, config_path) + return normalized_path + + +def remove_source(name, config_path=None): + """Remove a named source and return its path.""" + sources = read_sources(config_path) + if name not in sources: + raise SourceConfigError(f"Unknown source: {name}") + removed_path = sources.pop(name) + write_sources(sources, config_path) + return removed_path + + +def get_source_path(name, config_path=None): + """Return the configured path for a named source.""" + sources = read_sources(config_path) + if name not in sources: + raise SourceConfigError(f"Unknown source: {name}") + return sources[name] + + +def validate_source(name, config_path=None): + """Validate that a named source points to an existing directory.""" + source_path = get_source_path(name, config_path) + if not os.path.isdir(source_path): + raise SourceConfigError(f"Source path does not exist or is not a directory: {source_path}") + return source_path + + +def validate_source_name(name): + """Validate a source name used on the command line and in config.""" + if not name or name in {".", ".."}: + raise SourceConfigError("Source name cannot be empty, '.' or '..'") + if "/" in name or "\\" in name: + raise SourceConfigError("Source name cannot contain path separators") + + +def split_source_prefix(structure_definition, config_path=None): + """Split source-prefixed definitions like 'company/project/python'. + + Returns (source_name, remaining_definition, source_path) when the first path + segment is a configured source name. Otherwise returns (None, + structure_definition, None). + """ + if structure_definition.startswith(("file://", "/", "./", "../")): + return None, structure_definition, None + if "/" not in structure_definition: + return None, structure_definition, None + source_name, rest = structure_definition.split("/", 1) + sources = read_sources(config_path) + if source_name in sources: + return source_name, rest, validate_source(source_name, config_path) + return None, structure_definition, None + + +def resolve_structures_path(args, structure_definition=None): + """Resolve the effective structures path for commands. + + Precedence is: + 1. --structures-path or STRUCTKIT_STRUCTURES_PATH (already present on args) + 2. --source NAME + 3. source-prefixed structure names, for example company/project/python + 4. bundled contrib structures + """ + if getattr(args, "structures_path", None): + return args.structures_path, structure_definition + + source_name = getattr(args, "source", None) + if source_name: + return validate_source(source_name), structure_definition + + if structure_definition: + _, stripped_definition, source_path = split_source_prefix(structure_definition) + if source_path: + return source_path, stripped_definition + + return None, structure_definition diff --git a/tests/test_mcp_integration.py b/tests/test_mcp_integration.py index c3d0c7b..bb89565 100644 --- a/tests/test_mcp_integration.py +++ b/tests/test_mcp_integration.py @@ -43,6 +43,66 @@ def test_generate_structure_logic(self): ) self.assertIsInstance(text, str) + + def test_source_management_logic(self): + with tempfile.TemporaryDirectory() as temp_dir: + config_path = os.path.join(temp_dir, 'sources.json') + source_dir = os.path.join(temp_dir, 'source') + os.makedirs(source_dir) + + from unittest.mock import patch + with patch.dict(os.environ, {'STRUCTKIT_SOURCES_CONFIG': config_path}): + self.assertEqual(self.server._list_sources_logic(), 'No sources configured') + + added = self.server._add_source_logic('company', source_dir) + self.assertIn("Added source 'company'", added) + + listed = self.server._list_sources_logic() + self.assertIn('company', listed) + self.assertIn(source_dir, listed) + + shown = self.server._show_source_logic('company') + self.assertIn(source_dir, shown) + + validated = self.server._validate_source_logic('company') + self.assertIn("Source 'company' is valid", validated) + + removed = self.server._remove_source_logic('company') + self.assertIn("Removed source 'company'", removed) + + def test_source_management_logic_errors(self): + with tempfile.TemporaryDirectory() as temp_dir: + config_path = os.path.join(temp_dir, 'sources.json') + + from unittest.mock import patch + with patch.dict(os.environ, {'STRUCTKIT_SOURCES_CONFIG': config_path}): + self.assertIn('name is required', self.server._add_source_logic(None, temp_dir)) + self.assertIn('path_or_url is required', self.server._add_source_logic('company', None)) + self.assertIn('Only local filesystem', self.server._add_source_logic('remote', 'https://example.com/templates.git')) + self.assertIn('Unknown source', self.server._show_source_logic('missing')) + + def test_structure_logic_accepts_named_source(self): + with tempfile.TemporaryDirectory() as temp_dir: + config_path = os.path.join(temp_dir, 'sources.json') + source_dir = os.path.join(temp_dir, 'source') + os.makedirs(os.path.join(source_dir, 'project')) + structure_file = os.path.join(source_dir, 'project', 'python.yaml') + with open(structure_file, 'w') as f: + yaml.dump({'description': 'Custom Python structure', 'files': []}, f) + + from unittest.mock import patch + with patch.dict(os.environ, {'STRUCTKIT_SOURCES_CONFIG': config_path}): + self.server._add_source_logic('company', source_dir) + + listed = self.server._list_structures_logic(source='company') + self.assertIn('+ project/python', listed) + + explicit = self.server._get_structure_info_logic('project/python', source='company') + self.assertIn('Custom Python structure', explicit) + + prefixed = self.server._get_structure_info_logic('company/project/python') + self.assertIn('Custom Python structure', prefixed) + def test_validate_structure_logic(self): # Missing yaml_file text = self.server._validate_structure_logic(None) diff --git a/tests/test_sources_command.py b/tests/test_sources_command.py new file mode 100644 index 0000000..a943aa0 --- /dev/null +++ b/tests/test_sources_command.py @@ -0,0 +1,160 @@ +import argparse +import json +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +from structkit.commands.generate import GenerateCommand +from structkit.commands.list import ListCommand +from structkit.commands.sources import SourcesCommand +from structkit.sources import ( + SourceConfigError, + add_source, + get_source_path, + read_sources, + remove_source, + resolve_structures_path, + validate_source, +) + + +def test_source_config_read_write_add_show_remove(tmp_path): + config_path = tmp_path / "sources.json" + structures_dir = tmp_path / "templates" + structures_dir.mkdir() + + added_path = add_source("company", str(structures_dir), config_path) + + assert added_path == str(structures_dir.resolve()) + assert read_sources(config_path) == {"company": str(structures_dir.resolve())} + assert get_source_path("company", config_path) == str(structures_dir.resolve()) + assert validate_source("company", config_path) == str(structures_dir.resolve()) + assert remove_source("company", config_path) == str(structures_dir.resolve()) + assert read_sources(config_path) == {} + + +def test_add_source_rejects_invalid_local_path(tmp_path): + with pytest.raises(SourceConfigError, match="not a directory"): + add_source("missing", str(tmp_path / "missing"), tmp_path / "sources.json") + + +def test_add_source_rejects_remote_urls(tmp_path): + with pytest.raises(SourceConfigError, match="Only local filesystem"): + add_source("remote", "https://example.com/templates.git", tmp_path / "sources.json") + + +def test_missing_source_errors(tmp_path): + config_path = tmp_path / "sources.json" + config_path.write_text(json.dumps({"sources": {}})) + + with pytest.raises(SourceConfigError, match="Unknown source"): + get_source_path("missing", config_path) + + +def test_resolve_structures_path_uses_structures_path_before_source(tmp_path): + source_dir = tmp_path / "source" + source_dir.mkdir() + cli_dir = tmp_path / "cli" + cli_dir.mkdir() + config_path = tmp_path / "sources.json" + add_source("company", str(source_dir), config_path) + args = argparse.Namespace(structures_path=str(cli_dir), source="company") + + with patch.dict(os.environ, {"STRUCTKIT_SOURCES_CONFIG": str(config_path)}): + path, structure = resolve_structures_path(args, "project/python") + + assert path == str(cli_dir) + assert structure == "project/python" + + +def test_resolve_structures_path_uses_named_source(tmp_path): + source_dir = tmp_path / "source" + source_dir.mkdir() + config_path = tmp_path / "sources.json" + add_source("company", str(source_dir), config_path) + args = argparse.Namespace(structures_path=None, source="company") + + with patch.dict(os.environ, {"STRUCTKIT_SOURCES_CONFIG": str(config_path)}): + path, structure = resolve_structures_path(args, "project/python") + + assert path == str(source_dir.resolve()) + assert structure == "project/python" + + +def test_resolve_structures_path_uses_source_prefix(tmp_path): + source_dir = tmp_path / "source" + source_dir.mkdir() + config_path = tmp_path / "sources.json" + add_source("company", str(source_dir), config_path) + args = argparse.Namespace(structures_path=None, source=None) + + with patch.dict(os.environ, {"STRUCTKIT_SOURCES_CONFIG": str(config_path)}): + path, structure = resolve_structures_path(args, "company/project/python") + + assert path == str(source_dir.resolve()) + assert structure == "project/python" + + +def test_sources_command_list_add_show_validate_remove(tmp_path, capsys): + config_path = tmp_path / "sources.json" + source_dir = tmp_path / "source" + source_dir.mkdir() + parser = argparse.ArgumentParser() + SourcesCommand(parser) + + with patch.dict(os.environ, {"STRUCTKIT_SOURCES_CONFIG": str(config_path)}): + args = parser.parse_args(["add", "company", str(source_dir)]) + args.func(args) + assert "Added source 'company'" in capsys.readouterr().out + + args = parser.parse_args(["list"]) + args.func(args) + assert "company" in capsys.readouterr().out + + args = parser.parse_args(["show", "company"]) + args.func(args) + assert str(source_dir.resolve()) in capsys.readouterr().out + + args = parser.parse_args(["validate", "company"]) + args.func(args) + assert "is valid" in capsys.readouterr().out + + args = parser.parse_args(["remove", "company"]) + args.func(args) + assert "Removed source 'company'" in capsys.readouterr().out + + +def test_list_command_uses_named_source(tmp_path, capsys): + config_path = tmp_path / "sources.json" + source_dir = tmp_path / "source" + (source_dir / "project").mkdir(parents=True) + (source_dir / "project" / "python.yaml").write_text("files: []\n") + add_source("company", str(source_dir), config_path) + parser = argparse.ArgumentParser() + ListCommand(parser) + + with patch.dict(os.environ, {"STRUCTKIT_SOURCES_CONFIG": str(config_path)}): + args = parser.parse_args(["--source", "company", "--names-only"]) + args.func(args) + + assert "project/python" in capsys.readouterr().out + + +def test_generate_command_uses_source_prefix(tmp_path): + config_path = tmp_path / "sources.json" + source_dir = tmp_path / "source" + (source_dir / "project").mkdir(parents=True) + (source_dir / "project" / "python.yaml").write_text("files: []\n") + output_dir = tmp_path / "output" + add_source("company", str(source_dir), config_path) + parser = argparse.ArgumentParser() + command = GenerateCommand(parser) + + with patch.dict(os.environ, {"STRUCTKIT_SOURCES_CONFIG": str(config_path)}): + args = parser.parse_args(["company/project/python", str(output_dir)]) + command.execute(args) + + assert args.structures_path == str(source_dir.resolve()) + assert args.structure_definition == "project/python"