diff --git a/semantica/cli.py b/semantica/cli.py index c5c6de9f..0d001b81 100644 --- a/semantica/cli.py +++ b/semantica/cli.py @@ -5,51 +5,358 @@ This module provides the command-line interface for the Semantica framework, enabling users to interact with the framework via terminal commands. """ +import json +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Sequence + +import yaml + import click from rich.console import Console from rich.table import Table from . import __version__ -from .core.orchestrator import Semantica +from .core.config_manager import Config, ConfigManager +from .utils.exceptions import SemanticaError from .utils.logging import setup_logging +if TYPE_CHECKING: + from .core.orchestrator import Semantica + console = Console() -@click.group() -@click.version_option(version=__version__) -def main(): - """Semantica - Semantic Layer & Knowledge Engineering Framework""" - setup_logging() -@main.command() -def info(): - """Display information about Semantica.""" - console.print(f"[bold blue]Semantica Framework[/bold blue] v{__version__}") - console.print("A comprehensive Python framework for transforming unstructured data into semantic layers.") - - table = Table(title="Framework Components") - table.add_column("Component", style="cyan") - table.add_column("Status", style="green") - - table.add_row("Core Orchestrator", "Active") - table.add_row("Knowledge Graph Engine", "Active") - table.add_row("Pipeline Execution", "Active") - table.add_row("Vector Store Integration", "Active") - - console.print(table) +@dataclass +class CLIContext: + """Shared runtime context for all CLI commands.""" -@main.command() -@click.option("--source", "-s", multiple=True, help="Data sources to process.") -@click.option("--config", "-c", help="Path to configuration file.") -def build(source, config): - """Build a knowledge base from sources.""" - console.print(f"Initializing Semantica with {len(source)} sources...") + config_path: Optional[str] + config: Config + log_level: str + log_level_override: Optional[str] = None + framework: Optional["Semantica"] = None + + +def _require_ctx(cli_ctx: Optional[CLIContext]) -> CLIContext: + """Guard against uninitialized CLI context. + + Under normal Click operation (standalone_mode=True) ctx.obj is always set + before a subcommand runs. This guard protects embedded/library usage where + standalone_mode=False might leave ctx.obj as None. + """ + if cli_ctx is None: + raise click.ClickException( + "CLI context is uninitialized — this is a bug, please report it." + ) + return cli_ctx + + +def _run_with_error_handling(action: Callable[[], None]) -> None: + """Run a CLI action with consistent user-facing error formatting.""" try: - framework = Semantica(config=config) - # framework.build_knowledge_base(sources=list(source)) - console.print("[bold green]Success:[/bold green] Knowledge base construction initiated.") - except Exception as e: - console.print(f"[bold red]Error:[/bold red] {str(e)}") + action() + except click.ClickException: + raise + except SemanticaError as exc: + raise click.ClickException(str(exc)) from exc + except Exception as exc: + raise click.ClickException(f"Unexpected error: {exc}") from exc + + +def _load_config_data(file_path: Path) -> Dict[str, Any]: + """Load and validate raw YAML/JSON config data.""" + suffix = file_path.suffix.lower() + try: + if suffix in (".yaml", ".yml"): + with file_path.open("r", encoding="utf-8") as handle: + config_data = yaml.safe_load(handle) + elif suffix == ".json": + with file_path.open("r", encoding="utf-8") as handle: + config_data = json.load(handle) + else: + raise click.ClickException( + "Unsupported configuration file format: " + f"{suffix}. Supported formats: .yaml, .yml, .json" + ) + except (json.JSONDecodeError, yaml.YAMLError, UnicodeDecodeError) as exc: + raise click.ClickException( + f"Failed to parse configuration file '{file_path}': {exc}" + ) from exc + + if config_data is None: + config_data = {} + + if not isinstance(config_data, dict): + raise click.ClickException( + "Configuration file must contain a mapping/object at the root." + ) + + return config_data + + +def _build_runtime_config( + config_path: Optional[str], + log_level: Optional[str], +) -> Config: + """Resolve CLI config from file plus global flag overrides.""" + config_manager = ConfigManager() + + if config_path: + config_data = _load_config_data(Path(config_path)) + else: + config_data = {} + + logging_config = config_data.get("logging") + if logging_config is not None and not isinstance(logging_config, dict): + raise click.ClickException( + "Logging configuration section must contain a mapping/object." + ) + + if log_level: + if logging_config is None: + logging_config = {} + config_data["logging"] = logging_config + + logging_config["level"] = log_level.upper() + + # Keep validation disabled at CLI bootstrap to avoid blocking unrelated commands. + return config_manager.load_from_dict(config_data, validate=False) + + +def _get_framework(cli_ctx: CLIContext) -> "Semantica": + """Lazily initialize framework only when a command needs it.""" + if cli_ctx.framework is None: + from .core.orchestrator import Semantica + + cli_ctx.framework = Semantica(config=cli_ctx.config.to_dict()) + return cli_ctx.framework + + +def _run_build(cli_ctx: CLIContext, sources: Sequence[str]) -> None: + """Thin wrapper around existing build orchestration flow. + + build_knowledge_base is expected to return a dict of the form:: + + { + "statistics": { + "sources_processed": , + ... + }, + ... + } + + Both the top-level dict and the "statistics" key are optional; the + function degrades gracefully when either is absent or None. + """ + if not sources: + raise click.UsageError( + "At least one source is required. Use --source/-s one or more times." + ) + + framework = _get_framework(cli_ctx) + console.print(f"Initializing Semantica with {len(sources)} sources...") + result = framework.build_knowledge_base(sources=list(sources)) + + stats = result.get("statistics", {}) if isinstance(result, dict) else {} + processed = stats.get("sources_processed") + if processed is not None: + console.print( + "[bold green]Success:[/bold green] Knowledge base build completed " + f"for {processed} source(s)." + ) + else: + console.print( + "[bold green]Success:[/bold green] Knowledge base build completed." + ) + + +def _run_build_command( + cli_ctx: CLIContext, + source: Sequence[str], + command_config_path: Optional[str], +) -> None: + """Execute build command path with optional command-level config override. + + When a per-command config file is supplied, the logging configuration from + that file is re-applied (setup_logging clears existing handlers before + adding new ones, so there is no handler accumulation risk). + """ + if command_config_path: + cmd_config = _build_runtime_config( + command_config_path, cli_ctx.log_level_override + ) + # Re-apply logging so the command-level logging section takes effect. + setup_logging(config=cmd_config.get("logging", {})) + command_ctx = CLIContext( + config_path=command_config_path, + config=cmd_config, + log_level=cli_ctx.log_level, + # Preserve the global --log-level override so nested config + # lookups further down the call chain respect it. + log_level_override=cli_ctx.log_level_override, + ) + _run_build(command_ctx, source) + else: + _run_build(cli_ctx, source) + + +@click.group(context_settings={"help_option_names": ["-h", "--help"]}) +@click.version_option(version=__version__) +@click.option( + "--config", + "config_path", + type=click.Path(exists=True, dir_okay=False, resolve_path=True, path_type=str), + default=None, + help="Path to YAML/JSON config file.", +) +@click.option( + "--log-level", + type=click.Choice( + ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"], + case_sensitive=False, + ), + default=None, + help="Override logging level for this CLI invocation.", +) +@click.pass_context +def main(ctx: click.Context, config_path: Optional[str], log_level: Optional[str]): + """Semantica - Semantic Layer & Knowledge Engineering Framework""" + try: + config = _build_runtime_config(config_path=config_path, log_level=log_level) + setup_logging(config=config.get("logging", {})) + effective_log_level = config.get("logging.level", "INFO") + ctx.obj = CLIContext( + config_path=config_path, + config=config, + log_level=effective_log_level, + log_level_override=log_level.upper() if log_level else None, + ) + except click.ClickException: + raise + except SemanticaError as exc: + raise click.ClickException(str(exc)) from exc + except Exception as exc: + raise click.ClickException(f"Failed to initialize CLI: {exc}") from exc + + +@main.group(invoke_without_command=True) +@click.pass_context +def kg(ctx: click.Context) -> None: + """Knowledge graph and semantic build commands.""" + if ctx.invoked_subcommand is None: + click.echo(ctx.get_help()) + + +@main.group(invoke_without_command=True) +@click.pass_context +def pipeline(ctx: click.Context) -> None: + """Pipeline command group (foundation placeholder).""" + if ctx.invoked_subcommand is None: + click.echo(ctx.get_help()) + + +@main.group(name="services", invoke_without_command=True) +@click.pass_context +def services(ctx: click.Context) -> None: + """Service management commands (server, explorer, mcp — foundation placeholder). + + Subcommands will follow the spec layout:: + + semantica services server start|stop|status + semantica services explorer start|stop|status + semantica services mcp start|stop|status + """ + if ctx.invoked_subcommand is None: + click.echo(ctx.get_help()) + + +@main.group(name="config", invoke_without_command=True) +@click.pass_context +def config_group(ctx: click.Context) -> None: + """Configuration command group.""" + if ctx.invoked_subcommand is None: + click.echo(ctx.get_help()) + + +@main.command() +@click.pass_obj +def info(cli_ctx: CLIContext): + """Display information about Semantica.""" + cli_ctx = _require_ctx(cli_ctx) + + def _action() -> None: + console.print(f"[bold blue]Semantica Framework[/bold blue] v{__version__}") + console.print( + "A comprehensive Python framework for transforming unstructured data " + "into semantic layers." + ) + + table = Table(title="Framework Components") + table.add_column("Component", style="cyan") + table.add_column("Status", style="green") + + table.add_row("Core Orchestrator", "Active") + table.add_row("Knowledge Graph Engine", "Active") + table.add_row("Pipeline Execution", "Active") + table.add_row("Vector Store Integration", "Active") + table.add_row("CLI Config File", cli_ctx.config_path or "(none)") + table.add_row("CLI Log Level", cli_ctx.log_level) + + console.print(table) + + _run_with_error_handling(_action) + + +@kg.command("build") +@click.option("--source", "-s", multiple=True, help="Data sources to process.") +@click.option( + "-c", + "--config", + "command_config_path", + type=click.Path(exists=True, dir_okay=False, resolve_path=True, path_type=str), + default=None, + help="Path to YAML/JSON config file.", +) +@click.pass_obj +def kg_build( + cli_ctx: CLIContext, + source: Sequence[str], + command_config_path: Optional[str], +): + """Build a knowledge base from sources.""" + cli_ctx = _require_ctx(cli_ctx) + + def _action() -> None: + _run_build_command(cli_ctx, source, command_config_path) + + _run_with_error_handling(_action) + + +@main.command("build", hidden=True) +@click.option("--source", "-s", multiple=True, help="Data sources to process.") +@click.option( + "-c", + "--config", + "command_config_path", + type=click.Path(exists=True, dir_okay=False, resolve_path=True, path_type=str), + default=None, + help="Path to YAML/JSON config file.", +) +@click.pass_obj +def build_alias( + cli_ctx: CLIContext, + source: Sequence[str], + command_config_path: Optional[str], +): + """Backward-compatible alias for 'kg build'.""" + cli_ctx = _require_ctx(cli_ctx) + + def _action() -> None: + _run_build_command(cli_ctx, source, command_config_path) + + _run_with_error_handling(_action) + if __name__ == "__main__": main() diff --git a/tests/test_cli_foundation.py b/tests/test_cli_foundation.py new file mode 100644 index 00000000..e22992cb --- /dev/null +++ b/tests/test_cli_foundation.py @@ -0,0 +1,446 @@ +import click +import pytest +from click.testing import CliRunner, Result + +import semantica.cli as cli_module + + +@pytest.fixture +def runner() -> CliRunner: + # Click's CliRunner captures all output in result.output by default; error + # messages from ClickException / UsageError are included. If the project + # ever moves to a Click version that separates stderr (mix_stderr=False), + # update assertions that check error text to use result.stderr instead. + return CliRunner() + + +@pytest.fixture(autouse=True) +def disable_cli_logging(monkeypatch): + monkeypatch.setattr(cli_module, "setup_logging", lambda *args, **kwargs: None) + + +# --------------------------------------------------------------------------- +# Help surfaces +# --------------------------------------------------------------------------- + + +def test_root_help_shows_expected_groups(runner): + result = runner.invoke(cli_module.main, ["--help"]) + + assert result.exit_code == 0 + assert ( + "Semantica - Semantic Layer & Knowledge Engineering Framework" + in result.output + ) + assert "kg" in result.output + assert "pipeline" in result.output + assert "services" in result.output + + +def test_kg_group_help_shows_build_command(runner): + result = runner.invoke(cli_module.main, ["kg", "--help"]) + + assert result.exit_code == 0 + assert "Knowledge graph and semantic build commands." in result.output + assert "build" in result.output + + +def test_kg_build_help_shows_source_and_config_flags(runner): + result = runner.invoke(cli_module.main, ["kg", "build", "--help"]) + + assert result.exit_code == 0 + assert "--source" in result.output + assert "-s" in result.output + assert "--config" in result.output + assert "-c" in result.output + + +# --------------------------------------------------------------------------- +# info command +# --------------------------------------------------------------------------- + + +def test_info_command_shows_framework_components(runner): + result = runner.invoke(cli_module.main, ["info"]) + + assert result.exit_code == 0 + assert "Semantica Framework" in result.output + assert "Core Orchestrator" in result.output + assert "CLI Log Level" in result.output + + +def test_info_command_shows_config_path_when_supplied(runner): + # click.Path(resolve_path=True) turns cfg.yml into an absolute path, and + # Rich's table may truncate long paths with '…'. Assert the row is present + # and the "(none)" placeholder has been replaced by some path value. + with runner.isolated_filesystem(): + with open("cfg.yml", "w", encoding="utf-8") as handle: + handle.write("logging:\n level: INFO\n") + + result: Result = runner.invoke( + cli_module.main, ["--config", "cfg.yml", "info"] + ) + + assert result.exit_code == 0 + assert "CLI Config File" in result.output + assert "(none)" not in result.output + + +# --------------------------------------------------------------------------- +# Global --log-level override +# --------------------------------------------------------------------------- + + +def test_log_level_global_override_stores_in_context(runner, monkeypatch): + """--log-level at the root level propagates into CLIContext.""" + captured = {} + + def fake_run_build(cli_ctx, sources): + captured["log_level_override"] = cli_ctx.log_level_override + captured["log_level"] = cli_ctx.log_level + + monkeypatch.setattr(cli_module, "_run_build", fake_run_build) + + result = runner.invoke( + cli_module.main, + ["--log-level", "DEBUG", "kg", "build", "-s", "src.txt"], + ) + + assert result.exit_code == 0 + assert captured["log_level_override"] == "DEBUG" + + +# --------------------------------------------------------------------------- +# Command-level config +# --------------------------------------------------------------------------- + + +def test_command_config_preserves_global_log_level_override( + runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + """Global --log-level is forwarded into command_ctx.log_level_override.""" + captured: dict[str, object] = {} + + def fake_run_build(cli_ctx: cli_module.CLIContext, _: object) -> None: + captured["log_level_override"] = cli_ctx.log_level_override + + monkeypatch.setattr(cli_module, "_run_build", fake_run_build) + + with runner.isolated_filesystem(): + with open("cfg.yml", "w", encoding="utf-8") as handle: + handle.write("logging:\n level: INFO\n") + + result = runner.invoke( + cli_module.main, + ["--log-level", "DEBUG", "kg", "build", "-s", "src.txt", "-c", "cfg.yml"], + ) + + assert result.exit_code == 0 + assert captured["log_level_override"] == "DEBUG" + + +def test_command_config_keeps_own_logging_without_global_override(runner, monkeypatch): + captured = {} + + def fake_run_build(cli_ctx, sources): + captured["logging_level"] = cli_ctx.config.get("logging.level") + captured["sources"] = list(sources) + + monkeypatch.setattr(cli_module, "_run_build", fake_run_build) + + with runner.isolated_filesystem(): + with open("cfg.yml", "w", encoding="utf-8") as handle: + handle.write("logging:\n level: DEBUG\n") + + result = runner.invoke( + cli_module.main, + ["kg", "build", "-s", "README.md", "-c", "cfg.yml"], + ) + + assert result.exit_code == 0 + assert captured["sources"] == ["README.md"] + assert captured["logging_level"] == "DEBUG" + + +# --------------------------------------------------------------------------- +# Legacy / new command parity +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "argv", + [ + ["kg", "build", "-s", "README.md"], + ["build", "-s", "README.md"], + ], +) +def test_build_paths_invoke_shared_wrapper(runner, monkeypatch, argv): + captured = {} + + def fake_run_build(cli_ctx, sources): + captured["config_path"] = cli_ctx.config_path + captured["sources"] = list(sources) + + monkeypatch.setattr(cli_module, "_run_build", fake_run_build) + + result = runner.invoke(cli_module.main, argv) + + assert result.exit_code == 0 + assert captured["sources"] == ["README.md"] + + +@pytest.mark.parametrize( + "argv", + [ + ["kg", "build", "-s", "README.md", "-c", "cfg.yml"], + ["kg", "build", "-s", "README.md", "--config", "cfg.yml"], + ["build", "-s", "README.md", "-c", "cfg.yml"], + ["build", "-s", "README.md", "--config", "cfg.yml"], + ], +) +def test_build_config_short_and_long_flags_are_compatible(runner, monkeypatch, argv): + captured = {} + + def fake_run_build(cli_ctx, sources): + captured["config_path"] = cli_ctx.config_path + captured["sources"] = list(sources) + + monkeypatch.setattr(cli_module, "_run_build", fake_run_build) + + with runner.isolated_filesystem(): + with open("cfg.yml", "w", encoding="utf-8") as handle: + handle.write("logging:\n level: INFO\n") + + result = runner.invoke(cli_module.main, argv) + + assert result.exit_code == 0 + assert captured["sources"] == ["README.md"] + assert captured["config_path"] is not None + + +# --------------------------------------------------------------------------- +# build_knowledge_base result handling +# --------------------------------------------------------------------------- + + +class _MockFramework: + """Lightweight stand-in for Semantica used to test _run_build result paths.""" + + def __init__(self, return_value): + self._return_value = return_value + + def build_knowledge_base(self, sources): + return self._return_value + + +def test_build_result_with_stats_shows_source_count(runner, monkeypatch): + """When build returns statistics.sources_processed, that count appears in output.""" + monkeypatch.setattr( + cli_module, + "_get_framework", + lambda _: _MockFramework({"statistics": {"sources_processed": 3}}), + ) + + result = runner.invoke( + cli_module.main, + ["kg", "build", "-s", "a.txt", "-s", "b.txt", "-s", "c.txt"], + ) + + assert result.exit_code == 0 + assert "3 source(s)" in result.output + + +def test_build_result_without_stats_shows_generic_success(runner, monkeypatch): + """When build result has no statistics key, generic success message is shown.""" + monkeypatch.setattr( + cli_module, + "_get_framework", + lambda _: _MockFramework({}), + ) + + result = runner.invoke(cli_module.main, ["kg", "build", "-s", "src.txt"]) + + assert result.exit_code == 0 + assert "Knowledge base build completed" in result.output + assert "source(s)" not in result.output + + +def test_build_result_none_shows_generic_success(runner, monkeypatch): + """When build_knowledge_base returns None, generic success message is shown.""" + monkeypatch.setattr( + cli_module, + "_get_framework", + lambda _: _MockFramework(None), + ) + + result = runner.invoke(cli_module.main, ["kg", "build", "-s", "src.txt"]) + + assert result.exit_code == 0 + assert "Knowledge base build completed" in result.output + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "argv", + [ + ["kg", "build"], + ["build"], + ], +) +def test_missing_input_errors_are_clean_and_click_safe(runner, argv): + result = runner.invoke(cli_module.main, argv) + + assert result.exit_code != 0 + assert "At least one source is required" in result.output + assert "Traceback" not in result.output + + +def test_invalid_root_config_error_is_clean_and_click_safe(runner): + with runner.isolated_filesystem(): + with open("bad.md", "w", encoding="utf-8") as handle: + handle.write("not-a-config") + + result = runner.invoke(cli_module.main, ["--config", "bad.md", "info"]) + + assert result.exit_code != 0 + assert "Unsupported configuration file format" in result.output + assert "Traceback" not in result.output + + +def test_invalid_command_config_error_is_clean_and_click_safe(runner): + with runner.isolated_filesystem(): + with open("bad.md", "w", encoding="utf-8") as handle: + handle.write("not-a-config") + + result = runner.invoke( + cli_module.main, + ["kg", "build", "-s", "README.md", "-c", "bad.md"], + ) + + assert result.exit_code != 0 + assert "Unsupported configuration file format" in result.output + assert "Traceback" not in result.output + + +def test_empty_yaml_config_is_accepted(runner): + with runner.isolated_filesystem(): + with open("cfg.yml", "w", encoding="utf-8") as handle: + handle.write("") + + result = runner.invoke(cli_module.main, ["--config", "cfg.yml", "info"]) + + assert result.exit_code == 0 + + +def test_malformed_logging_section_is_clean_and_click_safe(runner): + with runner.isolated_filesystem(): + with open("cfg.yml", "w", encoding="utf-8") as handle: + handle.write("logging: []\n") + + result = runner.invoke( + cli_module.main, + ["--config", "cfg.yml", "--log-level", "INFO", "info"], + ) + + assert result.exit_code != 0 + assert ( + "Logging configuration section must contain a mapping/object" + in result.output + ) + assert "Traceback" not in result.output + + +@pytest.mark.parametrize( + "file_name, config_text, expected_error", + [ + ("cfg.json", "{not-json", "Failed to parse configuration file"), + ( + "cfg.yml", + "- item\n- item2\n", + "Configuration file must contain a mapping/object", + ), + ], +) +def test_command_config_parse_errors_are_clean_and_click_safe( + runner, + file_name, + config_text, + expected_error, +): + with runner.isolated_filesystem(): + with open(file_name, "w", encoding="utf-8") as handle: + handle.write(config_text) + + result = runner.invoke( + cli_module.main, + ["kg", "build", "-s", "README.md", "-c", file_name], + ) + + assert result.exit_code != 0 + assert expected_error in result.output + assert "Traceback" not in result.output + + +def test_runtime_errors_are_click_safe_without_traceback(runner, monkeypatch): + def boom(_cli_ctx, _sources): + raise RuntimeError("boom") + + monkeypatch.setattr(cli_module, "_run_build", boom) + + result = runner.invoke(cli_module.main, ["kg", "build", "-s", "README.md"]) + + assert result.exit_code != 0 + assert "Unexpected error: boom" in result.output + assert "Traceback" not in result.output + + +# --------------------------------------------------------------------------- +# Lazy initialization +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "argv", + [ + ["--help"], + ["kg", "--help"], + ["kg", "build", "--help"], + ["build", "--help"], + ], +) +def test_help_calls_do_not_initialize_framework(runner, monkeypatch, argv): + def fail_get_framework(_): + raise AssertionError("framework initialization must not happen on help") + + monkeypatch.setattr(cli_module, "_get_framework", fail_get_framework) + + result = runner.invoke(cli_module.main, argv) + + assert result.exit_code == 0 + + +# --------------------------------------------------------------------------- +# _require_ctx guard +# --------------------------------------------------------------------------- + + +def test_require_ctx_raises_click_exception_on_none(): + """_require_ctx converts None ctx into a clean ClickException.""" + with pytest.raises( + click.ClickException, + match="CLI context is uninitialized", + ): + cli_module._require_ctx(None) + + +def test_require_ctx_returns_ctx_unchanged(): + """_require_ctx is a pass-through when ctx is valid.""" + from semantica.core.config_manager import Config, ConfigManager + + cfg = ConfigManager().load_from_dict({}, validate=False) + ctx = cli_module.CLIContext(config_path=None, config=cfg, log_level="INFO") + assert cli_module._require_ctx(ctx) is ctx