java-topology/defects/openemu-0001/test/test_openemu_0001.py

"""
openemu-0001: SetupAssistant knownCores Array.contains O(N^2) in core dedup loop

Unit test verifying the O(N^2) defect and O(N) fix for the
knownCores dedup in SetupAssistant.swift.
"""

import time


# ---- model of CoreDownload (NSObject identity-based equality) ----

class CoreDownload:
    def __init__(self, bundle_id):
        self.bundle_id = bundle_id

    def __eq__(self, other):
        return self is other  # pointer equality, as NSObject

    def __hash__(self):
        return id(self)       # id-based, as NSObject

    def __repr__(self):
        return f"CoreDownload({self.bundle_id!r})"


class SetupCoreInfo:
    def __init__(self, core):
        self.core = core


# ---- defect: knownCores is a list (Array equivalent) ----

def setup_assistant_add_cores_defect(cores_to_download, core_list):
    """
    Defect version: knownCores is a list -- .contains is O(N) linear scan.
    Total: O(N^2) where N = len(core_list).
    Returns (result_list, op_count).
    """
    ops = 0
    known_cores = [info.core for info in cores_to_download if info.core is not None]  # list
    for core in core_list:
        # Array.contains: O(N) linear scan
        found = False
        for kc in known_cores:
            ops += 1
            if kc is core:
                found = True
                break
        if not found:
            cores_to_download.append(SetupCoreInfo(core=core))
            known_cores.append(core)
    return cores_to_download, ops


# ---- fix: knownCores is a set ----

def setup_assistant_add_cores_fixed(cores_to_download, core_list):
    """
    Fixed version: knownCores is a Set -- .contains is O(1).
    Total: O(N).
    Returns (result_list, op_count).
    """
    ops = 0
    known_cores = set(info.core for info in cores_to_download if info.core is not None)  # set
    for core in core_list:
        ops += 1  # O(1) hash lookup
        if core not in known_cores:
            cores_to_download.append(SetupCoreInfo(core=core))
            known_cores.add(core)
    return cores_to_download, ops


# ---- helpers ----

def make_core_list(n):
    return [CoreDownload(f"org.openemu.core.{i}") for i in range(n)]


# ---- tests ----

def test_defect_correctness():
    """Defect version correctly deduplicates cores."""
    all_cores = make_core_list(10)
    # Pre-populate with first 3
    initial = [SetupCoreInfo(core=c) for c in all_cores[:3]]
    result, _ = setup_assistant_add_cores_defect(initial, all_cores)
    assert len(result) == 10, f"Expected 10 entries, got {len(result)}"
    # All 10 cores present
    result_cores = {info.core for info in result}
    assert result_cores == set(all_cores)


def test_fixed_correctness():
    """Fixed version produces identical result to defect version."""
    all_cores = make_core_list(10)
    initial_d = [SetupCoreInfo(core=c) for c in all_cores[:3]]
    initial_f = [SetupCoreInfo(core=c) for c in all_cores[:3]]
    result_d, _ = setup_assistant_add_cores_defect(initial_d, all_cores)
    result_f, _ = setup_assistant_add_cores_fixed(initial_f, all_cores)
    assert len(result_d) == len(result_f), (
        f"Result length mismatch: defect={len(result_d)} fixed={len(result_f)}"
    )
    cores_d = {info.core for info in result_d}
    cores_f = {info.core for info in result_f}
    assert cores_d == cores_f, "Core sets differ between defect and fixed"


def test_no_duplicates():
    """Neither version produces duplicate entries."""
    all_cores = make_core_list(20)
    initial = [SetupCoreInfo(core=c) for c in all_cores[:5]]
    result_d, _ = setup_assistant_add_cores_defect(list(initial), all_cores)
    result_f, _ = setup_assistant_add_cores_fixed(list(initial), all_cores)
    for label, result in [("defect", result_d), ("fixed", result_f)]:
        seen = set()
        for info in result:
            assert info.core not in seen, f"{label}: duplicate core {info.core}"
            seen.add(info.core)


def test_op_count_ratio():
    """
    Defect op count grows as O(N^2); fixed op count grows as O(N).
    Ratio at N=35 (realistic core count) should be >= 10x.
    """
    n = 35
    all_cores = make_core_list(n)

    # Worst case: coresToDownload starts empty, all N cores are new
    _, ops_defect = setup_assistant_add_cores_defect([], all_cores)
    _, ops_fixed  = setup_assistant_add_cores_fixed([], all_cores)

    ratio = ops_defect / ops_fixed if ops_fixed > 0 else float("inf")
    print(f"  N={n}: defect ops={ops_defect}, fixed ops={ops_fixed}, ratio={ratio:.1f}x")
    assert ratio >= 10, f"Expected >=10x ratio at N={n}, got {ratio:.1f}x"


def test_timing_ratio():
    """Wall-clock timing confirms defect is slower at N=200 (stress)."""
    n = 200
    all_cores = make_core_list(n)
    runs = 500

    t0 = time.perf_counter()
    for _ in range(runs):
        setup_assistant_add_cores_defect([], list(all_cores))
    t_defect = time.perf_counter() - t0

    t0 = time.perf_counter()
    for _ in range(runs):
        setup_assistant_add_cores_fixed([], list(all_cores))
    t_fixed = time.perf_counter() - t0

    ratio = t_defect / t_fixed if t_fixed > 0 else float("inf")
    print(f"  defect={t_defect*1000:.1f}ms, fixed={t_fixed*1000:.1f}ms, ratio={ratio:.1f}x")
    assert ratio >= 5, f"Expected >=5x timing ratio, got {ratio:.1f}x"


if __name__ == "__main__":
    tests = [
        test_defect_correctness,
        test_fixed_correctness,
        test_no_duplicates,
        test_op_count_ratio,
        test_timing_ratio,
    ]
    passed = 0
    for t in tests:
        try:
            t()
            print(f"PASS  {t.__name__}")
            passed += 1
        except AssertionError as e:
            print(f"FAIL  {t.__name__}: {e}")

    print(f"\n{passed}/{len(tests)} passed")
    if passed != len(tests):
        raise SystemExit(1)