java-topology/defects/retroarch-0002/test/test_retroarch_0002.py

"""
retroarch-0002: core_info_database_supports_content_path O(C*(E+D)) linear scan

Unit test verifying the O(N^2) defect pattern and that the O(1) fix
eliminates the quadratic growth. Uses a Python model of the C logic.
"""

import time


# ---- model the defect: O(C * (E + D)) per query ----

class CoreInfo:
    def __init__(self, extensions, databases):
        self.extensions = extensions  # list of strings
        self.databases  = databases   # list of strings


def string_list_find_elem(lst, elem):
    """Models RetroArch string_list_find_elem: O(N) linear scan."""
    for item in lst:
        if item.lower() == elem.lower():
            return True
    return False


def core_info_database_supports_content_path_defect(cores, database, ext):
    """
    Defect version: O(C * (E + D)) -- iterates all cores, two linear scans
    per core (extensions then databases).
    """
    ops = 0
    for core in cores:
        # string_list_find_elem on supported_extensions_list
        for e in core.extensions:
            ops += 1
            if e.lower() == ext.lower():
                break
        else:
            continue
        # string_list_find_elem on databases_list
        for d in core.databases:
            ops += 1
            if d.lower() == database.lower():
                return True, ops
    return False, ops


# ---- model the fix: O(1) via hash map built at list-init time ----

def build_ext_db_index(cores):
    """
    Build a dict: extension -> set of supported database names.
    Called once at core list init; O(C * (E + D)) one-time cost.
    """
    index = {}
    for core in cores:
        for ext in core.extensions:
            key = ext.lower()
            if key not in index:
                index[key] = set()
            for db in core.databases:
                index[key].add(db.lower())
    return index


def core_info_database_supports_content_path_fixed(index, database, ext):
    """
    Fixed version: O(1) hash map lookups.
    """
    dbs = index.get(ext.lower())
    if dbs is None:
        return False
    return database.lower() in dbs


# ---- helpers ----

def build_cores(num_cores, exts_per_core=5, dbs_per_core=3):
    """Build a synthetic core list."""
    cores = []
    for i in range(num_cores):
        exts = [f"ext{j}" for j in range(exts_per_core)]
        dbs  = [f"db{i}", f"shared_db", f"db{i + num_cores}"][:dbs_per_core]
        cores.append(CoreInfo(exts, dbs))
    return cores


# ---- tests ----

def test_defect_correctness():
    """Defect version returns correct results."""
    cores = build_cores(10)
    # db0 is supported by core 0 with ext0
    found, _ = core_info_database_supports_content_path_defect(cores, "db0", "ext0")
    assert found, "Expected match for db0/ext0"
    found, _ = core_info_database_supports_content_path_defect(cores, "nosuchdb", "ext0")
    assert not found, "Expected no match for nosuchdb/ext0"


def test_fixed_correctness():
    """Fixed version returns same results as defect version."""
    cores = build_cores(10)
    index = build_ext_db_index(cores)

    for db_name in ["db0", "db5", "shared_db", "nosuchdb"]:
        for ext in ["ext0", "ext3", "noext"]:
            expected, _ = core_info_database_supports_content_path_defect(cores, db_name, ext)
            got          = core_info_database_supports_content_path_fixed(index, db_name, ext)
            assert expected == got, (
                f"Mismatch for db={db_name} ext={ext}: defect={expected} fixed={got}"
            )


def test_op_count_ratio():
    """
    Defect version op count grows as O(C*(E+D)); fixed version is O(1).
    Verify the ratio at a realistic core count is >= 20x.
    """
    num_cores = 100
    exts_per_core = 5
    dbs_per_core  = 3

    cores  = build_cores(num_cores, exts_per_core, dbs_per_core)
    index  = build_ext_db_index(cores)

    # Worst-case query: extension present in all cores, database not found
    total_defect_ops = 0
    queries = 1000
    for _ in range(queries):
        _, ops = core_info_database_supports_content_path_defect(
            cores, "nosuchdb", "ext0"
        )
        total_defect_ops += ops

    # Fixed version -- count hash-set membership checks as 1 op each
    total_fixed_ops = 0
    for _ in range(queries):
        dbs = index.get("ext0")
        total_fixed_ops += 1        # one dict lookup
        if dbs is not None:
            total_fixed_ops += 1    # one set membership test

    ratio = total_defect_ops / total_fixed_ops
    print(f"  defect ops={total_defect_ops}, fixed ops={total_fixed_ops}, ratio={ratio:.1f}x")
    assert ratio >= 20, f"Expected >=20x ratio, got {ratio:.1f}x"


def test_timing_ratio():
    """
    Wall-clock timing confirms defect is significantly slower at scale.
    """
    num_cores = 100
    cores  = build_cores(num_cores, exts_per_core=5, dbs_per_core=3)
    index  = build_ext_db_index(cores)
    queries = 5000

    t0 = time.perf_counter()
    for _ in range(queries):
        core_info_database_supports_content_path_defect(cores, "nosuchdb", "ext0")
    t_defect = time.perf_counter() - t0

    t0 = time.perf_counter()
    for _ in range(queries):
        core_info_database_supports_content_path_fixed(index, "nosuchdb", "ext0")
    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_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)