add import_file_path as an arg to bgp_status_precheck cli command

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)

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