1cc17cf9ac
- 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
304 lines
9.6 KiB
Python
304 lines
9.6 KiB
Python
"""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 |