mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
Merge pull request #576 from Sameer6305/feat/cli-foundation-base
CLI foundation: wire kg build path with legacy build compatibility and focused tests
This commit is contained in:
Mohd Kaif
+340
-33
@@ -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": <int>,
|
||||
...
|
||||
},
|
||||
...
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user