diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 134da94..194a0cc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -29,4 +29,4 @@ jobs: pip install -r requirements-dev.txt - name: Run tests - run: pytest -q + run: pytest -q --cov=. --cov-report=term-missing --cov-fail-under=80 diff --git a/README.md b/README.md index 1ea257a..6530c11 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,8 @@ Environment variables (optional): | `FC_INVENTORY_PORT` | `5000` | Local web UI port | | `FC_INVENTORY_PASSWORD` | _(unset)_ | Password for headless `collect` mode (avoids putting it on the command line) | | `FC_INVENTORY_DISABLE_UPDATE_CHECK` | _(unset)_ | Set to any value to disable the "new version available" check in the web UI | +| `FC_INVENTORY_VERIFY_SSL` | _(unset)_ | Set to `1`/`true` to verify the VRM's TLS certificate (off by default for self-signed certs) | +| `FC_INVENTORY_CA_BUNDLE` | _(unset)_ | Path to a CA bundle/cert used to verify the VRM (enables verification against your own trust chain) | Example: ```cmd diff --git a/app.py b/app.py index 7cfefda..177e6b8 100644 --- a/app.py +++ b/app.py @@ -13,6 +13,7 @@ import threading import webbrowser from datetime import datetime +from urllib.parse import urlparse from flask import Flask, render_template, request, jsonify, send_file @@ -58,6 +59,26 @@ } +@app.before_request +def _reject_cross_origin(): + """Lightweight CSRF / drive-by protection for state-changing requests. + + The tool has no login, so a malicious web page could otherwise POST to the + local server (CSRF). Browsers always attach an ``Origin`` header on + cross-origin requests; reject those whose origin host does not match the + request host. Non-browser clients (curl, the headless CLI) send no Origin + and are unaffected. + """ + if request.method in ("GET", "HEAD", "OPTIONS"): + return None + origin = request.headers.get("Origin") + if not origin: + return None + if urlparse(origin).netloc != request.host: + return jsonify({"error": "Cross-origin request rejected."}), 403 + return None + + def _run_collection(collector): """Background thread: collect data and build Excel file.""" try: @@ -104,13 +125,21 @@ def start_collection(): # Parse request body = request.get_json(silent=True) or {} host = body.get("host", "").strip() - port = int(body.get("port", 7443)) username = body.get("username", "").strip() password = body.get("password", "") if not host or not username or not password: return jsonify({"error": "Host, username, and password are required."}), 400 + # Validate the port: reject non-numeric or out-of-range values up front + # instead of failing deep inside the request handler. + try: + port = int(body.get("port", 7443)) + except (TypeError, ValueError): + return jsonify({"error": "Port must be a number."}), 400 + if not 1 <= port <= 65535: + return jsonify({"error": "Port must be between 1 and 65535."}), 400 + # Clean up old output file if current_job["output_file"] and os.path.exists(current_job["output_file"]): try: diff --git a/excel_builder.py b/excel_builder.py index 1bc00a1..b85a836 100644 --- a/excel_builder.py +++ b/excel_builder.py @@ -19,6 +19,23 @@ "vHost", "vCluster", "vDatastore", "vSwitch", ] +# Leading characters that spreadsheet apps may interpret as a formula. +# Inventory values come from FusionCompute (a VM name/description could be +# attacker-controlled), so neutralize them to prevent CSV/Excel injection. +_FORMULA_TRIGGERS = ("=", "+", "-", "@") + + +def _sanitize_cell(value): + """Defang spreadsheet-formula injection in string cell values. + + A string beginning with '=', '+', '-' or '@' is prefixed with a single + quote so Excel/Sheets render it as literal text instead of evaluating it. + Non-string values are returned unchanged. + """ + if isinstance(value, str) and value[:1] in _FORMULA_TRIGGERS: + return "'" + value + return value + def build_excel(data, output_path): """ @@ -52,9 +69,9 @@ def build_excel(data, output_path): headers.append(key) ws.append(headers) - # Write data rows + # Write data rows (sanitized against formula injection) for row in rows: - ws.append([row.get(h, "") for h in headers]) + ws.append([_sanitize_cell(row.get(h, "")) for h in headers]) # Apply header styling _style_header(ws) diff --git a/fc_client.py b/fc_client.py index af5f20b..4695907 100644 --- a/fc_client.py +++ b/fc_client.py @@ -5,19 +5,36 @@ import hashlib import logging +import os import requests import urllib3 logger = logging.getLogger(__name__) -# Suppress SSL warnings for self-signed certificates -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +def _resolve_verify(verify): + """Decide the requests TLS-verify setting. + + Precedence: explicit argument > FC_INVENTORY_CA_BUNDLE (path to a CA file) + > FC_INVENTORY_VERIFY_SSL (truthy enables verification) > default False. + + FusionCompute VRM commonly uses self-signed certs, so verification is OFF + by default for backward compatibility, but operators can opt in to a + proper trust chain to defend against man-in-the-middle attacks. + """ + if verify is not None: + return verify + ca_bundle = os.environ.get("FC_INVENTORY_CA_BUNDLE") + if ca_bundle: + return ca_bundle + flag = os.environ.get("FC_INVENTORY_VERIFY_SSL", "").strip().lower() + return flag in ("1", "true", "yes", "on") class FCClient: """Client for FusionCompute VRM REST API.""" - def __init__(self, host, username, password, port=7443): + def __init__(self, host, username, password, port=7443, verify=None): # Strip protocol prefix if user entered it (e.g. https://10.0.0.1) host = host.replace("https://", "").replace("http://", "").strip("/") self.host = host @@ -26,7 +43,10 @@ def __init__(self, host, username, password, port=7443): self.password = password self.base_url = f"https://{host}:{port}" self.session = requests.Session() - self.session.verify = False + self.session.verify = _resolve_verify(verify) + # Only suppress the insecure-request warning when verification is off. + if not self.session.verify: + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) self.session.headers.update({ "Content-Type": "application/json; charset=UTF-8", "Accept": "application/json;version=v1.0;charset=UTF-8", @@ -111,10 +131,6 @@ def login(self): ) logger.debug(f" -> HTTP {resp.status_code}") - try: - logger.debug(f" -> Body: {resp.text[:300]}") - except Exception: - pass if resp.status_code == 200: self.port = port @@ -122,6 +138,14 @@ def login(self): logger.info(f"Login OK! version={ver}, base_url={self.base_url}") return self._extract_token(resp, label) + # Log the body only on failure: a successful login + # response carries the session token, which must not + # be written to the log file. + try: + logger.debug(f" -> Body: {resp.text[:300]}") + except Exception: + pass + body = resp.text[:200] logger.warning(f" -> Failed ({resp.status_code}): {body}") last_error = f"HTTP {resp.status_code}: {body}" @@ -206,6 +230,24 @@ def _get(self, path, params=None): resp.raise_for_status() return resp.json() + @staticmethod + def _extract_list(data, *keys): + """Extract a list from an API response that may be a dict or bare list. + + FusionCompute responses vary by version: some endpoints return a + wrapper dict ({"vms": [...]}), others a top-level list. Tries each + candidate key in order on a dict, falls back to the response itself + when it is already a list, and always returns a list. + """ + if isinstance(data, list): + return data + if isinstance(data, dict): + for key in keys: + value = data.get(key) + if value is not None: + return value + return [] + def _get_all(self, path, result_key): """Fetch all pages of a paginated list endpoint.""" items = [] @@ -213,14 +255,14 @@ def _get_all(self, path, result_key): limit = 100 while True: data = self._get(path, params={"offset": offset, "limit": limit}) - # Try expected key, fallback to common alternatives + # If top-level is a list, use it directly (no pagination metadata). + if isinstance(data, list): + return data + # Try expected key, fallback to common alternatives. batch = data.get(result_key) if batch is None: batch = data.get("items", data.get("result", [])) if not batch: - # If top-level is a list, use it directly - if isinstance(data, list): - return data break items.extend(batch) total = data.get("total", len(items)) @@ -234,7 +276,7 @@ def _get_all(self, path, result_key): def get_sites(self): """Get all sites.""" data = self._get("/sites") - return data.get("sites", []) + return self._extract_list(data, "sites") # ── Cluster ────────────────────────────────────────── @@ -270,18 +312,18 @@ def get_vm_detail(self, vm_uri): def get_vm_nics(self, vm_uri): """Get network interfaces of a VM.""" data = self._get(f"{vm_uri}/nics") - return data.get("nics", data.get("items", [])) + return self._extract_list(data, "nics", "items") def get_vm_disks(self, vm_uri): """Get disk volumes of a VM.""" # Try /volumes first, fallback to /disks try: data = self._get(f"{vm_uri}/volumes") - return data.get("volumes", data.get("items", [])) + return self._extract_list(data, "volumes", "items") except Exception: try: data = self._get(f"{vm_uri}/disks") - return data.get("disks", data.get("items", [])) + return self._extract_list(data, "disks", "items") except Exception: return [] @@ -297,18 +339,14 @@ def get_dvswitches(self, site_uri): """Get all distributed virtual switches.""" data = self._get(f"{site_uri}/dvswitchs") logger.debug(f"DVSwitch response keys: {list(data.keys()) if isinstance(data, dict) else 'not a dict'}") - result = data.get("dvswitchs", data.get("dvSwitchs", data.get("items", []))) - if not result and isinstance(data, list): - result = data + result = self._extract_list(data, "dvswitchs", "dvSwitchs", "items") logger.info(f"Found {len(result)} DVSwitches") return result def get_portgroups(self, dvswitch_uri): """Get port groups of a DVSwitch. dvswitch_uri = full URI from dvswitch list.""" data = self._get(f"{dvswitch_uri}/portgroups") - result = data.get("portgroups", data.get("portGroups", data.get("items", []))) - if not result and isinstance(data, list): - result = data + result = self._extract_list(data, "portgroups", "portGroups", "items") logger.info(f"Found {len(result)} port groups under {dvswitch_uri}") return result @@ -316,9 +354,7 @@ def get_site_portgroups(self, site_uri): """Get all port groups in a site (fallback when DVSwitch listing is empty).""" try: data = self._get(f"{site_uri}/portgroups") - result = data.get("portgroups", data.get("portGroups", data.get("items", []))) - if not result and isinstance(data, list): - result = data + result = self._extract_list(data, "portgroups", "portGroups", "items") logger.info(f"Found {len(result)} port groups at site level") return result except Exception as e: diff --git a/requirements-dev.txt b/requirements-dev.txt index 044d133..4557cf3 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,2 +1,3 @@ -r requirements.txt pytest>=8,<9 +pytest-cov>=5,<6 diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..792ee56 --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,407 @@ +"""Tests for the Flask routes and the headless CLI entrypoint. + +The collector is always mocked so no FusionCompute connection is attempted. +""" + +import argparse + +import pytest + +import app as app_module +from app import app as flask_app + + +@pytest.fixture +def client(): + flask_app.config["TESTING"] = True + # Reset global job state between tests. + app_module.current_job = {"collector": None, "thread": None, "output_file": None} + with flask_app.test_client() as c: + yield c + + +class FakeThread: + def __init__(self, alive=True): + self._alive = alive + + def is_alive(self): + return self._alive + + +# ── / and /api/version ────────────────────────────────────── + + +def test_version_endpoint(client): + resp = client.get("/api/version") + assert resp.status_code == 200 + assert resp.get_json()["version"] == app_module.__version__ + + +# ── cross-origin / CSRF protection ────────────────────────── + + +def test_cross_origin_post_is_rejected(client): + resp = client.post( + "/api/collect", + json={"host": "h", "username": "u", "password": "p"}, + headers={"Origin": "http://evil.example"}, + ) + assert resp.status_code == 403 + + +def test_same_origin_post_is_allowed(client): + # Matching Origin host passes the guard (then hits normal validation). + resp = client.post( + "/api/collect", + json={"host": "", "username": "", "password": ""}, + headers={"Origin": "http://localhost"}, + base_url="http://localhost", + ) + # Not a 403 from the origin guard; fails later on missing fields instead. + assert resp.status_code == 400 + + +def test_get_requests_skip_origin_check(client): + resp = client.get("/api/progress", headers={"Origin": "http://evil.example"}) + assert resp.status_code == 200 + + +def test_index_renders(client): + resp = client.get("/") + assert resp.status_code == 200 + + +def test_changelog_page_renders(client): + resp = client.get("/changelog") + assert resp.status_code == 200 + + +# ── helpers ───────────────────────────────────────────────── + + +def test_find_resource_existing_and_missing(monkeypatch, tmp_path): + f = tmp_path / "thing.txt" + f.write_text("hi") + monkeypatch.setattr(app_module, "_base_dir", str(tmp_path)) + assert app_module._find_resource("thing.txt") == str(f) + assert app_module._find_resource("nope.txt") is None + + +def test_get_lan_ip_returns_string(): + ip = app_module._get_lan_ip() + assert isinstance(ip, str) + assert ip + + +# ── /api/collect ──────────────────────────────────────────── + + +def test_collect_requires_fields(client): + resp = client.post("/api/collect", json={"host": "", "username": "", "password": ""}) + assert resp.status_code == 400 + assert "required" in resp.get_json()["error"].lower() + + +def test_collect_conflict_when_already_running(client): + app_module.current_job["thread"] = FakeThread(alive=True) + resp = client.post("/api/collect", + json={"host": "h", "username": "u", "password": "p"}) + assert resp.status_code == 409 + + +def test_collect_rejects_invalid_port(client): + resp = client.post("/api/collect", + json={"host": "h", "username": "u", "password": "p", + "port": "notaport"}) + assert resp.status_code == 400 + + +def test_collect_rejects_out_of_range_port(client): + resp = client.post("/api/collect", + json={"host": "h", "username": "u", "password": "p", + "port": 70000}) + assert resp.status_code == 400 + + +def test_collect_starts_job(client, monkeypatch): + started = {} + + class FakeThreadObj: + def __init__(self, target, args, daemon): + started["target"] = target + started["args"] = args + self.daemon = daemon + + def start(self): + started["started"] = True + + def is_alive(self): + return True + + monkeypatch.setattr(app_module.threading, "Thread", FakeThreadObj) + monkeypatch.setattr(app_module, "InventoryCollector", + lambda *a, **kw: object()) + + resp = client.post("/api/collect", + json={"host": "h", "username": "u", "password": "p", "port": 7443}) + assert resp.status_code == 202 + assert resp.get_json()["status"] == "started" + assert started["started"] is True + + +# ── /api/progress ─────────────────────────────────────────── + + +def test_progress_idle_when_no_collector(client): + resp = client.get("/api/progress") + assert resp.status_code == 200 + assert resp.get_json()["status"] == "idle" + + +def test_progress_returns_collector_state(client): + class FakeCollector: + progress = {"status": "running", "percent": 42, "current_step": "x", "error": None} + + app_module.current_job["collector"] = FakeCollector() + resp = client.get("/api/progress") + body = resp.get_json() + assert body["status"] == "running" + assert body["percent"] == 42 + + +# ── /api/cancel ───────────────────────────────────────────── + + +def test_cancel_no_job(client): + resp = client.post("/api/cancel") + assert resp.status_code == 404 + + +def test_cancel_running_job(client): + class FakeCollector: + def __init__(self): + self.cancelled = False + + def cancel(self): + self.cancelled = True + + collector = FakeCollector() + app_module.current_job["collector"] = collector + app_module.current_job["thread"] = FakeThread(alive=True) + + resp = client.post("/api/cancel") + assert resp.status_code == 200 + assert resp.get_json()["status"] == "cancelling" + assert collector.cancelled is True + + +# ── /api/download ─────────────────────────────────────────── + + +def test_download_404_when_no_file(client): + resp = client.get("/api/download") + assert resp.status_code == 404 + + +def test_download_sends_file(client, tmp_path): + f = tmp_path / "FC_Inventory_x.xlsx" + f.write_bytes(b"PK\x03\x04fakexlsx") + app_module.current_job["output_file"] = str(f) + + resp = client.get("/api/download") + assert resp.status_code == 200 + assert b"fakexlsx" in resp.data + + +# ── /api/update-check ─────────────────────────────────────── + + +def test_update_check_disabled_by_env(client, monkeypatch): + monkeypatch.setenv("FC_INVENTORY_DISABLE_UPDATE_CHECK", "1") + resp = client.get("/api/update-check") + body = resp.get_json() + assert body["enabled"] is False + + +def test_update_check_reports_available(client, monkeypatch): + monkeypatch.delenv("FC_INVENTORY_DISABLE_UPDATE_CHECK", raising=False) + monkeypatch.setattr(app_module, "get_latest_release", + lambda repo: {"tag": "v99.0.0", "url": "https://x"}) + resp = client.get("/api/update-check") + body = resp.get_json() + assert body["update_available"] is True + assert body["latest"] == "99.0.0" + + +def test_update_check_degrades_on_error(client, monkeypatch): + monkeypatch.delenv("FC_INVENTORY_DISABLE_UPDATE_CHECK", raising=False) + + def boom(repo): + raise RuntimeError("offline") + + monkeypatch.setattr(app_module, "get_latest_release", boom) + resp = client.get("/api/update-check") + assert resp.status_code == 200 + body = resp.get_json() + assert body["enabled"] is True + assert "error" in body + + +# ── /api/changelog ────────────────────────────────────────── + + +def test_changelog_returns_content(client): + resp = client.get("/api/changelog") + assert resp.status_code == 200 + assert "Content-Type" in resp.headers + # Repo ships a CHANGELOG.md, so we expect real content. + assert len(resp.get_data(as_text=True)) > 0 + + +def test_changelog_404_when_missing(client, monkeypatch): + monkeypatch.setattr(app_module, "_find_resource", lambda name: None) + resp = client.get("/api/changelog") + assert resp.status_code == 404 + + +# ── _run_collection (background thread body) ──────────────── + + +def test_run_collection_success_builds_excel(monkeypatch, tmp_path): + monkeypatch.setattr(app_module, "_base_dir", str(tmp_path)) + built = {} + + class FakeCollector: + progress = {"status": "running"} + + def collect_all(self): + return {"vInfo": []} + + monkeypatch.setattr(app_module, "build_excel", + lambda data, path: built.update({"path": path, "data": data})) + + app_module._run_collection(FakeCollector()) + assert built["path"].endswith(".xlsx") + assert app_module.current_job["output_file"] == built["path"] + + +def test_run_collection_handles_interrupt(monkeypatch): + class FakeCollector: + progress = {"status": "running"} + + def collect_all(self): + raise InterruptedError() + + c = FakeCollector() + app_module._run_collection(c) + assert c.progress["status"] == "cancelled" + + +def test_run_collection_handles_error(monkeypatch): + class FakeCollector: + progress = {"status": "running"} + + def collect_all(self): + raise RuntimeError("kaboom") + + c = FakeCollector() + app_module._run_collection(c) + assert c.progress["status"] == "error" + assert "kaboom" in c.progress["error"] + + +# ── run_headless ──────────────────────────────────────────── + + +def _headless_args(**overrides): + defaults = dict(host="h", username="u", password="p", port=7443, out=None) + defaults.update(overrides) + return argparse.Namespace(**defaults) + + +def test_run_headless_missing_password_returns_2(monkeypatch): + monkeypatch.delenv("FC_INVENTORY_PASSWORD", raising=False) + monkeypatch.setattr(app_module.getpass, "getpass", lambda *a, **k: "") + code = app_module.run_headless(_headless_args(password=None)) + assert code == 2 + + +def test_run_headless_success(monkeypatch, tmp_path, capsys): + out = tmp_path / "result.xlsx" + + class FakeCollector: + def __init__(self, *a, **kw): + pass + + def collect_all(self): + return {"vInfo": []} + + monkeypatch.setattr(app_module, "InventoryCollector", FakeCollector) + monkeypatch.setattr(app_module, "build_excel", lambda data, path: None) + + code = app_module.run_headless(_headless_args(out=str(out))) + assert code == 0 + # Path is printed on its own line for scripts to capture. + assert str(out) in capsys.readouterr().out + + +def test_run_headless_password_from_env(monkeypatch, tmp_path): + monkeypatch.setenv("FC_INVENTORY_PASSWORD", "envpass") + seen = {} + + class FakeCollector: + def __init__(self, host, username, password, port=7443): + seen["password"] = password + + def collect_all(self): + return {} + + monkeypatch.setattr(app_module, "InventoryCollector", FakeCollector) + monkeypatch.setattr(app_module, "build_excel", lambda data, path: None) + + code = app_module.run_headless(_headless_args(password=None, out=str(tmp_path / "o.xlsx"))) + assert code == 0 + assert seen["password"] == "envpass" + + +def test_run_headless_collection_failure_returns_1(monkeypatch): + class FakeCollector: + def __init__(self, *a, **kw): + pass + + def collect_all(self): + raise RuntimeError("boom") + + monkeypatch.setattr(app_module, "InventoryCollector", FakeCollector) + code = app_module.run_headless(_headless_args()) + assert code == 1 + + +def test_run_headless_keyboard_interrupt_returns_130(monkeypatch): + class FakeCollector: + def __init__(self, *a, **kw): + pass + + def collect_all(self): + raise KeyboardInterrupt() + + monkeypatch.setattr(app_module, "InventoryCollector", FakeCollector) + code = app_module.run_headless(_headless_args()) + assert code == 130 + + +# ── arg parser ────────────────────────────────────────────── + + +def test_arg_parser_collect_subcommand(): + parser = app_module._build_arg_parser() + args = parser.parse_args(["collect", "--host", "h", "--username", "u"]) + assert args.command == "collect" + assert args.host == "h" + assert args.port == 7443 + + +def test_arg_parser_web_default(): + parser = app_module._build_arg_parser() + args = parser.parse_args([]) + assert args.command is None diff --git a/tests/test_collector_builders.py b/tests/test_collector_builders.py new file mode 100644 index 0000000..ddbc9b8 --- /dev/null +++ b/tests/test_collector_builders.py @@ -0,0 +1,366 @@ +"""Tests for the remaining collector helpers, sheet builders and orchestration. + +These complement test_collector.py and require no network access. +""" + +import pytest + +from collector import ( + InventoryCollector, + _flatten_dict, + _build_row, + _prettify_key, +) + + +def _make_collector(): + return InventoryCollector("host", "user", "pass", port=7443) + + +# ── _flatten_dict ─────────────────────────────────────────── + + +def test_flatten_dict_nested_and_scalar_lists(): + d = { + "a": 1, + "b": {"c": 2, "d": {"e": 3}}, + "tags": ["x", "y"], + "empty": [], + } + flat = _flatten_dict(d) + assert flat["a"] == 1 + assert flat["b.c"] == 2 + assert flat["b.d.e"] == 3 + assert flat["tags"] == "x, y" + assert flat["empty"] == "" + + +def test_flatten_dict_skips_list_of_dicts(): + d = {"nics": [{"ip": "1.1.1.1"}], "name": "vm"} + flat = _flatten_dict(d) + assert "nics" not in flat + assert flat["name"] == "vm" + + +def test_flatten_dict_non_dict_returns_empty(): + assert _flatten_dict(None) == {} + assert _flatten_dict([1, 2]) == {} + + +# ── _build_row / extras capture ───────────────────────────── + + +def test_build_row_maps_columns_and_captures_extras(): + from collections import OrderedDict + field_map = OrderedDict([("Name", ["name"]), ("CPU", ["cpu.quantity"])]) + data = {"name": "vm1", "cpu": {"quantity": 4}, "extraField": "kept"} + + row = _build_row(data, field_map, extras_allowed=True) + assert row["Name"] == "vm1" + assert row["CPU"] == 4 + # Unmapped field is captured under a prettified key. + assert any("kept" == v for v in row.values()) + + +def test_build_row_without_extras(): + from collections import OrderedDict + field_map = OrderedDict([("Name", ["name"])]) + data = {"name": "vm1", "ignored": "x"} + row = _build_row(data, field_map, extras_allowed=False) + assert list(row.keys()) == ["Name"] + + +def test_prettify_key_camel_and_dotted(): + assert _prettify_key("vmConfig.cpu.quantity") == "Cpu - Quantity" + assert _prettify_key("bmcIp") == "Bmc Ip" + + +# ── _build_vsummary ───────────────────────────────────────── + + +def test_build_vsummary_counts_power_and_clusters(): + c = _make_collector() + c.hosts = [{"urn": "h1"}] + c.clusters = [{"urn": "c1"}] + c.datastores = [] + c.dvswitches = [] + c.portgroups = [] + vinfo = [ + {"Power State": "ON", "Cluster": "A"}, + {"Power State": "OFF", "Cluster": "A"}, + {"Power State": "ON", "Cluster": "B"}, + ] + rows = c._build_vsummary(vinfo) + items = {r["Item"]: r["Count"] for r in rows} + assert items["Total VMs"] == 3 + assert items["Power ON"] == 2 + assert items["Power OFF"] == 1 + assert items["Total Hosts"] == 1 + # Per-cluster breakdown line for cluster A. + a_line = next(r for r in rows if r["Item"].strip() == "A") + assert "ON: 1" in a_line["Count"] and "OFF: 1" in a_line["Count"] + + +# ── _build_vcpu (socket math) ─────────────────────────────── + + +def test_build_vcpu_socket_calculation(): + c = _make_collector() + c.vms = [{"name": "vm1", "urn": "u1", "status": "running"}] + c.vm_details = { + "u1": {"vmConfig": {"cpu": {"quantity": 8, "coresPerSocket": 2}}} + } + rows = c._build_vcpu() + assert rows[0]["Total CPUs"] == 8 + assert rows[0]["Sockets"] == 4 + + +def test_build_vcpu_socket_handles_missing_values(): + c = _make_collector() + c.vms = [{"name": "vm1", "urn": "u1", "status": "running"}] + c.vm_details = {"u1": {"vmConfig": {"cpu": {}}}} + rows = c._build_vcpu() + assert rows[0]["Sockets"] == "" + + +# ── _build_vmemory ────────────────────────────────────────── + + +def test_build_vmemory_basic(): + c = _make_collector() + c.vms = [{"name": "vm1", "urn": "u1", "status": "stopped"}] + c.vm_details = {"u1": {"vmConfig": {"memory": {"quantityMB": 4096, "reservation": 1024}}}} + rows = c._build_vmemory() + assert rows[0]["Memory (MB)"] == 4096 + assert rows[0]["Reservation (MB)"] == 1024 + assert rows[0]["Power"] == "OFF" + + +# ── _build_vnetwork ───────────────────────────────────────── + + +def test_build_vnetwork_one_row_per_nic(): + c = _make_collector() + c.vms = [{"name": "vm1", "urn": "u1", "status": "running", "uuid": "uu"}] + c.vm_nics = {"u1": [ + {"name": "nic0", "mac": "aa", "ip": "10.0.0.1"}, + {"name": "nic1", "mac": "bb", "ip": "10.0.0.2"}, + ]} + rows = c._build_vnetwork() + assert len(rows) == 2 + assert rows[0]["MAC Address"] == "aa" + assert rows[1]["IP Address"] == "10.0.0.2" + assert rows[0]["VM UUID"] == "uu" + + +# ── _build_vhost ──────────────────────────────────────────── + + +def test_build_vhost_counts_running_vms_fallback(): + c = _make_collector() + c.hosts = [{"name": "esx1", "urn": "h1", "clusterUrn": "cl1"}] + c.host_details = {} + c.cluster_map = {"cl1": "Cluster-A"} + # Two VMs located on h1; host has no runningVmCount so fallback count is used. + c.vms = [{"locationUrn": "h1"}, {"locationUrn": "h1"}] + rows = c._build_vhost() + assert rows[0]["Host Name"] == "esx1" + assert rows[0]["Cluster"] == "Cluster-A" + assert rows[0]["Running VMs"] == 2 + + +# ── _build_vcluster ───────────────────────────────────────── + + +def test_build_vcluster_counts_hosts(): + c = _make_collector() + c.clusters = [{"name": "Cluster-A", "urn": "cl1", "isEnableHa": True}] + c.hosts = [{"clusterUrn": "cl1"}, {"clusterUrn": "cl1"}, {"clusterUrn": "cl2"}] + rows = c._build_vcluster() + assert rows[0]["Cluster Name"] == "Cluster-A" + assert rows[0]["HA Enabled"] is True + assert rows[0]["Total Hosts"] == 2 + + +# ── _build_vdatastore (unit conversion + used %) ──────────── + + +def test_build_vdatastore_gb_direct_and_used_pct(): + c = _make_collector() + c.datastores = [{ + "name": "DS1", "storageType": "SAN", + "capacityGB": 100, "freeSpaceGB": 25, "status": "normal", + }] + rows = c._build_vdatastore() + assert rows[0]["Capacity (GB)"] == 100 + assert rows[0]["Free (GB)"] == 25 + assert rows[0]["Used %"] == 75.0 + + +def test_build_vdatastore_converts_mb_to_gb(): + c = _make_collector() + c.datastores = [{"name": "DS2", "capacityMB": 2048, "freeSpaceMB": 1024}] + rows = c._build_vdatastore() + assert rows[0]["Capacity (GB)"] == 2.0 + assert rows[0]["Free (GB)"] == 1.0 + assert rows[0]["Used %"] == 50.0 + + +def test_build_vdatastore_handles_zero_capacity(): + c = _make_collector() + c.datastores = [{"name": "DS3", "capacityGB": 0, "freeSpaceGB": 0}] + rows = c._build_vdatastore() + # No divide-by-zero; used % left blank. + assert rows[0]["Used %"] == "" + + +# ── _build_vswitch ────────────────────────────────────────── + + +def test_build_vswitch_combines_dvswitches_and_portgroups(): + c = _make_collector() + c.dvswitches = [{"name": "dvs1", "mtu": 1500, "urn": "d1"}] + c.portgroups = [{"name": "pg1", "vlanId": 10, "_dvswitch_name": "dvs1"}] + rows = c._build_vswitch() + assert rows[0]["Type"] == "DVSwitch" + assert rows[1]["Type"] == "Port Group" + assert rows[1]["VLAN ID"] == 10 + assert rows[1]["Parent"] == "dvs1" + + +# ── _build_lookup_maps ────────────────────────────────────── + + +def test_build_lookup_maps(): + c = _make_collector() + c.hosts = [{"urn": "h1", "name": "esx1"}] + c.clusters = [{"urn": "c1", "name": "Cluster-A"}] + c.datastores = [{"urn": "d1", "name": "DS1"}] + c._build_lookup_maps() + assert c.host_map["h1"] == "esx1" + assert c.cluster_map["c1"] == "Cluster-A" + assert c.datastore_map["d1"] == "DS1" + + +# ── cancellation ──────────────────────────────────────────── + + +def test_cancel_raises_in_update_progress(): + c = _make_collector() + c.cancel() + assert c.cancelled is True + with pytest.raises(InterruptedError): + c._update_progress(10, "step") + + +# ── collect_all orchestration (fully mocked client) ───────── + + +class FakeClient: + """Minimal stand-in for FCClient covering the collect_all pipeline.""" + + def __init__(self): + self.logged_in = False + self.logged_out = False + + def login(self): + self.logged_in = True + + def logout(self): + self.logged_out = True + + def get_sites(self): + return [{"uri": "/site1"}] + + def get_clusters(self, site_uri): + return [{"urn": "cl1", "name": "Cluster-A"}] + + def get_hosts(self, site_uri): + return [{"uri": "/h1", "urn": "h1", "name": "esx1", "clusterUrn": "cl1"}] + + def get_host_detail(self, host_uri): + return {"ip": "10.0.0.10"} + + def get_datastores(self, site_uri): + return [{"urn": "d1", "name": "DS1", "capacityGB": 100, "freeSpaceGB": 50}] + + def get_dvswitches(self, site_uri): + return [{"uri": "/dvs1", "urn": "dv1", "name": "dvs1"}] + + def get_portgroups(self, dvs_uri): + return [{"name": "pg1", "vlanId": 10}] + + def get_site_portgroups(self, site_uri): + return [] + + def get_vms(self, site_uri): + return [{"uri": "/vm1", "urn": "vm1", "name": "vm-a", + "status": "running", "locationUrn": "h1", "clusterUrn": "cl1"}] + + def get_vm_detail(self, vm_uri): + return { + "uuid": "uuid-a", + "vmConfig": { + "cpu": {"quantity": 4, "coresPerSocket": 2}, + "memory": {"quantityMB": 8192}, + "nics": [{"name": "nic0", "ip": "10.0.0.50"}], + "disks": [{"name": "disk0", "quantityGB": 40}], + }, + } + + +def test_collect_all_happy_path(): + c = _make_collector() + c.client = FakeClient() + + result = c.collect_all() + + assert c.progress["status"] == "done" + assert c.progress["percent"] == 100 + assert c.client.logged_in and c.client.logged_out + + # All expected sheets present and populated. + assert set(result.keys()) == { + "vSummary", "vInfo", "vCPU", "vMemory", "vDisk", + "vNetwork", "vHost", "vCluster", "vDatastore", "vSwitch", + } + assert result["vInfo"][0]["VM Name"] == "vm-a" + assert result["vInfo"][0]["CPUs"] == 4 + assert result["vInfo"][0]["Total Disk (GB)"] == 40 + assert result["vDisk"][0]["Disk Name"] == "disk0" + assert result["vNetwork"][0]["IP Address"] == "10.0.0.50" + assert result["vHost"][0]["Host Name"] == "esx1" + + +def test_collect_all_propagates_and_marks_error(): + c = _make_collector() + + class BoomClient(FakeClient): + def get_sites(self): + raise RuntimeError("api down") + + c.client = BoomClient() + with pytest.raises(RuntimeError): + c.collect_all() + assert c.progress["status"] == "error" + assert "api down" in c.progress["error"] + assert c.client.logged_out # cleanup logout still attempted + + +def test_collect_all_cancelled_midway(): + c = _make_collector() + + class CancelClient(FakeClient): + def __init__(self, collector): + super().__init__() + self._collector = collector + + def get_sites(self): + # Cancel after login so the next _update_progress raises. + self._collector.cancel() + return super().get_sites() + + c.client = CancelClient(c) + with pytest.raises(InterruptedError): + c.collect_all() + assert c.progress["status"] == "cancelled" diff --git a/tests/test_excel_builder.py b/tests/test_excel_builder.py index a3fef51..ef2822a 100644 --- a/tests/test_excel_builder.py +++ b/tests/test_excel_builder.py @@ -2,7 +2,7 @@ from openpyxl import load_workbook -from excel_builder import build_excel, SHEET_ORDER +from excel_builder import build_excel, SHEET_ORDER, _sanitize_cell def test_build_excel_creates_all_sheets(tmp_path): @@ -45,3 +45,30 @@ def test_build_excel_empty_sheet_has_placeholder(tmp_path): build_excel({}, str(out)) ws = load_workbook(out)["vInfo"] assert ws["A1"].value == "No data collected" + + +def test_sanitize_cell_neutralizes_formula_triggers(): + assert _sanitize_cell("=1+1") == "'=1+1" + assert _sanitize_cell("+cmd") == "'+cmd" + assert _sanitize_cell("-2+3") == "'-2+3" + assert _sanitize_cell("@SUM(A1)") == "'@SUM(A1)" + # Safe values and non-strings pass through untouched. + assert _sanitize_cell("normal") == "normal" + assert _sanitize_cell(42) == 42 + assert _sanitize_cell("") == "" + + +def test_build_excel_defangs_formula_injection(tmp_path): + data = { + "vInfo": [ + {"VM Name": "=HYPERLINK(\"http://evil\",\"x\")", "CPUs": 2}, + ] + } + out = tmp_path / "out.xlsx" + build_excel(data, str(out)) + + ws = load_workbook(out)["vInfo"] + cell = ws["A2"] + # Stored as inert text, not a formula. + assert cell.data_type == "s" + assert cell.value.startswith("'=HYPERLINK") diff --git a/tests/test_fc_client.py b/tests/test_fc_client.py new file mode 100644 index 0000000..be339d7 --- /dev/null +++ b/tests/test_fc_client.py @@ -0,0 +1,450 @@ +"""Tests for the FusionCompute REST client. + +All network I/O is faked: each test installs a lightweight stand-in for the +``requests.Session`` the client builds in ``__init__``. No live VRM is needed. +""" + +import requests +import pytest + +from fc_client import FCClient, _resolve_verify + + +# ── Fakes ──────────────────────────────────────────────────── + + +class FakeResponse: + def __init__(self, status_code=200, headers=None, json_data=None, text=""): + self.status_code = status_code + self.headers = headers or {} + self._json = json_data + self.text = text + self.content = text.encode("utf-8") + + def json(self): + if self._json is None: + raise ValueError("no json") + return self._json + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.exceptions.HTTPError(f"HTTP {self.status_code}") + + +class FakeSession: + """Records calls and returns queued/scripted responses.""" + + def __init__(self): + self.headers = {} + self.verify = True + self.calls = [] + # Optional scripted handlers keyed by HTTP method (callables). + self.post_handler = None + self.put_handler = None + self.get_handler = None + self.delete_handler = None + + def _dispatch(self, handler, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + if handler is None: + return FakeResponse(200, json_data={}) + result = handler(url, **kwargs) + if isinstance(result, Exception): + raise result + return result + + def post(self, url, **kwargs): + return self._dispatch(self.post_handler, "POST", url, **kwargs) + + def put(self, url, **kwargs): + return self._dispatch(self.put_handler, "PUT", url, **kwargs) + + def get(self, url, **kwargs): + return self._dispatch(self.get_handler, "GET", url, **kwargs) + + def delete(self, url, **kwargs): + return self._dispatch(self.delete_handler, "DELETE", url, **kwargs) + + +def _make_client(**kwargs): + client = FCClient("10.0.0.1", "admin", "secret", **kwargs) + client.session = FakeSession() + return client + + +# ── __init__ / host normalization ─────────────────────────── + + +def test_init_strips_scheme_and_trailing_slash(): + c = FCClient("https://10.0.0.1/", "u", "p") + assert c.host == "10.0.0.1" + assert c.base_url == "https://10.0.0.1:7443" + + c2 = FCClient("http://vrm.example/", "u", "p", port=8443) + assert c2.host == "vrm.example" + assert c2.base_url == "https://vrm.example:8443" + + +def test_sha256_is_hex_digest(): + c = _make_client() + digest = c._sha256("secret") + assert len(digest) == 64 + assert all(ch in "0123456789abcdef" for ch in digest) + + +# ── TLS verification config ───────────────────────────────── + + +def test_verify_defaults_to_false(monkeypatch): + monkeypatch.delenv("FC_INVENTORY_VERIFY_SSL", raising=False) + monkeypatch.delenv("FC_INVENTORY_CA_BUNDLE", raising=False) + c = FCClient("10.0.0.1", "u", "p") + assert c.session.verify is False + + +def test_verify_explicit_argument_wins(): + c = FCClient("10.0.0.1", "u", "p", verify=True) + assert c.session.verify is True + + +def test_resolve_verify_env_flag(monkeypatch): + monkeypatch.delenv("FC_INVENTORY_CA_BUNDLE", raising=False) + monkeypatch.setenv("FC_INVENTORY_VERIFY_SSL", "true") + assert _resolve_verify(None) is True + monkeypatch.setenv("FC_INVENTORY_VERIFY_SSL", "0") + assert _resolve_verify(None) is False + + +def test_resolve_verify_ca_bundle_path(monkeypatch): + monkeypatch.setenv("FC_INVENTORY_CA_BUNDLE", "/etc/ssl/fc-ca.pem") + assert _resolve_verify(None) == "/etc/ssl/fc-ca.pem" + + +# ── login() ───────────────────────────────────────────────── + + +def test_login_success_token_in_header(): + c = _make_client() + + def handler(url, **kwargs): + return FakeResponse(200, headers={"X-Auth-Token": "tok-123"}, + json_data={"hello": "world"}) + + c.session.post_handler = handler + + result = c.login() + assert c.token == "tok-123" + assert c.session.headers["X-Auth-Token"] == "tok-123" + assert c.base_url.endswith("/service") + assert result == {"hello": "world"} + + +def test_login_success_token_in_body(): + c = _make_client() + + def handler(url, **kwargs): + # No token header -> must fall back to body. + return FakeResponse(200, headers={}, json_data={"accessSession": "body-tok"}) + + c.session.post_handler = handler + + c.login() + assert c.token == "body-tok" + + +def test_login_skips_to_next_version_on_version_error(): + c = _make_client() + seen_versions = [] + + def handler(url, **kwargs): + ver = c.session.headers["Accept"] + seen_versions.append(ver) + # First version rejected with the version-error code, then succeed. + if "v8.0" in ver: + return FakeResponse(403, text='{"errorCode":"10000022"}') + return FakeResponse(200, headers={"X-Auth-Token": "ok"}) + + c.session.post_handler = handler + + c.login() + assert c.token == "ok" + # The v8.0 attempt should not have been retried with other auth methods: + # exactly one v8.0 call before moving on. + v8_calls = [v for v in seen_versions if "v8.0" in v] + assert len(v8_calls) == 1 + + +def test_login_all_methods_fail_raises_connection_error(): + c = _make_client() + + def handler(url, **kwargs): + return FakeResponse(401, text="unauthorized") + + c.session.post_handler = handler + c.session.put_handler = handler + + with pytest.raises(ConnectionError): + c.login() + + +def test_login_connection_refused_moves_to_next_port(): + c = _make_client(port=9999) + attempted_ports = [] + + def handler(url, **kwargs): + # Record the port from the URL and always refuse. + attempted_ports.append(url.split(":")[2].split("/")[0]) + raise requests.exceptions.ConnectionError("refused") + + c.session.post_handler = handler + c.session.put_handler = handler + + with pytest.raises(ConnectionError): + c.login() + + # Should have tried the requested port plus the documented fallbacks. + assert "9999" in attempted_ports + assert "7443" in attempted_ports + assert "8443" in attempted_ports + + +# ── _extract_token ────────────────────────────────────────── + + +def test_extract_token_prefers_header(): + c = _make_client() + resp = FakeResponse(200, headers={"X-Auth-Token": "h"}, json_data={"token": "b"}) + c._extract_token(resp, "label") + assert c.token == "h" + + +def test_extract_token_body_fallback_keys(): + c = _make_client() + resp = FakeResponse(200, headers={}, json_data={"token": "from-body"}) + c._extract_token(resp, "label") + assert c.token == "from-body" + + +def test_extract_token_missing_raises(): + c = _make_client() + resp = FakeResponse(200, headers={}, json_data={"nope": 1}) + with pytest.raises(ConnectionError): + c._extract_token(resp, "label") + + +# ── logout ────────────────────────────────────────────────── + + +def test_logout_clears_token_and_swallows_errors(): + c = _make_client() + c.token = "tok" + + def handler(url, **kwargs): + raise requests.exceptions.ConnectionError("down") + + c.session.delete_handler = handler + c.logout() # must not raise + assert c.token is None + + +# ── _get / URL construction ───────────────────────────────── + + +def test_get_uses_base_url_for_relative_path(): + c = _make_client() + c.base_url = "https://10.0.0.1:7443/service" + captured = {} + + def handler(url, **kwargs): + captured["url"] = url + return FakeResponse(200, json_data={"ok": True}) + + c.session.get_handler = handler + assert c._get("/sites") == {"ok": True} + assert captured["url"] == "https://10.0.0.1:7443/service/sites" + + +def test_get_uses_host_port_for_service_prefixed_path(): + c = _make_client() + captured = {} + + def handler(url, **kwargs): + captured["url"] = url + return FakeResponse(200, json_data={}) + + c.session.get_handler = handler + c._get("/service/vms/123") + assert captured["url"] == "https://10.0.0.1:7443/service/vms/123" + + +def test_get_raises_on_http_error(): + c = _make_client() + c.session.get_handler = lambda url, **kw: FakeResponse(500, text="boom") + with pytest.raises(requests.exceptions.HTTPError): + c._get("/sites") + + +# ── _get_all pagination ───────────────────────────────────── + + +def test_get_all_accumulates_pages(monkeypatch): + c = _make_client() + pages = [ + {"clusters": [{"id": 1}, {"id": 2}], "total": 3}, + {"clusters": [{"id": 3}], "total": 3}, + ] + seen_offsets = [] + + def fake_get(path, params=None): + seen_offsets.append(params["offset"]) + return pages[len(seen_offsets) - 1] + + monkeypatch.setattr(c, "_get", fake_get) + items = c._get_all("/site/clusters", "clusters") + assert [i["id"] for i in items] == [1, 2, 3] + assert seen_offsets == [0, 100] + + +def test_get_all_falls_back_to_items_key(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", + lambda path, params=None: {"items": [{"id": 1}], "total": 1}) + assert c._get_all("/p", "clusters") == [{"id": 1}] + + +def test_get_all_falls_back_to_result_key(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", + lambda path, params=None: {"result": [{"id": 9}], "total": 1}) + assert c._get_all("/p", "clusters") == [{"id": 9}] + + +def test_get_all_stops_on_empty_batch(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path, params=None: {"clusters": []}) + assert c._get_all("/p", "clusters") == [] + + +def test_get_all_handles_top_level_list(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path, params=None: [{"id": 1}, {"id": 2}]) + assert c._get_all("/p", "clusters") == [{"id": 1}, {"id": 2}] + + +# ── _extract_list helper ──────────────────────────────────── + + +def test_extract_list_from_dict_key_order(): + assert FCClient._extract_list({"a": [1], "b": [2]}, "a", "b") == [1] + assert FCClient._extract_list({"b": [2]}, "a", "b") == [2] + + +def test_extract_list_from_bare_list(): + assert FCClient._extract_list([{"x": 1}], "a") == [{"x": 1}] + + +def test_extract_list_empty_value_is_returned(): + # Key present with an empty list should win over later candidates. + assert FCClient._extract_list({"a": [], "b": [9]}, "a", "b") == [] + + +def test_extract_list_fallbacks_to_empty(): + assert FCClient._extract_list({"nope": 1}, "a", "b") == [] + assert FCClient._extract_list(None, "a") == [] + + +# ── Getters with fallback keys ────────────────────────────── + + +def test_get_sites(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path, params=None: {"sites": [{"uri": "/s1"}]}) + assert c.get_sites() == [{"uri": "/s1"}] + + +def test_get_vm_disks_volumes_first(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: {"volumes": [{"id": "v1"}]}) + assert c.get_vm_disks("/vm/1") == [{"id": "v1"}] + + +def test_get_vm_disks_falls_back_to_disks(monkeypatch): + c = _make_client() + calls = {"n": 0} + + def fake_get(path): + calls["n"] += 1 + if path.endswith("/volumes"): + raise requests.exceptions.HTTPError("404") + return {"disks": [{"id": "d1"}]} + + monkeypatch.setattr(c, "_get", fake_get) + assert c.get_vm_disks("/vm/1") == [{"id": "d1"}] + + +def test_get_vm_disks_returns_empty_when_both_fail(monkeypatch): + c = _make_client() + + def fake_get(path): + raise requests.exceptions.HTTPError("404") + + monkeypatch.setattr(c, "_get", fake_get) + assert c.get_vm_disks("/vm/1") == [] + + +def test_get_dvswitches_key_fallbacks(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: {"dvSwitchs": [{"name": "dvs1"}]}) + assert c.get_dvswitches("/site") == [{"name": "dvs1"}] + + +def test_get_portgroups_key_fallback(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: {"portGroups": [{"name": "pg1"}]}) + assert c.get_portgroups("/dvs/1") == [{"name": "pg1"}] + + +def test_get_portgroups_list_fallback(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: [{"name": "pg1"}]) + assert c.get_portgroups("/dvs/1") == [{"name": "pg1"}] + + +def test_get_dvswitches_list_fallback(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: [{"name": "dvs1"}]) + assert c.get_dvswitches("/site") == [{"name": "dvs1"}] + + +def test_get_sites_list_fallback(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path, params=None: [{"uri": "/s1"}]) + assert c.get_sites() == [{"uri": "/s1"}] + + +def test_get_vm_nics_list_fallback(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: [{"mac": "aa"}]) + assert c.get_vm_nics("/vm/1") == [{"mac": "aa"}] + + +def test_get_vm_disks_list_fallback(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: [{"id": "v1"}]) + assert c.get_vm_disks("/vm/1") == [{"id": "v1"}] + + +def test_get_site_portgroups_list_fallback(monkeypatch): + c = _make_client() + monkeypatch.setattr(c, "_get", lambda path: [{"name": "pg1"}]) + assert c.get_site_portgroups("/site") == [{"name": "pg1"}] + + +def test_get_site_portgroups_swallows_errors(monkeypatch): + c = _make_client() + + def fake_get(path): + raise requests.exceptions.HTTPError("500") + + monkeypatch.setattr(c, "_get", fake_get) + assert c.get_site_portgroups("/site") == [] diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..8ec37c8 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,40 @@ +"""End-to-end pipeline test: mocked collection -> Excel -> reload and verify. + +This guards the seams between the collector and the workbook builder that the +focused unit tests don't exercise together. +""" + +from openpyxl import load_workbook + +from collector import InventoryCollector +from excel_builder import build_excel, SHEET_ORDER +from tests.test_collector_builders import FakeClient + + +def test_collect_to_excel_roundtrip(tmp_path): + collector = InventoryCollector("host", "user", "pass") + collector.client = FakeClient() + + data = collector.collect_all() + + out = tmp_path / "inventory.xlsx" + build_excel(data, str(out)) + assert out.exists() + + wb = load_workbook(out) + assert wb.sheetnames == SHEET_ORDER + + # vInfo sheet should carry the single collected VM. + vinfo = wb["vInfo"] + headers = [c.value for c in vinfo[1]] + assert "VM Name" in headers + name_col = headers.index("VM Name") + first_data_row = list(vinfo.iter_rows(min_row=2, values_only=True))[0] + assert first_data_row[name_col] == "vm-a" + + # vDatastore should reflect the computed Used %. + vds = wb["vDatastore"] + ds_headers = [c.value for c in vds[1]] + used_col = ds_headers.index("Used %") + ds_row = list(vds.iter_rows(min_row=2, values_only=True))[0] + assert ds_row[used_col] == 50.0 diff --git a/tests/test_version_utils_network.py b/tests/test_version_utils_network.py new file mode 100644 index 0000000..a431a2c --- /dev/null +++ b/tests/test_version_utils_network.py @@ -0,0 +1,56 @@ +"""Tests for the GitHub-release lookup wrapper in version_utils. + +The HTTP call is faked via monkeypatching ``version_utils.requests.get``. +""" + +import requests +import pytest + +import version_utils +from version_utils import get_latest_release + + +class FakeResponse: + def __init__(self, json_data=None, status_code=200): + self._json = json_data or {} + self.status_code = status_code + + def json(self): + return self._json + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.exceptions.HTTPError(f"HTTP {self.status_code}") + + +def test_get_latest_release_parses_tag_and_url(monkeypatch): + def fake_get(url, **kwargs): + assert "releases/latest" in url + return FakeResponse({"tag_name": "v2.3.0", "html_url": "https://x/rel"}) + + monkeypatch.setattr(version_utils.requests, "get", fake_get) + result = get_latest_release("owner/repo") + assert result == {"tag": "v2.3.0", "url": "https://x/rel"} + + +def test_get_latest_release_missing_keys_default_to_empty(monkeypatch): + monkeypatch.setattr(version_utils.requests, "get", + lambda url, **kw: FakeResponse({})) + result = get_latest_release("owner/repo") + assert result == {"tag": "", "url": ""} + + +def test_get_latest_release_raises_on_http_error(monkeypatch): + monkeypatch.setattr(version_utils.requests, "get", + lambda url, **kw: FakeResponse(status_code=404)) + with pytest.raises(requests.exceptions.HTTPError): + get_latest_release("owner/repo") + + +def test_get_latest_release_propagates_network_error(monkeypatch): + def boom(url, **kwargs): + raise requests.exceptions.ConnectionError("offline") + + monkeypatch.setattr(version_utils.requests, "get", boom) + with pytest.raises(requests.exceptions.ConnectionError): + get_latest_release("owner/repo")