Rajiv Bagai

36 papers A 2B 3C 4Misc 1Journal 11Unranked 14
YearRankTypeTitle / Venue / Authors
2023 J jnl
Trans. Data Priv.
Rajiv Bagai, Eric Weber, Vikas Thammanna Gowda
2021 conf
IDAACS
Vikas Thammanna Gowda, Rajiv Bagai, Gerald Spilinek, Spandana Vitalapura
2019 conf
UEMCON
Rajiv Bagai, Seth Layton, Srikanth Gampa, Kavitha Chandrashekar
2018 conf
CN
Huabo Lu, Rajiv Bagai
2017 J jnl
IEEE Trans. Inf. Forensics Secur.
Rajiv Bagai, Nafia Malik, Murtuza Jadliwala
2016 conf
CN
Rajiv Bagai, Huabo Lu
2015 J jnl
Int. J. Secur. Networks
Rajiv Bagai, Bin Tang, Ahsan A. Khan, Abdus Samad
2015 J jnl
Int. J. Granul. Comput. Rough Sets Intell. Syst.
Bin Tang, Rajiv Bagai, Huabo Lu
2015 conf
CICSyN
Rajiv Bagai, Kirankumar Chandrashekar
2014 conf
ADHOC-NOW Workshops
Bin Tang, Rajiv Bagai, F. N. U. Nilofar, Mehmet Bayram Yildirim
2013 B conf
SECON
Xinyu Xue, Xiang Hou, Bin Tang, Rajiv Bagai
2013 J jnl
IEEE Syst. J.
Rajiv Bagai, Bin Tang, Euna Kim
2012 B conf
TrustCom
Rajiv Bagai, Nan Jiang
2011 J jnl
Int. J. Secur. Networks
Neeraj Jaggi, Umesh Marappa Reddy, Rajiv Bagai
2011 A conf
PETS
Rajiv Bagai, Huabo Lu, Rong Li, Bin Tang
2011 B conf
AINA
Rajiv Bagai, Bin Tang
2010 C conf
EUC
Rajiv Bagai, Huabo Lu, Bin Tang
2006 conf
Computational Intelligence
Rajiv Bagai
2004 conf
Neural Networks and Computational Intelligence
Rajiv Bagai
2003 conf
Neural Networks and Computational Intelligence
Rajiv Bagai
2002 conf
Soft-Ware
Rajiv Bagai, Shellene J. Kelley
2000 J jnl
Inf. Syst.
Nicholas Q. Trân, Rajiv Bagai
2000 conf
IBERAMIA-SBIA
Rajiv Bagai
1999 C conf
ADBIS
Nicholas Q. Trân, Rajiv Bagai
1996 J jnl
J. Intell. Inf. Syst.
Rajiv Bagai, Rajshekhar Sunderraman
1996 J jnl
Int. J. Uncertain. Fuzziness Knowl. Based Syst.
Rajiv Bagai, Rajshekhar Sunderraman
1995 J jnl
Int. J. Comput. Math.
Rajiv Bagai, Rajshekhar Sunderraman
1995 conf
ACM Southeast Regional Conference
Rajiv Bagai, Mehmet A. Orgun
1995 conf
AMAST
Rajiv Bagai, Rajshekhar Sunderraman
1995 J jnl
Int. J. Comput. Math.
Rajiv Bagai, Mahesh Rathi
1995 Misc conf
COMAD
Rajshekhar Sunderraman, Rajiv Bagai
1994 conf
BNCOD
Rajiv Bagai, Rajshekhar Sunderraman
1994 A conf
ECOOP
Suad Alagic, Rajshekhar Sunderraman, Rajiv Bagai
1993 C conf
ISMIS
Rajiv Bagai, Vasant Shanbhogue, Jan M. Zytkow, Shang-Ching Chou
1993 C conf
ICCI
Rajiv Bagai, Vasant Shanbhogue, Jan M. Zytkow, Shang-Ching Chou
1990
Rajiv Bagai
tests/unit/test_decompile_analysis.py
← Index tests/unit/test_decompile_analysis.py python
"""Unit tests (mocked Binary Ninja) for analysis modules:
- bninja/analysis/cfg.py — CFGAnalysis
- bninja/analysis/disassembly.py — DisassemblyAnalysis
- bninja/analysis/low_level.py — LowLevelAnalysis
"""
import sys
import pytest
from unittest.mock import MagicMock

from tests.unit.conftest_binja_stubs import (
    install_binja_stubs,
    BranchType,
    InstructionTextTokenType,
    MockBasicBlock,
    MockEdge,
    MockFunction,
    MockToken,
    MockDisassemblyLine,
    MockBinaryView,
    MockSymbol,
    SymbolType,
    LowLevelILOperation,
)

install_binja_stubs()

from redb.extractors.decompiler.bninja.analysis.cfg import CFGAnalysis
from redb.extractors.decompiler.bninja.analysis.disassembly import DisassemblyAnalysis
from redb.extractors.decompiler.bninja.arch.x86 import Arch_x86


# ============================================================================
# 9a. CFGAnalysis
# ============================================================================


class TestCFGCyclomaticComplexity:
    def test_cyclomatic_complexity_linear(self):
        """Single block, no edges: E - N + 2 = 0 - 1 + 2 = 1."""
        block = MockBasicBlock(start=0x1000, end=0x1010, outgoing_edges=[])
        func = MockFunction(start=0x1000, basic_blocks=[block])
        cfg = CFGAnalysis(func)
        result = cfg.extract_function_cfg()
        assert result["cyclomatic_complexity"] == 1

    def test_cyclomatic_complexity_branch(self):
        """Diamond: 4 blocks, 4 edges -> 4 - 4 + 2 = 2."""
        entry = MockBasicBlock(start=0x1000, end=0x1010)
        true_b = MockBasicBlock(start=0x1010, end=0x1020)
        false_b = MockBasicBlock(start=0x1020, end=0x1030)
        merge = MockBasicBlock(start=0x1030, end=0x1040)

        entry.outgoing_edges = [MockEdge(target=true_b), MockEdge(target=false_b)]
        true_b.outgoing_edges = [MockEdge(target=merge)]
        false_b.outgoing_edges = [MockEdge(target=merge)]
        merge.outgoing_edges = []

        func = MockFunction(start=0x1000, basic_blocks=[entry, true_b, false_b, merge])
        cfg = CFGAnalysis(func)
        result = cfg.extract_function_cfg()
        assert result["cyclomatic_complexity"] == 2

    def test_cyclomatic_complexity_loop(self):
        """Loop: 3 blocks, 3 edges -> 3 - 3 + 2 = 2."""
        header = MockBasicBlock(start=0x1000, end=0x1010)
        body = MockBasicBlock(start=0x1010, end=0x1020)
        exit_b = MockBasicBlock(start=0x1020, end=0x1030)

        header.outgoing_edges = [MockEdge(target=body), MockEdge(target=exit_b)]
        body.outgoing_edges = [MockEdge(target=header)]
        exit_b.outgoing_edges = []

        func = MockFunction(start=0x1000, basic_blocks=[header, body, exit_b])
        cfg = CFGAnalysis(func)
        result = cfg.extract_function_cfg()
        assert result["cyclomatic_complexity"] == 2


class TestCFGExtractFunctionCFG:
    def _make_simple_cfg(self):
        """Create a simple two-block CFG for testing structure."""
        entry = MockBasicBlock(start=0x1000, end=0x1010)
        exit_b = MockBasicBlock(start=0x1010, end=0x1020)

        entry.outgoing_edges = [MockEdge(source=entry, target=exit_b, edge_type=BranchType.UnconditionalBranch)]
        exit_b.incoming_edges = [MockEdge(source=entry, target=exit_b)]
        exit_b.outgoing_edges = []
        entry.incoming_edges = []

        func = MockFunction(start=0x1000, basic_blocks=[entry, exit_b])
        return func

    def test_extract_function_cfg_structure(self):
        func = self._make_simple_cfg()
        cfg = CFGAnalysis(func)
        result = cfg.extract_function_cfg()
        assert "function_address" not in result
        # New schema: no "blocks" or "measures" nesting
        assert "blocks" not in result
        assert "measures" not in result

    def test_function_cfg_new_keys(self):
        """Assert all expected keys are present in the new output dict."""
        func = self._make_simple_cfg()
        cfg = CFGAnalysis(func)
        result = cfg.extract_function_cfg()
        expected_keys = [
            "cfg_topology_hash",
            "block_count",
            "edge_count",
            "llil_total_operations",
            "call_count",
            "cyclomatic_complexity",
            "loop_count",
            "max_depth",
            "max_fan_out",
            "md_index_topdown",
            "md_index_bottomup",
            "prime_product_llil",
            "cfg_feature_tlsh",
            "wl_minhash",
            "bb_features",
            "cfg_adjacency",
        ]
        for key in expected_keys:
            assert key in result, f"Missing key: {key}"

    def test_returns_none_for_empty_blocks(self):
        func = MockFunction(start=0x1000, basic_blocks=[])
        cfg = CFGAnalysis(func)
        assert cfg.extract_function_cfg() is None


class TestCFGTopologyHash:
    def _make_two_block_cfg(self):
        entry = MockBasicBlock(start=0x1000, end=0x1010)
        exit_b = MockBasicBlock(start=0x1010, end=0x1020)
        entry.outgoing_edges = [MockEdge(target=exit_b)]
        exit_b.outgoing_edges = []
        return MockFunction(start=0x1000, basic_blocks=[entry, exit_b])

    def test_topology_hash_is_16_bytes(self):
        func = self._make_two_block_cfg()
        cfg = CFGAnalysis(func)
        result = cfg.extract_function_cfg()
        assert isinstance(result["cfg_topology_hash"], bytes)
        assert len(result["cfg_topology_hash"]) == 16

    def test_topology_hash_deterministic(self):
        func = self._make_two_block_cfg()
        r1 = CFGAnalysis(func).extract_function_cfg()
        r2 = CFGAnalysis(func).extract_function_cfg()
        assert r1["cfg_topology_hash"] == r2["cfg_topology_hash"]


class TestCFGLoopCount:
    def test_no_loops(self):
        entry = MockBasicBlock(start=0x1000, end=0x1010)
        exit_b = MockBasicBlock(start=0x1010, end=0x1020)
        entry.outgoing_edges = [MockEdge(target=exit_b)]
        exit_b.outgoing_edges = []
        func = MockFunction(start=0x1000, basic_blocks=[entry, exit_b])
        result = CFGAnalysis(func).extract_function_cfg()
        assert result["loop_count"] == 0

    def test_single_loop(self):
        header = MockBasicBlock(start=0x1000, end=0x1010)
        body = MockBasicBlock(start=0x1010, end=0x1020)
        exit_b = MockBasicBlock(start=0x1020, end=0x1030)
        header.outgoing_edges = [MockEdge(target=body), MockEdge(target=exit_b)]
        body.outgoing_edges = [MockEdge(target=header)]
        exit_b.outgoing_edges = []
        func = MockFunction(start=0x1000, basic_blocks=[header, body, exit_b])
        result = CFGAnalysis(func).extract_function_cfg()
        assert result["loop_count"] == 1


class TestCFGMaxDepth:
    def test_max_depth_linear(self):
        entry = MockBasicBlock(start=0x1000, end=0x1010)
        b1 = MockBasicBlock(start=0x1010, end=0x1020)
        b2 = MockBasicBlock(start=0x1020, end=0x1030)
        entry.outgoing_edges = [MockEdge(target=b1)]
        b1.outgoing_edges = [MockEdge(target=b2)]
        b2.outgoing_edges = []
        func = MockFunction(start=0x1000, basic_blocks=[entry, b1, b2])
        result = CFGAnalysis(func).extract_function_cfg()
        assert result["max_depth"] == 2

    def test_max_depth_single_block(self):
        block = MockBasicBlock(start=0x1000, end=0x1010, outgoing_edges=[])
        func = MockFunction(start=0x1000, basic_blocks=[block])
        result = CFGAnalysis(func).extract_function_cfg()
        assert result["max_depth"] == 0


class TestCFGCollectBlockLlilOps:
    """Test that _collect_block_llil_ops correctly maps LLIL data to native blocks."""

    def test_llil_fields_nonzero_with_mock_llil(self):
        """When LLIL is available, llil_total_operations and call_count should be non-zero."""
        # Two native blocks
        entry = MockBasicBlock(start=0x1000, end=0x1010)
        exit_b = MockBasicBlock(start=0x1010, end=0x1020)
        entry.outgoing_edges = [MockEdge(target=exit_b)]
        exit_b.outgoing_edges = []

        # LLIL instructions: SET_REG, CALL in first block; STORE, RET in second
        llil_instrs_1 = [
            MockLLILInstruction(LowLevelILOperation.LLIL_SET_REG),
            MockLLILInstruction(LowLevelILOperation.LLIL_CALL),
        ]
        llil_instrs_2 = [
            MockLLILInstruction(LowLevelILOperation.LLIL_STORE),
            MockLLILInstruction(LowLevelILOperation.LLIL_RET),
        ]

        # LLIL basic blocks map back to native blocks via source_block
        llil_bb1 = MockLLILBasicBlock(llil_instrs_1, source_block=entry)
        llil_bb2 = MockLLILBasicBlock(llil_instrs_2, source_block=exit_b)
        llil_func = MockLLILFunction([llil_bb1, llil_bb2])

        func = MockFunction(start=0x1000, basic_blocks=[entry, exit_b], llil=llil_func)
        result = CFGAnalysis(func, llil_function=llil_func).extract_function_cfg()

        assert result["llil_total_operations"] == 4
        assert result["call_count"] == 1
        assert result["prime_product_llil"] != 0

    def test_llil_none_gives_zero_fields(self):
        """Without LLIL, LLIL-dependent fields should be zero."""
        block = MockBasicBlock(start=0x1000, end=0x1010, outgoing_edges=[])
        func = MockFunction(start=0x1000, basic_blocks=[block])
        result = CFGAnalysis(func).extract_function_cfg()

        assert result["llil_total_operations"] == 0
        assert result["call_count"] == 0
        assert result["prime_product_llil"] == 0

    def test_bb_features_with_llil(self):
        """bb_features should reflect LLIL instruction categories when LLIL is available."""
        block = MockBasicBlock(start=0x1000, end=0x1010, outgoing_edges=[])

        llil_instrs = [
            MockLLILInstruction(LowLevelILOperation.LLIL_ADD),
            MockLLILInstruction(LowLevelILOperation.LLIL_LOAD),
            MockLLILInstruction(LowLevelILOperation.LLIL_CALL),
        ]
        llil_bb = MockLLILBasicBlock(llil_instrs, source_block=block)
        llil_func = MockLLILFunction([llil_bb])

        func = MockFunction(start=0x1000, basic_blocks=[block])
        result = CFGAnalysis(func, llil_function=llil_func).extract_function_cfg()

        feats = result["bb_features"]
        assert len(feats) == 1
        assert feats[0][0] == 3  # instruction count = 3
        # At least one non-zero category count (not all OTHER)
        category_counts = feats[0][1:7]
        assert sum(category_counts) > 0


# ============================================================================
# 9b. DisassemblyAnalysis
# ============================================================================


class TestDisassemblyAnalysisGetJson:
    def _make_analysis(self, instructions=None, basic_blocks=None):
        arch = Arch_x86()
        if instructions is None:
            instructions = [
                (
                    [
                        MockToken("push", InstructionTextTokenType.InstructionToken),
                        MockToken(" ", InstructionTextTokenType.TextToken),
                        MockToken("rbp", InstructionTextTokenType.RegisterToken),
                    ],
                    0x1000,
                ),
                (
                    [
                        MockToken("mov", InstructionTextTokenType.InstructionToken),
                        MockToken(" ", InstructionTextTokenType.TextToken),
                        MockToken("rsp", InstructionTextTokenType.RegisterToken),
                    ],
                    0x1003,
                ),
            ]
        if basic_blocks is None:
            basic_blocks = [MockBasicBlock(
                start=0x1000, end=0x1010,
                disassembly_text=[MockDisassemblyLine([MockToken("push rbp")])]
            )]

        func = MockFunction(
            name="test_func",
            start=0x1000,
            basic_blocks=basic_blocks,
            instructions=instructions,
            symbol=MockSymbol(symbol_type=SymbolType.FunctionSymbol, name="test_func"),
            stack_adjustment=MagicMock(value=-8),
            mlil=None,
        )
        bv = MockBinaryView()
        logger = MagicMock()
        return DisassemblyAnalysis(arch, func, bv, logger)

    def test_get_json_basic_structure(self):
        da = self._make_analysis()
        result, errors = da.get_json()
        expected_keys = [
            "disassembled_function_hash",
            "disassembled_function",
            "disassembled_function_no_addresses",
            "disassembled_function_name",
            "disassembled_function_address",
            "instructions_count",
            "function_type",
            "instructions_types",
            "control_flow_count",
            "memory_access_pattern",
            "register_usage",
            "data_references_count",
        ]
        for key in expected_keys:
            assert key in result, f"Missing key: {key}"

    def test_get_json_hash_deterministic(self):
        da = self._make_analysis()
        r1, _ = da.get_json()
        da2 = self._make_analysis()
        r2, _ = da2.get_json()
        assert r1["disassembled_function_hash"] == r2["disassembled_function_hash"]


class TestDisassemblyCollectInstructionTypes:
    def test_collect_instruction_types(self):
        arch = Arch_x86()
        instructions = [
            ([MockToken("MOV", InstructionTextTokenType.InstructionToken)], 0x1000),
            ([MockToken("ADD", InstructionTextTokenType.InstructionToken)], 0x1001),
            ([MockToken("MOV", InstructionTextTokenType.InstructionToken)], 0x1002),
        ]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        types = da.collect_instruction_types()
        assert "DATA_MOVEMENT" in types
        assert "ARITHMETIC" in types

    def test_collect_instruction_types_empty(self):
        arch = Arch_x86()
        func = MockFunction(start=0x1000, instructions=[], symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        types = da.collect_instruction_types()
        assert types == {}


class TestDisassemblyMemoryPatterns:
    def _make_memory_instruction(self, tokens):
        return ([t for t in tokens], 0x1000)

    def test_collect_memory_patterns_stack(self):
        arch = Arch_x86()
        tokens = [
            MockToken("[", InstructionTextTokenType.BeginMemoryOperandToken),
            MockToken("RSP", InstructionTextTokenType.RegisterToken),
            MockToken("+0x8", InstructionTextTokenType.TextToken),
            MockToken("]", InstructionTextTokenType.EndMemoryOperandToken),
        ]
        instructions = [self._make_memory_instruction(tokens)]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        patterns = da.collect_memory_patterns()
        assert "MEM_STACK" in patterns

    def test_collect_memory_patterns_direct(self):
        arch = Arch_x86()
        tokens = [
            MockToken("[", InstructionTextTokenType.BeginMemoryOperandToken),
            MockToken("0x402000", InstructionTextTokenType.TextToken),
            MockToken("]", InstructionTextTokenType.EndMemoryOperandToken),
        ]
        instructions = [self._make_memory_instruction(tokens)]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        patterns = da.collect_memory_patterns()
        assert "MEM_DIRECT" in patterns

    def test_collect_memory_patterns_scaled(self):
        arch = Arch_x86()
        tokens = [
            MockToken("[", InstructionTextTokenType.BeginMemoryOperandToken),
            MockToken("RAX+RCX*4", InstructionTextTokenType.TextToken),
            MockToken("]", InstructionTextTokenType.EndMemoryOperandToken),
        ]
        instructions = [self._make_memory_instruction(tokens)]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        patterns = da.collect_memory_patterns()
        assert "MEM_SCALED_INDEX" in patterns

    def test_collect_memory_patterns_base_offset(self):
        arch = Arch_x86()
        tokens = [
            MockToken("[", InstructionTextTokenType.BeginMemoryOperandToken),
            MockToken("RAX+0x10", InstructionTextTokenType.TextToken),
            MockToken("]", InstructionTextTokenType.EndMemoryOperandToken),
        ]
        instructions = [self._make_memory_instruction(tokens)]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        patterns = da.collect_memory_patterns()
        assert "MEM_BASE_OFFSET" in patterns


class TestDisassemblyRegisterUsage:
    def test_collect_register_usage_gpr(self):
        arch = Arch_x86()
        instructions = [
            ([MockToken("RAX", InstructionTextTokenType.RegisterToken)], 0x1000),
        ]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        regs = da.collect_register_usage()
        assert "GPR" in regs

    def test_collect_register_usage_simd(self):
        arch = Arch_x86()
        instructions = [
            ([MockToken("XMM0", InstructionTextTokenType.RegisterToken)], 0x1000),
        ]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        regs = da.collect_register_usage()
        assert "SIMD" in regs

    def test_collect_register_usage_fpu(self):
        arch = Arch_x86()
        instructions = [
            ([MockToken("ST0", InstructionTextTokenType.RegisterToken)], 0x1000),
        ]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        regs = da.collect_register_usage()
        assert "FPU" in regs


class TestDisassemblyMisc:
    def test_count_data_references(self):
        arch = Arch_x86()
        func = MockFunction(start=0x1000, instructions=[], symbol=MockSymbol(), mlil=None)
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        assert da.count_data_references() == 0

    def test_compute_max_block_size(self):
        arch = Arch_x86()
        blocks = [
            MockBasicBlock(disassembly_text=[MockDisassemblyLine([]) for _ in range(3)]),
            MockBasicBlock(disassembly_text=[MockDisassemblyLine([]) for _ in range(5)]),
        ]
        func = MockFunction(start=0x1000, basic_blocks=blocks, instructions=[], symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        assert da.compute_max_block_size() == 5

    def test_compute_num_calls(self):
        arch = Arch_x86()
        instructions = [
            ([MockToken("CALL", InstructionTextTokenType.InstructionToken)], 0x1000),
            ([MockToken("MOV", InstructionTextTokenType.InstructionToken)], 0x1005),
            ([MockToken("CALL", InstructionTextTokenType.InstructionToken)], 0x1010),
        ]
        func = MockFunction(start=0x1000, instructions=instructions, symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        assert da.compute_num_calls() == 2

    def test_estimate_stack_size_value(self):
        arch = Arch_x86()
        stack = MagicMock()
        stack.value = -16
        func = MockFunction(start=0x1000, instructions=[], symbol=MockSymbol(), stack_adjustment=stack)
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        assert da.estimate_stack_size() == -16

    def test_estimate_stack_size_int(self):
        arch = Arch_x86()
        func = MockFunction(start=0x1000, instructions=[], symbol=MockSymbol(), stack_adjustment=-8)
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        assert da.estimate_stack_size() == -8

    def test_normalize_opcode(self):
        arch = Arch_x86()
        func = MockFunction(start=0x1000, instructions=[], symbol=MockSymbol())
        da = DisassemblyAnalysis(arch, func, MockBinaryView(), MagicMock())
        assert da.normalize_opcode("mov") == "MOV"
        assert da.normalize_opcode("PUSH") == "PUSH"


# ============================================================================
# 9c. LowLevelAnalysis (basic tests with mocked LLIL)
# ============================================================================


class MockLLILInstruction:
    """Mock LLIL instruction for low_level.py tests."""
    def __init__(self, operation, operands=None, address=0):
        self.operation = operation
        self.operands = operands or []
        self.address = address

    def __str__(self):
        return f"LLIL_{self.operation}"


class MockLLILBasicBlock:
    def __init__(self, instructions, source_block=None):
        self._instructions = instructions
        self.source_block = source_block

    def __iter__(self):
        return iter(self._instructions)


class MockLLILFunction:
    def __init__(self, basic_blocks):
        self.basic_blocks = basic_blocks
        self._instructions = []
        for bb in basic_blocks:
            self._instructions.extend(bb._instructions)

    @property
    def instructions(self):
        return iter(self._instructions)

    @property
    def source_function(self):
        mock = MagicMock()
        mock.start = 0x1000
        return mock


class TestLowLevelAnalysisCountControlFlow:
    def test_count_control_flow_instructions(self):
        from redb.extractors.decompiler.bninja.analysis.low_level import LowLevelAnalysis
        instrs = [
            MockLLILInstruction(LowLevelILOperation.LLIL_IF),
            MockLLILInstruction(LowLevelILOperation.LLIL_SET_REG),
            MockLLILInstruction(LowLevelILOperation.LLIL_CALL),
            MockLLILInstruction(LowLevelILOperation.LLIL_GOTO),
        ]
        bb = MockLLILBasicBlock(instrs)
        llil_func = MockLLILFunction([bb])

        func = MockFunction(start=0x1000, llil=llil_func, symbol=MockSymbol())
        func.low_level_il = None
        bv = MockBinaryView()
        bv.arch = MagicMock()
        bv.arch.stack_pointer = "sp"
        la = LowLevelAnalysis(func, bv, MagicMock())
        assert la.count_control_flow_instructions() == 3  # IF, CALL, GOTO


class TestLowLevelAnalysisNumCalls:
    def test_compute_num_calls_llil(self):
        from redb.extractors.decompiler.bninja.analysis.low_level import LowLevelAnalysis
        instrs = [
            MockLLILInstruction(LowLevelILOperation.LLIL_CALL),
            MockLLILInstruction(LowLevelILOperation.LLIL_TAILCALL),
            MockLLILInstruction(LowLevelILOperation.LLIL_SET_REG),
        ]
        bb = MockLLILBasicBlock(instrs)
        llil_func = MockLLILFunction([bb])

        func = MockFunction(start=0x1000, llil=llil_func, symbol=MockSymbol())
        func.low_level_il = None
        bv = MockBinaryView()
        la = LowLevelAnalysis(func, bv, MagicMock())
        assert la.compute_num_calls() == 2


class TestLowLevelAnalysisCollectNormalization:
    def test_collect_low_level(self):
        from redb.extractors.decompiler.bninja.analysis.low_level import LowLevelAnalysis
        instrs = [
            MockLLILInstruction(LowLevelILOperation.LLIL_SET_REG, address=0x1000),
            MockLLILInstruction(LowLevelILOperation.LLIL_STORE, address=0x1004),
        ]
        bb = MockLLILBasicBlock(instrs)
        llil_func = MockLLILFunction([bb])

        func = MockFunction(start=0x1000, llil=llil_func, symbol=MockSymbol())
        func.low_level_il = None
        bv = MockBinaryView()
        la = LowLevelAnalysis(func, bv, MagicMock())
        result, _ = la._collect_low_level_and_with_addr()
        assert len(result) == 2
        # Each item is a list of operation ints
        assert isinstance(result[0], list)

    def test_collect_low_level_with_addr_offset_clamping(self):
        from redb.extractors.decompiler.bninja.analysis.low_level import LowLevelAnalysis
        instrs = [
            MockLLILInstruction(LowLevelILOperation.LLIL_SET_REG, address=0x0FFF),  # Before function start
        ]
        bb = MockLLILBasicBlock(instrs)
        llil_func = MockLLILFunction([bb])

        func = MockFunction(start=0x1000, llil=llil_func, symbol=MockSymbol())
        func.low_level_il = None
        bv = MockBinaryView()
        la = LowLevelAnalysis(func, bv, MagicMock())
        _, result = la._collect_low_level_and_with_addr()
        assert len(result) == 1
        offset, _ = result[0]
        assert offset == 0  # Clamped to 0