fix: validate load target cda

cwmscli/load/root.py

@@ -4,14 +4,16 @@
 import logging
 from dataclasses import dataclass
 from typing import Optional
-from urllib.parse import urlparse
+from urllib.parse import urljoin, urlparse
 
 import click
+import requests
 
 from cwmscli import requirements as reqs
 from cwmscli.utils.deps import requires
 
 logger = logging.getLogger(__name__)
+CDA_PROBE_TIMEOUT_SECONDS = 2.5
 
 CONTEXT = dict(
     help_option_names=["-h", "--help"],
@@ -40,11 +42,74 @@ def _norm_office(o: Optional[str]) -> str:
     return (o or "").strip().upper()
 
 
+def _swagger_docs_url(api_root: str) -> str:
+    return urljoin(f"{api_root.rstrip('/')}/", "swagger-docs")
+
+
+def _looks_like_cda_landing_page(response: requests.Response) -> bool:
+    # Some local CDA builds serve the UI but not the generated OpenAPI route.
+    server = response.headers.get("Server", "").lower()
+    if "cwms-data-api" in server:
+        return True
+
+    content_type = response.headers.get("Content-Type", "").lower()
+    if "html" not in content_type:
+        return False
+    return "CDA - CWMS Data API" in response.text
+
+
+def _validate_cda_api_root(api_root: str, *, role: str) -> None:
+    parsed = urlparse(api_root)
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise click.ClickException(
+            f"{role} CDA URL must be an absolute http(s) URL, got {api_root!r}."
+        )
+
+    url = _swagger_docs_url(api_root)
+    try:
+        response = requests.get(
+            url,
+            headers={"Accept": "application/json"},
+            timeout=CDA_PROBE_TIMEOUT_SECONDS,
+        )
+        response.raise_for_status()
+        document = response.json()
+        if isinstance(document, dict) and (
+            document.get("openapi") or document.get("swagger")
+        ):
+            return
+    except (requests.RequestException, ValueError) as openapi_error:
+        logger.debug(
+            "CDA OpenAPI probe for %s at %s did not succeed: %s",
+            role,
+            url,
+            openapi_error,
+        )
+
+    try:
+        response = requests.get(api_root, timeout=CDA_PROBE_TIMEOUT_SECONDS)
+        response.raise_for_status()
+    except requests.Timeout as e:
+        raise click.ClickException(
+            f"Could not validate {role} CDA at {api_root}: timed out fetching {api_root}."
+        ) from e
+    except requests.RequestException as e:
+        raise click.ClickException(
+            f"Could not validate {role} CDA at {api_root}: failed to fetch {api_root}: {e}"
+        ) from e
+
+    if not _looks_like_cda_landing_page(response):
+        raise click.ClickException(
+            f"{role} URL {api_root} did not return a CDA OpenAPI document or CDA landing page."
+        )
+
+
 def validate_cda_targets(func):
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
         source_csv = kwargs.get("source_csv")
         target_csv = kwargs.get("target_csv")
+        skip_target_cda_check = kwargs.pop("skip_target_cda_check", False)
 
         if source_csv and target_csv:
             raise click.ClickException(
@@ -92,6 +157,10 @@ def wrapper(*args, **kwargs):
                 "This is allowed, but double-check intent.",
             )
 
+        # Dry-runs still need a real target; otherwise users can validate a bad load command.
+        if target_cda and not skip_target_cda_check:
+            _validate_cda_api_root(target_cda, role="Target")
+
         src_label = source_csv or source_cda or "-"
         tgt_label = target_csv or target_cda or "-"
         logger.info(
@@ -134,6 +203,14 @@ def shared_source_target_options(f):
         envvar="CDA_API_KEY",
         help="Target API key used when no saved cwms-cli login token is available.",
     )(f)
+    f = click.option(
+        "--skip-target-cda-check",
+        envvar="CWMS_CLI_SKIP_TARGET_CDA_CHECK",
+        is_flag=True,
+        default=False,
+        show_default=True,
+        help="Skip the preflight check that --target-cda points to a CDA service.",
+    )(f)
     f = click.option(
         "--dry-run/--no-dry-run",
         is_flag=True,

tests/commands/test_load_location_ids.py

@@ -267,6 +267,9 @@ def test_cli_rejects_source_csv_and_target_csv_together(tmp_path, monkeypatch):
     monkeypatch.setattr(
         "cwmscli.utils.get_saved_login_token", lambda *args, **kwargs: None
     )
+    monkeypatch.setattr(
+        "cwmscli.load.root._validate_cda_api_root", lambda *a, **k: None
+    )
     src = tmp_path / "in.csv"
     src.write_text("name,office-id,active\nLOC_A,SWT,True\n")
     out = tmp_path / "out.csv"
@@ -292,6 +295,9 @@ def test_cli_rejects_source_csv_with_explicit_source_cda(tmp_path, monkeypatch):
     monkeypatch.setattr(
         "cwmscli.utils.get_saved_login_token", lambda *args, **kwargs: None
     )
+    monkeypatch.setattr(
+        "cwmscli.load.root._validate_cda_api_root", lambda *a, **k: None
+    )
     src = tmp_path / "in.csv"
     src.write_text("name,office-id,active\nLOC_A,SWT,True\n")
 
@@ -312,3 +318,61 @@ def test_cli_rejects_source_csv_with_explicit_source_cda(tmp_path, monkeypatch):
     )
     assert result.exit_code != 0
     assert "mutually exclusive" in result.output
+
+
+def test_cli_can_skip_target_cda_check_with_flag(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        "cwmscli.load.root._validate_cda_api_root",
+        lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("target check should be skipped")
+        ),
+    )
+    src = tmp_path / "in.csv"
+    src.write_text("name,office-id,active\nLOC_A,SWT,True\n")
+
+    runner = CliRunner()
+    result = runner.invoke(
+        location_group,
+        [
+            "ids-all",
+            "--source-csv",
+            str(src),
+            "--source-office",
+            "SWT",
+            "--target-cda",
+            "http://not-cda.example/cwms-data",
+            "--skip-target-cda-check",
+            "--dry-run",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+
+
+def test_cli_can_skip_target_cda_check_with_env_var(tmp_path, monkeypatch):
+    monkeypatch.setenv("CWMS_CLI_SKIP_TARGET_CDA_CHECK", "1")
+    monkeypatch.setattr(
+        "cwmscli.load.root._validate_cda_api_root",
+        lambda *a, **k: (_ for _ in ()).throw(
+            AssertionError("target check should be skipped")
+        ),
+    )
+    src = tmp_path / "in.csv"
+    src.write_text("name,office-id,active\nLOC_A,SWT,True\n")
+
+    runner = CliRunner()
+    result = runner.invoke(
+        location_group,
+        [
+            "ids-all",
+            "--source-csv",
+            str(src),
+            "--source-office",
+            "SWT",
+            "--target-cda",
+            "http://not-cda.example/cwms-data",
+            "--dry-run",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output

tests/load/test_cda_target_validation.py

@@ -0,0 +1,94 @@
+import click
+import pytest
+
+from cwmscli.load.root import _validate_cda_api_root
+
+
+class FakeResponse:
+    def __init__(self, payload=None, error=None, headers=None, text=""):
+        self.payload = payload
+        self.error = error
+        self.headers = headers or {}
+        self.text = text
+
+    def raise_for_status(self):
+        if self.error:
+            raise self.error
+
+    def json(self):
+        if isinstance(self.payload, Exception):
+            raise self.payload
+        return self.payload
+
+
+def test_validate_cda_api_root_accepts_openapi_document(monkeypatch):
+    calls = []
+
+    def fake_get(url, headers=None, timeout=None):
+        calls.append((url, headers, timeout))
+        return FakeResponse({"openapi": "3.0.1", "info": {"title": "CWMS Data API"}})
+
+    monkeypatch.setattr("cwmscli.load.root.requests.get", fake_get)
+
+    _validate_cda_api_root("http://localhost:8082/cwms-data/", role="Target")
+
+    assert calls == [
+        (
+            "http://localhost:8082/cwms-data/swagger-docs",
+            {"Accept": "application/json"},
+            2.5,
+        )
+    ]
+
+
+def test_validate_cda_api_root_rejects_non_openapi_document(monkeypatch):
+    monkeypatch.setattr(
+        "cwmscli.load.root.requests.get",
+        lambda *a, **k: FakeResponse({"message": "not cda"}),
+    )
+
+    with pytest.raises(click.ClickException, match="did not return a CDA OpenAPI"):
+        _validate_cda_api_root("https://example.test/not-cda", role="Target")
+
+
+def test_validate_cda_api_root_accepts_cda_landing_page(monkeypatch):
+    calls = []
+
+    def fake_get(url, headers=None, timeout=None):
+        calls.append(url)
+        if url.endswith("/swagger-docs"):
+            return FakeResponse(ValueError("not json"))
+        return FakeResponse(
+            headers={"Content-Type": "text/html"},
+            text="<title>CDA - CWMS Data API</title>",
+        )
+
+    monkeypatch.setattr("cwmscli.load.root.requests.get", fake_get)
+
+    _validate_cda_api_root("http://localhost:7000/cwms-data", role="Target")
+
+    assert calls == [
+        "http://localhost:7000/cwms-data/swagger-docs",
+        "http://localhost:7000/cwms-data",
+    ]
+
+
+def test_validate_cda_api_root_rejects_invalid_url():
+    with pytest.raises(click.ClickException, match="absolute http"):
+        _validate_cda_api_root("sas-t7/sas-data/", role="Target")
+
+
+def test_validate_cda_api_root_reports_request_failure(monkeypatch):
+    class FakeRequestException(Exception):
+        pass
+
+    def fake_get(*args, **kwargs):
+        raise FakeRequestException("connection refused")
+
+    monkeypatch.setattr("cwmscli.load.root.requests.get", fake_get)
+    monkeypatch.setattr(
+        "cwmscli.load.root.requests.RequestException", FakeRequestException
+    )
+
+    with pytest.raises(click.ClickException, match="failed to fetch"):
+        _validate_cda_api_root("http://localhost:9999/cwms-data", role="Target")

tests/load/test_timeseries_data.py

@@ -12,6 +12,9 @@
 def test_load_timeseries_data_command_allows_group_without_category_filters(
     monkeypatch,
 ):
+    monkeypatch.setattr(
+        "cwmscli.load.root._validate_cda_api_root", lambda *a, **k: None
+    )
     calls = []
 
     def fake_load(**kwargs):