diff --git a/.containerignore b/.containerignore new file mode 100644 index 0000000..0f48a0e --- /dev/null +++ b/.containerignore @@ -0,0 +1,6 @@ +.git +.vscode +build +**/__pycache__ +**/*.pyc +**/*.so diff --git a/.github/workflows/system-tests.yml b/.github/workflows/system-tests.yml new file mode 100644 index 0000000..d28b122 --- /dev/null +++ b/.github/workflows/system-tests.yml @@ -0,0 +1,83 @@ +name: System tests + +on: + pull_request: + push: + branches: + - master + +permissions: + contents: read + +jobs: + snmpd-valgrind: + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - uses: actions/checkout@v4 + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y \ + build-essential \ + libsnmp-dev \ + python3-dev \ + python3-setuptools \ + snmp \ + snmpd \ + valgrind + + - name: Run API system tests and Valgrind + run: sh netsnmp/tests/system/run.sh + + coverage: + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - uses: actions/checkout@v4 + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y \ + build-essential \ + libsnmp-dev \ + python3-dev \ + python3-setuptools \ + snmp \ + snmpd + + - name: Run API system tests with C coverage + run: sh netsnmp/tests/system/run-coverage.sh + + - name: Upload C coverage report + uses: actions/upload-artifact@v4 + with: + name: client-intf-coverage + path: | + client_intf-coverage.txt + client_intf.c.gcov + + asan-ubsan: + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - uses: actions/checkout@v4 + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y \ + build-essential \ + libsnmp-dev \ + python3-dev \ + python3-setuptools \ + snmp \ + snmpd + + - name: Run API system tests with ASan and UBSan + run: sh netsnmp/tests/system/run-sanitizers.sh diff --git a/Containerfile.system-tests b/Containerfile.system-tests new file mode 100644 index 0000000..71fa576 --- /dev/null +++ b/Containerfile.system-tests @@ -0,0 +1,20 @@ +FROM ubuntu:latest + +ENV DEBIAN_FRONTEND=noninteractive + +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + build-essential \ + libsnmp-dev \ + python3 \ + python3-dev \ + python3-setuptools \ + snmp \ + snmpd \ + valgrind \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /src +COPY . . + +CMD ["sh", "netsnmp/tests/system/run.sh"] diff --git a/netsnmp/tests/system/README.md b/netsnmp/tests/system/README.md new file mode 100644 index 0000000..898afae --- /dev/null +++ b/netsnmp/tests/system/README.md @@ -0,0 +1,24 @@ +# System tests + +The system tests start a private SNMPv2c agent on `127.0.0.1:1161`, exercise +all convenience functions and `Session` methods, and repeat the test suite +under Valgrind. Each SNMP operation has a separate `test_*.py` file, and the +runner discovers new test files automatically. + +The normal and Valgrind suites each run all discovered tests in one process. +CI also builds the extension with GCC coverage and with ASan/UBSan. The +coverage job uploads the full C coverage report. + +Coverage also includes SNMPv1, multiple varbinds, protocol error paths, +repeated session and walk lifecycles, and the Python `Varbind` and `VarList` +containers. + +Run the same Ubuntu environment locally with Podman: + +```sh +podman build -f Containerfile.system-tests -t python3-netsnmp-system-tests . +podman run --rm python3-netsnmp-system-tests +``` + +Valgrind returns a non-zero status for memory errors and definite leaks. +GitHub Actions runs the same test script for every pull request. diff --git a/netsnmp/tests/system/common.py b/netsnmp/tests/system/common.py new file mode 100644 index 0000000..2c68326 --- /dev/null +++ b/netsnmp/tests/system/common.py @@ -0,0 +1,21 @@ +HOST = '127.0.0.1:1161' +READ_ARGS = { + 'Version': 2, + 'DestHost': HOST, + 'Community': 'public', + 'Timeout': 1000000, + 'Retries': 0, +} +WRITE_ARGS = READ_ARGS.copy() +WRITE_ARGS['Community'] = 'private' + +SYS_DESCR = '.1.3.6.1.2.1.1.1' +SYS_UPTIME = '.1.3.6.1.2.1.1.3' +SYS_LOCATION = '.1.3.6.1.2.1.1.6' +SYSTEM = '.1.3.6.1.2.1.1' + + +def assert_value(test_case, values): + test_case.assertIsInstance(values, tuple) + test_case.assertTrue(values) + test_case.assertIsNotNone(values[0]) diff --git a/netsnmp/tests/system/run-coverage.sh b/netsnmp/tests/system/run-coverage.sh new file mode 100755 index 0000000..1f4d55f --- /dev/null +++ b/netsnmp/tests/system/run-coverage.sh @@ -0,0 +1,17 @@ +#!/bin/sh +set -eu + +ROOT=$(CDPATH= cd -- "$(dirname "$0")/../../.." && pwd) +cd "$ROOT" + +rm -rf build +rm -f netsnmp/client_intf*.so + +CFLAGS='--coverage -O0' \ +LDFLAGS='--coverage' \ +RUN_VALGRIND=0 \ + sh netsnmp/tests/system/run.sh + +OBJECT_DIR=$(dirname "$(find build -name client_intf.o -print -quit)") +gcov -b -f -o "$OBJECT_DIR" netsnmp/client_intf.c \ + | tee client_intf-coverage.txt diff --git a/netsnmp/tests/system/run-sanitizers.sh b/netsnmp/tests/system/run-sanitizers.sh new file mode 100755 index 0000000..9395ab3 --- /dev/null +++ b/netsnmp/tests/system/run-sanitizers.sh @@ -0,0 +1,17 @@ +#!/bin/sh +set -eu + +ROOT=$(CDPATH= cd -- "$(dirname "$0")/../../.." && pwd) +cd "$ROOT" + +rm -rf build +rm -f netsnmp/client_intf*.so + +SANITIZERS=address,undefined +CFLAGS="-O1 -g -fno-omit-frame-pointer -fsanitize=$SANITIZERS" \ +LDFLAGS="-fsanitize=$SANITIZERS" \ +LD_PRELOAD=$(gcc -print-file-name=libasan.so) \ +ASAN_OPTIONS=detect_leaks=0:abort_on_error=1 \ +UBSAN_OPTIONS=halt_on_error=1:print_stacktrace=1 \ +RUN_VALGRIND=0 \ + sh netsnmp/tests/system/run.sh \ No newline at end of file diff --git a/netsnmp/tests/system/run.sh b/netsnmp/tests/system/run.sh new file mode 100755 index 0000000..6a24c37 --- /dev/null +++ b/netsnmp/tests/system/run.sh @@ -0,0 +1,60 @@ +#!/bin/sh +set -eu + +ROOT=$(CDPATH= cd -- "$(dirname "$0")/../../.." && pwd) +RUNTIME_DIR=$(mktemp -d) +SNMPD_PID= + +cleanup() { + if [ -n "$SNMPD_PID" ]; then + kill "$SNMPD_PID" 2>/dev/null || true + wait "$SNMPD_PID" 2>/dev/null || true + fi + rm -rf "$RUNTIME_DIR" +} +trap cleanup EXIT INT TERM + +cd "$ROOT" +python3 setup.py build_ext --inplace + +SNMPD_CONFIG="$RUNTIME_DIR/snmpd.conf" +cat "$ROOT/netsnmp/tests/system/snmpd.conf" >"$SNMPD_CONFIG" +printf '\npersistentDir %s\n' "$RUNTIME_DIR" >>"$SNMPD_CONFIG" + +snmpd -f -Lo -C \ + -c "$SNMPD_CONFIG" \ + -p "$RUNTIME_DIR/snmpd.pid" \ + >"$RUNTIME_DIR/snmpd.log" 2>&1 & +SNMPD_PID=$! + +attempt=0 +until snmpget -v2c -c public -t 1 -r 0 \ + 127.0.0.1:1161 .1.3.6.1.2.1.1.1.0 >/dev/null 2>&1; do + attempt=$((attempt + 1)) + if [ "$attempt" -ge 20 ] || ! kill -0 "$SNMPD_PID" 2>/dev/null; then + cat "$RUNTIME_DIR/snmpd.log" + exit 1 + fi +done + +python3 -m unittest discover -v -s netsnmp/tests/system -p 'test_*.py' + +if [ "${RUN_VALGRIND:-1}" = 0 ]; then + exit 0 +fi + +valgrind \ + --error-exitcode=99 \ + --leak-check=full \ + --show-leak-kinds=definite \ + --errors-for-leak-kinds=definite \ + --track-origins=yes \ + --log-file="$RUNTIME_DIR/valgrind.log" \ + python3 -m unittest discover -v -s netsnmp/tests/system \ + -p 'test_*.py' || { + status=$? + cat "$RUNTIME_DIR/valgrind.log" + exit "$status" + } + +cat "$RUNTIME_DIR/valgrind.log" diff --git a/netsnmp/tests/system/snmpd.conf b/netsnmp/tests/system/snmpd.conf new file mode 100644 index 0000000..e986783 --- /dev/null +++ b/netsnmp/tests/system/snmpd.conf @@ -0,0 +1,3 @@ +agentAddress udp:127.0.0.1:1161 +rocommunity public 127.0.0.1 +rwcommunity private 127.0.0.1 diff --git a/netsnmp/tests/system/test_errors.py b/netsnmp/tests/system/test_errors.py new file mode 100644 index 0000000..ea8376d --- /dev/null +++ b/netsnmp/tests/system/test_errors.py @@ -0,0 +1,39 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import READ_ARGS, SYS_DESCR, SYS_LOCATION + + +class ErrorPathTests(unittest.TestCase): + def test_unknown_oid_returns_no_value(self): + varbind = netsnmp.Varbind('.1.3.6.1.2.1.1.999', '0') + values = netsnmp.snmpget(varbind, **READ_ARGS) + + self.assertEqual(values, (None,)) + self.assertIn(varbind.type, ('NOSUCHOBJECT', 'NOSUCHINSTANCE')) + + def test_read_only_set_is_rejected(self): + session = netsnmp.Session(**READ_ARGS) + varlist = netsnmp.VarList( + netsnmp.Varbind( + SYS_LOCATION, '0', b'not-written', 'OCTETSTR')) + + self.assertEqual(session.set(varlist), 0) + self.assertTrue(session.ErrorStr) + self.assertNotEqual(session.ErrorNum, 0) + + def test_timeout_updates_session_error(self): + timeout_args = READ_ARGS.copy() + timeout_args['DestHost'] = '127.0.0.1:1162' + timeout_args['Timeout'] = 100000 + session = netsnmp.Session(**timeout_args) + varlist = netsnmp.VarList(netsnmp.Varbind(SYS_DESCR, '0')) + + self.assertEqual(session.get(varlist), (None,)) + self.assertTrue(session.ErrorStr) + self.assertNotEqual(session.ErrorInd, 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_get.py b/netsnmp/tests/system/test_get.py new file mode 100644 index 0000000..66036c4 --- /dev/null +++ b/netsnmp/tests/system/test_get.py @@ -0,0 +1,21 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import READ_ARGS, SYS_DESCR, assert_value + + +class GetTests(unittest.TestCase): + def test_convenience_function(self): + values = netsnmp.snmpget( + netsnmp.Varbind(SYS_DESCR, '0'), **READ_ARGS) + assert_value(self, values) + + def test_session_method(self): + session = netsnmp.Session(**READ_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYS_DESCR, '0')) + assert_value(self, session.get(varlist)) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_getbulk.py b/netsnmp/tests/system/test_getbulk.py new file mode 100644 index 0000000..1ab1096 --- /dev/null +++ b/netsnmp/tests/system/test_getbulk.py @@ -0,0 +1,21 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import READ_ARGS, SYSTEM, assert_value + + +class GetBulkTests(unittest.TestCase): + def test_convenience_function(self): + values = netsnmp.snmpgetbulk( + 0, 4, netsnmp.Varbind(SYSTEM), **READ_ARGS) + assert_value(self, values) + + def test_session_method(self): + session = netsnmp.Session(**READ_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, session.getbulk(0, 4, varlist)) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_getnext.py b/netsnmp/tests/system/test_getnext.py new file mode 100644 index 0000000..f94fd9e --- /dev/null +++ b/netsnmp/tests/system/test_getnext.py @@ -0,0 +1,21 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import READ_ARGS, SYSTEM, assert_value + + +class GetNextTests(unittest.TestCase): + def test_convenience_function(self): + varbind = netsnmp.Varbind(SYSTEM) + assert_value(self, netsnmp.snmpgetnext(varbind, **READ_ARGS)) + self.assertTrue(varbind.tag) + + def test_session_method(self): + session = netsnmp.Session(**READ_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, session.getnext(varlist)) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_lifecycle.py b/netsnmp/tests/system/test_lifecycle.py new file mode 100644 index 0000000..5a31bab --- /dev/null +++ b/netsnmp/tests/system/test_lifecycle.py @@ -0,0 +1,31 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import ( + READ_ARGS, + SYS_DESCR, + SYSTEM, + assert_value, +) + + +class SessionLifecycleTests(unittest.TestCase): + def test_repeated_session_creation(self): + for _ in range(25): + session = netsnmp.Session(**READ_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYS_DESCR, '0')) + assert_value(self, session.get(varlist)) + del session + + def test_repeated_walks(self): + session = netsnmp.Session(**READ_ARGS) + + for _ in range(10): + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, session.walk(varlist)) + self.assertGreater(len(varlist), 1) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_multivarbind.py b/netsnmp/tests/system/test_multivarbind.py new file mode 100644 index 0000000..ae130b9 --- /dev/null +++ b/netsnmp/tests/system/test_multivarbind.py @@ -0,0 +1,52 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import ( + READ_ARGS, + SYS_DESCR, + SYS_LOCATION, + SYS_UPTIME, + SYSTEM, + assert_value, +) + + +class MultiVarbindTests(unittest.TestCase): + def test_get_multiple_varbinds(self): + varlist = netsnmp.VarList( + netsnmp.Varbind(SYS_DESCR, '0'), + netsnmp.Varbind(SYS_UPTIME, '0'), + netsnmp.Varbind(SYS_LOCATION, '0'), + ) + values = netsnmp.Session(**READ_ARGS).get(varlist) + + self.assertEqual(len(values), 3) + self.assertTrue(all(value is not None for value in values)) + self.assertTrue(all(varbind.val is not None for varbind in varlist)) + + def test_getnext_multiple_varbinds(self): + varlist = netsnmp.VarList( + netsnmp.Varbind(SYS_DESCR), + netsnmp.Varbind(SYS_UPTIME), + netsnmp.Varbind(SYS_LOCATION), + ) + values = netsnmp.Session(**READ_ARGS).getnext(varlist) + + self.assertEqual(len(values), 3) + self.assertTrue(all(varbind.tag for varbind in varlist)) + + def test_getbulk_multiple_varbinds(self): + varlist = netsnmp.VarList( + netsnmp.Varbind(SYSTEM), + netsnmp.Varbind(SYS_UPTIME), + ) + values = netsnmp.Session(**READ_ARGS).getbulk(1, 3, varlist) + + assert_value(self, values) + self.assertGreater(len(values), 2) + self.assertEqual(len(varlist), len(values)) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_set.py b/netsnmp/tests/system/test_set.py new file mode 100644 index 0000000..2ff80b2 --- /dev/null +++ b/netsnmp/tests/system/test_set.py @@ -0,0 +1,35 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import ( + READ_ARGS, + SYS_LOCATION, + WRITE_ARGS, +) + + +class SetTests(unittest.TestCase): + def test_convenience_function(self): + value = b'convenience-api' + result = netsnmp.snmpset( + netsnmp.Varbind(SYS_LOCATION, '0', value, 'OCTETSTR'), + **WRITE_ARGS) + self.assertEqual(result, 1) + self.assertEqual(netsnmp.snmpget( + netsnmp.Varbind(SYS_LOCATION, '0'), **READ_ARGS)[0], value) + + def test_session_method(self): + read_session = netsnmp.Session(**READ_ARGS) + write_session = netsnmp.Session(**WRITE_ARGS) + value = b'session-api' + set_vars = netsnmp.VarList( + netsnmp.Varbind(SYS_LOCATION, '0', value, 'OCTETSTR')) + self.assertEqual(write_session.set(set_vars), 1) + + verify_vars = netsnmp.VarList(netsnmp.Varbind(SYS_LOCATION, '0')) + self.assertEqual(read_session.get(verify_vars)[0], value) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_v1.py b/netsnmp/tests/system/test_v1.py new file mode 100644 index 0000000..db8b338 --- /dev/null +++ b/netsnmp/tests/system/test_v1.py @@ -0,0 +1,53 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import ( + READ_ARGS, + SYS_DESCR, + SYS_LOCATION, + SYSTEM, + WRITE_ARGS, + assert_value, +) + + +V1_READ_ARGS = READ_ARGS.copy() +V1_READ_ARGS['Version'] = 1 +V1_WRITE_ARGS = WRITE_ARGS.copy() +V1_WRITE_ARGS['Version'] = 1 + + +class SnmpV1Tests(unittest.TestCase): + def test_get(self): + values = netsnmp.snmpget( + netsnmp.Varbind(SYS_DESCR, '0'), **V1_READ_ARGS) + assert_value(self, values) + + def test_getnext(self): + values = netsnmp.snmpgetnext( + netsnmp.Varbind(SYSTEM), **V1_READ_ARGS) + assert_value(self, values) + + def test_getbulk_is_not_supported(self): + session = netsnmp.Session(**V1_READ_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + self.assertIsNone(session.getbulk(0, 4, varlist)) + + def test_set(self): + value = b'snmp-v1' + result = netsnmp.snmpset( + netsnmp.Varbind(SYS_LOCATION, '0', value, 'OCTETSTR'), + **V1_WRITE_ARGS) + self.assertEqual(result, 1) + self.assertEqual(netsnmp.snmpget( + netsnmp.Varbind(SYS_LOCATION, '0'), **V1_READ_ARGS)[0], value) + + def test_walk(self): + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, netsnmp.snmpwalk(varlist, **V1_READ_ARGS)) + self.assertGreater(len(varlist), 1) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_varbind.py b/netsnmp/tests/system/test_varbind.py new file mode 100644 index 0000000..f3c2660 --- /dev/null +++ b/netsnmp/tests/system/test_varbind.py @@ -0,0 +1,37 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import SYS_DESCR + + +class VarbindTests(unittest.TestCase): + def test_complete_numeric_oid_is_preserved(self): + varbind = netsnmp.Varbind(SYS_DESCR + '.0') + + self.assertEqual(varbind.tag, SYS_DESCR + '.0') + self.assertEqual(varbind.iid, '') + + def test_explicit_iid_is_converted_to_string(self): + varbind = netsnmp.Varbind(SYS_DESCR, 0) + + self.assertEqual(varbind.tag, SYS_DESCR) + self.assertEqual(varbind.iid, '0') + + def test_value_preserves_its_type(self): + byte_value = netsnmp.Varbind(SYS_DESCR, '0', b'value') + integer_value = netsnmp.Varbind(SYS_DESCR, '0', 42) + + self.assertEqual(byte_value.val, b'value') + self.assertEqual(integer_value.val, 42) + + def test_print_str_returns_all_fields(self): + varbind = netsnmp.Varbind(SYS_DESCR, '0', b'value', 'OCTETSTR') + + self.assertEqual( + varbind.print_str(), + (SYS_DESCR, '0', b'value', 'OCTETSTR')) + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_varlist.py b/netsnmp/tests/system/test_varlist.py new file mode 100644 index 0000000..19c193f --- /dev/null +++ b/netsnmp/tests/system/test_varlist.py @@ -0,0 +1,41 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import SYS_DESCR, SYS_LOCATION + + +class VarListTests(unittest.TestCase): + def test_constructor_wraps_oid(self): + varlist = netsnmp.VarList(SYS_DESCR) + + self.assertEqual(len(varlist), 1) + self.assertIsInstance(varlist[0], netsnmp.Varbind) + self.assertEqual(varlist[0].tag, SYS_DESCR) + + def test_append_iterate_and_delete(self): + first = netsnmp.Varbind(SYS_DESCR, '0') + second = netsnmp.Varbind(SYS_LOCATION, '0') + varlist = netsnmp.VarList(first) + + varlist.append(second) + self.assertEqual(list(varlist), [first, second]) + + del varlist[0] + self.assertEqual(list(varlist), [second]) + + def test_append_rejects_non_varbind(self): + varlist = netsnmp.VarList() + + with self.assertRaises(TypeError): + varlist.append(SYS_DESCR) + + def test_assignment_rejects_non_varbind(self): + varlist = netsnmp.VarList(netsnmp.Varbind(SYS_DESCR)) + + with self.assertRaises(TypeError): + varlist[0] = SYS_LOCATION + + +if __name__ == '__main__': + unittest.main() diff --git a/netsnmp/tests/system/test_walk.py b/netsnmp/tests/system/test_walk.py new file mode 100644 index 0000000..ef36ced --- /dev/null +++ b/netsnmp/tests/system/test_walk.py @@ -0,0 +1,22 @@ +import unittest + +import netsnmp + +from netsnmp.tests.system.common import READ_ARGS, SYSTEM, assert_value + + +class WalkTests(unittest.TestCase): + def test_convenience_function(self): + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, netsnmp.snmpwalk(varlist, **READ_ARGS)) + self.assertGreater(len(varlist), 1) + + def test_session_method(self): + session = netsnmp.Session(**READ_ARGS) + varlist = netsnmp.VarList(netsnmp.Varbind(SYSTEM)) + assert_value(self, session.walk(varlist)) + self.assertGreater(len(varlist), 1) + + +if __name__ == '__main__': + unittest.main()