Add kernel selftest equivalent roadmap tests

Author: varun-r-mallya

pythonbpf/maps/__init__.py

@@ -1,5 +1,12 @@
-from .maps import HashMap, PerfEventArray, RingBuffer
+from .maps import ArrayMap, HashMap, PerfEventArray, RingBuffer
 from .maps_pass import maps_proc
 from .map_types import BPFMapType
 
-__all__ = ["HashMap", "PerfEventArray", "maps_proc", "RingBuffer", "BPFMapType"]
+__all__ = [
+    "ArrayMap",
+    "HashMap",
+    "PerfEventArray",
+    "maps_proc",
+    "RingBuffer",
+    "BPFMapType",
+]

pythonbpf/maps/maps.py

@@ -26,6 +26,26 @@ def update(self, key, value, flags=None):
             raise KeyError(f"Key {key} not found in map")
 
 
+class ArrayMap:
+    def __init__(self, key, value, max_entries):
+        self.key = key
+        self.value = value
+        self.max_entries = max_entries
+        self.entries = {}
+
+    def lookup(self, key):
+        return self.entries.get(key)
+
+    def update(self, key, value, flags=None):
+        self.entries[key] = value
+
+    def delete(self, key):
+        if key in self.entries:
+            del self.entries[key]
+        else:
+            raise KeyError(f"Key {key} not found in map")
+
+
 class PerfEventArray:
     def __init__(self, key_size, value_size):
         self.key_type = key_size

pythonbpf/maps/maps_pass.py

@@ -138,6 +138,14 @@ def process_hash_map(map_name, rval, compilation_context):
     return map_global
 
 
+@MapProcessorRegistry.register("ArrayMap")
+def process_array_map(map_name, rval, compilation_context):
+    """Document the planned BPF_ARRAY map support with an explicit failure."""
+    raise NotImplementedError(
+        "ArrayMap is not implemented yet; add BPF_MAP_TYPE_ARRAY metadata support"
+    )
+
+
 @MapProcessorRegistry.register("PerfEventArray")
 def process_perf_event_map(map_name, rval, compilation_context):
     """Process a BPF_PERF_EVENT_ARRAY map declaration"""

tests/README.md

@@ -80,6 +80,14 @@ All xfails use `strict = True`: if a test starts **passing** it shows up as **XP
 2. Run `make test` — the file is discovered and tested automatically at all levels.
 3. If the test is expected to fail, add it to `tests/test_config.toml` instead of `passing_tests/`.
 
+## Kernel selftest equivalents
+
+`tests/kernel_selftest_equivalent/` contains PythonBPF versions of important
+kernel BPF selftests from `bpf-next/tools/testing/selftests/bpf`. These tests
+describe features PythonBPF should grow next. They are collected by default and
+must be listed as strict expected failures in `tests/test_config.toml` until the
+corresponding feature lands.
+
 ## Directory structure
 
 ```
@@ -96,5 +104,6 @@ tests/
 │   ├── compiler.py            ← wrappers around compile_to_ir() + _run_llc()
 │   └── verifier.py            ← bpftool subprocess wrapper
 ├── passing_tests/             ← programs that should compile and verify cleanly
-└── failing_tests/             ← programs with known issues (declared in test_config.toml)
+├── failing_tests/             ← programs with known issues (declared in test_config.toml)
+└── kernel_selftest_equivalent/ ← kernel-selftest-inspired feature roadmap tests
 ```

tests/conftest.py

@@ -16,6 +16,7 @@
 """
 
 import logging
+import warnings
 
 import pytest
 
@@ -24,11 +25,15 @@
 # ── vmlinux availability ────────────────────────────────────────────────────
 
 try:
-    import vmlinux  # noqa: F401
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", DeprecationWarning)
+        import vmlinux  # noqa: F401
 
     VMLINUX_AVAILABLE = True
-except ImportError:
+    VMLINUX_SKIP_REASON = ""
+except Exception as exc:
     VMLINUX_AVAILABLE = False
+    VMLINUX_SKIP_REASON = f"vmlinux.py not usable for current kernel: {exc}"
 
 
 # ── pytest_generate_tests: parametrize on bpf_test_file ───────────────────
@@ -64,7 +69,10 @@ def pytest_collection_modifyitems(items):
         # vmlinux skip
         if case.needs_vmlinux and not VMLINUX_AVAILABLE:
             item.add_marker(
-                pytest.mark.skip(reason="vmlinux.py not available for current kernel")
+                pytest.mark.skip(
+                    reason=VMLINUX_SKIP_REASON
+                    or "vmlinux.py not available for current kernel"
+                )
             )
             continue
 

tests/framework/collector.py

@@ -33,7 +33,7 @@ def collect_all_test_files() -> list[BpfTestCase]:
     xfail_map: dict = config.get("xfail", {})
 
     cases = []
-    for subdir in ("passing_tests", "failing_tests"):
+    for subdir in ("passing_tests", "failing_tests", "kernel_selftest_equivalent"):
         for py_file in sorted((TESTS_DIR / subdir).rglob("*.py")):
             rel = str(py_file.relative_to(TESTS_DIR))
             needs_vmlinux = _is_vmlinux_test(rel)

tests/kernel_selftest_equivalent/maps/array_map_lookup_update.py

@@ -0,0 +1,35 @@
+# Adapted from bpf-next/tools/testing/selftests/bpf/progs/test_map_ops.c
+# and bpf-next/tools/testing/selftests/bpf/progs/bpf_iter_bpf_array_map.c.
+
+from ctypes import c_int32, c_uint64, c_void_p
+
+from pythonbpf import bpf, bpfglobal, compile, map, section
+from pythonbpf.maps import ArrayMap
+
+
+@bpf
+@map
+def counters() -> ArrayMap:
+    return ArrayMap(key=c_int32, value=c_uint64, max_entries=8)
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_getpid")
+def array_map_lookup_update(ctx: c_void_p) -> c_int32:
+    counters.update(0, 1)
+
+    current = counters.lookup(0)
+    if current:
+        next_value = current + 1
+        counters.update(0, next_value)
+
+    return c_int32(0)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()

tests/kernel_selftest_equivalent/ringbuf/reserve_submit_discard.py

@@ -0,0 +1,48 @@
+# Adapted from bpf-next/tools/testing/selftests/bpf/progs/test_ringbuf.c.
+
+from ctypes import c_int32, c_uint64, c_void_p
+
+from pythonbpf import bpf, bpfglobal, compile, map, section, struct
+from pythonbpf.helper import pid
+from pythonbpf.maps import RingBuffer
+
+
+@bpf
+@struct
+class sample_t:
+    pid: c_uint64
+    seq: c_uint64
+    value: c_uint64
+
+
+@bpf
+@map
+def events() -> RingBuffer:
+    return RingBuffer(max_entries=4096)
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_getpid")
+def ringbuf_reserve_submit_discard(ctx: c_void_p) -> c_int32:
+    first = events.reserve(24)
+    if first:
+        sample = sample_t(first)
+        sample.pid = pid()
+        sample.seq = 0
+        sample.value = 7
+        events.submit(first, 0)
+
+    second = events.reserve(24)
+    if second:
+        events.discard(second, 0)
+
+    return c_int32(0)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()

tests/test_config.toml

@@ -20,3 +20,7 @@
 "failing_tests/vmlinux/assignment_handling.py" = {reason = "Assigning vmlinux enum value (XDP_PASS) to a local variable not yet supported", level = "ir"}
 
 "failing_tests/xdp_pass.py" = {reason = "XDP program using vmlinux structs (struct_xdp_md) and complex map/struct interaction not yet supported", level = "ir"}
+
+"kernel_selftest_equivalent/maps/array_map_lookup_update.py" = {reason = "ArrayMap / BPF_MAP_TYPE_ARRAY support is planned but not implemented yet", level = "ir"}
+
+"kernel_selftest_equivalent/ringbuf/reserve_submit_discard.py" = {reason = "RingBuffer reserve/typed record/discard workflow is planned but not implemented yet", level = "ir"}