From ad6d00009868307d89199cbb46fb7caff923f718 Mon Sep 17 00:00:00 2001 From: Ali Bahaloo Date: Mon, 27 Jul 2026 12:39:40 -0700 Subject: [PATCH] feat: extract the assessment into a shared, headless surface (4.1.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The verdict rules, the Cisco client and the run orchestration were reachable only through `analyzer.run()`, which resolves credentials from the per-user config file, discovers an inventory in the working folder, prints to a terminal, writes an HTML file and exits the process. A second consumer could not call the assessment without inheriting all of that, so it would have had to reimplement the rules — and a reimplementation of Constitution IV is a second, divergent answer about the same firewall. Add `caia/assessment.py` holding `analyze(inventory, client)`: the whole check with no user interface attached. It takes a list of per-device dicts (whatever produced them) and a `psirt.Client` the caller owns; it prints nothing, writes nothing, never exits the process and never consults the working folder. `_load_catalogues`, `_run_checks` and `_assemble_devices` move there unchanged. Together with `matching`, `psirt` and `inventory`, this is now the package's documented shared surface. `analyzer` keeps the terminal run flow and supplies its output through the new `on_plan` and `on_progress` callbacks, so the printed run is byte-identical to before. Two properties are deliberate and covered by tests: - A catalogue fetch that fails raises, rather than returning an inventory of Needs review. Access failure must not read as a data-quality problem. - A failure once the catalogues are in hand still returns `partial_cause` with every answer already obtained, so a run is never thrown away. `analyze` also accepts pre-built catalogues, so a long-running consumer can hold them rather than spend requests re-fetching what never changes. `inventory` no longer calls `sys.exit` when openpyxl is missing; it raises ImportError. A library must not terminate its host process on import. No behaviour change for `caia` users: same prompts, same output, same report, same exit codes. 246 existing tests pass untouched; 25 added for the library path. Co-Authored-By: Claude Opus 5 --- README.md | 33 +++-- VERSION | 2 +- caia/analyzer.py | 117 +++------------- caia/assessment.py | 172 +++++++++++++++++++++++ caia/inventory.py | 9 +- tests/test_assessment.py | 294 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 516 insertions(+), 111 deletions(-) create mode 100644 caia/assessment.py create mode 100644 tests/test_assessment.py diff --git a/README.md b/README.md index 3f2a20c..57d9a3f 100644 --- a/README.md +++ b/README.md @@ -394,17 +394,28 @@ to suppress it everywhere. **Package layout** (`caia/`) -| Module | Responsibility | -| ------ | -------------- | -| `cli.py` | single entry point; refuses removed options, then dispatches `--help`/`--version`/`--update`/`--uninstall`/`--config` and the run | -| `analyzer.py` | run flow + interactive prompts (working-folder inventory + `output/`) | -| `inventory.py` | read the inventory workbook into per-firewall records | -| `psirt.py` | the only module that talks to Cisco's advisory API | -| `matching.py` | pure verdict logic: release normalisation, model→platform mapping, verdicts | -| `html_report.py` | render the self-contained HTML report | -| `config.py` | per-user configuration (paths, precedence, `--config`) | -| `version.py` | installed version, GitHub release discovery, uv-based update | -| `ui.py` | terminal colors + prompts (no dependencies) | +| Module | Responsibility | Shared surface | +| ------ | -------------- | -------------- | +| `assessment.py` | the assessment itself — `analyze()`, with no user interface attached | ✅ | +| `matching.py` | pure verdict logic: release normalisation, model→platform mapping, verdicts | ✅ | +| `psirt.py` | the only module that talks to Cisco's advisory API | ✅ | +| `inventory.py` | read the inventory workbook into per-firewall records | ✅ | +| `cli.py` | single entry point; refuses removed options, then dispatches `--help`/`--version`/`--update`/`--uninstall`/`--config` and the run | | +| `analyzer.py` | terminal run flow + interactive prompts (working-folder inventory + `output/`) | | +| `html_report.py` | render the self-contained HTML report | | +| `config.py` | per-user configuration (paths, precedence, `--config`) | | +| `version.py` | installed version, GitHub release discovery, uv-based update | | +| `ui.py` | terminal colors + prompts (no dependencies) | | + +**Shared surface.** The four modules marked above are the ones a consumer other than this +CLI is expected to import; the rest are the terminal program built on top of them. The +entry point is `assessment.analyze(inventory, client)` — it takes a list of per-device +dicts (whatever produced them) plus a `psirt.Client` the caller owns, prints nothing, +writes nothing, and never exits the process. `analyzer.run()` is one caller of it; the +terminal output you see is supplied by the `on_plan` and `on_progress` callbacks it passes +in. The split exists so that a second consumer cannot drift from this tool's verdicts: +the rules that read Cisco's data live in one place, and anything that reimplements them +reimplements the safety property in Constitution IV. The version is single-sourced from the committed `VERSION` file into the package metadata and reported at runtime via `importlib.metadata`. Devices sharing a product family, hardware diff --git a/VERSION b/VERSION index fcdb2e1..ee74734 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -4.0.0 +4.1.0 diff --git a/caia/analyzer.py b/caia/analyzer.py index bc0eac4..c10ff5f 100644 --- a/caia/analyzer.py +++ b/caia/analyzer.py @@ -7,9 +7,13 @@ PSIRT advisory API — the same data behind Cisco's public Software Checker — and the answer is reported directly. Writes one timestamped, self-contained HTML report. -This module holds the run flow; the command-line entry point is `caia.cli`. Inputs are -read from, and the report is written to, the current working folder (an `output/` -subfolder), so the tool works from any folder once installed. +This module holds the *terminal* run flow: credential resolution, inventory discovery and +its prompts, progress output, and writing the report into the current working folder (an +`output/` subfolder), so the tool works from any folder once installed. The command-line +entry point is `caia.cli`. + +The assessment itself lives in `caia.assessment` and is shared with any other consumer of +this package, so a verdict cannot differ between the terminal and anything else. """ from __future__ import annotations @@ -18,7 +22,7 @@ import sys from pathlib import Path -from caia import config, html_report, matching, psirt +from caia import assessment, config, html_report, matching, psirt from caia import inventory as inv from caia import ui @@ -138,36 +142,6 @@ def _resolve_cisco_credentials(): return client_id, client_secret -def _families_in(inventory): - """Distinct lower-cased product families present in the inventory.""" - seen = [] - for device in inventory: - family = str(device.get("family") or "").strip().lower() - if family and family not in seen: - seen.append(family) - return seen - - -def _load_catalogues(client, families): - """Fetch the platform and release catalogues Cisco publishes, per family. - - Only families Cisco can answer about are fetched; anything else is resolved locally to - Needs review, so an inventory containing switches does not fail the run. - """ - catalogues = matching.Catalogues() - for family in families: - if not matching.family_supported(family): - continue - catalogues.add_platforms(family, client.get_platforms(family)) - catalogues.add_releases(family, client.get_os_data(family)) - return catalogues - - -def _describe_check(check): - os_type, alias, release = check - return f"{os_type}/{alias or '-'}/{release}" - - def run(args): """Assess the whole inventory against Cisco's published advisory data. @@ -196,8 +170,21 @@ def run(args): client = psirt.Client(client_id, client_secret) ui.plain() ui.system("Fetching Cisco platform and release catalogues ...") + + def report_plan(check_count, unresolved_count): + ui.ok(f"{len(inventory)} firewalls collapse to {ui.bold(str(check_count))} distinct " + f"checks.") + if unresolved_count: + ui.info(f"{unresolved_count} firewall(s) need review before Cisco can be asked.") + + def report_progress(index, total, check): + ui.plain(ui.dim(f" [{index}/{total}] {assessment.describe_check(check)}")) + + # Only the catalogue fetch can raise here; a failure once they are in hand comes back + # as `partial_cause` so the answers already obtained survive (assessment.analyze). try: - catalogues = _load_catalogues(client, _families_in(inventory)) + devices, partial_cause = assessment.analyze( + inventory, client, on_plan=report_plan, on_progress=report_progress) except psirt.PsirtAuthError as e: die(f"Cisco rejected the configured credentials: {e}", hint="Re-check them with 'caia --config'.") @@ -205,16 +192,7 @@ def run(args): die(f"Could not reach Cisco: {e}", hint="Check your connection or proxy and try again.") - checks, unresolved = matching.build_checks(inventory, catalogues) - ui.ok(f"{len(inventory)} firewalls collapse to {ui.bold(str(len(checks)))} distinct " - f"checks.") - if unresolved: - ui.info(f"{len(unresolved)} firewall(s) need review before Cisco can be asked.") - - results, partial_cause = _run_checks(client, checks) - # 4. Verdicts. - devices = _assemble_devices(inventory, catalogues, checks, unresolved, results) summary = _build_summary(inv_path, devices, partial_cause) _print_verdict_tally(summary) @@ -230,57 +208,6 @@ def run(args): return 0 -def _run_checks(client, checks): - """Ask Cisco once per check. Returns ({check: (outcome, advisories)}, partial_cause). - - A failure part-way through stops further requests but keeps every answer already - obtained: the remaining devices become Needs review and the report is flagged partial - rather than the run being thrown away (FR-016). - """ - results = {} - total = len(checks) - for index, check in enumerate(checks, 1): - ui.plain(ui.dim(f" [{index}/{total}] {_describe_check(check)}")) - os_type, alias, release = check - try: - outcome, advisories = client.get_advisories(os_type, release, alias) - except psirt.PsirtAuthError as e: - return results, f"Cisco rejected the access token part-way through: {e}" - except psirt.PsirtUnreachableError as e: - return results, f"Cisco became unreachable part-way through: {e}" - results[check] = (outcome, advisories) - return results, None - - -def _assemble_devices(inventory, catalogues, checks, unresolved, results): - """Attach a verdict, reason and advisories to every inventory row.""" - reasons = {id(device): reason for device, reason in unresolved} - by_device = {} - for check, members in checks.items(): - outcome, advisories = results.get(check, (None, None)) - for device in members: - by_device[id(device)] = (check, outcome, advisories) - - devices = [] - for device in inventory: - entry = dict(device) - reason = reasons.get(id(device)) - if reason is not None: - entry["verdict"] = matching.NEEDS_REVIEW - entry["reason"] = reason - entry["advisories"] = [] - else: - check, outcome, advisories = by_device.get(id(device), (None, None, None)) - verdict, reason = matching.verdict_from_outcome( - check, catalogues, outcome, advisories, - raw_release=device.get("version")) - entry["verdict"] = verdict - entry["reason"] = reason - entry["advisories"] = list(advisories or []) if verdict == matching.IMPACTED else [] - devices.append(entry) - return devices - - def _build_summary(inv_path, devices, partial_cause): counts = {matching.IMPACTED: 0, matching.NOT_IMPACTED: 0, matching.NEEDS_REVIEW: 0} for device in devices: diff --git a/caia/assessment.py b/caia/assessment.py new file mode 100644 index 0000000..ad4c1b1 --- /dev/null +++ b/caia/assessment.py @@ -0,0 +1,172 @@ +"""The assessment itself, with no user interface attached. + +`analyze()` is the whole check: fetch Cisco's catalogues, collapse the inventory into the +distinct questions Cisco needs to be asked, ask them, and attach a verdict to every device. +It prints nothing, writes no file, never exits the process, and never consults the working +folder — so the same code path serves the terminal and any other consumer. + +This module and `matching`, `psirt` and `inventory` are the package's **shared surface**: +the modules a consumer other than the CLI is expected to import. `analyzer`, `cli`, `ui`, +`config`, `version` and `html_report` are the terminal program built on top of them, and +are not part of that surface. + +That split exists so a second consumer cannot drift from the CLI's verdicts. Cisco's +published data is the authority (Constitution IV) and the rules that read it live here +exactly once; anything that reimplements them is reimplementing the safety property. +""" + +from __future__ import annotations + +from caia import matching, psirt + + +# --------------------------------------------------------------------------- # +# Inspecting an inventory +# --------------------------------------------------------------------------- # +def families_in(inventory): + """Distinct lower-cased product families present in the inventory. + + Order-preserving, so a caller fetching catalogues per family does so predictably. + """ + seen = [] + for device in inventory: + family = str(device.get("family") or "").strip().lower() + if family and family not in seen: + seen.append(family) + return seen + + +def describe_check(check): + """A short label for a check, e.g. `ftd/ISA3000/7.4.2.3`. For progress reporting.""" + os_type, alias, release = check + return f"{os_type}/{alias or '-'}/{release}" + + +# --------------------------------------------------------------------------- # +# Catalogues +# --------------------------------------------------------------------------- # +def load_catalogues(client, families): + """Fetch the platform and release catalogues Cisco publishes, per family. + + Only families Cisco can answer about are fetched; anything else is resolved locally to + Needs review, so an inventory containing switches does not fail the run. + + These catalogues are identical for every caller and change rarely, so a long-running + consumer may fetch them once and pass the result to `analyze(catalogues=...)` instead + of paying for them on every assessment. + """ + catalogues = matching.Catalogues() + for family in families: + if not matching.family_supported(family): + continue + catalogues.add_platforms(family, client.get_platforms(family)) + catalogues.add_releases(family, client.get_os_data(family)) + return catalogues + + +# --------------------------------------------------------------------------- # +# The assessment +# --------------------------------------------------------------------------- # +def analyze(inventory, client, catalogues=None, on_plan=None, on_progress=None): + """Assess an inventory against Cisco's published advisory data. + + Args: + inventory: list of per-device dicts, as produced by `inventory.load_inventory` — + each carrying at least `name`, `model`, `family` and `version`. Any other keys are + carried through onto the corresponding result entry untouched, so a caller's own + columns survive into a report. + client: a `psirt.Client`, or anything exposing its `get_platforms`, `get_os_data` + and `get_advisories` methods. The caller owns its credentials and its pacing. + catalogues: an optional pre-built `matching.Catalogues`. When omitted, catalogues are + fetched from Cisco for the families present in `inventory`. + on_plan: optional callable(check_count, unresolved_count), called once after + deduplication and before the first advisory request. `check_count` is exactly how + many requests the run will make, so a caller can estimate how long it will take. + on_progress: optional callable(index, total, check), called before each advisory + request. `index` is 1-based; pass `check` to `describe_check` for a label. + + Returns: + (devices, partial_cause). + + `devices` holds one dict per inventory row, in inventory order — a copy of the input + row plus `verdict`, `reason`, and `advisories` (populated only for an Impacted + verdict). Every row appears exactly once; none is ever dropped. + + `partial_cause` is None for a complete run, or a sentence explaining why it stopped + early. A run that stops early keeps every answer already obtained and leaves the + devices it never reached as Needs review (FR-016) — it is never discarded. + + Raises: + psirt.PsirtAuthError, psirt.PsirtUnreachableError: when the catalogues cannot be + fetched. Without them nothing can be confirmed, and an inventory of Needs review + would read as a data-quality problem rather than the access failure it is — so + this fails loudly instead. A failure *after* the catalogues are in hand is + reported through `partial_cause`, never raised. + """ + if catalogues is None: + catalogues = load_catalogues(client, families_in(inventory)) + + checks, unresolved = matching.build_checks(inventory, catalogues) + if on_plan is not None: + on_plan(len(checks), len(unresolved)) + + results, partial_cause = _run_checks(client, checks, on_progress) + devices = _assemble_devices(inventory, catalogues, checks, unresolved, results) + return devices, partial_cause + + +def _run_checks(client, checks, on_progress=None): + """Ask Cisco once per check. Returns ({check: (outcome, advisories)}, partial_cause). + + A failure part-way through stops further requests but keeps every answer already + obtained: the remaining devices become Needs review and the run is reported partial + rather than being thrown away (FR-016). + """ + results = {} + total = len(checks) + for index, check in enumerate(checks, 1): + if on_progress is not None: + on_progress(index, total, check) + os_type, alias, release = check + try: + outcome, advisories = client.get_advisories(os_type, release, alias) + except psirt.PsirtAuthError as e: + return results, f"Cisco rejected the access token part-way through: {e}" + except psirt.PsirtUnreachableError as e: + return results, f"Cisco became unreachable part-way through: {e}" + results[check] = (outcome, advisories) + return results, None + + +def _assemble_devices(inventory, catalogues, checks, unresolved, results): + """Attach a verdict, reason and advisories to every inventory row. + + Devices are tracked by object identity, which is why `analyze` owns the whole span + from `build_checks` to here: the same dicts must flow through both. Callers only ever + hand in a list and receive a list back, so they cannot break that. + """ + reasons = {id(device): reason for device, reason in unresolved} + by_device = {} + for check, members in checks.items(): + outcome, advisories = results.get(check, (None, None)) + for device in members: + by_device[id(device)] = (check, outcome, advisories) + + devices = [] + for device in inventory: + entry = dict(device) + reason = reasons.get(id(device)) + if reason is not None: + entry["verdict"] = matching.NEEDS_REVIEW + entry["reason"] = reason + entry["advisories"] = [] + else: + check, outcome, advisories = by_device.get(id(device), (None, None, None)) + verdict, reason = matching.verdict_from_outcome( + check, catalogues, outcome, advisories, + raw_release=device.get("version")) + entry["verdict"] = verdict + entry["reason"] = reason + entry["advisories"] = list(advisories or []) if verdict == matching.IMPACTED else [] + devices.append(entry) + return devices diff --git a/caia/inventory.py b/caia/inventory.py index a9194d3..0562784 100644 --- a/caia/inventory.py +++ b/caia/inventory.py @@ -9,12 +9,13 @@ from __future__ import annotations -import sys - try: import openpyxl -except ImportError: # pragma: no cover - sys.exit("openpyxl is required. Reinstall the tool with uv.") +except ImportError as exc: # pragma: no cover - only a broken install reaches this + # Raised, not exited: this module is imported by consumers other than the command-line + # program, and a library must never terminate its host process on import. + raise ImportError("openpyxl is required to read an inventory workbook. " + "Reinstall the tool with uv.") from exc class InventoryError(Exception): diff --git a/tests/test_assessment.py b/tests/test_assessment.py new file mode 100644 index 0000000..e6d9ddb --- /dev/null +++ b/tests/test_assessment.py @@ -0,0 +1,294 @@ +"""Tests for the shared assessment surface (`caia.assessment`). No network access. + +These exercise `analyze()` the way a consumer other than the CLI does: an inventory built +in memory rather than read from a workbook, credentials the caller already holds, and no +terminal to print to. `test_api_flow` covers the same engine reached through the terminal +program, so between them the CLI path and the library path are both pinned. + +The contract asserted here is what keeps a second consumer honest: `analyze()` prints +nothing, exits nothing, writes nothing, does not mutate its input, and either returns a +verdict for every row or raises — it never returns an inventory of Needs review that could +be mistaken for a data-quality problem when Cisco access is what actually failed. +""" + +from __future__ import annotations + +import contextlib +import io +import unittest + +from caia import assessment +from caia import matching as m +from caia import psirt +from tests.fixtures import psirt_responses as fx + + +def device(name, model, family, version, priority=2, **extra): + """One device dict in exactly the shape `inventory.load_inventory` produces.""" + row = {"name": name, "model": model, "family": family, "version": version, + "priority": priority, "raw": {"FirewallName": name}} + row.update(extra) + return row + + +# One row per path a consumer must be able to reach without a workbook. +INVENTORY = [ + device("WEB-IMPACT-1", "ISA-3000", "FTD", "7.4.2.3", 1), + # Shares (ftd, ISA3000, 7.4.2.3) with the row above -> one request serves both. + device("WEB-IMPACT-2", "ISA-3000", "FTD", "7.4.2.3", 2), + device("WEB-CLEAN", "ISA-3000", "FTD", "7.9.9", 3), + # FMC takes no platform alias at all. + device("WEB-FMC", "Secure FMC 1600", "FMC", "7.4.2.3", 2), + # Parenthetical shorthand must normalise before Cisco is asked. + device("WEB-RELUNKNOWN", "ASA 5525", "ASA", "9.14(4)24", 2), + # Resolved locally, so these two never reach Cisco. + device("WEB-BADMODEL", "Unknown Appliance 9", "ASA", "9.16.4.67", 2), + device("WEB-BADFAMILY", "Catalyst 9300", "SWITCH", "17.9.4", 3), +] + + +class FakeClient: + """Answers from the fixtures, recording catalogue and advisory calls separately.""" + + def __init__(self, fail_after=None, catalogue_error=None): + self.asked = [] + self.catalogue_calls = [] + self.fail_after = fail_after + self.catalogue_error = catalogue_error + + def get_platforms(self, os_type): + self.catalogue_calls.append(("platforms", os_type)) + if self.catalogue_error is not None: + raise self.catalogue_error + return fx.PLATFORMS.get(os_type, []) + + def get_os_data(self, os_type): + self.catalogue_calls.append(("os_data", os_type)) + if self.catalogue_error is not None: + raise self.catalogue_error + return fx.OS_DATA.get(os_type, []) + + def get_advisories(self, os_type, release, platform_alias=None): + self.asked.append((os_type, platform_alias, release)) + if self.fail_after is not None and len(self.asked) > self.fail_after: + raise psirt.PsirtUnreachableError("Cisco became unreachable (URLError).") + if os_type == "asa" and release.startswith("9.14."): + return m.INVALID_VERSION, [] + if release == "7.9.9": + return m.NO_DATA, [] + return m.HAS_ADVISORIES, fx.advisories_response( + 2, platform_name=_display_name(os_type, platform_alias))["advisories"] + + +def _display_name(os_type, alias): + for row in fx.PLATFORMS.get(os_type, []): + if row["platformAlias"] == alias: + return row["name"] + return "Cisco Secure Firewall Management Center (FMC) Appliances" + + +def verdicts(devices): + return {d["name"]: d["verdict"] for d in devices} + + +class AnalyzeTests(unittest.TestCase): + """The library path: an in-memory inventory and a caller-owned client.""" + + def test_every_row_gets_exactly_one_verdict_in_inventory_order(self): + devices, partial = assessment.analyze(INVENTORY, FakeClient()) + self.assertIsNone(partial) + self.assertEqual([d["name"] for d in devices], [d["name"] for d in INVENTORY]) + for entry in devices: + self.assertIn(entry["verdict"], + (m.IMPACTED, m.NOT_IMPACTED, m.NEEDS_REVIEW)) + + def test_verdicts_match_the_outcome_table(self): + devices, _ = assessment.analyze(INVENTORY, FakeClient()) + self.assertEqual(verdicts(devices), { + "WEB-IMPACT-1": m.IMPACTED, + "WEB-IMPACT-2": m.IMPACTED, + "WEB-CLEAN": m.NOT_IMPACTED, + "WEB-FMC": m.IMPACTED, + "WEB-RELUNKNOWN": m.NEEDS_REVIEW, + "WEB-BADMODEL": m.NEEDS_REVIEW, + "WEB-BADFAMILY": m.NEEDS_REVIEW, + }) + + def test_identical_devices_cost_one_request_between_them(self): + client = FakeClient() + assessment.analyze(INVENTORY, client) + self.assertEqual(client.asked.count(("ftd", "ISA3000", "7.4.2.3")), 1) + + def test_locally_resolved_devices_never_reach_cisco(self): + client = FakeClient() + assessment.analyze(INVENTORY, client) + self.assertNotIn("switch", [os_type for os_type, _a, _r in client.asked]) + self.assertEqual([r for _o, _a, r in client.asked].count("9.16.4.67"), 0) + + def test_advisories_are_attached_only_to_impacted_devices(self): + devices, _ = assessment.analyze(INVENTORY, FakeClient()) + for entry in devices: + if entry["verdict"] == m.IMPACTED: + self.assertTrue(entry["advisories"]) + else: + self.assertEqual(entry["advisories"], []) + + def test_caller_keys_are_carried_through_untouched(self): + inventory = [device("WEB-EXTRA", "ISA-3000", "FTD", "7.4.2.3", + site="Vancouver", client_id=42)] + devices, _ = assessment.analyze(inventory, FakeClient()) + self.assertEqual(devices[0]["site"], "Vancouver") + self.assertEqual(devices[0]["client_id"], 42) + self.assertEqual(devices[0]["raw"], {"FirewallName": "WEB-EXTRA"}) + + def test_the_callers_inventory_is_not_mutated(self): + before = [dict(d) for d in INVENTORY] + assessment.analyze(INVENTORY, FakeClient()) + self.assertEqual(INVENTORY, before) + for entry in INVENTORY: + self.assertNotIn("verdict", entry) + + def test_nothing_is_printed(self): + buf = io.StringIO() + with contextlib.redirect_stdout(buf): + assessment.analyze(INVENTORY, FakeClient()) + self.assertEqual(buf.getvalue(), "") + + +class CallbackTests(unittest.TestCase): + """`on_plan` and `on_progress` are what a caller builds a progress display from.""" + + def test_on_plan_reports_request_count_and_unresolved_count(self): + seen = [] + assessment.analyze(INVENTORY, FakeClient(), + on_plan=lambda checks, unresolved: seen.append((checks, unresolved))) + self.assertEqual(len(seen), 1) + check_count, unresolved_count = seen[0] + # WEB-BADMODEL and WEB-BADFAMILY resolve locally; the rest collapse to 4 questions. + self.assertEqual(unresolved_count, 2) + self.assertEqual(check_count, 4) + + def test_on_plan_runs_before_any_advisory_request(self): + client = FakeClient() + asked_at_plan = [] + assessment.analyze(INVENTORY, client, + on_plan=lambda *_a: asked_at_plan.append(len(client.asked))) + self.assertEqual(asked_at_plan, [0]) + + def test_on_plan_check_count_equals_the_requests_actually_made(self): + client = FakeClient() + counts = [] + assessment.analyze(INVENTORY, client, on_plan=lambda c, _u: counts.append(c)) + self.assertEqual(counts[0], len(client.asked)) + + def test_on_progress_is_called_once_per_check_with_a_1_based_index(self): + calls = [] + assessment.analyze(INVENTORY, FakeClient(), + on_progress=lambda i, total, check: calls.append((i, total, check))) + totals = {total for _i, total, _c in calls} + self.assertEqual(totals, {len(calls)}) + self.assertEqual([i for i, _t, _c in calls], list(range(1, len(calls) + 1))) + + def test_analyze_works_with_no_callbacks_at_all(self): + devices, partial = assessment.analyze(INVENTORY, FakeClient()) + self.assertIsNone(partial) + self.assertEqual(len(devices), len(INVENTORY)) + + +class CataloguesTests(unittest.TestCase): + """A long-running consumer may hold catalogues rather than re-fetch them.""" + + def test_supplied_catalogues_are_used_and_nothing_is_fetched(self): + primed = FakeClient() + catalogues = assessment.load_catalogues( + primed, assessment.families_in(INVENTORY)) + + client = FakeClient() + devices, partial = assessment.analyze(INVENTORY, client, catalogues=catalogues) + + self.assertEqual(client.catalogue_calls, []) + self.assertIsNone(partial) + self.assertEqual(verdicts(devices), verdicts( + assessment.analyze(INVENTORY, FakeClient())[0])) + + def test_catalogues_are_fetched_only_for_supported_families(self): + client = FakeClient() + assessment.analyze(INVENTORY, client) + fetched = {os_type for _kind, os_type in client.catalogue_calls} + self.assertEqual(fetched, {"ftd", "fmc", "asa"}) + self.assertNotIn("switch", fetched) + + def test_families_in_is_lowercased_deduplicated_and_ordered(self): + self.assertEqual(assessment.families_in(INVENTORY), + ["ftd", "fmc", "asa", "switch"]) + + def test_families_in_ignores_blank_families(self): + self.assertEqual( + assessment.families_in([device("X", "ISA-3000", "", "7.4.2.3"), + device("Y", "ISA-3000", " ", "7.4.2.3")]), + []) + + +class FailureTests(unittest.TestCase): + """Access failure must never be mistaken for a clean or merely-unassessable fleet.""" + + def test_catalogue_auth_failure_raises_rather_than_returning_needs_review(self): + client = FakeClient( + catalogue_error=psirt.PsirtAuthError("Cisco rejected the credentials.")) + with self.assertRaises(psirt.PsirtAuthError): + assessment.analyze(INVENTORY, client) + self.assertEqual(client.asked, []) + + def test_catalogue_unreachable_raises(self): + client = FakeClient( + catalogue_error=psirt.PsirtUnreachableError("Cisco could not be reached.")) + with self.assertRaises(psirt.PsirtUnreachableError): + assessment.analyze(INVENTORY, client) + + def test_failure_part_way_through_keeps_earlier_answers_and_reports_partial(self): + devices, partial = assessment.analyze(INVENTORY, FakeClient(fail_after=1)) + + self.assertIsNotNone(partial) + self.assertIn("unreachable", partial.lower()) + self.assertEqual(len(devices), len(INVENTORY)) + + got = verdicts(devices) + # The first question was answered, so the devices behind it keep their verdict. + self.assertEqual(got["WEB-IMPACT-1"], m.IMPACTED) + self.assertEqual(got["WEB-IMPACT-2"], m.IMPACTED) + # Everything never asked about is Needs review, never Not impacted. + self.assertEqual(got["WEB-CLEAN"], m.NEEDS_REVIEW) + + def test_a_partial_run_never_yields_not_impacted_for_an_unasked_device(self): + devices, partial = assessment.analyze(INVENTORY, FakeClient(fail_after=0)) + self.assertIsNotNone(partial) + self.assertNotIn(m.NOT_IMPACTED, [d["verdict"] for d in devices]) + self.assertEqual(len(devices), len(INVENTORY)) + + def test_every_needs_review_device_carries_a_reason(self): + devices, _ = assessment.analyze(INVENTORY, FakeClient(fail_after=1)) + for entry in devices: + if entry["verdict"] == m.NEEDS_REVIEW: + self.assertIsInstance(entry["reason"], m.Reason) + self.assertTrue(entry["reason"].kind) + + def test_an_empty_inventory_is_not_an_error(self): + client = FakeClient() + devices, partial = assessment.analyze([], client) + self.assertEqual(devices, []) + self.assertIsNone(partial) + self.assertEqual(client.asked, []) + + +class DescribeCheckTests(unittest.TestCase): + def test_includes_family_alias_and_release(self): + self.assertEqual(assessment.describe_check(("ftd", "ISA3000", "7.4.2.3")), + "ftd/ISA3000/7.4.2.3") + + def test_a_family_without_a_platform_alias_shows_a_placeholder(self): + self.assertEqual(assessment.describe_check(("fmc", None, "7.4.2.3")), + "fmc/-/7.4.2.3") + + +if __name__ == "__main__": # pragma: no cover + unittest.main()