SchemaLock adapter: support multiple base URLs (staging + prod)

stellargate.example.yaml

@@ -11,7 +11,15 @@ tools:
   schemalock:
     enabled: true
     config: ./schemalock.yaml
-    base_url: http://127.0.0.1:8000
+    # Single URL (legacy) — findings are tagged with this URL:
+    #   base_url: http://127.0.0.1:8000
+    # OR run the same contract checks against multiple environments; each URL
+    # is tested once and findings are labelled by environment:
+    base_urls:
+      staging: http://127.0.0.1:8000
+      prod: https://api.example.com
+    # base_urls may also be a plain list of URLs; then each URL is used as its
+    # own environment label (e.g. ["http://a:8000", "http://b:8000"]).
 
   vaultsweep:
     enabled: true

stellargate/adapters/schemalock.py

@@ -36,12 +36,24 @@
 DEFAULT_SEVERITY = "medium"
 
 
-def run(options: dict) -> list[Finding]:
-    config_path = options.get("config")
-    if not config_path:
-        raise AdapterError(f"{TOOL_NAME}: 'config' option (path to schemalock.yaml) is required")
-    base_url = options.get("base_url", "http://127.0.0.1:8000")
-
+def _resolve_base_urls(options: dict) -> dict[str, str]:
+    """Resolve configured base URLs into an {env_label: url} mapping.
+
+    Accepts three shapes (last one wins if several are present):
+      * options["base_urls"] as a dict {"env": "url"}  -> labels are the env keys
+      * options["base_urls"] as a list of url strings   -> labels are the URLs
+      * options["base_url"] as a single url string        -> legacy, label defaults to url
+    """
+    urls = options.get("base_urls")
+    if isinstance(urls, dict):
+        return {str(label): str(url) for label, url in urls.items()}
+    if isinstance(urls, (list, tuple)):
+        return {str(url): str(url) for url in urls}
+    single = options.get("base_url")
+    return {str(single): str(single) for single in [single]}
+
+
+def _run_one(config_path: str, base_url: str) -> dict:
     with tempfile.TemporaryDirectory() as tmp:
         report_path = Path(tmp) / "schemalock-report.json"
         cmd = [
@@ -73,11 +85,31 @@ def run(options: dict) -> list[Finding]:
 
         try:
             with open(report_path) as f:
-                data = json.load(f)
+                return json.load(f)
         except json.JSONDecodeError as e:
             raise AdapterError(f"{TOOL_NAME}: report file was not valid JSON ({e})")
 
-    return parse_report(data)
+
+def run(options: dict) -> list[Finding]:
+    config_path = options.get("config")
+    if not config_path:
+        raise AdapterError(f"{TOOL_NAME}: 'config' option (path to schemalock.yaml) is required")
+
+    environments = _resolve_base_urls(options)
+
+    findings: list[Finding] = []
+    for label, url in environments.items():
+        data = _run_one(config_path, url)
+        for finding in parse_report(data):
+            # Prepend the environment label so a reviewer can tell which
+            # environment a contract check failed against, e.g.
+            # "staging: GET /escrows/{id}". The legacy single-URL path uses
+            # a URL label, keeping the finding self-describing but unchanged
+            # in shape.
+            if label:
+                finding.location = f"{label}: {finding.location}" if finding.location else label
+            findings.append(finding)
+    return findings
 
 
 def parse_report(data: dict) -> list[Finding]:

tests/test_adapters.py

@@ -132,39 +132,101 @@ def test_vaultsweep_crash_with_no_output_raises_not_zero_findings():
             vaultsweep.run({"path": "."})
 
 
-def test_schemalock_mixed_report_maps_all_failed_checks():
-    data = json.loads((FIXTURES / "schemalock_mixed_report.json").read_text())
-    findings = schemalock.parse_report(data)
-    # 7 checks in fixture, 4 failed (auth_required, status, error_envelope,
-    # unknown) -> only 4 findings; passed checks are skipped.
-    assert len(findings) == 4
-
-    severity_by_rule = {f.rule_id: f.severity for f in findings}
-    assert severity_by_rule == {
-        "CONTRACT-AUTH_REQUIRED": "critical",
-        "CONTRACT-STATUS": "high",
-        "CONTRACT-ERROR_ENVELOPE": "medium",
-        "CONTRACT-RATE_LIMIT": "medium",
+def test_schemalock_run_multiple_base_urls_runs_each_and_labels_findings():
+    """base_urls list -> one CLI invocation per URL, findings labelled by URL."""
+    data = {
+        "checks": [
+            {
+                "name": "get_escrow",
+                "check_type": "status",
+                "passed": False,
+                "endpoint": "GET /escrows/{id}",
+                "detail": "got 500, expected 200",
+            }
+        ]
     }
+    calls = []
+    reports = [data, data]
 
-    # every failed check becomes a Finding
-    assert all(f.tool == "schemalock" for f in findings)
-    # passed checks never appear as findings
-    assert not any(f for f in findings if f.raw.get("check_type") and f.raw["passed"])
+    def _side_effect(cmd, *args, **kwargs):
+        calls.append(cmd)
+        idx = cmd.index("--json-report")
+        Path(cmd[idx + 1]).write_text(json.dumps(reports.pop(0)))
 
+    with patch("subprocess.run", side_effect=_side_effect):
+        findings = schemalock.run(
+            {"config": "./schemalock.yaml", "base_urls": ["http://staging:9000", "http://prod:9000"]}
+        )
 
-def test_schemalock_mixed_report_only_failed_checks():
-    data = json.loads((FIXTURES / "schemalock_mixed_report.json").read_text())
-    findings = schemalock.parse_report(data)
-    failed_types = [
-        c["check_type"].lower()
-        for c in data["checks"]
-        if not c.get("passed", True)
+    assert len(calls) == 2
+    assert calls[0][calls[0].index("--base-url") + 1] == "http://staging:9000"
+    assert calls[1][calls[1].index("--base-url") + 1] == "http://prod:9000"
+    assert [f.location for f in findings] == [
+        "http://staging:9000: GET /escrows/{id}",
+        "http://prod:9000: GET /escrows/{id}",
     ]
-    assert {f.rule_id for f in findings} == {
-        f"CONTRACT-{t.upper()}" for t in failed_types
+
+
+def test_schemalock_legacy_single_base_url_calls_once():
+    """Legacy base_url (single string) still invokes the CLI once, URL-tagged."""
+    data = {
+        "checks": [
+            {
+                "name": "get_escrow",
+                "check_type": "status",
+                "passed": False,
+                "endpoint": "GET /escrows/{id}",
+                "detail": "got 500, expected 200",
+            }
+        ]
     }
-    assert all(not f.raw.get("passed", False) for f in findings)
+    calls = []
+
+    def _side_effect(cmd, *args, **kwargs):
+        calls.append(cmd)
+        idx = cmd.index("--json-report")
+        Path(cmd[idx + 1]).write_text(json.dumps(data))
+
+    with patch("subprocess.run", side_effect=_side_effect):
+        findings = schemalock.run(
+            {"config": "./schemalock.yaml", "base_url": "http://127.0.0.1:8000"}
+        )
+
+    assert len(calls) == 1
+    assert calls[0][calls[0].index("--base-url") + 1] == "http://127.0.0.1:8000"
+    assert len(findings) == 1
+
+
+def test_schemalock_base_urls_dict_uses_env_keys_as_labels():
+    data = {
+        "checks": [
+            {
+                "name": "get_escrow",
+                "check_type": "status",
+                "passed": False,
+                "endpoint": "GET /escrows/{id}",
+                "detail": "got 500, expected 200",
+            }
+        ]
+    }
+    reported = [data, data]
+
+    def _side_effect(cmd, *args, **kwargs):
+        idx = cmd.index("--json-report")
+        Path(cmd[idx + 1]).write_text(json.dumps(reported.pop(0)))
+
+    with patch("subprocess.run", side_effect=_side_effect):
+        findings = schemalock.run(
+            {
+                "config": "./schemalock.yaml",
+                "base_urls": {"staging": "http://staging:9000", "prod": "http://prod:9000"},
+            }
+        )
+
+    assert [f.location for f in findings] == [
+        "staging: GET /escrows/{id}",
+        "prod: GET /escrows/{id}",
+    ]
 
 
 def test_schemalock_severity_lookup_is_case_insensitive():