Xiao Wang

"""
Pytest configuration and shared fixtures for REDB unit tests.
"""
import os
import sys
import pytest
import logging
import tempfile
import hashlib
from unittest.mock import Mock, MagicMock, patch
from pathlib import Path

# Add the project root to the path
sys.path.insert(0, str(Path(__file__).parent.parent))

# Test files directory
TEST_FILES_DIR = Path(__file__).parent.parent / "test_files"


# ============================================================================
# Logger Fixtures
# ============================================================================

@pytest.fixture
def mock_logger():
    """Create a mock logger for testing."""
    logger = Mock()
    logger.debug = Mock()
    logger.info = Mock()
    logger.warning = Mock()
    logger.error = Mock()
    return logger


@pytest.fixture
def real_logger():
    """Create a real logger for testing with debug output."""
    logger = logging.getLogger("test_logger")
    logger.setLevel(logging.DEBUG)
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setLevel(logging.DEBUG)
        formatter = logging.Formatter('%(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
    return logger


# ============================================================================
# Test Binary Files Fixtures
# ============================================================================

@pytest.fixture
def elf_binary_path():
    """Path to the ELF test binary."""
    path = TEST_FILES_DIR / "hello"
    if not path.exists():
        pytest.skip(f"Test file not found: {path}")
    return str(path)


@pytest.fixture
def elf_advanced_binary_path():
    """Path to the advanced ELF test binary."""
    path = TEST_FILES_DIR / "hello_advanced"
    if not path.exists():
        pytest.skip(f"Test file not found: {path}")
    return str(path)


@pytest.fixture
def pe_binary_path():
    """Path to the PE test binary (.NET PE)."""
    path = TEST_FILES_DIR / "d8637bdbcfc9112fcb1f0167b398e771"
    if not path.exists():
        pytest.skip(f"Test file not found: {path}")
    return str(path)


@pytest.fixture
def elf_binary_content(elf_binary_path):
    """Read the ELF test binary content."""
    with open(elf_binary_path, "rb") as f:
        return f.read()


@pytest.fixture
def pe_binary_content(pe_binary_path):
    """Read the PE test binary content."""
    with open(pe_binary_path, "rb") as f:
        return f.read()


# ============================================================================
# Temporary Files Fixtures
# ============================================================================

@pytest.fixture
def temp_binary_file():
    """Create a temporary binary file for testing."""
    with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as f:
        # Write some test data
        f.write(b"\x7fELF" + b"\x00" * 100 + b"test data")
        temp_path = f.name

    yield temp_path

    # Cleanup
    if os.path.exists(temp_path):
        os.unlink(temp_path)


@pytest.fixture
def temp_text_file():
    """Create a temporary text file for testing."""
    with tempfile.NamedTemporaryFile(delete=False, suffix=".txt", mode="w") as f:
        f.write("This is a test file with some text content.")
        temp_path = f.name

    yield temp_path

    # Cleanup
    if os.path.exists(temp_path):
        os.unlink(temp_path)


@pytest.fixture
def temp_pe_like_file():
    """Create a temporary file with PE-like header for testing."""
    with tempfile.NamedTemporaryFile(delete=False, suffix=".exe") as f:
        # MZ header
        f.write(b"MZ" + b"\x00" * 58)
        # e_lfanew pointing to PE header at offset 64
        f.write(b"\x40\x00\x00\x00")
        # PE signature at offset 64
        f.write(b"PE\x00\x00")
        # Minimal COFF header (20 bytes)
        f.write(b"\x4c\x01")  # Machine (i386)
        f.write(b"\x01\x00")  # NumberOfSections
        f.write(b"\x00" * 16)  # Rest of COFF header
        temp_path = f.name

    yield temp_path

    # Cleanup
    if os.path.exists(temp_path):
        os.unlink(temp_path)


# ============================================================================
# Mock Exporter Fixtures
# ============================================================================

@pytest.fixture
def mock_print_exporter(mock_logger):
    """Create a mock PrintExporter."""
    from redb.extractors.database_exporters import PrintExporter
    exporter = PrintExporter(mock_logger, "test_index")
    return exporter


@pytest.fixture
def mock_elasticsearch_exporter(mock_logger):
    """Create a mock ElasticsearchExporter with mocked client."""
    with patch('redb.settings.get_elasticsearch_client') as mock_client:
        mock_client.return_value = Mock()
        from redb.extractors.database_exporters import ElasticsearchExporter
        exporter = ElasticsearchExporter(mock_logger, "test_index")
        return exporter


@pytest.fixture
def mock_clickhouse_exporter(mock_logger):
    """Create a mock ClickHouseExporter with mocked client."""
    with patch('redb.settings.create_clickhouse_client') as mock_client:
        mock_client.return_value = Mock()
        from redb.extractors.database_exporters import ClickHouseExporter
        exporter = ClickHouseExporter(mock_logger, "test_index", client=Mock())
        return exporter


# ============================================================================
# Mock Settings Fixtures
# ============================================================================

@pytest.fixture
def mock_settings():
    """Mock the redb.settings module."""
    with patch.multiple(
        'redb.settings',
        ELASTIC_BINARIES_COLLECTION="test_collection",
        get_elasticsearch_client=Mock(return_value=Mock()),
        create_clickhouse_client=Mock(return_value=Mock()),
    ):
        yield


# ============================================================================
# Hash Fixtures
# ============================================================================

@pytest.fixture
def known_hashes(elf_binary_content):
    """Compute known hashes for the ELF test binary."""
    return {
        'md5': hashlib.md5(elf_binary_content).hexdigest(),
        'sha1': hashlib.sha1(elf_binary_content).hexdigest(),
        'sha256': hashlib.sha256(elf_binary_content).hexdigest(),
    }


@pytest.fixture
def pe_known_hashes(pe_binary_content):
    """Compute known hashes for the PE test binary."""
    return {
        'md5': hashlib.md5(pe_binary_content).hexdigest(),
        'sha1': hashlib.sha1(pe_binary_content).hexdigest(),
        'sha256': hashlib.sha256(pe_binary_content).hexdigest(),
    }


# ============================================================================
# Sample Data Fixtures
# ============================================================================

@pytest.fixture
def sample_binary_data():
    """Sample binary data for entropy and hash testing."""
    return b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f" * 16


@pytest.fixture
def high_entropy_data():
    """High entropy random-like data for testing."""
    import random
    random.seed(42)  # Reproducible
    return bytes(random.randint(0, 255) for _ in range(256))


@pytest.fixture
def low_entropy_data():
    """Low entropy repetitive data for testing."""
    return b"\x00" * 256


# ============================================================================
# PE Object Fixtures
# ============================================================================

@pytest.fixture
def pe_object(pe_binary_path):
    """Create a pefile PE object for testing."""
    import pefile
    try:
        pe = pefile.PE(pe_binary_path)
        yield pe
        pe.close()
    except Exception as e:
        pytest.skip(f"Failed to parse PE file: {e}")


# ============================================================================
# ELF Object Fixtures
# ============================================================================

@pytest.fixture
def elf_object(elf_binary_path):
    """Create an ELFFile object for testing."""
    from elftools.elf.elffile import ELFFile
    try:
        with open(elf_binary_path, 'rb') as f:
            elf = ELFFile(f)
            # We need to keep the file open for ELFFile to work
            yield elf
    except Exception as e:
        pytest.skip(f"Failed to parse ELF file: {e}")


# ============================================================================
# Dataclass Fixtures
# ============================================================================

@pytest.fixture
def sample_hash_dataclass():
    """Create a sample Hash dataclass."""
    from redb.models.dataclasses import Hash
    return Hash(
        md5="d41d8cd98f00b204e9800998ecf8427e",
        sha1="da39a3ee5e6b4b0d3255bfef95601890afd80709",
        sha256="e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    )


@pytest.fixture
def sample_hashes_dataclass():
    """Create a sample Hashes dataclass."""
    from redb.models.dataclasses import Hashes
    return Hashes(
        md5="d41d8cd98f00b204e9800998ecf8427e",
        sha1="da39a3ee5e6b4b0d3255bfef95601890afd80709",
        sha256="e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
        ssdeep_hash="3::",
        tlsh_hash=""
    )


@pytest.fixture
def sample_basic_properties():
    """Create a sample BasicProperties dataclass."""
    from redb.models.dataclasses import BasicProperties
    return BasicProperties(
        filename="test.exe",
        sample_name="test.exe",
        filesize=1024,
        filetype="PE32 executable",
        filetype_mime="application/x-dosexec",
        filetype_magika="pebin",
        file_entropy=6.5
    )


# ============================================================================
# Utility Functions for Tests
# ============================================================================

def compute_entropy(data):
    """Compute entropy of data for verification in tests."""
    import math
    from collections import Counter

    if not data:
        return 0.0

    if isinstance(data, str):
        counts = Counter(data)
        frequencies = (i / len(data) for i in counts.values())
        return -sum(f * math.log(f, 2) for f in frequencies)
    else:
        occurrences = Counter(bytearray(data))
        entropy = 0
        for x in occurrences.values():
            p_x = float(x) / len(data)
            entropy -= p_x * math.log(p_x, 2)
        return entropy


# Make the function available as a fixture
@pytest.fixture
def entropy_calculator():
    """Return the entropy calculation function."""
    return compute_entropy


# ============================================================================
# Skip Markers
# ============================================================================

# Mark tests that require external tools
requires_die = pytest.mark.skipif(
    not os.path.exists("/usr/bin/diec") and not os.path.exists("/usr/local/bin/diec"),
    reason="Detect It Easy (diec) not installed"
)

requires_capa = pytest.mark.skipif(
    not os.path.exists("/usr/bin/capa") and not os.path.exists("/usr/local/bin/capa"),
    reason="capa not installed"
)

requires_floss = pytest.mark.skipif(
    not os.path.exists("/usr/bin/floss") and not os.path.exists("/usr/local/bin/floss"),
    reason="floss not installed"
)