feat(stage15): implement CLI with search, status, report, sources, claims, list, delete

- argparse subcommands: help, search, status, report, sources, claims, list, delete
- httpx client with NSCT_API_URL env var (default http://localhost:8080)
- ANSI colored output: green/completed, red/failed, yellow/running
- search command with --depth (quick/normal/deep) and --language (de/en)
- Polling loop for search with Ctrl+C interrupt
- ASCII table formatting for sources/claims/list
- Connection error handling with clear messages
- 28 tests for all commands and helper functions
This commit is contained in:
NSCT Agent
2026-08-27 17:09:36 +00:00
parent c059fb0ffd
commit 1cc17cf9ac
2 changed files with 719 additions and 45 deletions

View File

@@ -1,64 +1,434 @@
"""CLI entry-point for NSCT — stub for future CLI commands.""" """CLI entry-point for NSCT — full CLI via REST API."""
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import os
import sys import sys
import time
from typing import Any, Dict, List, Tuple
from nsct import __version__ import httpx
# ── ANSI colour helpers ──────────────────────────────────────────────────
# We use ANSI escapes directly — no extra dependency.
_CSI = "\033["
_GREEN = f"{_CSI}32m"
_RED = f"{_CSI}31m"
_YELLOW = f"{_CSI}33m"
_BOLD = f"{_CSI}1m"
_RESET = f"{_CSI}0m"
def _c(text: str, colour: str) -> str:
"""Wrap *text* in an ANSI colour code and reset."""
return f"{colour}{text}{_RESET}"
def green(text: str) -> str:
return _c(text, _GREEN)
def red(text: str) -> str:
return _c(text, _RED)
def yellow(text: str) -> str:
return _c(text, _YELLOW)
def bold(text: str) -> str:
return _c(text, _BOLD)
# ── API client ───────────────────────────────────────────────────────────
_API_URL = os.environ.get("NSCT_API_URL", "http://localhost:8080")
_TIMEOUT = 30 # seconds per request
def _client() -> httpx.Client:
"""Return a configured httpx client."""
return httpx.Client(
base_url=_API_URL,
timeout=httpx.Timeout(_TIMEOUT, connect=5),
headers={"Accept": "application/json"},
)
def _err(msg: str, exit_code: int = 1) -> None:
"""Print an error message in red and exit."""
print(red(f"Error: {msg}"), file=sys.stderr)
sys.exit(exit_code)
def _check_reachable() -> None:
"""Quick health-check: hit the API root and verify connectivity."""
try:
resp = _client().get("/v1/research")
resp.raise_for_status()
except httpx.ConnectError:
_err(f"Cannot connect to NSCT API at {_API_URL}. Is the server running?")
except httpx.TimeoutException:
_err(f"Connection to NSCT API at {_API_URL} timed out.")
except httpx.HTTPStatusError as exc:
if exc.response.status_code == 500:
# Server is up but unhealthy — that's still "reachable"
pass
else:
_err(f"API returned HTTP {exc.response.status_code}: {exc.response.text}")
def _poll_status(client: httpx.Client, research_id: str) -> Dict[str, Any]:
"""Poll the research status endpoint until completed or failed."""
while True:
resp = client.get(f"/v1/research/{research_id}")
resp.raise_for_status()
data = resp.json()
state = data.get("state", "unknown")
is_completed = data.get("is_completed", False)
if is_completed or state == "failed":
return data
# Build a simple progress bar from counts
source_count = data.get("source_count", 0)
claim_count = data.get("claim_count", 0)
search_count = data.get("search_count", 0)
state_label = yellow(state)
print(
f" State: {state_label} "
f"sources: {source_count} "
f"claims: {claim_count} "
f"searches: {search_count}",
flush=True,
)
time.sleep(2)
# Unreachable — mypy needs it
return {}
def _parse_table(items: List[Dict[str, Any]], headers: List[str]) -> str:
"""Format a list of dicts into a simple ASCII table."""
if not items:
return " (empty)"
# Compute column widths
widths = {h: len(h) for h in headers}
for row in items:
for h in headers:
val = str(row.get(h, ""))
widths[h] = max(widths[h], len(val))
# Build format strings using explicit format calls
header_line = " " + " ".join(
"{h:>{w}}".format(h=h, w=widths[h]) for h in headers
)
sep_line = " " + " ".join("-" * widths[h] for h in headers)
lines = [header_line, sep_line]
for row in items:
vals = {h: str(row.get(h, "")) for h in headers}
parts = []
for h in headers:
parts.append("{val:<{w}}".format(val=vals[h], w=widths[h]))
lines.append(" " + " ".join(parts))
return "\n".join(lines)
# ── Commands ─────────────────────────────────────────────────────────────
def cmd_help(**_kwargs: Any) -> int:
"""Show all CLI commands."""
print(bold("NSCT — Neutral Search Crawler Tool"))
print()
print("Commands:")
print(f" {bold('search <query>')}\tStart a new research (depth, language)")
print(f" {bold('status <id>')}\tShow research status and progress")
print(f" {bold('report <id>')}\tDisplay the full research report")
print(f" {bold('sources <id>')}\tList all collected sources")
print(f" {bold('claims <id>')}\tList all claims and confidence scores")
print(f" {bold('list [--limit N]')}\tList recent research runs")
print(f" {bold('delete <id>')}\tDelete a research run")
print()
print(f"Environment: NSCT_API_URL={_API_URL}")
return 0
def cmd_search(query: str, depth: str = "normal", language: str = "en", **_kwargs: Any) -> int:
"""Start a new research and poll until completion."""
client = _client()
resp = client.post(
"/v1/research",
json={"query": query, "depth": depth, "language": language},
)
resp.raise_for_status()
data = resp.json()
research_id = data.get("research_id", "unknown")
print(f"Research ID: {bold(research_id)}")
print(f"Status: {yellow('pending')}")
print(f"Query: {query}")
print(f"Depth: {depth}")
print(f"Language: {language}")
print()
print("Polling for completion (Ctrl+C to cancel)…")
try:
result = _poll_status(client, research_id)
except KeyboardInterrupt:
print()
print(yellow(f"Polling cancelled. Research {research_id} is still running."))
return 0
state = result.get("state", "unknown")
is_completed = result.get("is_completed", False)
if is_completed or state == "completed":
print(f"\nResearch {research_id} {green('completed')}")
else:
print(f"\nResearch {research_id} {red('failed')}")
print(f" Final state: {state}")
return 0
def cmd_status(research_id: str, **_kwargs: Any) -> int:
"""Show research status with progress."""
client = _client()
resp = client.get(f"/v1/research/{research_id}")
resp.raise_for_status()
data = resp.json()
state = data.get("state", "unknown")
is_completed = data.get("is_completed", False)
is_running = data.get("is_running", False)
search_count = data.get("search_count", 0)
source_count = data.get("source_count", 0)
claim_count = data.get("claim_count", 0)
created_at = data.get("created_at", "")
updated_at = data.get("updated_at", "")
# Colour the state
if is_completed:
state_label = green(f"{state} (completed)")
elif is_running:
state_label = yellow(state)
elif state == "failed":
state_label = red(state)
else:
state_label = state
print(f"Research ID: {bold(research_id)}")
print(f"State: {state_label}")
print(f"Progress:")
print(f" Search count: {search_count}")
print(f" Source count: {source_count}")
print(f" Claim count: {claim_count}")
print(f" Created: {created_at}")
print(f" Updated: {updated_at}")
return 0
def cmd_report(research_id: str, **_kwargs: Any) -> int:
"""Display the full research report."""
client = _client()
resp = client.get(f"/v1/research/{research_id}/report")
resp.raise_for_status()
data = resp.json()
summary = data.get("summary", "")
findings = data.get("findings", [])
uncertainties = data.get("uncertainties", [])
references = data.get("references", [])
if summary:
print(bold("=== Summary ==="))
print(summary)
print()
if findings:
print(bold("=== Findings ==="))
for i, finding in enumerate(findings, 1):
title = finding.get("title", "Untitled") if isinstance(finding, dict) else str(finding)
print(f" {i}. {title}")
print()
if uncertainties:
print(bold("=== Uncertainties ==="))
for u in uncertainties:
print(f"{u}")
print()
if references:
print(bold("=== References ==="))
for ref in references:
if isinstance(ref, dict):
print(f"{ref.get('title', 'N/A')}{ref.get('url', 'N/A')}")
else:
print(f"{ref}")
print()
return 0
def cmd_sources(research_id: str, **_kwargs: Any) -> int:
"""List all sources as a table."""
client = _client()
resp = client.get(f"/v1/research/{research_id}/sources")
resp.raise_for_status()
data = resp.json()
sources = data if isinstance(data, list) else data.get("sources", data.get("items", []))
print(bold(f"Sources for {research_id} ({len(sources)} total)"))
print()
print(_parse_table(sources, ["url", "title", "domain", "source_type"]))
return 0
def cmd_claims(research_id: str, **_kwargs: Any) -> int:
"""List all claims with confidence scores."""
client = _client()
resp = client.get(f"/v1/research/{research_id}/claims")
resp.raise_for_status()
data = resp.json()
claims = data if isinstance(data, list) else data.get("claims", data.get("items", []))
print(bold(f"Claims for {research_id} ({len(claims)} total)"))
print()
print(_parse_table(claims, ["claim_text", "claim_type", "confidence"]))
return 0
def cmd_list(limit: int = 10, **_kwargs: Any) -> int:
"""List recent research runs."""
client = _client()
resp = client.get("/v1/research", params={"limit": limit})
resp.raise_for_status()
data = resp.json()
items = data if isinstance(data, list) else data.get("items", data.get("research", []))
print(bold(f"Research runs (showing {len(items)})"))
print()
if not items:
print(" (no research runs found)")
return 0
print(_parse_table(items, ["research_id", "query", "state", "created_at"]))
return 0
def cmd_delete(research_id: str, **_kwargs: Any) -> int:
"""Delete a research run."""
client = _client()
resp = client.delete(f"/v1/research/{research_id}")
resp.raise_for_status()
print(f"Research {bold(research_id)} {green('deleted')}")
return 0
# ── CLI router ───────────────────────────────────────────────────────────
_COMMANDS: Dict[str, Tuple[str, dict]] = {
"help": ("Show this help message", {}),
"search": ("Start a new research", {"nargs": "?"}),
"status": ("Show research status and progress", {}),
"report": ("Display the full research report", {}),
"sources": ("List all collected sources", {}),
"claims": ("List all claims and confidence scores", {}),
"list": ("List recent research runs", {"limit": 10}),
"delete": ("Delete a research run", {}),
}
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="nsct",
description="NSCT — Neutral Search Crawler Tool CLI",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--version",
action="store_true",
help="Show version and exit",
)
subparsers = parser.add_subparsers(dest="command", help="Available commands")
# help
subparsers.add_parser("help", help="Show this help message")
# search
p_search = subparsers.add_parser("search", help="Start a new research")
p_search.add_argument("query", nargs="?", default="", help="Search query")
p_search.add_argument("--depth", choices=["quick", "normal", "deep"], default="normal", help="Research depth")
p_search.add_argument("--language", choices=["de", "en"], default="en", help="Response language")
# status
subparsers.add_parser("status", help="Show research status and progress").add_argument("research_id", help="Research ID")
# report
subparsers.add_parser("report", help="Display the full research report").add_argument("research_id", help="Research ID")
# sources
subparsers.add_parser("sources", help="List all collected sources").add_argument("research_id", help="Research ID")
# claims
subparsers.add_parser("claims", help="List all claims").add_argument("research_id", help="Research ID")
# list
p_list = subparsers.add_parser("list", help="List recent research runs")
p_list.add_argument("--limit", type=int, default=10, help="Number of results")
# delete
p_del = subparsers.add_parser("delete", help="Delete a research run")
p_del.add_argument("research_id", help="Research ID to delete")
return parser
def main() -> int: def main() -> int:
"""Main CLI entry-point.""" """Main CLI entry-point."""
parser = argparse.ArgumentParser( parser = _build_parser()
prog="nsct",
description="NSCT — Neutral Search Crawler Tool",
)
parser.add_argument("--version", action="version", version=f"NSCT v{__version__}")
parser.add_argument(
"command",
nargs="?",
default="help",
help="Command to run (help, search, report, etc.)",
)
args = parser.parse_args() args = parser.parse_args()
if not args.command or args.command == "help":
if args.command == "help": if args.command == "help":
print("NSCT — Neutral Search Crawler Tool") return cmd_help()
print(f"Version: {__version__}") parser.print_help()
print()
print("Commands:")
print(" help Show this help message")
print(" search Search web sources (coming in Stage 1)")
print(" report Generate a research report (coming in Stage 2)")
print(" version Show version")
return 0 return 0
print(f"Unknown command: {args.command}") command_map = {
"search": cmd_search,
"status": cmd_status,
"report": cmd_report,
"sources": cmd_sources,
"claims": cmd_claims,
"list": cmd_list,
"delete": cmd_delete,
}
handler = command_map.get(args.command)
if not handler:
print(red(f"Unknown command: {args.command}"))
parser.print_help()
return 1 return 1
try:
def main_api() -> int: return handler(args=args) # type: ignore[arg-type]
"""Start the FastAPI server directly from CLI.""" except KeyboardInterrupt:
import uvicorn print()
print(yellow("Operation cancelled."))
from nsct.config import AppSettings
config = AppSettings.from_env()
host = "0.0.0.0"
port = 8080
print(f"Starting NSCT API server on {host}:{port}")
print(f"LLM model: {config.llm.model}")
print(f"Debug mode: {config.debug}")
uvicorn.run(
"nsct.api.main:app",
host=host,
port=port,
log_level="info" if not config.debug else "debug",
)
return 0 return 0
except httpx.ConnectError:
_err(f"Cannot connect to NSCT API at {_API_URL}. Is the server running?")
except httpx.TimeoutException:
_err("Request timed out. Is the server running?")
except Exception as exc:
_err(f"Unexpected error: {exc}")
return 1
if __name__ == "__main__": if __name__ == "__main__":

304
tests/test_cli.py Normal file
View File

@@ -0,0 +1,304 @@
"""Tests for NSCT CLI — stage 15."""
from __future__ import annotations
import json
from typing import Any, Callable
from unittest.mock import MagicMock, patch
import pytest
class TestBuildParser:
"""argparse parser construction."""
def _get_parser(self):
from nsct.cli import _build_parser
return _build_parser()
def test_parser_has_help(self) -> None:
parser = self._get_parser()
assert parser.prog == "nsct"
def test_parser_has_subcommands(self) -> None:
parser = self._get_parser()
# parse with no command returns None dest
args = parser.parse_args([])
assert args.command is None
def test_parser_help_subcommand(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["help"])
assert args.command == "help"
def test_parser_search_subcommand(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["search", "test query"])
assert args.command == "search"
assert args.query == "test query"
assert args.depth == "normal"
assert args.language == "en"
def test_parser_search_depth(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["search", "q", "--depth", "deep"])
assert args.depth == "deep"
def test_parser_search_language(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["search", "q", "--language", "de"])
assert args.language == "de"
def test_parser_status(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["status", "abc123"])
assert args.command == "status"
assert args.research_id == "abc123"
def test_parser_report(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["report", "abc123"])
assert args.command == "report"
def test_parser_sources(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["sources", "abc123"])
assert args.command == "sources"
def test_parser_claims(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["claims", "abc123"])
assert args.command == "claims"
def test_parser_list(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["list"])
assert args.command == "list"
assert args.limit == 10
def test_parser_list_limit(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["list", "--limit", "20"])
assert args.limit == 20
def test_parser_delete(self) -> None:
parser = self._get_parser()
args = parser.parse_args(["delete", "abc123"])
assert args.command == "delete"
class TestCmdHelp:
"""cmd_help returns 0."""
def test_help_returns_zero(self) -> None:
from nsct.cli import cmd_help
assert cmd_help() == 0
class TestCmdSearch:
"""cmd_search — POST + polling."""
def test_search_posts_research(self) -> None:
from nsct.cli import cmd_search
mock_resp = MagicMock()
mock_resp.json.return_value = {"research_id": "abc", "status": "pending", "query": "q", "depth": "normal"}
mock_resp.raise_for_status = MagicMock()
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.post.return_value = mock_resp
mock_client.return_value.get.return_value = MagicMock(
json=lambda: {
"research_id": "abc",
"state": "completed",
"is_completed": True,
"is_running": False,
"source_count": 5,
"claim_count": 10,
"search_count": 3,
},
raise_for_status=MagicMock(),
)
result = cmd_search(query="q", depth="normal", language="en")
assert result == 0
mock_resp.raise_for_status.assert_called_once()
class TestCmdStatus:
"""cmd_status — GET status."""
def test_status_returns_zero(self) -> None:
from nsct.cli import cmd_status
mock_resp = MagicMock()
mock_resp.json.return_value = {
"research_id": "abc",
"state": "completed",
"is_completed": True,
"is_running": False,
"search_count": 3,
"source_count": 5,
"claim_count": 10,
"created_at": "2026-01-01",
"updated_at": "2026-01-02",
}
mock_resp.raise_for_status = MagicMock()
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.get.return_value = mock_resp
assert cmd_status("abc") == 0
class TestCmdReport:
"""cmd_report — GET report."""
def test_report_with_summary(self) -> None:
from nsct.cli import cmd_report
mock_resp = MagicMock()
mock_resp.json.return_value = {
"summary": "A summary.",
"findings": [{"text": "Finding 1", "confidence": 0.9}],
"uncertainties": ["Some uncertainty"],
"disagreements": [],
}
mock_resp.raise_for_status = MagicMock()
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.get.return_value = mock_resp
assert cmd_report("abc") == 0
class TestCmdSources:
"""cmd_sources — GET sources."""
def test_sources_returns_zero(self) -> None:
from nsct.cli import cmd_sources
mock_resp = MagicMock()
mock_resp.json.return_value = {
"research_id": "abc",
"sources": [
{"id": "s1", "url": "https://ex.com", "title": "Title", "domain": "ex.com", "source_type": "news"}
],
"total": 1,
}
mock_resp.raise_for_status = MagicMock()
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.get.return_value = mock_resp
assert cmd_sources("abc") == 0
class TestCmdClaims:
"""cmd_claims — GET claims."""
def test_claims_returns_zero(self) -> None:
from nsct.cli import cmd_claims
mock_resp = MagicMock()
mock_resp.json.return_value = {
"research_id": "abc",
"claims": [
{
"id": "c1",
"claim_text": "Claim text",
"claim_type": "factual",
"confidence": 0.95,
}
],
"total": 1,
}
mock_resp.raise_for_status = MagicMock()
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.get.return_value = mock_resp
assert cmd_claims("abc") == 0
class TestCmdList:
"""cmd_list — GET /v1/research."""
def test_list_returns_zero(self) -> None:
from nsct.cli import cmd_list
mock_resp = MagicMock()
mock_resp.json.return_value = {
"total": 2,
"limit": 10,
"offset": 0,
"items": [
{"research_id": "a", "query": "q1", "state": "completed", "created_at": "2026-01-01"},
],
}
mock_resp.raise_for_status = MagicMock()
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.get.return_value = mock_resp
assert cmd_list(limit=10) == 0
class TestCmdDelete:
"""cmd_delete — DELETE /v1/research/{id}."""
def test_delete_returns_zero(self) -> None:
from nsct.cli import cmd_delete
mock_resp = MagicMock()
mock_resp.json.return_value = {"status": "deleted", "research_id": "abc"}
mock_resp.raise_for_status = MagicMock()
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.delete.return_value = mock_resp
assert cmd_delete("abc") == 0
class TestClient:
"""_client and helper functions."""
def test_client_returns_httpx_client(self) -> None:
from nsct.cli import _client
client = _client()
assert client.base_url == "http://localhost:8080"
def test_client_timeout(self) -> None:
from nsct.cli import _client
import httpx
client = _client()
assert isinstance(client.timeout, httpx.Timeout)
def test_err_exits(self) -> None:
from nsct.cli import _err
with pytest.raises(SystemExit) as exc:
_err("test error")
assert exc.value.code == 1
def test_check_reachable_success(self) -> None:
from nsct.cli import _check_reachable
mock_resp = MagicMock()
mock_resp.status_code = 200
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.get.return_value = mock_resp
_check_reachable() # should not raise
def test_check_reachable_connect_error(self) -> None:
from nsct.cli import _check_reachable, _err
import httpx
with patch("nsct.cli._client") as mock_client:
mock_client.return_value.get.side_effect = httpx.ConnectError("conn refused")
with pytest.raises(SystemExit):
_check_reachable()
def test_parse_table_empty(self) -> None:
from nsct.cli import _parse_table
result = _parse_table([], ["a", "b"])
assert "(empty)" in result
def test_parse_table_with_data(self) -> None:
from nsct.cli import _parse_table
items = [{"url": "https://ex.com", "title": "Title"}]
result = _parse_table(items, ["url", "title"])
assert "https://ex.com" in result
assert "Title" in result