diff --git a/gso/cli/lso_calls.py b/gso/cli/lso_calls.py
index e7086d44fec3044d104ce2f25b3f8054f1795806..c8f228efc18d1839d3f3b75fb708c47525d87842 100644
--- a/gso/cli/lso_calls.py
+++ b/gso/cli/lso_calls.py
@@ -3,6 +3,7 @@
import json
import logging
+from pathlib import Path
import click
import httpx
@@ -20,6 +21,15 @@ from gso.utils.types.lso_response import ExecutableRunResponse
logger = structlog.get_logger(__name__)
app = typer.Typer()
+_IMPORT_FILE_ARG = typer.Argument(
+ ...,
+ exists=True,
+ file_okay=True,
+ dir_okay=False,
+ readable=True,
+ help="Path to the JSON import file to embed in the check",
+)
+
def _validate_partner(partner: str) -> None:
if not filter_partners_by_name(name=partner, case_sensitive=True):
@@ -27,16 +37,31 @@ def _validate_partner(partner: str) -> None:
raise typer.Exit(1)
-def _call_lso(host: str, partner: str) -> ExecutableRunResponse:
+def _load_import_file(import_file_path: Path) -> str:
+ """Read a JSON file from the given path, return it as a compact JSON string, Exits on error."""
+ try:
+ with import_file_path.open("r", encoding="utf-8") as f:
+ data = json.load(f)
+ return json.dumps(data, separators=(",", ":"))
+ except Exception as e:
+ logger.exception("Failed to read import file")
+ typer.echo(f"Error: could not read or parse '{import_file_path}': {e}")
+ raise typer.Exit(2) from e
+
+
+def _call_lso(
+ host: str,
+ partner: str,
+ import_json_str: str,
+) -> ExecutableRunResponse:
oss = settings.load_oss_params()
proxy = oss.PROVISIONING_PROXY
url = f"{proxy.scheme}://{proxy.api_base}/api/execute/"
payload = {
"executable_name": "bgp_status_pre_check.py",
- "args": [host, partner],
+ "args": [host, partner, import_json_str],
"is_async": False,
}
-
try:
resp = httpx.post(url, json=payload, timeout=30)
resp.raise_for_status()
@@ -77,7 +102,11 @@ def _print_parsed_output(exec_resp: ExecutableRunResponse) -> None:
typer.echo(output_str)
-def _maybe_save(host: str, partner: str, exec_resp: ExecutableRunResponse) -> None:
+def _maybe_save(
+ host: str,
+ partner: str,
+ exec_resp: ExecutableRunResponse,
+) -> None:
prompt = (
f"\nIf you are happy with the above output for router '{host}' "
f"(partner: {partner}), shall we save it to the database?"
@@ -106,10 +135,12 @@ def _maybe_save(host: str, partner: str, exec_resp: ExecutableRunResponse) -> No
def bgp_status_precheck(
host: str = typer.Argument(..., help="FQDN of the router to pre-check"),
partner: str = typer.Argument(..., help="Partner name for import file path"),
+ import_file_path: Path = _IMPORT_FILE_ARG,
) -> None:
"""Trigger the bgp_status_pre-check script on LSO, print results, and optionally save."""
_validate_partner(partner)
- exec_resp = _call_lso(host, partner)
+ import_json_str = _load_import_file(import_file_path)
+ exec_resp = _call_lso(host, partner, import_json_str)
_print_full(exec_resp)
_print_parsed_output(exec_resp)
_maybe_save(host, partner, exec_resp)
diff --git a/test/cli/test_lso_calls.py b/test/cli/test_lso_calls.py
index 51bc2d6cf6c9783e60686ff9e4e3ab2639d3cd9b..86eec0b6cace3f633644847a7c779632f0f832d4 100644
--- a/test/cli/test_lso_calls.py
+++ b/test/cli/test_lso_calls.py
@@ -35,6 +35,14 @@ class DummyResponse:
raise httpx.HTTPStatusError("error", request=None, response=self) # noqa: RUF100,EM101
+@pytest.fixture()
+def import_file(tmp_path):
+ """Create a temporary JSON import file."""
+ path = tmp_path / "import.json"
+ path.write_text(json.dumps({"foo": "bar"}))
+ return path
+
+
@pytest.fixture()
def mock_http_success(monkeypatch):
"""Return a successful dummy response for httpx.post."""
@@ -60,42 +68,55 @@ def mock_http_error(monkeypatch):
def mock_http_bad_shape(monkeypatch):
"""Return JSON that does not fit ExecutableRunResponse."""
bad = {"unexpected": "data"}
- # create a DummyResponse and patch httpx.post
dummy_resp = DummyResponse(bad)
monkeypatch.setattr("httpx.post", lambda *args, **kwargs: dummy_resp) # noqa: ARG005
return dummy_resp
-def test_invalid_partner_name(mock_http_success):
+def test_invalid_partner_name(mock_http_success, import_file):
"""Step 0: unknown partner should abort before any HTTP call."""
- result = runner.invoke(app, ["rt1.ams.geant.org", "UNKNOWN"], input="")
+ result = runner.invoke(
+ app,
+ ["rt1.ams.geant.org", "UNKNOWN", str(import_file)],
+ input="",
+ )
assert result.exit_code == 1
assert "partner 'unknown' not found" in result.stdout.lower()
assert db.session.query(BgpStatusPreCheckTable).count() == 0
-def test_no_save_leaves_table_empty(mock_http_success, partner_factory):
+def test_no_save_leaves_table_empty(mock_http_success, partner_factory, import_file):
"""If user declines save, table remains empty."""
partner_factory("SURF")
- result = runner.invoke(app, ["rt1.example.com", "SURF"], input="n\n")
+ result = runner.invoke(
+ app,
+ ["rt1.example.com", "SURF", str(import_file)],
+ input="n\n",
+ )
assert result.exit_code == 0
assert "not saving" in result.stdout.lower()
- # Table should be empty
assert db.session.query(BgpStatusPreCheckTable).count() == 0
-def test_prompt_save_yes_persists_record(mock_http_success, partner_factory):
+def test_prompt_save_yes_persists_record(mock_http_success, partner_factory, import_file):
"""Typing 'y' at prompt should also persist."""
partner_factory("SURF")
- result = runner.invoke(app, ["rt1.example.com", "SURF"], input="y\n")
+ result = runner.invoke(
+ app,
+ ["rt1.example.com", "SURF", str(import_file)],
+ input="y\n",
+ )
assert result.exit_code == 0
assert db.session.query(BgpStatusPreCheckTable).count() == 1
-def test_http_failure_aborts(mock_http_error, partner_factory):
+def test_http_failure_aborts(mock_http_error, partner_factory, import_file):
"""Network/timeout errors should abort with exit code 1."""
partner_factory("SURF")
- result = runner.invoke(app, ["rt1.example.com", "SURF"])
+ result = runner.invoke(
+ app,
+ ["rt1.example.com", "SURF", str(import_file)],
+ )
assert result.exit_code == 1
# Now stderr is separately captured:
assert "error: failed to call lso: timeout" in result.stdout.lower()
@@ -104,16 +125,19 @@ def test_http_failure_aborts(mock_http_error, partner_factory):
assert db.session.query(BgpStatusPreCheckTable).count() == 0
-def test_invalid_shape_aborts(mock_http_bad_shape, partner_factory):
+def test_invalid_shape_aborts(mock_http_bad_shape, partner_factory, import_file):
"""Malformed top-level JSON shape should abort."""
partner_factory("SURF")
- result = runner.invoke(app, ["rt1.example.com", "SURF"])
+ result = runner.invoke(
+ app,
+ ["rt1.example.com", "SURF", str(import_file)],
+ )
assert result.exit_code == 1
assert "invalid JSON returned by LSO" in result.stdout
assert db.session.query(BgpStatusPreCheckTable).count() == 0
-def test_parse_output_nonjson(mock_http_success, partner_factory):
+def test_parse_output_nonjson(monkeypatch, partner_factory, import_file):
"""If output is not valid JSON, we still complete without saving."""
partner_factory("SURF")
# Patch BASE_RESPONSE to use non-JSON output
@@ -126,17 +150,16 @@ def test_parse_output_nonjson(mock_http_success, partner_factory):
_orig = _httpx.post
_httpx.post = lambda *args, **kwargs: DummyResponse(bad) # noqa: ARG005
try:
- result = runner.invoke(app, ["rt1.example.com", "SURF"], input="n\n")
+ result = runner.invoke(app, ["rt1.example.com", "SURF", str(import_file)], input="n\n")
assert result.exit_code == 0
assert "(not valid JSON, raw string below)" in result.stdout
finally:
_httpx.post = _orig
-def test_pagination_on_large_output(mock_http_success, monkeypatch, partner_factory):
+def test_pagination_on_large_output(monkeypatch, partner_factory, import_file):
"""Parsed output >50 lines should trigger click.echo_via_pager."""
partner_factory("SURF")
- # Build huge object
big = {"x": ["line"] * 100}
payload = dict(BASE_RESPONSE)
payload["result"] = dict(payload["result"])
@@ -150,6 +173,31 @@ def test_pagination_on_large_output(mock_http_success, monkeypatch, partner_fact
paged = True
monkeypatch.setattr(click, "echo_via_pager", fake_pager)
- result = runner.invoke(app, ["rt1.example.com", "SURF"], input="n\n")
+ result = runner.invoke(
+ app,
+ ["rt1.example.com", "SURF", str(import_file)],
+ input="n\n",
+ )
assert result.exit_code == 0
assert paged, "Expected parsed-output pager for large JSON"
+
+
+def test_invalid_import_file(tmp_path, partner_factory):
+ """Invalid JSON import file should abort with exit code 2 and no DB write."""
+ # create invalid JSON file
+ bad_file = tmp_path / "bad_import.json"
+ bad_file.write_text("{invalid_json}")
+ partner_factory("SURF")
+
+ # Invoke with the malformed JSON file
+ result = runner.invoke(
+ app,
+ ["rt1.example.com", "SURF", str(bad_file)],
+ )
+
+ # Expect exit code 2 from _load_import_file
+ assert result.exit_code == 2
+ assert "Error: could not read or parse" in result.stdout
+
+ # Ensure no record was written
+ assert db.session.query(BgpStatusPreCheckTable).count() == 0