brain-of-reese/tests/unit/test_core_memory.py

"""Unit tests: ``app.core.memory`` — process RSS helpers (phase 187, task 01).

Every branch of the four helpers, with the filesystem stubbed through the
module's path constants (``PROC_STATUS_PATH`` / ``CGROUP_PEAK_PATH`` →
``tmp_path`` files, so ``open`` is never mocked globally) and ``resource``
stubbed through ``memory._resource`` (the guarded-import binding):

* the REAL ``/proc/self/status`` read on a Linux runner — ``rss_mb()`` is a
  positive int and provably comes from the ``VmRSS`` parse, not the
  fallback (the fallback is pinned to a sentinel that never surfaces);
* the kB→MB floor arithmetic (1500 kB → 1 MB);
* missing file, unparseable value, value-less ``VmRSS:`` line, and a file
  without any ``VmRSS`` line — each falls back to ``rss_peak_mb()``;
* the non-Linux path: ``/proc`` open raises AND ``getrusage`` is patched →
  the proxy value; both unavailable → ``None``;
* ``rss_peak_mb`` arithmetic, the no-``resource`` ``None``, and the
  never-raise wrapper when ``getrusage`` itself raises;
* ``cgroup_peak_mb`` present (bytes→MB), absent, garbage, empty → the
  latter three are ``None``;
* ``fmt_mb(None) == "?"``, ``fmt_mb(0) == "0"`` (0 is a value, not
  missing), plain ints as digits;
* the purity contract: stdlib imports only — no ``app.`` import — and the
  docstring carries the "Pure stdlib" line the phase completion criteria
  require.
"""
from __future__ import annotations

import ast
import importlib
import inspect
import sys
from types import SimpleNamespace
from typing import Any

import pytest

import app.core.memory as memory


#: Stub whose ``getrusage`` answers with a fixed ``ru_maxrss`` (kB on
#: Linux) so fallback assertions are deterministic.
def _fake_resource(maxrss_kb: int) -> Any:
    return SimpleNamespace(
        RUSAGE_SELF=0,
        getrusage=lambda who: SimpleNamespace(ru_maxrss=maxrss_kb),
    )


#: Stub whose ``getrusage`` raises — the never-raise contract probe.
def _raising_resource(exc: Exception) -> Any:
    def getrusage(who: int) -> Any:
        raise exc

    return SimpleNamespace(RUSAGE_SELF=0, getrusage=getrusage)


# ---------------------------------------------------------------------------
# rss_mb — the real /proc read (Linux) and the stubbed failure modes
# ---------------------------------------------------------------------------


@pytest.mark.skipif(not sys.platform.startswith("linux"), reason="needs /proc/self/status")
def test_rss_mb_reads_real_vm_rss_on_linux(monkeypatch: pytest.MonkeyPatch) -> None:
    # Pin the FALLBACK to a negative sentinel: a positive int can therefore
    # only come from the real VmRSS parse, not from getrusage.
    monkeypatch.setattr(memory, "rss_peak_mb", lambda: -1)

    value = memory.rss_mb()

    assert value is not None
    assert isinstance(value, int)
    assert value > 0


def test_rss_mb_parses_vm_rss_kb_flooring_to_mb(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    status = tmp_path / "status"
    status.write_text("Name:\tpython\nVmRSS:\t   1500 kB\nVmSize:\t999999 kB\n")
    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
    monkeypatch.setattr(memory, "_resource", _fake_resource(999_999))

    assert memory.rss_mb() == 1  # 1500 kB // 1024 — integer floor, not round


def test_rss_mb_missing_file_falls_back_to_getrusage(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(tmp_path / "nope"))
    monkeypatch.setattr(memory, "_resource", _fake_resource(9216))  # 9 MB

    assert memory.rss_mb() == 9


def test_rss_mb_unparseable_value_falls_back(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    status = tmp_path / "status"
    status.write_text("VmRSS:\tnot-a-number kB\n")
    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
    monkeypatch.setattr(memory, "_resource", _fake_resource(2048))

    assert memory.rss_mb() == 2


def test_rss_mb_value_less_line_falls_back(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    # "VmRSS:" with no value: the split()[1] raises IndexError internally,
    # the suppress turns it into the rss_peak_mb() proxy.
    status = tmp_path / "status"
    status.write_text("VmRSS:\n")
    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
    monkeypatch.setattr(memory, "_resource", _fake_resource(3072))

    assert memory.rss_mb() == 3


def test_rss_mb_file_without_vm_rss_line_falls_back(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    status = tmp_path / "status"
    status.write_text("Name:\tpython\nVmSize:\t999999 kB\nVmHWM:\t888888 kB\n")
    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(status))
    monkeypatch.setattr(memory, "_resource", _fake_resource(1024))

    assert memory.rss_mb() == 1


def test_rss_mb_returns_none_when_proc_and_resource_both_unavailable(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    monkeypatch.setattr(memory, "PROC_STATUS_PATH", str(tmp_path / "nope"))
    monkeypatch.setattr(memory, "_resource", None)

    assert memory.rss_mb() is None


# ---------------------------------------------------------------------------
# rss_peak_mb — arithmetic, the guarded import, the never-raise wrapper
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    ("maxrss_kb", "expected_mb"),
    [(1024, 1), (1023, 0), (1500, 1), (4_294_967_296, 4_194_304)],
)
def test_rss_peak_mb_arithmetic(
    monkeypatch: pytest.MonkeyPatch, maxrss_kb: int, expected_mb: int
) -> None:
    monkeypatch.setattr(memory, "_resource", _fake_resource(maxrss_kb))

    assert memory.rss_peak_mb() == expected_mb


def test_rss_peak_mb_none_without_resource_module(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(memory, "_resource", None)

    assert memory.rss_peak_mb() is None


def test_guarded_import_degrades_to_none_when_resource_is_missing(
    tmp_path: Any,
) -> None:
    # The guarded-import branch itself: a platform WITHOUT the Unix-only
    # `resource` module must still import this module, and every helper
    # must answer None (no /proc, no proxy). `sys.modules[name] = None`
    # makes `import resource` raise ImportError, so re-executing the
    # module exercises the except arm for real (no pragma).
    saved = sys.modules.get("resource")
    sys.modules["resource"] = None  # type: ignore[assignment]
    try:
        importlib.reload(memory)

        assert memory._resource is None
        assert memory.rss_peak_mb() is None
        memory.PROC_STATUS_PATH = str(tmp_path / "nope")  # reset by reload
        assert memory.rss_mb() is None
    finally:
        if saved is not None:
            sys.modules["resource"] = saved
        else:  # pragma: no cover - non-Unix runners never had it
            sys.modules.pop("resource", None)
        importlib.reload(memory)  # restore the real bindings for later tests


def test_module_reloads_with_real_resource_after_the_missing_module_probe() -> None:
    # Sibling pin: the finally-reload above restored the module, so the
    # guarded import took the happy arm again (order-independent: this
    # only passes if _resource is the live module on this platform).
    if not sys.platform.startswith("linux"):
        pytest.skip("resource exists on Linux")

    assert memory._resource is not None


def test_rss_peak_mb_never_raises_when_getrusage_raises(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(memory, "_resource", _raising_resource(OSError("boom")))

    assert memory.rss_peak_mb() is None


def test_rss_peak_mb_real_call_on_linux_yields_positive_int() -> None:
    if not sys.platform.startswith("linux"):
        pytest.skip("resource.getrusage is the Linux deploy path")
    value = memory.rss_peak_mb()

    assert value is not None
    assert isinstance(value, int)
    assert value > 0


# ---------------------------------------------------------------------------
# cgroup_peak_mb — present / absent / garbage / empty
# ---------------------------------------------------------------------------


def test_cgroup_peak_mb_present_converts_bytes_to_mb(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    # 2,306,867,200 bytes == 2200 MiB — the incident box's ~2.16 GB of
    # cgroup page-cache-inflated "peak" (module docstring caveat).
    peak = tmp_path / "memory.peak"
    peak.write_text("2306867200\n")
    monkeypatch.setattr(memory, "CGROUP_PEAK_PATH", str(peak))

    assert memory.cgroup_peak_mb() == 2200


def test_cgroup_peak_mb_absent_is_none(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
) -> None:
    monkeypatch.setattr(memory, "CGROUP_PEAK_PATH", str(tmp_path / "memory.peak"))

    assert memory.cgroup_peak_mb() is None


@pytest.mark.parametrize("body", ["max", "", "   \n", "1048576 kB"])
def test_cgroup_peak_mb_garbage_is_none(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Any, body: str
) -> None:
    peak = tmp_path / "memory.peak"
    peak.write_text(body)
    monkeypatch.setattr(memory, "CGROUP_PEAK_PATH", str(peak))

    assert memory.cgroup_peak_mb() is None


# ---------------------------------------------------------------------------
# fmt_mb — the "?" rendering
# ---------------------------------------------------------------------------


def test_fmt_mb_renders_none_as_question_mark_and_values_as_digits() -> None:
    assert memory.fmt_mb(None) == "?"
    assert memory.fmt_mb(0) == "0"  # 0 is a VALUE, not missing
    assert memory.fmt_mb(512) == "512"


# ---------------------------------------------------------------------------
# Purity contract (completion criterion: stdlib only, no app. imports)
# ---------------------------------------------------------------------------


def test_module_imports_stdlib_only_and_declares_pure_stdlib() -> None:
    tree = ast.parse(inspect.getsource(memory))
    imported: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            imported.update(alias.name.split(".")[0] for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            if node.level:  # relative import — still inside the app package
                imported.add("app")
            elif node.module:
                imported.add(node.module.split(".")[0])

    assert "app" not in imported
    # __future__, contextlib, types are static; `resource` is the guarded
    # Unix-only stdlib import — everything must stay standard library.
    assert imported <= {"__future__", "contextlib", "resource", "types"}, imported
    assert "Pure stdlib" in (memory.__doc__ or "")