diff --git a/docs/commands/README.md b/docs/commands/README.md index ae22af3..17b34a6 100644 --- a/docs/commands/README.md +++ b/docs/commands/README.md @@ -42,5 +42,6 @@ Commands: | [prechange](nexus-dashboard/prechange.md) | Analyse a candidate config against current fabric state | | [delta](nexus-dashboard/delta.md) | Compare two snapshots and report changes | | [snapshots](nexus-dashboard/snapshots.md) | Resolve a snapshot ID (CI baseline pinning) | +| [analyze](nexus-dashboard/analyze.md) | Trigger an assurance analysis and print the snapshot it produces | | [compliance](nexus-dashboard/compliance.md) | Report compliance rule status | | [doctor](nexus-dashboard/doctor.md) | Check connectivity, credentials, fabric visibility | diff --git a/docs/commands/nexus-dashboard/README.md b/docs/commands/nexus-dashboard/README.md index 93c2c01..f4a9730 100644 --- a/docs/commands/nexus-dashboard/README.md +++ b/docs/commands/nexus-dashboard/README.md @@ -13,6 +13,7 @@ Product-level help (configuration variables and verb list) is in [Nexus Dashboar | [prechange](prechange.md) | Gate a planned change — Terraform plan JSON or APIC MO JSON | | [delta](delta.md) | Post-change comparison between two snapshots | | [snapshots](snapshots.md) | Print a snapshot ID for pipeline pinning | +| [analyze](analyze.md) | Trigger an assurance analysis and print the snapshot it produces | | [compliance](compliance.md) | Fabric compliance rules; optional `--fail-on-violations` | | [doctor](doctor.md) | Read-only connectivity and credential check | diff --git a/docs/commands/nexus-dashboard/analyze.md b/docs/commands/nexus-dashboard/analyze.md new file mode 100644 index 0000000..ce9b53f --- /dev/null +++ b/docs/commands/nexus-dashboard/analyze.md @@ -0,0 +1,34 @@ +# analyze + +[← Nexus Dashboard commands](README.md) · [Command reference](../README.md) + +Trigger an assurance analysis on a fabric, wait for it to finish, and print the ID of the snapshot it produced. This is the API behind the GUI's "Analyze now". + +Nexus Dashboard collects online fabrics on its own schedule — every 2–4 hours depending on fabric size. A pipeline that pins [`snapshots latest`](snapshots.md) before a change and compares against `latest` afterwards races that schedule: if no collection happened in between, both selectors resolve to the same record and [`delta`](delta.md) reports nothing. `analyze` removes the race by producing a snapshot on demand. + +## Examples + +```bash +nac-analytics nd analyze +``` + +Bracket a configuration push with two real snapshots: + +```bash +PRE=$(nac-analytics nd analyze) +terraform apply plan.tfplan +POST=$(nac-analytics nd analyze) +nac-analytics nd delta "$PRE" "$POST" +``` + +Trigger without waiting — prints the **analysis job ID**, which is not a snapshot ID and cannot be passed to `delta`: + +```bash +nac-analytics nd analyze --no-wait +``` + +`-output json` (or `yaml`) emits the whole snapshot record, including the `analysisJobId` that ties it back to the trigger. + +## Notes + +Run `nac-analytics nd analyze --help` for all flags and options. diff --git a/docs/commands/nexus-dashboard/snapshots.md b/docs/commands/nexus-dashboard/snapshots.md index a492e89..be421b7 100644 --- a/docs/commands/nexus-dashboard/snapshots.md +++ b/docs/commands/nexus-dashboard/snapshots.md @@ -6,6 +6,8 @@ Resolve a fabric snapshot selector and print its ID. Use in CI to pin baselines Selectors: `latest`, `latest-N`, or a concrete `snapshotId`. Combine with `-since` / `-until` when the API's 50-record listing cap requires narrowing the window. +This command only reads what ND has already collected, and ND collects online fabrics every 2–4 hours. To produce a snapshot on demand instead, use [analyze](analyze.md). + ## Examples ```bash diff --git a/docs/nexus-dashboard.md b/docs/nexus-dashboard.md index 398e0e8..5a39c0d 100644 --- a/docs/nexus-dashboard.md +++ b/docs/nexus-dashboard.md @@ -38,6 +38,7 @@ Commands: prechange Analyse a candidate configuration against a fabric's current state. delta Compare two snapshots of a fabric and report what changed. snapshots Resolve a fabric snapshot and print its ID (for CI baseline pinning). + analyze Trigger an assurance analysis and print the snapshot ID it produces. compliance Report compliance rule status for a fabric (or every fabric with --all). doctor Check connectivity, credentials, and fabric visibility. ``` @@ -49,5 +50,6 @@ Commands: | [prechange](commands/nexus-dashboard/prechange.md) | Gate a planned change — Terraform plan JSON or APIC MO JSON | | [delta](commands/nexus-dashboard/delta.md) | Post-change comparison between two snapshots | | [snapshots](commands/nexus-dashboard/snapshots.md) | Print a snapshot ID for pipeline pinning | +| [analyze](commands/nexus-dashboard/analyze.md) | Trigger an assurance analysis and print the snapshot it produces | | [compliance](commands/nexus-dashboard/compliance.md) | Fabric compliance rules; optional `--fail-on-violations` | | [doctor](commands/nexus-dashboard/doctor.md) | Read-only connectivity and credential check | diff --git a/nac_analytics/products/nexus_dashboard/cli.py b/nac_analytics/products/nexus_dashboard/cli.py index 846ca45..6e2069c 100644 --- a/nac_analytics/products/nexus_dashboard/cli.py +++ b/nac_analytics/products/nexus_dashboard/cli.py @@ -413,6 +413,28 @@ def _emit_snapshot(record: dict[str, object], output: str) -> None: typer.echo(serialize_structured(record, output)) +def _emit_notices(client: NDClient) -> None: + """Print a client's operational warnings to stderr, keeping stdout clean.""" + for warning in client.notices: + typer.secho(f"warning: {warning}", fg=typer.colors.YELLOW, err=True) + + +def _analysis_trigger_error(exc: ApiError) -> Exception: + """Explain a refused assurance analysis trigger. + + Triggering needs the super-admin, fabric-admin or support-engineer role, so + a 403 is an RBAC fact the operator can act on rather than a transport + failure, and it exits as an auth error. + """ + if "HTTP 403" not in str(exc): + return exc + return AuthError( + "Not permitted to trigger an assurance analysis. This API requires the " + "super-admin, fabric-admin or support-engineer role; an observer " + f"account can read snapshots but cannot start one. ({exc})" + ) + + def _resolve_pre_post( pre: str | None, post: str | None, @@ -815,8 +837,86 @@ def snapshots( record = client.resolve_snapshot( config.fabric, selector, start_date=since, end_date=until ) - for warning in client.notices: - typer.secho(f"warning: {warning}", fg=typer.colors.YELLOW, err=True) + _emit_notices(client) + _emit_snapshot(record, output) + except Exception as exc: + raise _fail(exc, verbose) from exc + + +@app.command() +def analyze( + host: HostOpt = None, + username: UserOpt = None, + password: PasswordOpt = None, + domain: DomainOpt = DEFAULT_DOMAIN, + fabric: FabricOpt = None, + no_wait: Annotated[ + bool, + typer.Option( + "--no-wait", + help=( + "Print the analysis job ID and exit instead of waiting. This is " + "a job ID, not a snapshotId; it cannot be passed to delta." + ), + ), + ] = False, + output: Annotated[ + str, + typer.Option( + "--output", + "-o", + help="Output format: text (snapshot ID only), json, or yaml.", + ), + ] = "text", + timeout: TimeoutOpt = 30, + poll_interval: PollOpt = 15, + verify_ssl: VerifyOpt = True, + ca_bundle: CaBundleOpt = None, + verbose: VerboseOpt = False, +) -> None: + """Trigger an assurance analysis and print the snapshot ID it produces.""" + _configure_logging(verbose) + try: + if output not in ("text", "json", "yaml"): + raise InputError( + f"Unknown output format '{output}'. Choose from: text, json, yaml." + ) + config = _build_config( + host=host, + username=username, + password=password, + domain=domain, + fabric=fabric, + verify_ssl=verify_ssl, + ca_bundle=ca_bundle, + timeout=timeout, + poll_interval=poll_interval, + ) + note(_connect_message(config)) + with NDClient(config) as client: + client.validate_fabric(config.fabric) + # Taken before triggering so a snapshot that already existed can + # never be mistaken for the one this run produces. + baseline = ( + None if no_wait else client.latest_collection_timestamp(config.fabric) + ) + note(f"Triggering an assurance analysis on {config.fabric}...") + try: + job_id = client.trigger_assurance_analysis(config.fabric) + except ApiError as exc: + raise _analysis_trigger_error(exc) from exc + if no_wait: + _emit_notices(client) + typer.echo(job_id) + return + note( + f"Waiting for analysis {job_id} to produce a snapshot " + f"(up to {timeout} minutes)..." + ) + record = client.wait_for_analysis_snapshot( + config.fabric, job_id, baseline=baseline + ) + _emit_notices(client) _emit_snapshot(record, output) except Exception as exc: raise _fail(exc, verbose) from exc diff --git a/nac_analytics/products/nexus_dashboard/client.py b/nac_analytics/products/nexus_dashboard/client.py index 0dfe83f..8a013f4 100644 --- a/nac_analytics/products/nexus_dashboard/client.py +++ b/nac_analytics/products/nexus_dashboard/client.py @@ -48,6 +48,16 @@ {"FAILED", "STOPPED", "ABORTED", "PARTIALLY_FAILED", "UNAVAILABLE"} ) +# Assurance analysis reports the same upper-case vocabulary as delta. A live +# ND 4.2.1 run reports COMPLETE; SUCCESS is accepted because the API schema +# lists both. The job is never filtered by type: the same trigger yields +# ONLINE-ANALYSIS, ONLINE-ANALYSIS-ACI or ONLINE-ANALYSIS-NX depending on the +# fabric, so only the job ID is a reliable filter. +ANALYSIS_SUCCEEDED = frozenset({"SUCCESS", "COMPLETE"}) +ANALYSIS_FAILED = frozenset( + {"FAILED", "STOPPED", "ABORTED", "PARTIALLY_FAILED", "UNAVAILABLE"} +) + # How long a job may stay absent before it is treated as non-existent. # /jobs/summary returns HTTP 200 with an empty `entries` array for a job that # does not exist, so absence is indistinguishable from a job not yet listed. @@ -253,6 +263,48 @@ def resolve_snapshot_ids( return prior_id, later_id +def snapshot_newer_than( + snapshots: list[dict[str, Any]], baseline: str | None +) -> dict[str, Any] | None: + """Return the newest snapshot collected strictly after `baseline`. + + A `baseline` of None means the fabric had no snapshot to compare against, + so any record qualifies. Records with an unparseable + `collectionTimestamp` are skipped rather than guessed at. + """ + if baseline is None: + ordered = sort_snapshots(snapshots) + return ordered[0] if ordered else None + floor = parse_timestamp(baseline) + if floor is None: + return None + for record in sort_snapshots(snapshots): + collected = parse_timestamp(str(record.get("collectionTimestamp", ""))) + if collected is not None and collected > floor: + return record + return None + + +def snapshot_for_job( + snapshots: list[dict[str, Any]], job_id: str, *, newer_than: str | None +) -> dict[str, Any] | None: + """Return the newest snapshot this analysis job produced. + + Both conditions matter. `analysisJobId` is not unique: the recurring + scheduled analysis keeps one job ID across cycles, so a single ID can name + several snapshots hours apart. And should a trigger ever be coalesced into + an already-running job, its ID would match a snapshot collected *before* + the trigger — the stale baseline this command exists to avoid. So a match + must also be newer than what was there beforehand. + """ + matching = [ + record + for record in snapshots + if str(record.get("analysisJobId", "")) == job_id and job_id + ] + return snapshot_newer_than(matching, newer_than) + + # -- jobs ------------------------------------------------------------------ @@ -290,6 +342,17 @@ def select_job(entries: list[dict[str, Any]], job_id: str) -> dict[str, Any] | N return None +def analysis_job_id(body: dict[str, Any]) -> str: + """Return the job ID `POST /jobs/assuranceAnalysis` reports. + + The ID reappears verbatim as the resulting snapshot's `analysisJobId`. + """ + job_id = body.get("jobId") + if not job_id: + raise JobError("Assurance analysis was triggered but returned no jobId.") + return str(job_id) + + class AbsenceWindow: """Bounds how long a job may stay absent before it is called non-existent. @@ -615,6 +678,125 @@ def resolve_snapshot( ) return select_snapshot(snapshots, selector) + # -- assurance analysis ------------------------------------------------ + + def latest_collection_timestamp(self, fabric: str) -> str | None: + """Return the newest finished snapshot's `collectionTimestamp`. + + None when the fabric has never produced one. Callers take this before + triggering an analysis so a snapshot that predates the trigger is not + mistaken for its result. + """ + snapshots = finished_snapshots(self.list_snapshots(fabric)) + if not snapshots: + return None + return str(snapshots[0].get("collectionTimestamp", "")) or None + + def trigger_assurance_analysis(self, fabric: str) -> str: + """Start an on-demand assurance analysis and return its job ID. + + This is the API behind the GUI's "Analyze now". It requires the + super-admin, fabric-admin or support-engineer role; an observer + account is refused with HTTP 403. + """ + body = self.post_json( + f"{ANALYZE}/jobs/assuranceAnalysis", json={"fabricName": fabric} + ) + job_id = analysis_job_id(body) + logger.debug("Assurance analysis %s started on %s", job_id, fabric) + return job_id + + def wait_for_analysis_snapshot( + self, fabric: str, job_id: str, *, baseline: str | None + ) -> dict[str, Any]: + """Wait for an assurance analysis to produce a finished snapshot. + + The job is polled before the snapshot list because no snapshot is + visible until the job is terminal, and because a failed job explains + itself where an absent snapshot cannot. + """ + deadline = time.monotonic() + self.config.job_timeout_minutes * 60 + window = AbsenceWindow(self.config.poll_interval_seconds) + job_done = False + while True: + if not job_done: + job_done = self._analysis_job_finished(job_id, window) + if job_done: + record = self._analysis_snapshot(fabric, job_id, baseline) + if record is not None: + return record + logger.debug( + "Analysis %s finished; waiting for its snapshot...", job_id + ) + if time.monotonic() > deadline: + raise JobError( + f"Assurance analysis {job_id} did not produce a finished " + f"snapshot within {self.config.job_timeout_minutes} minutes. " + "A full fabric collection can take considerably longer than " + "the default; raise --timeout." + ) + time.sleep(self.config.poll_interval_seconds) + + def _analysis_job_finished(self, job_id: str, window: AbsenceWindow) -> bool: + """True once the analysis job has succeeded; raises if it failed. + + The job is filtered by ID alone. Its `jobType` varies with the fabric + (ONLINE-ANALYSIS, -ACI or -NX), so filtering on type would drop it. + """ + job = select_job(self.job_summary(job_id=job_id), job_id) + if job is None: + if window.missing(): + raise JobError( + f"Assurance analysis {job_id} was never reported by " + f"{ANALYZE}/jobs/summary after {window.polls} polls " + f"(~{window.approx_seconds}s). The endpoint answers 200 " + "with no entries for a job that does not exist." + ) + logger.debug( + "Assurance analysis %s not visible yet (%d/%d)...", + job_id, + window.polls, + window.limit, + ) + return False + window.seen() + status = str(job.get("status", "")).upper() + if status in ANALYSIS_SUCCEEDED: + return True + if status in ANALYSIS_FAILED: + parts = [f"Assurance analysis {job_id} ended {status}."] + message = str(job.get("errorMessage", "")).strip() + if message: + parts.append(message) + if status == "PARTIALLY_FAILED": + parts.append( + "A partial collection produces a snapshot whose status is " + "not 'finished', so it cannot be used for delta analysis." + ) + raise JobError(" ".join(parts)) + logger.debug("Assurance analysis %s is %s...", job_id, status or "pending") + return False + + def _analysis_snapshot( + self, fabric: str, job_id: str, baseline: str | None + ) -> dict[str, Any] | None: + """Find the snapshot a finished analysis produced, if it has landed.""" + snapshots = finished_snapshots(self.list_snapshots(fabric)) + record = snapshot_for_job(snapshots, job_id, newer_than=baseline) + if record is not None: + return record + fallback = snapshot_newer_than(snapshots, baseline) + if fallback is not None: + self.notice( + "Snapshot %s is newer than the analysis was triggered but " + "carries analysisJobId %r, not %r. Reporting it anyway; verify " + "it describes the change you expect.", + fallback.get("snapshotId"), + str(fallback.get("analysisJobId", "")), + job_id, + ) + return fallback + # -- pre-change analysis ----------------------------------------------- def create_prechange_analysis( diff --git a/tests/unit/test_analyze.py b/tests/unit/test_analyze.py new file mode 100644 index 0000000..72ab12f --- /dev/null +++ b/tests/unit/test_analyze.py @@ -0,0 +1,390 @@ +"""Triggering an assurance analysis and waiting for the snapshot it produces. + +The endpoint, the ID vocabulary and the timing here were all confirmed against +a live ND 4.2.1 cluster: `POST /jobs/assuranceAnalysis` answers with a `jobId` +that reappears verbatim as the resulting snapshot's `analysisJobId`. +""" + +from __future__ import annotations + +import json + +import httpx +import pytest +from typer.testing import CliRunner + +from nac_analytics.cli import app +from nac_analytics.core.exceptions import AuthError, JobError +from nac_analytics.products.nexus_dashboard.client import ( + NDClient as RealNDClient, +) +from nac_analytics.products.nexus_dashboard.client import ( + analysis_job_id, + snapshot_for_job, + snapshot_newer_than, +) +from tests.conftest import Lab, json_response + +runner = CliRunner() + +TRIGGER_PATH = "/api/v1/analyze/jobs/assuranceAnalysis" +SUMMARY_PATH = "/api/v1/analyze/jobs/summary" +SNAPSHOTS_PATH = "/api/v1/analyze/fabricSnapshots" +FABRICS_PATH = "/api/v1/manage/fabrics" + +# The live prefix. The published spec's response example shortens it to +# `ANALYSIS-ACI-`, which does not match what the cluster returns. +JOB_ID = "ONLINE-ANALYSIS-ACI-c6ddb3c8-97b3-11f1-82e2-024ba9b9c180" +# The recurring scheduled analysis keeps one job ID across cycles, so this one +# names several snapshots collected hours apart. +RECURRING_JOB_ID = "ONLINE-ANALYSIS-ACI-c9635808-5e21-11f1-9161-62c5f98b2c1d" + +BASELINE = "2026-08-14T07:29:57Z" + +ENV = { + "ND_HOST": "nd.example.com", + "ND_USER": "admin", + "ND_PASSWORD": "s3cr3t", + "ND_FABRIC": "FABRIC-A", + "ND_VERIFY_SSL": "false", +} + + +def snapshot( + snapshot_id: str, + collected: str, + *, + job_id: str = RECURRING_JOB_ID, + status: str = "finished", +) -> dict: + return { + "snapshotId": snapshot_id, + "analysisJobId": job_id, + "collectionTimestamp": collected, + "analysisTimestamp": collected.replace(":57Z", ":59Z"), + "snapshotType": "online", + "status": status, + "fabricName": "FABRIC-A", + } + + +OLD = snapshot("snap-old", "2026-08-14T05:29:57Z") +BASE = snapshot("snap-base", BASELINE) +FRESH = snapshot("snap-fresh", "2026-08-14T07:46:36Z", job_id=JOB_ID) + + +def job(status: str, **extra: object) -> dict: + return { + "jobId": JOB_ID, + "jobType": "ONLINE-ANALYSIS-ACI", + "status": status, + **extra, + } + + +def analysis_lab( + *, + snapshots: list[dict] | list[list[dict]], + jobs: list[dict] | None = None, + trigger: httpx.Response | None = None, +) -> Lab: + """A cluster that answers the trigger, the job poll and the snapshot list. + + `snapshots` may be a list of records, or a list of successive listings so a + test can make a snapshot appear part-way through the polling. `jobs=[]` + means the job summary knows nothing of the job; the default is a job that + has already finished. + """ + if snapshots and isinstance(snapshots[0], list): + listings = [json_response({"snapshots": page}) for page in snapshots] + else: + listings = [json_response({"snapshots": snapshots})] + entries = [job("COMPLETE")] if jobs is None else jobs + return Lab( + { + FABRICS_PATH: json_response( + {"fabrics": [{"name": "FABRIC-A", "management": {"type": "aci"}}]} + ), + TRIGGER_PATH: trigger or json_response({"jobId": JOB_ID}), + SUMMARY_PATH: json_response({"entries": entries}), + SNAPSHOTS_PATH: listings, + } + ) + + +@pytest.fixture(autouse=True) +def virtual_clock(monkeypatch: pytest.MonkeyPatch) -> None: + """Make sleeping advance a fake clock instead of doing nothing. + + The shared `no_sleep` fixture alone would leave a wait that never succeeds + spinning against the mock cluster for a real timeout's worth of seconds. + Here a poll costs its own interval and nothing else, so the timeout + branches are both instant and exact. + """ + now = [0.0] + + def advance(seconds: float) -> None: + now[0] += seconds + + monkeypatch.setattr( + "nac_analytics.products.nexus_dashboard.client.time.monotonic", + lambda: now[0], + ) + monkeypatch.setattr( + "nac_analytics.products.nexus_dashboard.client.time.sleep", advance + ) + + +# -- the trigger ----------------------------------------------------------- + + +def test_the_trigger_posts_only_the_fabric_name(make_client) -> None: + lab = analysis_lab(snapshots=[FRESH]) + client = make_client(lab) + + assert client.trigger_assurance_analysis("FABRIC-A") == JOB_ID + + request = lab.requests_to(TRIGGER_PATH)[0] + assert request.method == "POST" + assert json.loads(request.content) == {"fabricName": "FABRIC-A"} + + +def test_a_response_without_a_job_id_is_a_job_error() -> None: + with pytest.raises(JobError, match="returned no jobId"): + analysis_job_id({}) + + +def test_the_job_id_is_read_verbatim() -> None: + assert analysis_job_id({"jobId": JOB_ID}) == JOB_ID + + +# -- matching a job to its snapshot ---------------------------------------- + + +def test_the_snapshot_is_matched_on_analysis_job_id() -> None: + found = snapshot_for_job([BASE, FRESH, OLD], JOB_ID, newer_than=BASELINE) + + assert found is not None + assert found["snapshotId"] == "snap-fresh" + + +def test_the_newest_snapshot_wins_when_a_job_id_names_several() -> None: + """The recurring analysis reuses one job ID, so a match is not unique.""" + newer = snapshot("snap-newer", "2026-08-14T09:29:57Z") + + found = snapshot_for_job([OLD, newer, BASE], RECURRING_JOB_ID, newer_than=BASELINE) + + assert found is not None + assert found["snapshotId"] == "snap-newer" + + +def test_a_matching_snapshot_older_than_the_baseline_is_refused() -> None: + """Guards the stale-snapshot bug this command exists to prevent. + + Were a trigger ever coalesced into an already-running scheduled job, its + ID would match snapshots collected before the trigger. + """ + assert snapshot_for_job([OLD, BASE], RECURRING_JOB_ID, newer_than=BASELINE) is None + + +def test_any_snapshot_qualifies_when_the_fabric_had_none() -> None: + found = snapshot_for_job([FRESH], JOB_ID, newer_than=None) + + assert found is not None + assert found["snapshotId"] == "snap-fresh" + + +def test_the_newest_snapshot_past_the_baseline_is_the_fallback() -> None: + assert snapshot_newer_than([OLD, BASE], BASELINE) is None + found = snapshot_newer_than([OLD, BASE, FRESH], BASELINE) + assert found is not None + assert found["snapshotId"] == "snap-fresh" + + +# -- waiting --------------------------------------------------------------- + + +def test_the_job_is_polled_by_id_and_not_by_type(make_client) -> None: + """`jobType` varies with the fabric, so only the ID is a reliable filter.""" + client = make_client(analysis_lab(snapshots=[FRESH, BASE])) + + client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + params = lab_params(client, SUMMARY_PATH) + assert params["jobId"] == JOB_ID + assert "jobTypes" not in params + + +def lab_params(client, path: str) -> httpx.QueryParams: + transport = client.client._transport + lab = transport.handler + return lab.requests_to(path)[0].url.params + + +@pytest.mark.parametrize("status", ["COMPLETE", "SUCCESS"]) +def test_both_success_values_are_accepted(make_client, status: str) -> None: + lab = analysis_lab(snapshots=[FRESH, BASE], jobs=[job(status)]) + client = make_client(lab) + + record = client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + assert record["snapshotId"] == "snap-fresh" + + +def test_a_failed_job_reports_the_api_error_message(make_client) -> None: + lab = analysis_lab( + snapshots=[BASE], + jobs=[job("FAILED", errorMessage="leaf-101 unreachable")], + ) + client = make_client(lab) + + with pytest.raises(JobError, match="leaf-101 unreachable"): + client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + +def test_a_partial_collection_explains_why_its_snapshot_is_unusable( + make_client, +) -> None: + lab = analysis_lab(snapshots=[BASE], jobs=[job("PARTIALLY_FAILED")]) + client = make_client(lab) + + with pytest.raises(JobError, match="not 'finished'"): + client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + +def test_a_job_that_is_never_listed_is_a_job_error(make_client) -> None: + """/jobs/summary answers 200 with no entries for a job that does not exist.""" + lab = analysis_lab(snapshots=[BASE], jobs=[]) + client = make_client(lab) + + with pytest.raises(JobError, match="never reported"): + client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + +def test_the_wait_continues_until_the_snapshot_lands(make_client) -> None: + """No snapshot is visible until the job is terminal, so the list lags.""" + lab = analysis_lab(snapshots=[[BASE], [BASE], [FRESH, BASE]]) + client = make_client(lab) + + record = client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + assert record["snapshotId"] == "snap-fresh" + assert len(lab.requests_to(SNAPSHOTS_PATH)) == 3 + + +def test_an_unfinished_snapshot_is_never_returned(make_client) -> None: + running = snapshot( + "snap-running", "2026-08-14T07:46:36Z", job_id=JOB_ID, status="inProgress" + ) + lab = analysis_lab(snapshots=[running, BASE]) + client = make_client(lab, job_timeout_minutes=1, poll_interval_seconds=30) + + with pytest.raises(JobError, match="did not produce a finished snapshot"): + client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + +def test_the_timeout_message_suggests_raising_it(make_client) -> None: + lab = analysis_lab(snapshots=[BASE]) + client = make_client(lab, job_timeout_minutes=1, poll_interval_seconds=30) + + with pytest.raises(JobError, match="raise --timeout"): + client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + +def test_an_unmatched_but_newer_snapshot_is_reported_with_a_warning( + make_client, +) -> None: + """Belt-and-braces: live runs match exactly, but a mismatch must be loud.""" + stranger = snapshot("snap-stranger", "2026-08-14T07:46:36Z", job_id="OTHER-JOB") + lab = analysis_lab(snapshots=[stranger, BASE]) + client = make_client(lab) + + record = client.wait_for_analysis_snapshot("FABRIC-A", JOB_ID, baseline=BASELINE) + + assert record["snapshotId"] == "snap-stranger" + assert any("OTHER-JOB" in item and JOB_ID in item for item in client.notices) + + +# -- the command ----------------------------------------------------------- + + +@pytest.fixture +def use_lab(monkeypatch: pytest.MonkeyPatch) -> object: + def install(lab: Lab) -> None: + def factory(config: object, **_: object) -> RealNDClient: + http = httpx.Client(transport=httpx.MockTransport(lab)) + return RealNDClient(config, http=http) # type: ignore[arg-type] + + monkeypatch.setattr( + "nac_analytics.products.nexus_dashboard.cli.NDClient", factory + ) + + return install + + +def test_analyze_prints_only_the_snapshot_id(use_lab, tmp_path, monkeypatch) -> None: + monkeypatch.chdir(tmp_path) + # The first listing is the pre-trigger baseline; the analysis lands after. + use_lab(analysis_lab(snapshots=[[BASE], [FRESH, BASE]])) + + result = runner.invoke(app, ["nd", "analyze"], env=ENV) + + assert result.exit_code == 0, result.output + assert result.output.strip().splitlines()[-1] == "snap-fresh" + + +def test_analyze_emits_the_whole_record_as_json(use_lab, tmp_path, monkeypatch) -> None: + monkeypatch.chdir(tmp_path) + use_lab(analysis_lab(snapshots=[[BASE], [FRESH, BASE]])) + + result = runner.invoke(app, ["nd", "analyze", "-o", "json"], env=ENV) + + assert result.exit_code == 0, result.output + assert JOB_ID in result.output + + +def test_a_snapshot_that_already_existed_is_never_reported( + use_lab, tmp_path, monkeypatch +) -> None: + """The baseline is read before the trigger, so `latest` cannot satisfy it.""" + monkeypatch.chdir(tmp_path) + use_lab(analysis_lab(snapshots=[FRESH, BASE])) + + result = runner.invoke(app, ["nd", "analyze"], env=ENV) + + assert result.exit_code == JobError.exit_code + assert "did not produce a finished snapshot" in result.output + + +def test_no_wait_prints_the_job_id_without_polling( + use_lab, tmp_path, monkeypatch +) -> None: + monkeypatch.chdir(tmp_path) + lab = analysis_lab(snapshots=[FRESH, BASE]) + use_lab(lab) + + result = runner.invoke(app, ["nd", "analyze", "--no-wait"], env=ENV) + + assert result.exit_code == 0, result.output + assert result.output.strip().splitlines()[-1] == JOB_ID + assert lab.requests_to(SUMMARY_PATH) == [] + # Only the fabric validation reads snapshots; no baseline, no polling. + assert lab.requests_to(SNAPSHOTS_PATH) == [] + + +def test_a_forbidden_trigger_names_the_roles_required( + use_lab, tmp_path, monkeypatch +) -> None: + monkeypatch.chdir(tmp_path) + use_lab( + analysis_lab( + snapshots=[BASE], + trigger=json_response({"message": "access denied"}, 403), + ) + ) + + result = runner.invoke(app, ["nd", "analyze"], env=ENV) + + assert result.exit_code == AuthError.exit_code + assert "fabric-admin" in result.output + assert "observer" in result.output