ocp/tests/test_handles.py

"""The handle model (occt_handle.h) — the spike that gates all surface work.

OCCT transients are reference-counted intrusively, and this binding gives each
Python wrapper exactly one of those references, released at deallocation.  The
properties below are what the rest of the binding is allowed to assume:

  * an object crossing to Python and back keeps its identity while alive;
  * either side may drop its reference first without taking the object away
    from the other;
  * counts stay balanced when a call raises, and across many cycles.
"""

from __future__ import annotations

import gc
import os

import pytest
from n3xd_ocp import _debug
from OCP.BRep import BRep_Tool
from OCP.Poly import Poly_Triangulation
from OCP.TopAbs import TopAbs_FACE
from OCP.TopExp import TopExp
from OCP.TopLoc import TopLoc_Location
from OCP.TopoDS import TopoDS
from OCP.TopTools import TopTools_IndexedMapOfShape


@pytest.fixture
def triangulation(fixture_shapes):
    faces = TopTools_IndexedMapOfShape()
    TopExp.MapShapes_s(fixture_shapes["box_meshed"], TopAbs_FACE, faces)
    tri = BRep_Tool.Triangulation_s(TopoDS.Face_s(faces.FindKey(1)),
                                    TopLoc_Location())
    assert tri is not None, "fixture should carry a mesh"
    return tri


def test_python_constructed_transient_is_heap_owned():
    # nb::init<> would place the object inside the Python instance and let OCCT
    # delete storage it does not own; ocp_new heap-allocates instead.
    tri = Poly_Triangulation(4, 2, False, False)
    assert tri.NbNodes() == 4
    assert _debug.refcount(tri) >= 1


def test_kernel_returned_transient_survives_its_shape(fixture_shapes):
    faces = TopTools_IndexedMapOfShape()
    TopExp.MapShapes_s(fixture_shapes["box_meshed"], TopAbs_FACE, faces)
    tri = BRep_Tool.Triangulation_s(TopoDS.Face_s(faces.FindKey(1)),
                                    TopLoc_Location())
    del faces
    gc.collect()
    assert tri.NbNodes() > 0  # touches the memory


def test_roundtrip_preserves_wrapper_identity(triangulation):
    assert _debug.roundtrip(triangulation) is triangulation


def test_cpp_side_reference_keeps_the_object_alive():
    tri = Poly_Triangulation(8, 4, False, False)
    slot = _debug.hold(tri)

    del tri
    gc.collect()

    assert _debug.held_is_alive(slot)
    assert _debug.held_nb_nodes(slot) == 8  # dereferences it
    _debug.release_held()


def test_python_reference_survives_the_cpp_side_dropping_its_own():
    tri = Poly_Triangulation(6, 2, False, False)
    _debug.hold(tri)
    _debug.release_held()

    gc.collect()
    assert tri.NbNodes() == 6


def test_refcount_is_balanced_across_conversions(triangulation):
    before = _debug.refcount(triangulation)
    for _ in range(1000):
        _debug.roundtrip(triangulation)
    gc.collect()
    assert _debug.refcount(triangulation) == before


def test_refcount_is_balanced_when_a_call_raises(triangulation):
    before = _debug.refcount(triangulation)
    for _ in range(100):
        with pytest.raises(TypeError):
            # Argument converts, then the call fails on arity.
            _debug.roundtrip(triangulation, "unexpected")
    gc.collect()
    assert _debug.refcount(triangulation) == before


def _rss_kb() -> int:
    with open("/proc/self/status") as fh:
        for line in fh:
            if line.startswith("VmRSS:"):
                return int(line.split()[1])
    raise RuntimeError("VmRSS not found")


@pytest.mark.skipif(
    os.environ.get("N3XD_OCP_ASAN") == "1",
    reason="ASAN quarantines freed memory, so RSS is not a leak signal there",
)
def test_many_create_destroy_cycles_do_not_leak():
    # ASAN covers memory *safety*; this covers *growth*.  One leaked handle per
    # cycle would keep every triangulation alive — visible as RSS climbing with
    # the iteration count rather than settling.
    def cycle(n: int) -> None:
        for _ in range(n):
            tri = Poly_Triangulation(64, 32, False, False)
            _debug.roundtrip(tri)
            del tri

    cycle(2_000)  # let allocator arenas reach steady state first
    gc.collect()
    baseline = _rss_kb()

    cycle(50_000)
    gc.collect()
    growth = _rss_kb() - baseline

    # Leaking these would cost megabytes; a few hundred KB of allocator noise
    # is expected.
    assert growth < 4_000, f"RSS grew {growth} KB across 50k cycles"


def test_null_handle_maps_to_none(fixture_shapes):
    # An unmeshed face has no triangulation; OCCT returns a null handle and the
    # caster must present that as None rather than a wrapper around nullptr.
    faces = TopTools_IndexedMapOfShape()
    TopExp.MapShapes_s(fixture_shapes["box"], TopAbs_FACE, faces)
    tri = BRep_Tool.Triangulation_s(TopoDS.Face_s(faces.FindKey(1)),
                                    TopLoc_Location())
    assert tri is None


def test_none_converts_to_a_null_handle():
    assert _debug.roundtrip(None) is None
    assert _debug.refcount(None) == 0