test: exercise every Cloud read command against a live stack

.github/workflows/cloud-read.yml

@@ -30,21 +30,28 @@ jobs:
           python -m venv .venv
           .venv/bin/python -m pip install -e ".[dev]"
 
-      - name: Run read-only Cloud canary
+      - name: Cloud reads (every catalogued read command)
         run: >-
-          .venv/bin/pytest -m "integration and cloud and read"
-          -vv --tb=short --junitxml=cloud-read.xml
+          .venv/bin/pytest tests/integration/cloud/read/test_catalog.py
+          -m "integration and cloud and read"
+          -vv --color=yes --tb=short --junitxml=cloud-read.xml
+
+      - name: Cloud ACS operations (below the CLI)
+        run: >-
+          .venv/bin/pytest tests/integration/cloud/read/test_acs_operations.py
+          -m "integration and cloud and read"
+          -vv --color=yes --tb=short --junitxml=cloud-acs.xml
 
       - name: Publish test summary
         if: always()
         uses: test-summary/action@37b508cfee6d4d080eedd00b5bb240a6a784a6a5 # v2
         with:
-          paths: cloud-read.xml
+          paths: cloud-*.xml
 
       - name: Upload test report
         if: always()
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
         with:
           name: cloud-read-${{ github.run_id }}
-          path: cloud-read.xml
+          path: cloud-*.xml
           if-no-files-found: ignore

CHANGELOG.md

@@ -15,6 +15,10 @@ project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - Install and test with only the standard library's `venv` and `pip`. Setup no
   longer needs a separate package manager, and the documented commands are the
   same ones CI runs on macOS, Linux, and Windows.
+- Add a Splunk Cloud read catalog that mirrors the Enterprise one: it runs every
+  catalogued read against a live stack, requires the ACS-backed reads to
+  succeed, and requires every other read to fail with a typed error envelope
+  rather than guessing at an endpoint the stack does not serve.
 - Add catalog-driven live coverage for all 61 Enterprise reads and 93 writes,
   including cleanup, global-state restoration, restart/reconnect, a container
   CI gate on code pull requests and `main` merges, a manual Cloud read canary,

pyproject.toml

@@ -63,7 +63,11 @@ include = ["src", "tests", "VERSION", "README.md", "CHANGELOG.md", "LICENSE"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
-addopts = "-ra --strict-markers"
+# importlib import mode lets the Enterprise and Cloud suites share the
+# `test_catalog.py` filename; `pythonpath` keeps the shared `cli_catalog` helper
+# importable without relying on a conftest side effect.
+addopts = "-ra --strict-markers --import-mode=importlib"
+pythonpath = ["tests"]
 markers = [
     "integration: end-to-end test needing a live Splunk (set SPLUNK_INTEGRATION_TEST=true)",
     "enterprise: tests against Splunk Enterprise",

tests/integration/cloud/conftest.py

@@ -0,0 +1,40 @@
+"""Shared gates for live Splunk Cloud tests.
+
+Mirrors `tests/integration/enterprise/conftest.py`: skip a local run that did
+not opt in, but fail loudly when a run opts in and is misconfigured, so a broken
+canary can never pass by silently skipping every test.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+from vct_splunk.core.backends import deduce_backend
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _require_cloud_target() -> None:
+    """Require the opt-in, a real Cloud URL, and an ACS token."""
+    if os.environ.get("SPLUNK_ACS_LIVE_TEST") != "true":
+        pytest.skip("set SPLUNK_ACS_LIVE_TEST=true to run Splunk Cloud integration tests")
+
+    url = os.environ.get("SPLUNK_URL")
+    if not url:
+        pytest.fail("SPLUNK_URL is required for Splunk Cloud integration tests", pytrace=False)
+
+    # The backend is deduced from the URL, never chosen, so a non-Cloud URL here
+    # would quietly test Enterprise instead of what this suite claims to cover.
+    if deduce_backend(url) != "cloud":
+        pytest.fail(
+            f"SPLUNK_URL={url!r} is not a Splunk Cloud host; "
+            "expected something like https://<stack>.splunkcloud.com",
+            pytrace=False,
+        )
+
+    if not os.environ.get("SPLUNK_ACS_TOKEN"):
+        pytest.fail(
+            "SPLUNK_ACS_TOKEN is required for Splunk Cloud integration tests",
+            pytrace=False,
+        )

tests/integration/test_acs_live.py → tests/integration/cloud/read/test_acs_operations.py

@@ -1,8 +1,12 @@
-"""Credential-gated read-only ACS canary."""
+"""Read-only ACS operations against a live Cloud stack, below the CLI layer.
 
-from __future__ import annotations
+The catalog suite next door drives the same reads through the public CLI. This
+module exercises the ACS client and operations directly, so a shape change in
+the upstream API surfaces here rather than only as a CLI symptom. Credentials
+are gated once, in this package's conftest.
+"""
 
-import os
+from __future__ import annotations
 
 import pytest
 
@@ -20,8 +24,7 @@
 
 @pytest.fixture
 def acs_client():
-    if os.environ.get("SPLUNK_ACS_LIVE_TEST") != "true":
-        pytest.skip("set SPLUNK_ACS_LIVE_TEST=true with ACS credentials")
+    """Open an ACS client from the environment for one test."""
     with AcsClient(acs_config_from_env(cloud_stack_from_url())) as client:
         yield client
 

tests/integration/cloud/read/test_catalog.py

@@ -0,0 +1,82 @@
+"""Execute every catalogued read against a live Splunk Cloud stack.
+
+The Cloud mirror of `tests/integration/enterprise/read/test_catalog.py`. It runs
+the same read leaves from the same catalog, and asserts the Cloud contract:
+
+* A read that Cloud actually serves (via ACS) must succeed and return a typed
+  ``{data, meta}`` envelope.
+* Every other read must still honor the documented contract — a typed error
+  envelope and a documented exit code — never a traceback, and never a silent
+  fallthrough to an endpoint the stack does not serve.
+
+That second rule is the point of the suite. Cloud coverage is read-only and
+partial, so the guarantee worth proving is that an unsupported command fails
+*cleanly* instead of guessing.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+from click.testing import CliRunner
+
+from cli_catalog import CATALOG, Case
+from vct_splunk.cli import cli
+from vct_splunk.commands.dispatch import has_cloud_list
+
+pytestmark = [pytest.mark.integration, pytest.mark.cloud, pytest.mark.read]
+
+READ_CASES = tuple(case for case in CATALOG if case.kind == "read")
+
+#: Reads that must succeed on Cloud. Derived from the real dispatch table, so
+#: adding a Cloud route extends this suite automatically instead of drifting.
+CLOUD_BACKED = tuple(
+    case
+    for case in READ_CASES
+    if (case.path[-1] == "list" and has_cloud_list(case.path[0])) or case.path == ("inspect",)
+)
+
+#: The documented exit codes (see the README). Anything else means the CLI
+#: crashed or invented a status.
+CONTRACT_EXITS = (0, 1, 3, 4, 5)
+
+
+def _invoke(case: Case):
+    """Run one catalogued read through the public CLI, asking for JSON."""
+    args = case.live_argv if case.live_argv is not None else case.argvs[0]
+    return CliRunner().invoke(cli, [*case.path, *args, "--output", "json"])
+
+
+@pytest.mark.parametrize("case", CLOUD_BACKED, ids=lambda case: " ".join(case.path))
+def test_cloud_backed_read_succeeds(case: Case) -> None:
+    """A read Cloud serves must succeed and return the typed data envelope."""
+    result = _invoke(case)
+
+    assert result.exit_code == 0, (
+        f"{' '.join(case.path)} exited {result.exit_code}: {result.output}"
+    )
+    payload = json.loads(result.output)
+    assert set(payload) == {"data", "meta"}
+    assert payload["meta"]["target"]
+
+
+@pytest.mark.parametrize("case", READ_CASES, ids=lambda case: " ".join(case.path))
+def test_every_cloud_read_leaf_honors_the_contract(case: Case) -> None:
+    """Every read either succeeds or fails cleanly — never a crash or a guess."""
+    result = _invoke(case)
+
+    assert result.exit_code in CONTRACT_EXITS, (
+        f"{' '.join(case.path)} exited {result.exit_code}, "
+        f"outside the documented contract: {result.output}"
+    )
+    payload = json.loads(result.output)
+
+    if result.exit_code in (0, 5):
+        assert set(payload) == {"data", "meta"}
+        return
+
+    # A failure must still be a typed envelope that a script can branch on.
+    assert set(payload) == {"error"}
+    assert payload["error"]["code"]
+    assert payload["error"]["message"]