Coverage/all branches

tests/test_git_history_branches.py

@@ -0,0 +1,42 @@
+"""Additional branch-coverage tests for git_history defensive paths.
+
+These target lines the main test_git_history.py does not exercise:
+- GitPython-absent guard (_require_gitpython raises)
+- _push_repo warning path when no remote is configured
+
+NOTE: the bare-repo working-tree-None guards in save_config_snapshot /
+rollback_config (git_history.py:255, :445) are NOT covered here. They are
+currently unreachable: init_git_repo(bare=True) itself raises inside GitPython
+before returning, so those guards are effectively dead code (see flagged bug).
+"""
+
+from pathlib import Path
+
+import pytest
+
+from audnet.exceptions import GitHistoryError
+from audnet import git_history
+from audnet.git_history import _push_repo, init_git_repo
+
+
+class TestRequireGitPython:
+    def test_raises_when_gitpython_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Simulate GitPython not being importable.
+        monkeypatch.setattr(git_history, "gitpython", None)
+        # Every public entrypoint goes through _require_gitpython().
+        with pytest.raises(GitHistoryError, match="GitPython is not installed"):
+            git_history.init_git_repo(Path("/tmp/should-not-be-created"))
+
+
+class TestPushRepoWarning:
+    def test_push_without_remote_logs_warning(
+        self, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        import logging
+
+        repo_path = tmp_path / "git-repo"
+        repo = init_git_repo(repo_path)
+        # No remote configured -> _push_repo should swallow the error and warn.
+        with caplog.at_level(logging.WARNING, logger="audnet.git_history"):
+            _push_repo(repo, "origin")
+        assert any("Failed to push" in rec.message for rec in caplog.records)

tests/test_netbox_inventory_branches.py

@@ -0,0 +1,185 @@
+"""Additional branch-coverage tests for the NetBox inventory source.
+
+Targets defensive branches not exercised by test_netbox_inventory.py:
+- _same_origin mismatch paths (scheme / host)
+- _normalize_device edge types (primary_ip weird type, audnet_ctx not a dict)
+- pagination guards (_MAX_PAGES cap, off-origin next URL)
+- non-JSON response body
+- list-shaped response body
+- "no valid devices could be built" error
+"""
+
+import json
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from audnet.exceptions import ConfigError
+from audnet.inventory_sources.netbox import (
+    _normalize_device,
+    _same_origin,
+    fetch_netbox_devices,
+)
+
+
+def _resp(body: bytes) -> MagicMock:
+    mock_resp = MagicMock()
+    mock_resp.read.return_value = body
+    mock_resp.__enter__ = MagicMock(return_value=mock_resp)
+    mock_resp.__exit__ = MagicMock(return_value=False)
+    cm = MagicMock()
+    cm.return_value = mock_resp
+    return cm
+
+
+def _page(devices: list[dict[str, Any]], next_url: str | None = None) -> bytes:
+    return json.dumps({"count": len(devices), "results": devices, "next": next_url}).encode()
+
+
+# ---------------------------------------------------------------------------
+# _same_origin
+# ---------------------------------------------------------------------------
+
+
+class TestSameOrigin:
+    def test_non_http_scheme_returns_false(self) -> None:
+        assert _same_origin("https://nb.example.com", "ftp://nb.example.com") is False
+
+    def test_scheme_mismatch_returns_false(self) -> None:
+        assert _same_origin("https://nb.example.com", "http://nb.example.com") is False
+
+    def test_host_mismatch_returns_false(self) -> None:
+        assert _same_origin("https://nb.example.com", "https://other.example.com") is False
+
+    def test_port_mismatch_returns_false(self) -> None:
+        assert _same_origin("https://nb.example.com:8443", "https://nb.example.com") is False
+
+    def test_same_origin_returns_true(self) -> None:
+        assert _same_origin("https://nb.example.com", "https://nb.example.com/api/") is True
+
+
+# ---------------------------------------------------------------------------
+# _normalize_device edge types
+# ---------------------------------------------------------------------------
+
+
+class TestNormalizeDeviceEdges:
+    def test_weird_primary_ip_type_yields_empty_host(self) -> None:
+        raw = {
+            "name": "rtr01",
+            "primary_ip": 12345,  # neither dict nor str
+            "platform": {"slug": "ios"},
+        }
+        kw = _normalize_device(raw)
+        assert kw["host"] == ""
+
+    def test_audnet_ctx_not_a_dict_falls_back(self) -> None:
+        raw = {
+            "name": "rtr01",
+            "primary_ip": {"address": "10.0.0.1/24"},
+            "platform": {"slug": "ios"},
+            "config_context": {"audnet": "not-a-dict"},
+        }
+        kw = _normalize_device(raw)
+        # Without a real audnet dict, username falls back to default "admin".
+        assert kw["username"] == "admin"
+
+
+# ---------------------------------------------------------------------------
+# fetch_netbox_devices defensive paths
+# ---------------------------------------------------------------------------
+
+
+class TestFetchNetboxBranches:
+    @patch("audnet.inventory_sources.netbox.urlopen")
+    def test_non_json_response_raises(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.side_effect = _resp(b"<!DOCTYPE html><html>oops</html>")
+        with pytest.raises(ConfigError, match="non-JSON"):
+            fetch_netbox_devices("https://nb.example.com", token="t")
+
+    @patch("audnet.inventory_sources.netbox.urlopen")
+    def test_list_shaped_body_extends(self, mock_urlopen: MagicMock) -> None:
+        # NetBox can return a bare list instead of an envelope.
+        body = json.dumps(
+            [
+                {
+                    "name": "rtr01",
+                    "primary_ip": {"address": "10.0.0.1/24"},
+                    "platform": {"slug": "ios"},
+                }
+            ]
+        ).encode()
+        mock_urlopen.side_effect = _resp(body)
+        devices = fetch_netbox_devices("https://nb.example.com", token="t")
+        assert len(devices) == 1
+        assert devices[0].name == "rtr01"
+
+    @patch("audnet.inventory_sources.netbox.urlopen")
+    def test_no_valid_devices_raises(self, mock_urlopen: MagicMock) -> None:
+        # All returned devices are invalid -> "No valid devices" error.
+        bad = [
+            {"name": "x", "primary_ip": None, "platform": None},
+            {"name": "y", "primary_ip": None, "platform": None},
+        ]
+        mock_urlopen.side_effect = _resp(_page(bad))
+        with pytest.raises(ConfigError, match="No valid devices"):
+            fetch_netbox_devices("https://nb.example.com", token="t")
+
+    @patch("audnet.inventory_sources.netbox.urlopen")
+    def test_off_origin_next_url_rejected(self, mock_urlopen: MagicMock) -> None:
+        first = _page(
+            [
+                {
+                    "name": "rtr01",
+                    "primary_ip": {"address": "10.0.0.1/24"},
+                    "platform": {"slug": "ios"},
+                }
+            ],
+            next_url="https://evil.example.com/api/dcim/devices/?limit=50",
+        )
+        mock_urlopen.side_effect = _resp(first)
+        with pytest.raises(ConfigError, match="off-origin"):
+            fetch_netbox_devices("https://nb.example.com", token="t")
+
+    @patch("audnet.inventory_sources.netbox.urlopen")
+    def test_pagination_exceeds_max_pages(self, mock_urlopen: MagicMock) -> None:
+        # Every page points back to the same on-origin URL -> loops until the
+        # _MAX_PAGES cap is hit.
+        body = _page(
+            [
+                {
+                    "name": "rtr01",
+                    "primary_ip": {"address": "10.0.0.1/24"},
+                    "platform": {"slug": "ios"},
+                }
+            ],
+            next_url="https://nb.example.com/api/dcim/devices/?limit=50&offset=50",
+        )
+        mock_urlopen.side_effect = _resp(body)
+        with pytest.raises(ConfigError, match="pagination exceeded"):
+            fetch_netbox_devices("https://nb.example.com", token="t")
+
+    @patch("audnet.inventory_sources.netbox.urlopen")
+    def test_allow_http_env_flag_suppresses_warning_path(self, mock_urlopen: MagicMock) -> None:
+        # http:// URL with AUDNET_NETBOX_ALLOW_HTTP set should fetch without
+        # raising (exercises the env-flag branch instead of the warning branch).
+        mock_urlopen.side_effect = _resp(
+            _page(
+                [
+                    {
+                        "name": "rtr01",
+                        "primary_ip": {"address": "10.0.0.1/24"},
+                        "platform": {"slug": "ios"},
+                    }
+                ]
+            )
+        )
+        import os
+
+        os.environ["AUDNET_NETBOX_ALLOW_HTTP"] = "1"
+        try:
+            devices = fetch_netbox_devices("http://nb.example.com", token="t")
+            assert len(devices) == 1
+        finally:
+            del os.environ["AUDNET_NETBOX_ALLOW_HTTP"]

tests/test_realtime_branches.py

@@ -0,0 +1,165 @@
+"""Additional branch-coverage tests for the real-time listener module.
+
+Targets defensive / branch paths not exercised by test_realtime.py:
+- AlertManager.send_alert dry_run suppression
+- AlertManager._get_client lazy creation + close()
+- aiosmtplib-unavailable guard in _send_email
+- pysnmp-unavailable guard in SnmpTrapReceiver
+- _on_snmp_trap callback (sync + async on_trap)
+- _handle_change baseline-enrichment path (pass + fail compliance)
+"""
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import asyncio
+import pytest
+
+from audnet.models import Device
+from audnet.realtime import (
+    AlertConfig,
+    AlertManager,
+    ChangeEvent,
+    RealtimeListener,
+    SnmpTrapReceiver,
+)
+
+
+def _event(**kw: Any) -> ChangeEvent:
+    base = dict(
+        device_name="rtr01",
+        source_ip="10.0.0.1",
+        event_type="syslog",
+        timestamp=1234567890.0,
+        raw_message="%SYS-5-CONFIG: changed",
+        change_summary="config changed",
+    )
+    base.update(kw)
+    return ChangeEvent(**base)
+
+
+class TestSendAlertDryRun:
+    @pytest.mark.asyncio
+    async def test_dry_run_suppresses_alert(self) -> None:
+        config = AlertConfig(dry_run=True)
+        manager = AlertManager(config)
+        await manager.send_alert(_event())
+        # dry_run returns before incrementing the counter
+        assert manager._alert_count == 0
+
+
+class TestClientLifecycle:
+    @pytest.mark.asyncio
+    async def test_get_client_lazy_creation_and_close(self) -> None:
+        config = AlertConfig(webhook_url="https://hooks.example.com/audit")
+        manager = AlertManager(config)
+        # _get_client creates the httpx client on first call (line 146)
+        client = await manager._get_client()
+        assert client is not None
+        assert manager._httpx_client is client
+        # close() releases it (lines 159-160)
+        await manager.close()
+        assert manager._httpx_client is None
+
+
+class TestSendEmailGuards:
+    @pytest.mark.asyncio
+    async def test_email_skipped_without_from(self) -> None:
+        config = AlertConfig(smtp_host="smtp.example.com", email_to=["ops@example.com"])
+        manager = AlertManager(config)
+        # No email_from -> returns immediately (line 287-288)
+        await manager._send_email(_event())
+
+    @pytest.mark.asyncio
+    async def test_email_skipped_when_aiosmtplib_unavailable(self) -> None:
+        config = AlertConfig(
+            smtp_host="smtp.example.com",
+            email_from="audnet@example.com",
+            email_to=["ops@example.com"],
+        )
+        manager = AlertManager(config)
+        with patch("audnet.realtime._AIOSMTPLIB_AVAILABLE", False):
+            # Should log error and return without raising (lines 290-292)
+            await manager._send_email(_event())
+
+
+class TestSnmpTrapReceiverGuard:
+    def test_raises_when_pysnmp_unavailable(self) -> None:
+        config = AlertConfig()
+        with patch("audnet.realtime._PYSNMP_AVAILABLE", False):
+            with pytest.raises(ImportError, match="pysnmp"):
+                SnmpTrapReceiver(config, lambda *a: None, {})
+
+
+class TestOnSnmpTrap:
+    @pytest.mark.asyncio
+    async def test_on_snmp_trap_schedules_alert(self) -> None:
+        config = AlertConfig()
+        manager = AlertManager(config)
+        listener = RealtimeListener(config, manager, {})
+        # _on_snmp_trap builds a ChangeEvent and schedules _handle_change,
+        # which (no baseline) forwards to send_alert -> alert counter bumps.
+        listener._on_snmp_trap("rtr01", "10.0.0.1", "linkDown")
+        await asyncio.sleep(0.1)
+        assert manager._alert_count == 1
+
+    @pytest.mark.asyncio
+    async def test_on_snmp_trap_runs_with_baseline(self) -> None:
+        config = AlertConfig()
+        manager = AlertManager(config)
+        device = Device(name="rtr01", host="10.0.0.1", username="admin", password="x")
+        listener = RealtimeListener(config, manager, {"rtr01": device}, baseline={"rtr01": {}})
+        fake_snapshot = MagicMock()
+        fake_snapshot.collection_error = None
+        with patch(
+            "audnet.collector_async.collect_device_async",
+            new=AsyncMock(return_value=fake_snapshot),
+        ):
+            listener._on_snmp_trap("rtr01", "10.0.0.1", "linkDown")
+            await asyncio.sleep(0.1)
+        assert manager._alert_count == 1
+
+
+async def _await_pending() -> None:
+    import asyncio
+
+    await asyncio.sleep(0.05)
+
+
+class TestHandleChangeBaseline:
+    @pytest.mark.asyncio
+    async def test_enriches_with_compliance_results(self) -> None:
+        config = AlertConfig()
+        manager = AlertManager(config)
+        device = Device(name="rtr01", host="10.0.0.1", username="admin", password="x")
+        listener = RealtimeListener(config, manager, {"rtr01": device}, baseline={"rtr01": {}})
+
+        fake_snapshot = MagicMock()
+        fake_snapshot.collection_error = None
+        fake_results = [
+            MagicMock(check_name="ssh_v2_only", passed=True, severity="high", detail="ok"),
+            MagicMock(check_name="strong_crypto", passed=False, severity="high", detail="weak"),
+        ]
+        with patch(
+            "audnet.collector_async.collect_device_async",
+            new=AsyncMock(return_value=fake_snapshot),
+        ):
+            with patch("audnet.compliance.run_checks", return_value=fake_results):
+                event = _event()
+                await listener._handle_change(event)
+
+        assert event.compliance_results  # enriched
+        # At least one failing check -> severity escalated to high
+        assert event.severity == "high"
+        await manager.send_alert(event)
+        assert manager._alert_count == 1
+
+    @pytest.mark.asyncio
+    async def test_skips_enrichment_when_device_unknown(self) -> None:
+        config = AlertConfig()
+        manager = AlertManager(config)
+        listener = RealtimeListener(config, manager, {}, baseline={"rtr01": {}})
+        event = _event(device_name="unknown-rtr")
+        # device not in inventory -> baseline enrichment branch skipped
+        await listener._handle_change(event)
+        assert event.compliance_results == []