diff --git a/src/audnet/git_history.py b/src/audnet/git_history.py index c7a32ca..41bd818 100644 --- a/src/audnet/git_history.py +++ b/src/audnet/git_history.py @@ -105,9 +105,7 @@ def sanitize_config(raw_config: str, device_name: str) -> str: return "".join(sanitized) -def sanitize_config_to_file( - raw_config: str, device_name: str, output_path: Path -) -> None: +def sanitize_config_to_file(raw_config: str, device_name: str, output_path: Path) -> None: """Sanitize a device config and write directly to a file. Processes the raw config line-by-line, writing each line to *output_path* @@ -169,17 +167,19 @@ def init_git_repo( repo_path.mkdir(parents=True, exist_ok=True) repo = gitpython.Repo.init(repo_path, bare=bare) - - # Create an initial commit so the repo has a valid HEAD + # Configure user identity even for bare repos (config lives in git_dir). _configure_repo(repo) - initial_file = repo_path / ".gitkeep" - initial_file.touch() - repo.index.add([str(initial_file.relative_to(repo_path))]) - repo.index.commit( - "chore: initialize audnet Git config history repo", - commit_date=_commit_dt(), - author_date=_commit_dt(), - ) + if not bare: + # Seed an initial commit so the repo has a valid HEAD. A bare repo has + # no working tree, so no seed file/commit is created there. + initial_file = repo_path / ".gitkeep" + initial_file.touch() + repo.index.add([str(initial_file.relative_to(repo_path))]) + repo.index.commit( + "chore: initialize audnet Git config history repo", + commit_date=_commit_dt(), + author_date=_commit_dt(), + ) logger.info("Initialized new Git config history repo at %s", repo_path) return repo diff --git a/src/audnet/vendor_registry.py b/src/audnet/vendor_registry.py index 4874cd5..9b0c034 100644 --- a/src/audnet/vendor_registry.py +++ b/src/audnet/vendor_registry.py @@ -352,11 +352,19 @@ async def detect_vendor_snmp( except ImportError: from pysnmp.hlapi.asyncio import getCmd as _get_cmd # pysnmp 4.x + # pysnmp 7.x exposes UdpTransportTarget as an awaitable factory + # (UdpTransportTarget.create((host, port), timeout=, retries=)); the old + # pysnmp 4.x positional-constructor form is kept as a fallback. + try: + transport_target = await UdpTransportTarget.create((host, port), timeout=timeout, retries=1) + except (AttributeError, TypeError): # pragma: no cover - pysnmp 4.x path + transport_target = UdpTransportTarget((host, port), timeout=timeout, retries=1) + try: iterator = _get_cmd( SnmpEngine(), CommunityData(community), - UdpTransportTarget((host, port), timeout=timeout, retries=1), + transport_target, ContextData(), ObjectType(ObjectIdentity(_SYS_DESCR_OID)), ) diff --git a/tests/test_git_history.py b/tests/test_git_history.py index a51010b..150ba01 100644 --- a/tests/test_git_history.py +++ b/tests/test_git_history.py @@ -316,6 +316,29 @@ def test_default_path(self, tmp_path: Path, monkeypatch): repo = init_git_repo(tmp_path / "sub") assert repo.working_tree_dir is not None + def test_creates_bare_repo(self, tmp_path: Path): + # Bug fix regression: bare repos must initialize without crashing + # (no seed .gitkeep commit, since a bare repo has no working tree). + repo_path = tmp_path / "bare-repo.git" + repo = init_git_repo(repo_path, bare=True) + assert repo.bare is True + assert repo.working_tree_dir is None + assert (repo_path / "config").exists() + + def test_save_on_bare_repo_raises(self, tmp_path: Path): + # The previously-dead working-tree guard in save_config_snapshot is now + # reachable: a bare repo has no working tree, so saving configs errors. + repo_path = tmp_path / "bare-repo.git" + init_git_repo(repo_path, bare=True) + with pytest.raises(GitHistoryError): + save_config_snapshot({"rtr01": "hostname rtr01\n"}, history_dir=repo_path) + + def test_rollback_on_bare_repo_raises(self, tmp_path: Path): + repo_path = tmp_path / "bare-repo.git" + init_git_repo(repo_path, bare=True) + with pytest.raises(GitHistoryError): + rollback_config("rtr01", "HEAD", history_dir=repo_path) + # --------------------------------------------------------------------------- # save_config_snapshot diff --git a/tests/test_vendor_registry_snmp.py b/tests/test_vendor_registry_snmp.py index 10ced1d..4232114 100644 --- a/tests/test_vendor_registry_snmp.py +++ b/tests/test_vendor_registry_snmp.py @@ -1,9 +1,9 @@ """Tests for the SNMP-based vendor auto-detection path in vendor_registry. -NOTE: these unit tests isolate the external SNMP transport -(``pysnmp.hlapi.asyncio.UdpTransportTarget``) because the project's current -call signature is incompatible with pysnmp 7.x (see flagged issue). The tests -exercise ``detect_vendor_snmp``'s branching logic, not the live transport. +The transport target is isolated because a real SNMP query would hit the +network. These tests exercise ``detect_vendor_snmp``'s branching logic with a +mocked ``get_cmd``; a separate live test confirms the pysnmp 7.x +``await UdpTransportTarget.create(...)`` call shape works without crashing. """ from typing import Any @@ -18,24 +18,25 @@ class _DummyTransportTarget: - """Stand-in for UdpTransportTarget so the code under test can run without - hitting the pysnmp 7.x constructor incompatibility. Accepts the same - call shape the project uses: ``UdpTransportTarget((host, port), timeout=, retries=)``.""" + """Stand-in for UdpTransportTarget mirroring the pysnmp 7.x awaitable + factory: ``await UdpTransportTarget.create((host, port), timeout=, retries=)``.""" def __init__(self, address: Any, timeout: float = 1, retries: int = 5, **_kwargs: Any) -> None: self.address = address + @classmethod + async def create( + cls, address: Any, timeout: float = 1, retries: int = 5, **_kwargs: Any + ) -> "_DummyTransportTarget": + return cls(address, timeout=timeout, retries=retries) + def _patch_snmp( monkeypatch: MonkeyPatch, get_cmd_result: _SnmpResult | None = None, raise_exc: Exception | None = None, ) -> None: - """Patch the in-function imports used by ``detect_vendor_snmp``. - - Either ``get_cmd_result`` (a pysnmp 4-tuple) is returned when awaited, or - ``raise_exc`` is raised when the (mocked) get_cmd is invoked. - """ + """Patch the in-function imports used by ``detect_vendor_snmp``.""" async def _fake_get_cmd(*_args: Any, **_kwargs: Any) -> _SnmpResult: if raise_exc is not None: @@ -104,3 +105,13 @@ async def _fake_getCmd(*_args: Any, **_kwargs: Any) -> _SnmpResult: monkeypatch.setattr(_asyncio_hlapi, "getCmd", _fake_getCmd, raising=False) monkeypatch.setattr(_asyncio_hlapi, "UdpTransportTarget", _DummyTransportTarget) assert await detect_vendor_snmp("10.0.0.1") == "cisco_ios" + + async def test_live_pysnmp7_create_does_not_crash(self, monkeypatch: MonkeyPatch) -> None: + # Bug fix regression: pysnmp 7.x requires `await UdpTransportTarget.create(...)`. + # Use the REAL UdpTransportTarget (not the dummy) so .create() is exercised. + async def _fake_get_cmd(*_args: Any, **_kwargs: Any) -> _SnmpResult: + return ("RequestTimedOut", 0, 0, []) + + monkeypatch.setattr(_asyncio_hlapi, "get_cmd", _fake_get_cmd) + result = await detect_vendor_snmp("192.0.2.1", timeout=1, port=161) + assert result == "cisco_ios" diff --git a/tests/test_vendors_expanded.py b/tests/test_vendors_expanded.py index 0af4c50..8543899 100644 --- a/tests/test_vendors_expanded.py +++ b/tests/test_vendors_expanded.py @@ -326,26 +326,34 @@ async def test_snmp_import_compatibility(self): @pytest.mark.asyncio async def test_snmp_detection_with_mock(self): - """SNMP detection parses sysDescr and returns correct vendor.""" + """SNMP detection parses sysDescr and returns correct vendor. + + Hermetic: the transport target is mocked so no real network call is + made. The fixture sysDescr is Cisco IOS-XE, so the expected vendor is + cisco_xe (previously this test passed only because the broken pysnmp + 4.x transport constructor raised and the broad except fell back). + """ from unittest.mock import AsyncMock, patch from audnet.vendor_registry import detect_vendor_snmp - # Mock the pysnmp response: simulate a Cisco IOS-XE sysDescr mock_var_bind = (None, "Cisco IOS-XE Software, Version 17.15.1") mock_result = (None, 0, 0, [mock_var_bind]) - # pysnmp 7.x exports get_cmd; 4.x exports getCmd - try: - from pysnmp.hlapi.asyncio import get_cmd # noqa: F401 + class _DummyTransportTarget: + def __init__(self, address, timeout=1, retries=5, **_kw): + self.address = address - pysnmp_path = "pysnmp.hlapi.asyncio.get_cmd" - except ImportError: - pysnmp_path = "pysnmp.hlapi.asyncio.getCmd" + @classmethod + async def create(cls, address, timeout=1, retries=5, **_kw): + return cls(address, timeout=timeout, retries=retries) - with patch(pysnmp_path, new_callable=AsyncMock, return_value=mock_result): + with ( + patch("pysnmp.hlapi.asyncio.get_cmd", new_callable=AsyncMock, return_value=mock_result), + patch("pysnmp.hlapi.asyncio.UdpTransportTarget", _DummyTransportTarget), + ): result = await detect_vendor_snmp("10.0.0.1", timeout=1) - assert result == "cisco_ios" + assert result == "cisco_xe" @pytest.mark.asyncio async def test_snmp_detection_error(self):