diff --git a/CHANGES.md b/CHANGES.md index 408554e..98f264f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -3,6 +3,7 @@ ## Unreleased - Add Renode device support. +- Add opt-in resource leak detection fixture. ## 0.0.3 diff --git a/Documentation/usage.rst b/Documentation/usage.rst index 531383d..923c1e7 100644 --- a/Documentation/usage.rst +++ b/Documentation/usage.rst @@ -229,6 +229,80 @@ Remaining tests are skipped if all retries fail. max_retries: 3 # reboot attempts before skipping remaining tests base_delay: 2.0 # seconds between retries (doubles each attempt) +.. _resource-monitoring: + +Resource Monitoring +=================== + +Tests that request the ``resources`` fixture get their resource usage +sampled before and after the test body. A delta that does not settle fails +the test at teardown. The feature is disabled unless enabled here and needs +``CONFIG_FS_PROCFS`` on the target. + +.. code-block:: yaml + + config: + resources: + enabled: True + probes: [tasks, fds] # default + settle_retries: 3 # re-samples while a delta remains + settle_delay: 1.0 # seconds between re-samples + heap_tolerance: 0 # bytes of heap growth tolerated + cpuload_max: 10 # percent, cpuload probe only + +Each probe runs one NSH command and compares the two samples: + +.. list-table:: + :header-rows: 1 + + * - Probe + - Command + - Reports + - Requires + * - ``heap`` + - ``free`` + - ``used`` grew past ``heap_tolerance`` or ``nused`` grew, per heap + - ``CONFIG_NSH_DISABLE_FREE`` unset + * - ``tasks`` + - ``ps`` + - a task or thread present after the test but not before + - ``CONFIG_NSH_DISABLE_PS`` unset + * - ``fds`` + - ``cat /proc/self/group/fd`` + - a descriptor opened in the shell task group and not closed + - ``CONFIG_NSH_DISABLE_CAT`` unset + * - ``iob`` + - ``cat /proc/iobinfo`` + - fewer free network buffers + - ``CONFIG_MM_IOB`` + * - ``mempool`` + - ``cat /proc/mempool`` + - ``nused`` grew, per pool + - ``CONFIG_MM_HEAP_MEMPOOL`` + * - ``net`` + - ``cat /proc/net/tcp``, ``cat /proc/net/udp`` + - a new connection + - ``CONFIG_NET`` + * - ``cpuload`` + - ``cat /proc/cpuload`` + - load above ``cpuload_max`` after the test + - ``CONFIG_SCHED_CPULOAD`` + +A probe whose Kconfig requirements (including the matching +``CONFIG_FS_PROCFS_EXCLUDE_*`` option) are not met on every product is +dropped with a warning. The final delta is written to the device log and to +the ``resources`` user property of the test report. Tests that crash or +busy-loop are not checked. + +``tasks`` keys entries by TID and command name; set ``CONFIG_TASK_NAME_SIZE`` +above zero so a leaked task is reported by name rather than ````. + +``heap`` is not a default: first-use allocations (stdio buffers, work +queues) and shell state (environment variables, ``OLDPWD``) show up as +growth, so it reports more than true leaks. Enable it per test with the +``resources`` marker and a ``heap_tolerance`` that fits, or repeat the run +with ``--loops`` and look for growth that recurs on every loop. + Signal Handlers =============== diff --git a/Documentation/writing-test-cases.rst b/Documentation/writing-test-cases.rst index 1f3a30e..2742c84 100644 --- a/Documentation/writing-test-cases.rst +++ b/Documentation/writing-test-cases.rst @@ -250,7 +250,26 @@ NTFC provides built-in fixtures and pytest marks for test development. Built-in Fixtures: -TODO +.. list-table:: + :header-rows: 1 + + * - Fixture + - Purpose + * - ``core`` + - Current core name, parametrized with ``--run_in_cores`` + * - ``switch_to_core`` + - Switch to the parametrized core in SMP mode and back afterwards + * - ``resources`` + - Fail the test if resources are not released (see + :ref:`resource-monitoring`). List it after ``switch_to_core`` so + sampling happens on the switched core. + +.. code-block:: python + + @pytest.mark.resources(probes=["heap", "tasks"], heap_tolerance=64) + def test_no_leak(resources): + ret = pytest.product.sendCommand("app", ["done"], timeout=15) + assert ret == 0 Custom Fixtures: @@ -300,6 +319,9 @@ Available Marks: * - ``@pytest.mark.skip()`` - Skip test conditionally - ``@pytest.mark.skip(reason="not ready")`` + * - ``@pytest.mark.resources()`` + - Override resource monitor options for one test + - ``@pytest.mark.resources(probes=["heap"])`` * - ``@pytest.mark.parser_binary()`` - Discover and run test framework tests (cmocka / custom) as pytest items - ``@pytest.mark.parser_binary("bin", list_pattern=..., result_pattern=...)`` diff --git a/src/ntfc/envconfig.py b/src/ntfc/envconfig.py index e742da8..1eb6412 100644 --- a/src/ntfc/envconfig.py +++ b/src/ntfc/envconfig.py @@ -98,6 +98,28 @@ def heartbeat(self) -> Dict[str, Any]: # Merge with defaults return {**default_config, **heartbeat_cfg} + @property + def resources(self) -> Dict[str, Any]: + """Return resource monitoring configuration. + + :return: Dictionary with keys: 'enabled' (bool), 'probes' (list), + 'settle_retries' (int), 'settle_delay' (float), + 'heap_tolerance' (int), 'cpuload_max' (float). Defaults to + disabled if not configured. + """ + default_config = { + "enabled": False, + "probes": ["tasks", "fds"], + "settle_retries": 3, + "settle_delay": 1.0, + "heap_tolerance": 0, + "cpuload_max": 10, + } + config = self._cfg_values.get("config", {}) + resources_cfg = config.get("resources", {}) + + return {**default_config, **resources_cfg} + @property def recovery(self) -> Dict[str, Any]: """Return device recovery configuration. diff --git a/src/ntfc/log/report.py b/src/ntfc/log/report.py index 1739c51..81d58c3 100644 --- a/src/ntfc/log/report.py +++ b/src/ntfc/log/report.py @@ -20,6 +20,7 @@ """Report generation module for NTFC.""" +import copy import glob import html import os @@ -118,22 +119,10 @@ def _split_xml_by_module( # noqa: C901 new_testsuite.set("time", str(time_total)) new_testsuite.set("name", module_name) - # Add all test cases + # Add all test cases, including nested elements such as + # for testcase in testcases: - # Deep copy the testcase element - new_testcase = ElementTree.SubElement( - new_testsuite, "testcase" - ) - for attr, value in testcase.attrib.items(): - new_testcase.set(attr, value) - - # Copy all child elements - for child in testcase: - new_child = ElementTree.SubElement(new_testcase, child.tag) - for attr, value in child.attrib.items(): - new_child.set(attr, value) - if child.text: - new_child.text = child.text + new_testsuite.append(copy.deepcopy(testcase)) # Write per-module XML file output_file = os.path.join( diff --git a/src/ntfc/pytest/configure.py b/src/ntfc/pytest/configure.py index 2520fee..8528cde 100644 --- a/src/ntfc/pytest/configure.py +++ b/src/ntfc/pytest/configure.py @@ -143,6 +143,8 @@ def pytest_configure(self, config: pytest.Config) -> None: "extra_opts: Additional parameters for testing " "(e.g., --run_in_cores=cpu1,cpu2)", "config_value_check: Validate complex configuration strings", + "resources: Override resource monitor options for one test " + "(e.g., probes=['heap'], heap_tolerance=64)", ] for m in markers: diff --git a/src/ntfc/pytest/mypytest.py b/src/ntfc/pytest/mypytest.py index 5a1c15d..18847d2 100644 --- a/src/ntfc/pytest/mypytest.py +++ b/src/ntfc/pytest/mypytest.py @@ -42,6 +42,7 @@ from .configure import PytestConfigPlugin from .gtest_plugin import GtestParserPlugin from .parsers import ParserPlugin +from .resources_plugin import ResourcePlugin from .runner import RunnerPlugin from .signal_plugin import SignalPlugin @@ -390,6 +391,9 @@ def runner( # run pytest with our custom test plugin runner = RunnerPlugin(nologs) + # resource leak detection fixture + resources = ResourcePlugin(self._config, pytest.products) + # optional debug plugins (coredump collection) debug = get_debug_setup(pytest.products) self._ptconfig.set_restart_gdb(debug.restart_all_controllers) @@ -400,7 +404,9 @@ def runner( try: # attach GDB controllers after the devices are up debug.start(pytest.result_dir) - return self._run(opt, [runner, collector, *debug.plugins]) + return self._run( + opt, [runner, collector, resources, *debug.plugins] + ) finally: debug.stop() self._device_stop() diff --git a/src/ntfc/pytest/resources_plugin.py b/src/ntfc/pytest/resources_plugin.py new file mode 100644 index 0000000..55f9e23 --- /dev/null +++ b/src/ntfc/pytest/resources_plugin.py @@ -0,0 +1,201 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Pytest plugin providing the ``resources`` leak detection fixture.""" + +from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional + +import pytest + +from ntfc.log.logger import logger +from ntfc.resmon.base import clean_output +from ntfc.resmon.monitor import ResourceMonitor, Sampler +from ntfc.resmon.probes import PROBES, make_probes + +if TYPE_CHECKING: + from ntfc.core import ProductCore + from ntfc.envconfig import EnvConfig + from ntfc.product import Product + +############################################################################### +# Class: ResourcePlugin +############################################################################### + + +class ResourcePlugin: + """Sample target resources around tests that request ``resources``. + + The fixture records a baseline before the test body, re-samples after + it and raises :class:`~ntfc.resmon.monitor.ResourceLeakError` when a + probe reports a delta that does not settle. Probes whose Kconfig + gates fail on any product are dropped with a warning. + + :param config: Environment configuration. + :param products: Products under test. + """ + + COMMAND_TIMEOUT = 10 + + def __init__(self, config: "EnvConfig", products: "List[Product]"): + """Initialize the plugin.""" + self._cfg = config.resources + self._products = products + self._cores: "Dict[str, ProductCore]" = {} + + def _options(self, marker: Optional[Any]) -> Dict[str, Any]: + """Merge the ``resources`` marker kwargs over the configuration.""" + return {**self._cfg, **(marker.kwargs if marker else {})} + + @staticmethod + def _gate_failures(product: "Product", name: str) -> List[str]: + """Return the Kconfig gates of a probe that fail on a product.""" + conf = product.conf + core = conf.primary_test_core_index + return [ + symbol + for symbol, expected in PROBES[name].kconfig + if bool(conf.kv_check(symbol, core)) != expected + ] + + def probe_names(self, marker: Optional[Any]) -> List[str]: + """Return the probes available on every product. + + :param marker: ``resources`` marker of the test, if any. + :raises ValueError: Unknown probe name. + """ + names = [] + for name in self._options(marker)["probes"]: + if name not in PROBES: + raise ValueError(f"unknown resource probe: {name}") + failures = { + product.name: self._gate_failures(product, name) + for product in self._products + } + failed = {p: f for p, f in failures.items() if f} + if failed: + logger.warning( + f"resources: probe '{name}' unavailable: {failed}" + ) + continue + names.append(name) + return names + + def _sampler(self, core: "ProductCore") -> Sampler: + """Return a callable that runs one command on a core.""" + + def send(cmd: str) -> str: + ret = core.sendCommandReadUntilPattern( + cmd, timeout=self.COMMAND_TIMEOUT + ) + return clean_output(ret.output, cmd, core.prompt.decode()) + + return send + + def targets(self) -> Dict[str, Sampler]: + """Return the cores to sample, keyed by ``product/core``.""" + targets = {} + for product in self._products: + conf = product.conf + indices = ( + [conf.primary_test_core_index] + if conf.is_smp + else conf.active_core_indices + ) + for idx in indices: + core = product.core(idx) + name = f"{product.name}/{core.name}" + self._cores[name] = core + targets[name] = self._sampler(core) + return targets + + def monitor(self, marker: Optional[Any]) -> ResourceMonitor: + """Build the monitor for one test. + + :param marker: ``resources`` marker of the test, if any. + """ + options = self._options(marker) + if not options["enabled"]: + return ResourceMonitor([], {}) + + missing = [ + product.name + for product in self._products + if not product.conf.kv_check( + "CONFIG_FS_PROCFS", product.conf.primary_test_core_index + ) + ] + if missing: + logger.warning( + f"resources: CONFIG_FS_PROCFS not set on {missing}, " + "resource monitoring disabled" + ) + return ResourceMonitor([], {}) + + return ResourceMonitor( + make_probes(self.probe_names(marker), options), + self.targets(), + settle_retries=int(options["settle_retries"]), + settle_delay=float(options["settle_delay"]), + ) + + def _unhealthy(self) -> bool: + """Return True when any product is in a fault state.""" + return any( + p.crash or p.busyloop or p.flood or p.notalive + for p in self._products + ) + + def _record(self, request: Any, monitor: ResourceMonitor) -> None: + """Attach the final delta to the report and the device logs.""" + diff = monitor.last_diff + request.node.user_properties.append(("resources", diff)) + for target, probes in diff.items(): + for probe, lines in probes.items(): + for line in lines: + self._cores[target].device.log_event( + f"resources {probe}: {line}" + ) + + @pytest.fixture # type: ignore + def resources(self, request: Any) -> Iterator[ResourceMonitor]: + """Check that the test releases the resources it used. + + Request this fixture to sample the configured probes before the + test body and again after it. A remaining delta fails the test at + teardown. Use ``@pytest.mark.resources(probes=[...], ...)`` to + override the configuration for one test. List it after + ``switch_to_core`` so sampling happens on the switched core. + + :yield: :class:`~ntfc.resmon.monitor.ResourceMonitor` + """ + monitor = self.monitor(request.node.get_closest_marker("resources")) + monitor.baseline() + + yield monitor + + if not monitor.active: + return + if self._unhealthy(): + logger.warning("resources: product unhealthy, skipping check") + return + try: + monitor.check() + finally: + self._record(request, monitor) diff --git a/src/ntfc/resmon/__init__.py b/src/ntfc/resmon/__init__.py new file mode 100644 index 0000000..b3cb56b --- /dev/null +++ b/src/ntfc/resmon/__init__.py @@ -0,0 +1,35 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Resource monitoring: probes that sample target resources over NSH.""" + +from .base import Probe, clean_output, parse_table +from .monitor import ResourceLeakError, ResourceMonitor +from .probes import PROBES, make_probes + +__all__ = [ + "PROBES", + "Probe", + "ResourceLeakError", + "ResourceMonitor", + "clean_output", + "make_probes", + "parse_table", +] diff --git a/src/ntfc/resmon/base.py b/src/ntfc/resmon/base.py new file mode 100644 index 0000000..8cea557 --- /dev/null +++ b/src/ntfc/resmon/base.py @@ -0,0 +1,244 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Resource probe base classes and output parsing helpers.""" + +import re +from abc import ABC, abstractmethod +from typing import Any, Dict, List, Mapping, Optional, Sequence, Set, Tuple + +_LINE_END_RE = re.compile(r"\r*\n") + +############################################################################### +# Function: clean_output +############################################################################### + + +def clean_output(raw: str, cmd: str, prompt: str) -> str: + """Strip the echoed command and the trailing prompt from NSH output. + + :param raw: Output as returned by the device. + :param cmd: Command that was sent. + :param prompt: Target shell prompt. + :return: Output lines with normalized line endings. + """ + text = _LINE_END_RE.sub("\n", raw).replace("\r", "") + lines = text.split("\n") + # leading stale prompts and the echo, with or without its prompt + echo = {cmd.strip(), prompt.strip() + " " + cmd.strip(), prompt.strip()} + while lines and lines[0].strip() in echo: + lines = lines[1:] + if lines and lines[-1].strip() == prompt.strip(): + lines[-1] = "" + return "\n".join(lines) + + +############################################################################### +# Function: parse_table +############################################################################### + + +def find_header(lines: List[str], required: Sequence[str]) -> Optional[int]: + """Return the index of the first line naming every required column.""" + for idx, line in enumerate(lines): + tokens = line.split() + if all(col in tokens for col in required): + return idx + return None + + +def parse_table( + text: str, + columns: Optional[List[str]] = None, + skip: int = 0, + required: Optional[Sequence[str]] = None, +) -> List[Dict[str, str]]: + """Parse a whitespace separated table into rows keyed by column name. + + The last column absorbs the rest of the line, so values with spaces + (for example a task command line) are kept intact. Short rows are + padded with empty strings. + + :param text: Table text. + :param columns: Column names. When omitted the header line provides + them. + :param skip: Number of leading non-empty lines to drop. + :param required: Column names that identify the header line. Lines + before it (for example shell diagnostics) are ignored; without a + matching line the table is empty. + :return: List of rows. + """ + lines = [line for line in text.split("\n") if line.strip()][skip:] + if required is not None: + header = find_header(lines, required) + if header is None: + return [] + lines = lines[header:] + if columns is None: + if not lines: + return [] + columns = lines[0].split() + lines = lines[1:] + elif required is not None: + lines = lines[1:] + + rows = [] + for line in lines: + parts = [p.strip() for p in line.split(None, len(columns) - 1)] + parts += [""] * (len(columns) - len(parts)) + rows.append(dict(zip(columns, parts))) + return rows + + +############################################################################### +# Class: Probe +############################################################################### + + +class Probe(ABC): + """One resource sampled before and after a test.""" + + name = "" + commands: Tuple[str, ...] = () + kconfig: Tuple[Tuple[str, bool], ...] = () + + def __init__(self, options: Mapping[str, Any]) -> None: + """Initialize probe. + + :param options: Resource monitor configuration. + """ + self._options = options + + @abstractmethod + def parse(self, outputs: List[str]) -> Any: + """Parse command outputs, one per entry in :attr:`commands`.""" + + @abstractmethod + def diff(self, before: Any, after: Any) -> List[str]: + """Compare two parsed samples. + + :return: Human readable leak descriptions, empty when clean. + """ + + +############################################################################### +# Class: CounterProbe +############################################################################### + + +class CounterProbe(Probe): + """Numeric table probe: rows keyed by a column, counters compared.""" + + key_column: Optional[str] = None + # column -> option holding the allowed growth, None for zero tolerance + grow_columns: Dict[str, Optional[str]] = {} + shrink_columns: Tuple[str, ...] = () + + def _required(self) -> Tuple[str, ...]: + """Return the column names that identify the table header.""" + columns = (self.key_column, *self.grow_columns, *self.shrink_columns) + return tuple(col for col in columns if col) + + def _rows(self, text: str) -> List[Dict[str, str]]: + """Return table rows from the command output.""" + return parse_table(text, required=self._required()) + + def _valid(self, text: str) -> bool: + """Return True when the output contains the expected header.""" + lines = [line for line in text.split("\n") if line.strip()] + return find_header(lines, self._required()) is not None + + def parse(self, outputs: List[str]) -> Dict[str, Dict[str, int]]: + """Parse the table into ``{row key: {column: value}}``.""" + if not self._valid(outputs[0]): + raise ValueError(f"{self.name}: unexpected output: {outputs[0]!r}") + parsed = {} + for row in self._rows(outputs[0]): + key = row[self.key_column] if self.key_column else self.name + parsed[key] = { + col: int(val) for col, val in row.items() if val.isdigit() + } + return parsed + + def _tolerance(self, column: str) -> int: + """Return the allowed growth for a column.""" + option = self.grow_columns[column] + return int(self._options.get(option, 0)) if option else 0 + + def diff( + self, + before: Dict[str, Dict[str, int]], + after: Dict[str, Dict[str, int]], + ) -> List[str]: + """Report rows that appeared and counters that moved.""" + result = [] + for key, row in after.items(): + if key not in before: + result.append(f"{key} new") + continue + for col in self.grow_columns: + delta = row[col] - before[key][col] + if delta > self._tolerance(col): + result.append( + f"{key} {col} {before[key][col]} -> {row[col]} " + f"({delta:+d})" + ) + for col in self.shrink_columns: + delta = row[col] - before[key][col] + if delta < 0: + result.append( + f"{key} {col} {before[key][col]} -> {row[col]} " + f"({delta:+d})" + ) + return result + + +############################################################################### +# Class: SetProbe +############################################################################### + + +class SetProbe(Probe): + """Set probe: entries present after but not before are leaks.""" + + key_columns: Tuple[str, ...] = () + + def _entries(self, text: str) -> Set[Tuple[str, ...]]: + """Return the entry keys from the command output.""" + return { + tuple(row[col] for col in self.key_columns) + for row in parse_table(text, required=self.key_columns) + } + + def parse(self, outputs: List[str]) -> Set[Tuple[str, ...]]: + """Parse all outputs into one set of entries.""" + entries: Set[Tuple[str, ...]] = set() + for output in outputs: + entries |= self._entries(output) + return entries + + def diff( + self, before: Set[Tuple[str, ...]], after: Set[Tuple[str, ...]] + ) -> List[str]: + """Report entries added since the baseline.""" + return [ + f"new {self.name}: {' '.join(entry)}" + for entry in sorted(after - before) + ] diff --git a/src/ntfc/resmon/monitor.py b/src/ntfc/resmon/monitor.py new file mode 100644 index 0000000..26eccaf --- /dev/null +++ b/src/ntfc/resmon/monitor.py @@ -0,0 +1,142 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Resource monitor: baseline, settle and compare samples.""" + +import time +from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional + +from ntfc.log.logger import logger + +if TYPE_CHECKING: + from .base import Probe + +Sampler = Callable[[str], str] +Sample = Dict[str, Dict[str, Any]] +Diff = Dict[str, Dict[str, List[str]]] + +############################################################################### +# Class: ResourceLeakError +############################################################################### + + +class ResourceLeakError(AssertionError): + """Raised when resources were not released after a test.""" + + +############################################################################### +# Class: ResourceMonitor +############################################################################### + + +class ResourceMonitor: + """Sample probes on a set of targets and compare against a baseline. + + :param probes: Probes to run. + :param targets: Mapping of target name to a callable that sends one + NSH command and returns its cleaned output. + :param settle_retries: Re-samples allowed while a delta remains. + :param settle_delay: Seconds between re-samples. + :param sleep: Sleep function, injectable for tests. + """ + + def __init__( + self, + probes: "List[Probe]", + targets: Dict[str, Sampler], + settle_retries: int = 3, + settle_delay: float = 1.0, + sleep: Callable[[float], Any] = time.sleep, + ) -> None: + """Initialize the monitor.""" + self._probes = probes + self._targets = targets + self._retries = settle_retries + self._delay = settle_delay + self._sleep = sleep + self._baseline: Optional[Sample] = None + self.last_diff: Diff = {} + + @property + def active(self) -> bool: + """Return True when there is something to sample.""" + return bool(self._probes) and bool(self._targets) + + def snapshot(self) -> Sample: + """Sample every probe on every target.""" + sample: Sample = {} + for target, send in self._targets.items(): + sample[target] = { + probe.name: probe.parse([send(cmd) for cmd in probe.commands]) + for probe in self._probes + } + return sample + + def baseline(self) -> None: + """Record the baseline sample.""" + self._baseline = self.snapshot() if self.active else {} + + def _diff(self, after: Sample) -> Diff: + """Compare a sample against the baseline, keeping only leaks.""" + assert self._baseline is not None + result: Diff = {} + for target, probes in after.items(): + for probe in self._probes: + lines = probe.diff( + self._baseline[target][probe.name], probes[probe.name] + ) + if lines: + result.setdefault(target, {})[probe.name] = lines + return result + + def check(self) -> Diff: + """Re-sample until clean or retries run out. + + :return: Empty dict when clean. + :raises RuntimeError: No baseline recorded. + :raises ResourceLeakError: A delta remained after settling. + """ + if self._baseline is None: + raise RuntimeError("resource monitor has no baseline") + if not self.active: + return {} + + diff = self._diff(self.snapshot()) + for _ in range(self._retries): + if not diff: + break + logger.debug(f"resources: delta remains, settling: {diff}") + self._sleep(self._delay) + diff = self._diff(self.snapshot()) + + self.last_diff = diff + if diff: + raise ResourceLeakError(self.format_diff(diff)) + return diff + + @staticmethod + def format_diff(diff: Diff) -> str: + """Render a diff as one line per leak.""" + lines = [] + for target, probes in diff.items(): + for probe, items in probes.items(): + for item in items: + lines.append(f"{target} {probe}: {item}") + return "resource leak detected:\n" + "\n".join(lines) diff --git a/src/ntfc/resmon/probes.py b/src/ntfc/resmon/probes.py new file mode 100644 index 0000000..c877109 --- /dev/null +++ b/src/ntfc/resmon/probes.py @@ -0,0 +1,247 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Built-in resource probes.""" + +import re +from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, Type + +from .base import CounterProbe, Probe, SetProbe, find_header, parse_table + +_PROCFS = ("CONFIG_FS_PROCFS", True) +_PROCESS = ("CONFIG_FS_PROCFS_EXCLUDE_PROCESS", False) +_CAT = ("CONFIG_NSH_DISABLE_CAT", False) + +############################################################################### +# Class: HeapProbe +############################################################################### + + +class HeapProbe(CounterProbe): + """Heap usage from ``free``.""" + + name = "heap" + commands = ("free",) + kconfig = ( + _PROCFS, + ("CONFIG_FS_PROCFS_EXCLUDE_MEMINFO", False), + ("CONFIG_NSH_DISABLE_FREE", False), + ) + key_column = "name" + grow_columns = {"used": "heap_tolerance", "nused": None} + + +############################################################################### +# Class: TasksProbe +############################################################################### + + +class TasksProbe(SetProbe): + """Tasks and threads from ``ps``.""" + + name = "tasks" + commands = ("ps",) + kconfig = (_PROCFS, _PROCESS, ("CONFIG_NSH_DISABLE_PS", False)) + + def _entries(self, text: str) -> Set[Tuple[str, ...]]: + """Key tasks by TID and command line. + + Columns such as EVENT can be blank, so the command is taken from + the header offset of ``COMMAND`` rather than by splitting. + """ + lines = [line for line in text.split("\n") if line.strip()] + header = find_header(lines, ("TID", "COMMAND")) + if header is None: + return set() + start = lines[header].index("COMMAND") + return { + (line.split()[0], line[start:].strip()) + for line in lines[header + 1 :] + } + + +############################################################################### +# Class: FdsProbe +############################################################################### + + +class FdsProbe(SetProbe): + """File descriptors open in the shell task group.""" + + name = "fds" + commands = ("cat /proc/self/group/fd",) + kconfig = (_PROCFS, _PROCESS, _CAT) + key_columns = ("FD", "TYPE", "PATH") + + +############################################################################### +# Class: IobProbe +############################################################################### + + +class IobProbe(CounterProbe): + """Network I/O buffer pool from ``/proc/iobinfo``.""" + + name = "iob" + commands = ("cat /proc/iobinfo",) + kconfig = ( + _PROCFS, + _CAT, + ("CONFIG_MM_IOB", True), + ("CONFIG_FS_PROCFS_EXCLUDE_IOBINFO", False), + ) + shrink_columns = ("nfree",) + + +############################################################################### +# Class: MempoolProbe +############################################################################### + + +class MempoolProbe(CounterProbe): + """Memory pools from ``/proc/mempool``.""" + + name = "mempool" + commands = ("cat /proc/mempool",) + kconfig = ( + _PROCFS, + _CAT, + ("CONFIG_MM_HEAP_MEMPOOL", True), + ("CONFIG_FS_PROCFS_EXCLUDE_MEMPOOL", False), + ) + key_column = "name" + grow_columns = {"nused": None} + _COLUMNS = [ + "name", + "total", + "bsize", + "nused", + "nfree", + "nifree", + "nwaiter", + ] + + def _required(self) -> Tuple[str, ...]: + """Return the counter columns; the first header column is unnamed.""" + return ("nused", "nfree") + + def _rows(self, text: str) -> List[Dict[str, str]]: + """Parse rows; the header has an unnamed first column.""" + rows = parse_table( + text, columns=self._COLUMNS, required=self._required() + ) + for row in rows: + row["name"] = row["name"].rstrip(":") + return rows + + +############################################################################### +# Class: NetProbe +############################################################################### + + +class NetProbe(SetProbe): + """Socket connections from ``/proc/net/tcp`` and ``/proc/net/udp``.""" + + name = "net" + commands = ("cat /proc/net/tcp", "cat /proc/net/udp") + kconfig = ( + _PROCFS, + _CAT, + ("CONFIG_NET", True), + ("CONFIG_FS_PROCFS_EXCLUDE_NET", False), + ) + + def _entries(self, text: str) -> Set[Tuple[str, ...]]: + """Key connections by local and remote endpoint.""" + entries: Set[Tuple[str, ...]] = set() + for line in text.split("\n"): + tokens = line.split() + if len(tokens) >= 3 and tokens[0].endswith(":"): + entries.add((tokens[-2], tokens[-1])) + return entries + + +############################################################################### +# Class: CpuloadProbe +############################################################################### + + +class CpuloadProbe(Probe): + """CPU load from ``/proc/cpuload`` checked against a limit.""" + + name = "cpuload" + commands = ("cat /proc/cpuload",) + kconfig = ( + _PROCFS, + _CAT, + ("CONFIG_SCHED_CPULOAD", True), + ("CONFIG_FS_PROCFS_EXCLUDE_CPULOAD", False), + ) + _LOAD_RE = re.compile(r"([\d.]+)%") + DEFAULT_MAX = 10.0 + + def parse(self, outputs: List[str]) -> Optional[float]: + """Return the load percentage, or ``None`` when unreadable.""" + match = self._LOAD_RE.search(outputs[0]) + return float(match.group(1)) if match else None + + def diff( + self, before: Optional[float], after: Optional[float] + ) -> List[str]: + """Report a load above the configured limit.""" + del before + limit = float(self._options.get("cpuload_max", self.DEFAULT_MAX)) + if after is None or after <= limit: + return [] + return [f"cpuload {after:g}% > {limit:g}%"] + + +############################################################################### +# Registry +############################################################################### + +PROBES: Dict[str, Type[Probe]] = { + probe.name: probe + for probe in ( + HeapProbe, + TasksProbe, + FdsProbe, + IobProbe, + MempoolProbe, + NetProbe, + CpuloadProbe, + ) +} + + +def make_probes(names: List[str], options: Mapping[str, Any]) -> List[Probe]: + """Instantiate probes by name. + + :param names: Probe names in the order they should run. + :param options: Resource monitor configuration. + :raises ValueError: Unknown probe name. + """ + probes = [] + for name in names: + if name not in PROBES: + raise ValueError(f"unknown resource probe: {name}") + probes.append(PROBES[name](options)) + return probes diff --git a/tests/log/test_report.py b/tests/log/test_report.py index c34712c..357140b 100644 --- a/tests/log/test_report.py +++ b/tests/log/test_report.py @@ -367,6 +367,30 @@ def test_split_xml_by_module(): assert os.path.exists(module_file) +def test_split_xml_by_module_keeps_nested_properties(): + """Test that testcase properties survive the per-module split.""" + with tempfile.TemporaryDirectory() as tmpdir: + xml_file = os.path.join(tmpdir, "report.xml") + testsuites = Element("testsuites") + testsuite = SubElement(testsuites, "testsuite") + tc = SubElement(testsuite, "testcase") + tc.set("name", "test_one") + tc.set("classname", "test_module1.py::test_one") + tc.set("time", "1.0") + props = SubElement(tc, "properties") + prop = SubElement(props, "property") + prop.set("name", "resources") + prop.set("value", "{}") + ElementTree(testsuites).write(xml_file) + + module_files = Reporter()._split_xml_by_module(xml_file, tmpdir) + + root = ElementTree(file=module_files["test_module1"]).getroot() + found = root.findall(".//testcase/properties/property") + assert len(found) == 1 + assert found[0].get("name") == "resources" + + def test_split_xml_by_module_nonexistent_file(): """Test _split_xml_by_module with nonexistent file.""" with tempfile.TemporaryDirectory() as tmpdir: diff --git a/tests/pytest/test_mypytest.py b/tests/pytest/test_mypytest.py index 6686b4f..2ec336f 100644 --- a/tests/pytest/test_mypytest.py +++ b/tests/pytest/test_mypytest.py @@ -25,6 +25,7 @@ from ntfc.envconfig import EnvConfig from ntfc.pytest.mypytest import MyPytest +from ntfc.pytest.resources_plugin import ResourcePlugin def test_collector_collect_file(config_sim, device_dummy): @@ -333,6 +334,7 @@ def _run(_opt, plugins): monkeypatch.setattr(_pytest, "products", [], raising=False) assert p.runner("tests", {}, nologs=True) == 0 + assert any(isinstance(pl, ResourcePlugin) for pl in run_plugins) assert debug_plugin in run_plugins debug.start.assert_called_once_with("") diff --git a/tests/pytest/test_resources_plugin.py b/tests/pytest/test_resources_plugin.py new file mode 100644 index 0000000..223fea6 --- /dev/null +++ b/tests/pytest/test_resources_plugin.py @@ -0,0 +1,294 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Tests for the resources pytest plugin.""" + +import logging +from typing import Any, Dict, List +from unittest.mock import MagicMock + +import pytest + +from ntfc.device.cmds import CmdReturn, CmdStatus +from ntfc.envconfig import EnvConfig +from ntfc.pytest.resources_plugin import ResourcePlugin +from ntfc.resmon.monitor import ResourceLeakError + +FREE_HDR = ( + " total used free maxused maxfree nused nfree" + " name\n" +) +FREE_A = FREE_HDR + " 1000 100 900 100 900 6 3 Umem\n" +FREE_B = FREE_HDR + " 1000 116 884 116 884 7 3 Umem\n" +PS_A = " TID PID COMMAND\n 4 4 nsh_main\n" +FD_A = "FD OFLAGS TYPE POS PATH\n0 2 1 0 /dev/console\n" + +KV_ALL = { + "CONFIG_FS_PROCFS": True, + "CONFIG_MM_IOB": True, + "CONFIG_NET": True, +} + + +def _core(name: str, script: Dict[str, List[str]]) -> MagicMock: + core = MagicMock() + core.name = name + core.prompt = b"nsh> " + core.calls: List[str] = [] + + def send(cmd: str, timeout: int = 30) -> CmdReturn: + core.calls.append(cmd) + outputs = script[cmd] + out = outputs.pop(0) if len(outputs) > 1 else outputs[0] + return CmdReturn(CmdStatus.SUCCESS, None, f"{cmd}\r\n{out}nsh> ") + + core.sendCommandReadUntilPattern.side_effect = send + return core + + +def _product( + name: str = "prod", + kv: Dict[str, Any] = KV_ALL, + script: Dict[str, List[str]] | None = None, + cores: int = 1, + smp: bool = False, +) -> MagicMock: + script = script or {"free": [FREE_A], "ps": [PS_A]} + product = MagicMock() + product.name = name + product.crash = False + product.busyloop = False + product.flood = False + product.notalive = False + product.conf.kv_check.side_effect = lambda sym, core=0: kv.get(sym, False) + product.conf.primary_test_core_index = 0 + product.conf.active_core_indices = list(range(cores)) + product.conf.is_smp = smp + product_cores = [ + _core("main" if i == 0 else f"cpu{i}", dict(script)) + for i in range(cores) + ] + product.core.side_effect = lambda i=0: product_cores[i] + product.test_cores = product_cores + return product + + +def _config(resources: Dict[str, Any] | None = None) -> EnvConfig: + cfg: Dict[str, Any] = { + "config": {}, + "product": {"name": "prod", "cores": {}}, + } + if resources is not None: + cfg["config"]["resources"] = resources + return EnvConfig(cfg) + + +def _request(marker_kwargs: Dict[str, Any] | None = None) -> MagicMock: + request = MagicMock() + request.node.user_properties = [] + marker = None + if marker_kwargs is not None: + marker = MagicMock() + marker.kwargs = marker_kwargs + request.node.get_closest_marker.return_value = marker + return request + + +def _fixture_gen(plugin: ResourcePlugin, request: MagicMock) -> Any: + """Return the unwrapped resources fixture generator.""" + fixture = ResourcePlugin.resources.__wrapped__ # type: ignore + return fixture(plugin, request) + + +def _run_fixture(plugin: ResourcePlugin, request: MagicMock) -> Any: + """Drive the resources fixture through setup and teardown.""" + gen = _fixture_gen(plugin, request) + monitor = next(gen) + try: + next(gen) + except StopIteration: + pass + return monitor + + +class TestEnvConfigResources: + def test_defaults(self) -> None: + assert _config().resources == { + "enabled": False, + "probes": ["tasks", "fds"], + "settle_retries": 3, + "settle_delay": 1.0, + "heap_tolerance": 0, + "cpuload_max": 10, + } + + def test_merge(self) -> None: + cfg = _config({"enabled": True, "probes": ["heap"]}) + assert cfg.resources["enabled"] is True + assert cfg.resources["probes"] == ["heap"] + assert cfg.resources["settle_retries"] == 3 + + +class TestProbeSelection: + def test_disabled_is_inactive(self) -> None: + product = _product() + plugin = ResourcePlugin(_config({"enabled": False}), [product]) + assert plugin.monitor(None).active is False + assert product.test_cores[0].calls == [] + + def test_missing_procfs_is_inactive( + self, caplog: pytest.LogCaptureFixture + ) -> None: + product = _product(kv={}) + plugin = ResourcePlugin(_config({"enabled": True}), [product]) + with caplog.at_level(logging.WARNING): + assert plugin.monitor(None).active is False + assert "CONFIG_FS_PROCFS" in caplog.text + + def test_default_probes(self) -> None: + plugin = ResourcePlugin(_config({"enabled": True}), [_product()]) + assert plugin.probe_names(None) == ["tasks", "fds"] + + def test_gated_probe_dropped( + self, caplog: pytest.LogCaptureFixture + ) -> None: + kv = {**KV_ALL, "CONFIG_FS_PROCFS_EXCLUDE_PROCESS": True} + plugin = ResourcePlugin(_config({"enabled": True}), [_product(kv=kv)]) + with caplog.at_level(logging.WARNING): + assert plugin.probe_names(None) == [] + assert "'tasks'" in caplog.text + assert "'fds'" in caplog.text + assert "CONFIG_FS_PROCFS_EXCLUDE_PROCESS" in caplog.text + + def test_probe_must_be_available_on_every_product(self) -> None: + kv = {**KV_ALL, "CONFIG_NSH_DISABLE_PS": True} + plugin = ResourcePlugin( + _config({"enabled": True}), [_product("a"), _product("b", kv=kv)] + ) + assert plugin.probe_names(None) == ["fds"] + + def test_marker_overrides_probes(self) -> None: + plugin = ResourcePlugin(_config({"enabled": True}), [_product()]) + marker = MagicMock() + marker.kwargs = {"probes": ["heap"]} + assert plugin.probe_names(marker) == ["heap"] + + def test_unknown_probe_in_config(self) -> None: + plugin = ResourcePlugin( + _config({"enabled": True, "probes": ["nope"]}), [_product()] + ) + with pytest.raises(ValueError, match="nope"): + plugin.probe_names(None) + + +class TestTargets: + def test_amp_samples_every_active_core(self) -> None: + product = _product(cores=2) + plugin = ResourcePlugin(_config({"enabled": True}), [product]) + targets = plugin.targets() + assert sorted(targets) == ["prod/cpu1", "prod/main"] + + def test_smp_samples_primary_core_only(self) -> None: + product = _product(cores=2, smp=True) + plugin = ResourcePlugin(_config({"enabled": True}), [product]) + assert list(plugin.targets()) == ["prod/main"] + + def test_sampler_cleans_output(self) -> None: + product = _product() + plugin = ResourcePlugin(_config({"enabled": True}), [product]) + send = plugin.targets()["prod/main"] + assert send("free") == FREE_A + assert product.test_cores[0].calls == ["free"] + + +class TestFixture: + def test_clean_test_passes_and_records(self) -> None: + product = _product( + script={ + "free": [FREE_A], + "ps": [PS_A], + "cat /proc/self/group/fd": [FD_A], + } + ) + plugin = ResourcePlugin(_config({"enabled": True}), [product]) + request = _request() + monitor = _run_fixture(plugin, request) + assert monitor.active is True + assert request.node.user_properties == [("resources", {})] + assert product.test_cores[0].calls.count("ps") == 2 + + def test_leak_raises_and_logs(self) -> None: + product = _product(script={"free": [FREE_A, FREE_B]}) + plugin = ResourcePlugin( + _config( + {"enabled": True, "probes": ["heap"], "settle_retries": 0} + ), + [product], + ) + request = _request() + with pytest.raises(ResourceLeakError, match="Umem used 100 -> 116"): + _run_fixture(plugin, request) + key, diff = request.node.user_properties[0] + assert key == "resources" + assert diff["prod/main"]["heap"][0] == "Umem used 100 -> 116 (+16)" + product.test_cores[0].device.log_event.assert_called() + msg = product.test_cores[0].device.log_event.call_args[0][0] + assert "heap" in msg and "Umem" in msg + + def test_unhealthy_product_skips_check( + self, caplog: pytest.LogCaptureFixture + ) -> None: + product = _product(script={"free": [FREE_A, FREE_B]}) + plugin = ResourcePlugin( + _config({"enabled": True, "probes": ["heap"]}), [product] + ) + request = _request() + gen = _fixture_gen(plugin, request) + next(gen) + product.crash = True + with caplog.at_level(logging.WARNING): + with pytest.raises(StopIteration): + next(gen) + assert "skipping" in caplog.text + assert request.node.user_properties == [] + assert product.test_cores[0].calls == ["free"] + + def test_disabled_fixture_is_noop(self) -> None: + product = _product() + plugin = ResourcePlugin(_config(), [product]) + request = _request() + monitor = _run_fixture(plugin, request) + assert monitor.active is False + assert product.test_cores[0].calls == [] + assert request.node.user_properties == [] + + def test_marker_tolerance_applies(self) -> None: + product = _product(script={"free": [FREE_A, FREE_B]}) + plugin = ResourcePlugin( + _config( + {"enabled": True, "probes": ["heap"], "settle_retries": 0} + ), + [product], + ) + request = _request({"heap_tolerance": 100}) + with pytest.raises(ResourceLeakError) as exc: + _run_fixture(plugin, request) + assert "nused" in str(exc.value) + assert " used " not in str(exc.value) diff --git a/tests/resmon/test_monitor.py b/tests/resmon/test_monitor.py new file mode 100644 index 0000000..f03cd93 --- /dev/null +++ b/tests/resmon/test_monitor.py @@ -0,0 +1,154 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Tests for the resource monitor.""" + +from typing import Dict, List + +import pytest + +from ntfc.resmon.monitor import ResourceLeakError, ResourceMonitor +from ntfc.resmon.probes import make_probes + +FREE_HDR = ( + " total used free maxused maxfree nused nfree" + " name\n" +) +FREE_A = FREE_HDR + " 1000 100 900 100 900 6 3 Umem\n" +FREE_B = FREE_HDR + " 1000 116 884 116 884 7 3 Umem\n" + + +class FakeTarget: + """Scripted command responder: returns outputs in sequence per cmd.""" + + def __init__(self, script: Dict[str, List[str]]) -> None: + self.script = script + self.calls: List[str] = [] + + def __call__(self, cmd: str) -> str: + self.calls.append(cmd) + outputs = self.script[cmd] + return outputs.pop(0) if len(outputs) > 1 else outputs[0] + + +def _monitor( + script: Dict[str, List[str]], retries: int = 3, delay: float = 1.0 +) -> tuple[ResourceMonitor, FakeTarget, List[float]]: + target = FakeTarget(script) + sleeps: List[float] = [] + mon = ResourceMonitor( + make_probes(["heap"], {}), + {"p/main": target}, + settle_retries=retries, + settle_delay=delay, + sleep=sleeps.append, + ) + return mon, target, sleeps + + +class TestResourceMonitor: + def test_clean_check_returns_empty(self) -> None: + mon, target, sleeps = _monitor({"free": [FREE_A]}) + mon.baseline() + assert mon.check() == {} + assert target.calls == ["free", "free"] + assert sleeps == [] + + def test_leak_raises_with_details(self) -> None: + mon, _, _ = _monitor({"free": [FREE_A, FREE_B]}, retries=0) + mon.baseline() + with pytest.raises(ResourceLeakError) as exc: + mon.check() + msg = str(exc.value) + assert "p/main" in msg + assert "heap" in msg + assert "Umem used 100 -> 116 (+16)" in msg + + def test_last_diff_kept_after_leak(self) -> None: + mon, _, _ = _monitor({"free": [FREE_A, FREE_B]}, retries=0) + mon.baseline() + with pytest.raises(ResourceLeakError): + mon.check() + assert mon.last_diff == { + "p/main": { + "heap": [ + "Umem used 100 -> 116 (+16)", + "Umem nused 6 -> 7 (+1)", + ] + } + } + + def test_settle_clears_transient_delta(self) -> None: + mon, target, sleeps = _monitor( + {"free": [FREE_A, FREE_B, FREE_A]}, retries=3, delay=0.5 + ) + mon.baseline() + assert mon.check() == {} + assert target.calls == ["free"] * 3 + assert sleeps == [0.5] + + def test_retries_exhausted(self) -> None: + mon, target, sleeps = _monitor( + {"free": [FREE_A, FREE_B]}, retries=2, delay=0.1 + ) + mon.baseline() + with pytest.raises(ResourceLeakError): + mon.check() + # baseline + first sample + 2 retries + assert len(target.calls) == 4 + assert sleeps == [0.1, 0.1] + + def test_check_without_baseline(self) -> None: + mon, _, _ = _monitor({"free": [FREE_A]}) + with pytest.raises(RuntimeError, match="baseline"): + mon.check() + + def test_inactive_without_probes(self) -> None: + target = FakeTarget({"free": [FREE_A]}) + mon = ResourceMonitor([], {"p/main": target}) + assert mon.active is False + mon.baseline() + assert mon.check() == {} + assert target.calls == [] + + def test_inactive_without_targets(self) -> None: + mon = ResourceMonitor(make_probes(["heap"], {}), {}) + assert mon.active is False + + def test_active(self) -> None: + mon, _, _ = _monitor({"free": [FREE_A]}) + assert mon.active is True + + def test_snapshot_exposes_parsed_values(self) -> None: + mon, _, _ = _monitor({"free": [FREE_A]}) + snap = mon.snapshot() + assert snap["p/main"]["heap"]["Umem"]["used"] == 100 + + def test_multi_command_probe_feeds_all_outputs(self) -> None: + target = FakeTarget( + { + "cat /proc/net/tcp": ["TCP sl\n 0: x 1.1.1.1:1 2.2.2.2:2\n"], + "cat /proc/net/udp": ["UDP sl\n"], + } + ) + mon = ResourceMonitor(make_probes(["net"], {}), {"t": target}) + mon.baseline() + assert mon.check() == {} + assert target.calls[:2] == ["cat /proc/net/tcp", "cat /proc/net/udp"] diff --git a/tests/resmon/test_probes.py b/tests/resmon/test_probes.py new file mode 100644 index 0000000..37f0e52 --- /dev/null +++ b/tests/resmon/test_probes.py @@ -0,0 +1,407 @@ +############################################################################ +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. The +# ASF licenses this file to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance with the +# License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +############################################################################ + +"""Tests for resource probes.""" + +import pytest + +from ntfc.resmon.base import clean_output, parse_table +from ntfc.resmon.probes import ( + PROBES, + CpuloadProbe, + FdsProbe, + HeapProbe, + IobProbe, + MempoolProbe, + NetProbe, + TasksProbe, + make_probes, +) + +FREE_HDR = ( + " total used free maxused maxfree nused nfree" + " name\n" +) +FREE_UMEM = ( + " 67108864 1787800 65321064 1788496 65257512 66 3" + " Umem\n" +) +FREE_KMEM = ( + " 1048576 100000 948576 100000 948576 10 1" + " Kmem\n" +) +FREE_A = FREE_HDR + FREE_UMEM +FREE_B = FREE_HDR + ( + " 67108864 1787900 65320964 1788496 65257512 67 3" + " Umem\n" +) +FREE_TWO_HEAPS = FREE_HDR + FREE_KMEM + FREE_UMEM + +_SIG = "0000000000000000" +PS_A = ( + " TID PID PPID PRI POLICY TYPE NPX STATE EVENT " + "SIGMASK STACK COMMAND\n" + f" 0 0 0 0 FIFO Kthread - Ready {_SIG}" + " 0069584 Idle_Task\n" + f" 1 0 0 224 FIFO Kthread - Waiting Semaphore {_SIG}" + " 0067456 sim_loop_wq 0x7f36774003f0 0x7f3677400470\n" + f" 4 4 0 100 FIFO Task - Running {_SIG}" + " 0067496 nsh_main\n" +) +PS_B = PS_A + ( + f" 5 5 4 100 FIFO Task - Waiting Signal {_SIG}" + " 0067480 sh -c sleep\n" +) +PS_SMP = ( + " TID PID PPID CPU PRI POLICY TYPE NPX STATE EVENT " + "SIGMASK STACK COMMAND\n" + " 0 0 0 0 0 FIFO Kthread - Assigned " + f"{_SIG} 0002032 CPU0 IDLE\n" + " 4 4 0 1 100 FIFO Task - Running " + f"{_SIG} 0003008 nsh_main\n" +) + +FD_A = ( + "\n" + "FD OFLAGS TYPE POS PATH \n" + "0 2 1 0 /dev/console \n" + "1 2 1 0 /dev/console \n" + "3 524288 3 0 \n" +) +FD_B = FD_A + "4 0 1 0 /tmp/leak \n" +FD_A_MOVED = FD_A.replace("1 2 1 0 ", "1 2 1 12 ") + +IOB_HDR = " ntotal nfree nwait nthrottle\n" +IOB_A = IOB_HDR + " 36 36 0 0\n" +IOB_B = IOB_HDR + " 36 30 0 0\n" + +MEMPOOL_A = ( + " total bsize nused nfree nifree nwaiter\n" + " Umem: 4096 32 10 20 0 0\n" +) +MEMPOOL_B = MEMPOOL_A.replace(" 10 20", " 12 18") + +NET_A = ( + "TCP sl st flg ref tmr uack nrt rxsz local_address remote_address\n" + " 0: 04 000 1 0 0 0 1500 0.0.0.0:80 0.0.0.0:0 \n" +) +NET_B = NET_A + ( + " 1: 04 000 1 3 1 0 1500 10.0.0.2:80 10.0.0.1:5000 \n" +) +NET_A_TICK = NET_A.replace(" 1 0 0 0 ", " 1 7 2 0 ") +UDP_EMPTY = "UDP sl local_address remote_address\n" + + +class TestCleanOutput: + def test_strips_echo_and_prompt(self) -> None: + raw = "free\r\n total used\r\n 1 2\r\nnsh> " + assert clean_output(raw, "free", "nsh> ") == " total used\n 1 2\n" + + def test_keeps_output_without_echo(self) -> None: + raw = " total used\n 1 2\nnsh> " + assert clean_output(raw, "free", "nsh> ") == " total used\n 1 2\n" + + def test_handles_missing_prompt(self) -> None: + assert clean_output("free\n 1 2", "free", "nsh> ") == " 1 2" + + def test_strips_echo_with_prompt(self) -> None: + raw = "nsh> free\n 1 2\nnsh> " + assert clean_output(raw, "free", "nsh> ") == " 1 2\n" + + def test_strips_stale_prompt_before_echo(self) -> None: + raw = "nsh> \nfree\n 1 2\nnsh> " + assert clean_output(raw, "free", "nsh> ") == " 1 2\n" + + +JUNK = "nxposix_spawn_exec: ERROR: exec failed: 2\n" + + +class TestParseTable: + def test_junk_before_header_skipped(self) -> None: + rows = parse_table(JUNK + FD_A, required=["FD", "PATH"]) + assert rows[0]["FD"] == "0" + assert rows[0]["PATH"] == "/dev/console" + + def test_required_header_missing(self) -> None: + assert parse_table(JUNK, required=["FD"]) == [] + + def test_rows_keyed_by_header(self) -> None: + rows = parse_table(FREE_A) + assert rows == [ + { + "total": "67108864", + "used": "1787800", + "free": "65321064", + "maxused": "1788496", + "maxfree": "65257512", + "nused": "66", + "nfree": "3", + "name": "Umem", + } + ] + + def test_last_column_keeps_spaces(self) -> None: + rows = parse_table(PS_A) + assert ( + rows[1]["COMMAND"] == "sim_loop_wq 0x7f36774003f0 0x7f3677400470" + ) + + def test_short_rows_padded(self) -> None: + rows = parse_table(FD_A) + assert rows[2]["FD"] == "3" + assert rows[2]["PATH"] == "" + + def test_explicit_columns(self) -> None: + rows = parse_table("a b\n1 2\n", columns=["x", "y"], skip=1) + assert rows == [{"x": "1", "y": "2"}] + + def test_empty(self) -> None: + assert parse_table("") == [] + + +class TestHeapProbe: + def test_metadata(self) -> None: + probe = HeapProbe({}) + assert probe.name == "heap" + assert probe.commands == ("free",) + assert ("CONFIG_FS_PROCFS", True) in probe.kconfig + assert ("CONFIG_FS_PROCFS_EXCLUDE_MEMINFO", False) in probe.kconfig + assert ("CONFIG_NSH_DISABLE_FREE", False) in probe.kconfig + + def test_junk_before_header(self) -> None: + probe = HeapProbe({}) + assert probe.parse([JUNK + FREE_A]) == probe.parse([FREE_A]) + + def test_no_diff_when_equal(self) -> None: + probe = HeapProbe({}) + assert probe.diff(probe.parse([FREE_A]), probe.parse([FREE_A])) == [] + + def test_used_growth_reported(self) -> None: + probe = HeapProbe({}) + diff = probe.diff(probe.parse([FREE_A]), probe.parse([FREE_B])) + assert any("Umem" in d and "used" in d and "+100" in d for d in diff) + assert any("nused" in d and "+1" in d for d in diff) + + def test_tolerance_hides_small_used_growth(self) -> None: + probe = HeapProbe({"heap_tolerance": 100}) + diff = probe.diff(probe.parse([FREE_A]), probe.parse([FREE_B])) + assert not any(" used " in d for d in diff) + assert any("nused" in d for d in diff) + + def test_shrink_is_not_a_leak(self) -> None: + probe = HeapProbe({}) + assert probe.diff(probe.parse([FREE_B]), probe.parse([FREE_A])) == [] + + def test_multiple_heaps(self) -> None: + probe = HeapProbe({}) + before = probe.parse([FREE_TWO_HEAPS]) + after = probe.parse( + [ + FREE_TWO_HEAPS.replace( + "100000 948576 100000", + "100016 948560 100016", + ) + ] + ) + diff = probe.diff(before, after) + assert len(diff) == 1 + assert diff[0].startswith("Kmem") + + def test_unexpected_output_raises(self) -> None: + probe = HeapProbe({}) + with pytest.raises(ValueError, match="heap: unexpected output"): + probe.parse(["nsh: free: command not found\n"]) + + def test_new_row_after_is_reported(self) -> None: + probe = HeapProbe({}) + diff = probe.diff(probe.parse([FREE_A]), probe.parse([FREE_TWO_HEAPS])) + assert any("Kmem" in d for d in diff) + + +class TestTasksProbe: + def test_metadata(self) -> None: + probe = TasksProbe({}) + assert probe.name == "tasks" + assert probe.commands == ("ps",) + assert ("CONFIG_FS_PROCFS_EXCLUDE_PROCESS", False) in probe.kconfig + assert ("CONFIG_NSH_DISABLE_PS", False) in probe.kconfig + + def test_new_task_reported(self) -> None: + probe = TasksProbe({}) + diff = probe.diff(probe.parse([PS_A]), probe.parse([PS_B])) + assert diff == ["new tasks: 5 sh -c sleep"] + + def test_state_change_ignored(self) -> None: + probe = TasksProbe({}) + after = PS_A.replace("Running ", "Waiting ") + assert probe.diff(probe.parse([PS_A]), probe.parse([after])) == [] + + def test_gone_task_ignored(self) -> None: + probe = TasksProbe({}) + assert probe.diff(probe.parse([PS_B]), probe.parse([PS_A])) == [] + + def test_junk_before_header(self) -> None: + probe = TasksProbe({}) + assert probe.parse([JUNK + PS_A]) == probe.parse([PS_A]) + + def test_blank_event_column(self) -> None: + probe = TasksProbe({}) + assert ("0", "Idle_Task") in probe.parse([PS_A]) + + def test_missing_header_is_empty(self) -> None: + probe = TasksProbe({}) + assert probe.parse(["nsh: ps: command not found\n"]) == set() + + def test_smp_layout(self) -> None: + probe = TasksProbe({}) + parsed = probe.parse([PS_SMP]) + assert ("4", "nsh_main") in parsed + assert ("0", "CPU0 IDLE") in parsed + + +class TestFdsProbe: + def test_metadata(self) -> None: + probe = FdsProbe({}) + assert probe.name == "fds" + assert probe.commands == ("cat /proc/self/group/fd",) + assert ("CONFIG_NSH_DISABLE_CAT", False) in probe.kconfig + + def test_new_fd_reported(self) -> None: + probe = FdsProbe({}) + diff = probe.diff(probe.parse([FD_A]), probe.parse([FD_B])) + assert diff == ["new fds: 4 1 /tmp/leak"] + + def test_junk_before_header(self) -> None: + probe = FdsProbe({}) + assert probe.parse([JUNK + FD_A]) == probe.parse([FD_A]) + + def test_position_change_ignored(self) -> None: + probe = FdsProbe({}) + assert probe.diff(probe.parse([FD_A]), probe.parse([FD_A_MOVED])) == [] + + +class TestIobProbe: + def test_metadata(self) -> None: + probe = IobProbe({}) + assert probe.name == "iob" + assert ("CONFIG_MM_IOB", True) in probe.kconfig + + def test_junk_before_header(self) -> None: + probe = IobProbe({}) + assert probe.parse([JUNK + IOB_A]) == probe.parse([IOB_A]) + + def test_free_shrink_reported(self) -> None: + probe = IobProbe({}) + diff = probe.diff(probe.parse([IOB_A]), probe.parse([IOB_B])) + assert diff == ["iob nfree 36 -> 30 (-6)"] + + def test_empty_output_raises(self) -> None: + with pytest.raises(ValueError, match="iob: unexpected"): + IobProbe({}).parse([""]) + + def test_equal_is_clean(self) -> None: + probe = IobProbe({}) + assert probe.diff(probe.parse([IOB_A]), probe.parse([IOB_A])) == [] + + +class TestMempoolProbe: + def test_metadata(self) -> None: + probe = MempoolProbe({}) + assert probe.name == "mempool" + assert ("CONFIG_MM_HEAP_MEMPOOL", True) in probe.kconfig + + def test_unexpected_output_raises(self) -> None: + with pytest.raises(ValueError, match="mempool: unexpected"): + MempoolProbe({}).parse(["nsh: cat: open failed\n"]) + + def test_junk_before_header(self) -> None: + probe = MempoolProbe({}) + assert probe.parse([JUNK + MEMPOOL_A]) == probe.parse([MEMPOOL_A]) + + def test_nused_growth_reported(self) -> None: + probe = MempoolProbe({}) + diff = probe.diff(probe.parse([MEMPOOL_A]), probe.parse([MEMPOOL_B])) + assert diff == ["Umem nused 10 -> 12 (+2)"] + + +class TestNetProbe: + def test_metadata(self) -> None: + probe = NetProbe({}) + assert probe.name == "net" + assert probe.commands == ("cat /proc/net/tcp", "cat /proc/net/udp") + assert ("CONFIG_NET", True) in probe.kconfig + + def test_new_connection_reported(self) -> None: + probe = NetProbe({}) + before = probe.parse([NET_A, UDP_EMPTY]) + after = probe.parse([NET_B, UDP_EMPTY]) + assert probe.diff(before, after) == [ + "new net: 10.0.0.2:80 10.0.0.1:5000" + ] + + def test_counters_ignored(self) -> None: + probe = NetProbe({}) + before = probe.parse([NET_A, UDP_EMPTY]) + after = probe.parse([NET_A_TICK, UDP_EMPTY]) + assert probe.diff(before, after) == [] + + +class TestCpuloadProbe: + def test_metadata(self) -> None: + probe = CpuloadProbe({}) + assert probe.name == "cpuload" + assert ("CONFIG_SCHED_CPULOAD", True) in probe.kconfig + + def test_above_limit_reported(self) -> None: + probe = CpuloadProbe({"cpuload_max": 10}) + diff = probe.diff(probe.parse([" 1.0%\n"]), probe.parse([" 57.5%\n"])) + assert diff == ["cpuload 57.5% > 10%"] + + def test_below_limit_clean(self) -> None: + probe = CpuloadProbe({"cpuload_max": 10}) + assert ( + probe.diff(probe.parse([" 1.0%\n"]), probe.parse([" 9.9%\n"])) + == [] + ) + + def test_unparsable_is_clean(self) -> None: + probe = CpuloadProbe({}) + assert probe.diff(probe.parse(["x\n"]), probe.parse(["garbage"])) == [] + + +class TestRegistry: + def test_all_probes_registered(self) -> None: + assert set(PROBES) == { + "heap", + "tasks", + "fds", + "iob", + "mempool", + "net", + "cpuload", + } + + def test_make_probes_keeps_order(self) -> None: + probes = make_probes(["tasks", "heap"], {}) + assert [p.name for p in probes] == ["tasks", "heap"] + + def test_make_probes_unknown_name(self) -> None: + with pytest.raises(ValueError, match="unknown resource probe"): + make_probes(["nope"], {})