Xiantao Li

"""
Unit test configuration.

Replaces the redb.extractors package with a minimal stub so that
``from redb.extractors.database_exporters import ...`` works without
pulling in the heavy __init__.py import chain (pefile, elftools,
signify, magic, etc.).

Also stubs out redb.settings so that Elasticsearch / ClickHouse
clients are never actually created during unit tests.

Additionally installs binaryninja stubs for decompiler unit tests.

Stubs heavy third-party packages (pefile, magic, psutil, etc.) so that
redb.workers, redb.queries, redb.ingestor, and redb.s3_utils can all
be imported and patched in unit tests without requiring the real packages.
"""
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock

# ---------------------------------------------------------------------------
# 0. Install binaryninja stubs FIRST — must happen before any decompiler
#    module is imported (triggered by __init__.py chain).
# ---------------------------------------------------------------------------
from tests.unit.conftest_binja_stubs import install_binja_stubs as _install_binja_stubs
_install_binja_stubs()

# ---------------------------------------------------------------------------
# 1. Stub external packages required by database_exporters.py itself,
#    the decompiler chain, and the workers/ingestor modules.
# ---------------------------------------------------------------------------
_heavy_deps = [
    # Database / networking
    "elasticsearch", "clickhouse_connect", "dotenv",
    # Binary analysis — top-level packages
    "pefile", "magic", "machofile",
    # Archive handling
    "py7zr", "pyzipper",
    # File type detection
    "magika",
    # System monitoring
    "psutil",
    # S3 / MinIO (submodule paths needed for `from minio.error import S3Error`)
    "minio", "minio.error",
    # YARA
    "yara_x",
    # Binary parsing — elftools tree
    "elftools", "elftools.elf", "elftools.elf.elffile",
    "elftools.common", "elftools.common.exceptions",
    # Code signing — signify tree
    "signify", "signify.fingerprinter", "signify.authenticode",
    # LIEF
    "lief",
    # Fuzzy hashing
    "ppdeep",
    # .NET analysis
    "dotnetfile",
    # Crypto / ASN.1
    "asn1crypto", "asn1crypto.cms", "asn1crypto.pem", "asn1crypto.x509", "asn1crypto.core",
    # Hashing
    "xxhash", "blake3", "mmh3", "numpy",
    # APK analysis
    "androguard", "androguard.core", "androguard.core.apk",
    "androguard.core.dex", "androguard.core.api_specific_resources",
    "androguard.misc",
    "permhash",
]
for _mod in _heavy_deps:
    if _mod not in sys.modules:
        try:
            __import__(_mod)
        except (ImportError, ModuleNotFoundError):
            sys.modules[_mod] = MagicMock()

# Stub urllib3 with a real Warning subclass (warnings.filterwarnings needs a class)
if "urllib3" not in sys.modules or isinstance(sys.modules.get("urllib3"), MagicMock):
    try:
        import urllib3  # noqa: F401
    except (ImportError, ModuleNotFoundError):
        _urllib3_mod = types.ModuleType("urllib3")
        _urllib3_exc = types.ModuleType("urllib3.exceptions")

        class _InsecureRequestWarning(UserWarning):
            pass

        _urllib3_exc.InsecureRequestWarning = _InsecureRequestWarning
        _urllib3_mod.exceptions = _urllib3_exc
        sys.modules["urllib3"] = _urllib3_mod
        sys.modules["urllib3.exceptions"] = _urllib3_exc

# Provide a functional ppdeep stub (tests call ppdeep.hash() and expect a string).
# Must overwrite the MagicMock that the loop above may have installed.
_ppdeep_existing = sys.modules.get("ppdeep")
if _ppdeep_existing is None or isinstance(_ppdeep_existing, MagicMock):
    _ppdeep_mod = types.ModuleType("ppdeep")
    _ppdeep_mod.__name__ = "ppdeep"

    def _ppdeep_hash(data):
        """Fake ssdeep hash — returns a deterministic placeholder."""
        if isinstance(data, str):
            data = data.encode("utf-8")
        if len(data) < 50:
            return None
        import hashlib
        h = hashlib.md5(data).hexdigest()
        return f"3:{h[:16]}:{h[16:]}"

    _ppdeep_mod.hash = _ppdeep_hash
    sys.modules["ppdeep"] = _ppdeep_mod

# Stub mmh3 if the C extension is not available
_mmh3_existing = sys.modules.get("mmh3")
if _mmh3_existing is None or isinstance(_mmh3_existing, MagicMock):
    _mmh3_mod = types.ModuleType("mmh3")
    _mmh3_mod.__name__ = "mmh3"

    def _mmh3_hash(data, seed=0):
        """Fake MurmurHash3 — deterministic hash using built-in hashlib."""
        import hashlib
        if isinstance(data, str):
            data = data.encode("utf-8")
        seed_bytes = (seed & 0xFFFFFFFF).to_bytes(8, "little")
        h = hashlib.md5(data + seed_bytes).digest()
        return int.from_bytes(h[:4], "little", signed=True)

    _mmh3_mod.hash = _mmh3_hash
    sys.modules["mmh3"] = _mmh3_mod

# Stub tlsh if the C extension is not available
if "tlsh" not in sys.modules:
    try:
        import tlsh as _real_tlsh  # noqa: F401
    except (ImportError, ModuleNotFoundError):
        _tlsh_mock = types.ModuleType("tlsh")
        _tlsh_mock.__name__ = "tlsh"

        def _tlsh_hash(data):
            """Fake TLSH hash — returns a deterministic placeholder.

            Returns empty string for data < 50 bytes or data with very
            low entropy (e.g. repeating patterns), mimicking the real
            TLSH library's "TNULL" behaviour.
            """
            if isinstance(data, str):
                data = data.encode("utf-8")
            if len(data) < 50:
                return ""
            # Real TLSH returns "" for low-entropy input
            if len(set(data)) < 8:
                return ""
            import hashlib
            return "T1" + hashlib.sha256(data).hexdigest()[:70].upper()

        _tlsh_mock.hash = _tlsh_hash
        sys.modules["tlsh"] = _tlsh_mock

# ---------------------------------------------------------------------------
# 2. Stub redb.settings (package) so module-level code never runs
# ---------------------------------------------------------------------------
_settings_mock = MagicMock()
_settings_mock.__name__ = "redb.settings"
sys.modules["redb.settings"] = _settings_mock
sys.modules["redb.settings.elasticsearch"] = MagicMock()
sys.modules["redb.settings.clickhouse"] = MagicMock()

# ---------------------------------------------------------------------------
# 3. Replace redb.extractors with a minimal package that does NOT eagerly
#    import every extractor (and their heavy deps).  Sub-module imports
#    like ``from redb.extractors.database_exporters import ...`` still work
#    because __path__ points to the real package directory.
# ---------------------------------------------------------------------------
_repo_root = Path(__file__).resolve().parents[2]
_extractors_pkg = types.ModuleType("redb.extractors")
_extractors_pkg.__path__ = [str(_repo_root / "redb" / "extractors")]
_extractors_pkg.__package__ = "redb.extractors"
sys.modules["redb.extractors"] = _extractors_pkg

# ---------------------------------------------------------------------------
# 4. Force-import redb submodules so that @patch('redb.workers.xxx'),
#    @patch('redb.queries.xxx'), etc. can resolve at decoration time.
#    This works because all heavy deps are already stubbed above.
# ---------------------------------------------------------------------------
import redb.logging_utils   # noqa: F401, E402
import redb.s3_utils         # noqa: F401, E402
import redb.queries          # noqa: F401, E402
import redb.workers          # noqa: F401, E402
import redb.ingestor         # noqa: F401, E402