Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 13 additions & 13 deletions src/audnet/git_history.py
Original file line number Diff line number Diff line change
Expand Up @@ -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*
Expand Down Expand Up @@ -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

Expand Down
10 changes: 9 additions & 1 deletion src/audnet/vendor_registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)),
)
Expand Down
23 changes: 23 additions & 0 deletions tests/test_git_history.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
35 changes: 23 additions & 12 deletions tests/test_vendor_registry_snmp.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -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"
28 changes: 18 additions & 10 deletions tests/test_vendors_expanded.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
Loading