Kanglong Liu

13 papers Journal 12Unranked 1
YearRankTypeTitle / Venue / Authors
2026 J jnl
CoRR
Zhuohang Jiang, Xu Yuan, Haohao Qu, Shanru Lin, Kanglong Liu, Wenqi Fan, Qing Li
2025 J jnl
Cogn. Comput.
Pengpeng Xue, Jing Xiong, Liang Tan, Zhongzhu Liu, Kanglong Liu
2025 J jnl
Digit. Scholarsh. Humanit.
Han Xu, Jinghang Gu, Kanglong Liu, Qinyi Li
2025 J jnl
IEEE Trans. Circuits Syst. Video Technol.
Weiqing Yan, Kanglong Liu, Wujie Zhou, Chang Tang
2025 J jnl
Educ. Inf. Technol.
Simin Xu, Yanfang Su, Kanglong Liu
2025 J jnl
J. Quant. Linguistics
Ziqiao Su, Han Xu, Kanglong Liu
2024 conf
ICWL
Shuangshuang Zheng, Kanglong Liu
2024 J jnl
Glottometrics
Ho Ling Kwok, Riccardo Moratto, Kanglong Liu
2024 J jnl
Digit. Scholarsh. Humanit.
Zhongliang Wang, Andrew K. F. Cheung, Kanglong Liu
2024 J jnl
CoRR
Simin Xu, Yanfang Su, Kanglong Liu
2024 J jnl
CoRR
Lulu Wang, Simin Xu, Kanglong Liu
2024 J jnl
Appl. Artif. Intell.
Zhongliang Wang, Ming Liu, Kanglong Liu
2011 J jnl
Lit. Linguistic Comput.
Defeng Li, Chunling Zhang, Kanglong Liu
tests/integration/test_pe_extractor.py
← Index tests/integration/test_pe_extractor.py python
"""
Integration tests for the PEExtractor base class.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock

pytestmark = [pytest.mark.integration, pytest.mark.pe]


class TestPEExtractorInitialization:
    """Tests for PEExtractor initialization."""

    def test_pe_extractor_with_valid_pe(self, pe_binary_path, mock_logger):
        """Test PEExtractor initialization with valid PE file."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)

            assert extractor.pe is not None
            assert extractor.filepath == pe_binary_path

    def test_pe_extractor_with_provided_pe_object(self, pe_binary_path, mock_logger, pe_object):
        """Test PEExtractor initialization with pre-existing PE object."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger, pe=pe_object)

            # Should use the provided PE object, not create a new one
            assert extractor.pe is pe_object

    def test_pe_extractor_with_invalid_file(self, elf_binary_path, mock_logger):
        """Test PEExtractor initialization with non-PE file."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(elf_binary_path, mock_logger)

            # Should return None for pe when file is not a valid PE
            assert extractor.pe is None
            # Logger should have recorded an error
            assert mock_logger.error.called


class TestPEExtractorMethods:
    """Tests for PEExtractor helper methods."""

    def test_check_dotnet_true(self, pe_binary_path, mock_logger):
        """Test _check_dotnet returns True for .NET PE."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)

            # The test PE file is a .NET binary
            if extractor.pe is not None:
                result = extractor._check_dotnet()
                # d8637bdbcfc9112fcb1f0167b398e771 is a .NET binary
                assert result is True

    def test_check_dotnet_false(self, elf_binary_path, mock_logger):
        """Test _check_dotnet returns False for non-PE file."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(elf_binary_path, mock_logger)

            # Should return False when PE is None
            result = extractor._check_dotnet()
            assert result is False

    def test_is_signed_false(self, pe_binary_path, mock_logger):
        """Test _is_signed method."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)

            if extractor.pe is not None:
                result = extractor._is_signed()
                # Result depends on the actual binary
                assert isinstance(result, bool)

    def test_has_overlay(self, pe_binary_path, mock_logger):
        """Test _has_overlay method."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)

            if extractor.pe is not None:
                result = extractor._has_overlay()
                assert isinstance(result, bool)


class TestGeneratePEFileObject:
    """Tests for _generate_pefile_object method."""

    def test_generate_pefile_object_valid(self, pe_binary_path, mock_logger):
        """Test _generate_pefile_object with valid PE file."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)

            # The PE object should be created successfully
            assert extractor.pe is not None

    def test_generate_pefile_object_invalid(self, temp_text_file, mock_logger):
        """Test _generate_pefile_object with invalid file."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(temp_text_file, mock_logger)

            # Should return None for invalid PE
            assert extractor.pe is None


class TestGenerateDotNetFileObject:
    """Tests for _generate_dotnetfile_object method."""

    def test_generate_dotnetfile_object_valid(self, pe_binary_path, mock_logger):
        """Test _generate_dotnetfile_object with .NET PE file."""
        from redb.extractors.pe_extractors.pe_dotnet import PEDotNetExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEDotNetExtractor(pe_binary_path, mock_logger)

            if extractor.pe is not None and extractor._check_dotnet():
                dotnet, error = extractor._generate_dotnetfile_object()
                # Should create dotnet object for .NET PE
                assert dotnet is not None
                assert error is None

    def test_generate_dotnetfile_object_invalid(self, elf_binary_path, mock_logger):
        """Test _generate_dotnetfile_object with non-.NET file."""
        from redb.extractors.pe_extractors.pe_dotnet import PEDotNetExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEDotNetExtractor(elf_binary_path, mock_logger)

            dotnet, error = extractor._generate_dotnetfile_object()
            # Should return error for non-.NET file
            assert dotnet is None
            assert error is not None


class TestPEExtractorTag:
    """Tests for tag method in PE extractors."""

    def test_pe_features_tag(self, pe_binary_path, mock_logger):
        """Test that PEFeaturesExtractor returns correct tag."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor
        from redb.extractors.enum import Tag

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)
            assert extractor.tag() == Tag.PE_FEATURES.value

    def test_pe_import_tag(self, pe_binary_path, mock_logger):
        """Test that PEImportExtractor returns correct tag."""
        from redb.extractors.pe_extractors.pe_imports import PEImportExtractor
        from redb.extractors.enum import Tag

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEImportExtractor(pe_binary_path, mock_logger)
            assert extractor.tag() == Tag.PE_IMPORT.value

    def test_pe_section_tag(self, pe_binary_path, mock_logger):
        """Test that PESectionExtractor returns correct tag."""
        from redb.extractors.pe_extractors.pe_sections import PESectionExtractor
        from redb.extractors.enum import Tag

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PESectionExtractor(pe_binary_path, mock_logger)
            assert extractor.tag() == Tag.PE_SECTION.value


class TestPEExtractorClickHouseTable:
    """Tests for get_clickhouse_table method in PE extractors."""

    def test_pe_features_clickhouse_table(self, pe_binary_path, mock_logger):
        """Test that PEFeaturesExtractor returns correct ClickHouse table name."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)
            assert extractor.get_clickhouse_table() == "redb_pe_features"

    def test_pe_import_clickhouse_table(self, pe_binary_path, mock_logger):
        """Test that PEImportExtractor returns correct ClickHouse table name."""
        from redb.extractors.pe_extractors.pe_imports import PEImportExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEImportExtractor(pe_binary_path, mock_logger)
            assert extractor.get_clickhouse_table() == "redb_pe_imports"

    def test_pe_section_clickhouse_table(self, pe_binary_path, mock_logger):
        """Test that PESectionExtractor returns correct ClickHouse table name."""
        from redb.extractors.pe_extractors.pe_sections import PESectionExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PESectionExtractor(pe_binary_path, mock_logger)
            assert extractor.get_clickhouse_table() == "redb_pe_sections"


class TestPEExtractorPrepareExportData:
    """Tests for prepare_export_data method in PE extractors."""

    def test_pe_features_prepare_export_elasticsearch(self, pe_binary_path, mock_logger):
        """Test prepare_export_data for Elasticsearch."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)

            if extractor.pe is not None:
                # First extract
                extractor.extract()
                result = extractor.prepare_export_data("ElasticsearchExporter")
                # Should return the pe_features dataclass
                assert result is extractor.pe_features

    def test_pe_features_prepare_export_clickhouse(self, pe_binary_path, mock_logger):
        """Test prepare_export_data for ClickHouse."""
        from redb.extractors.pe_extractors.pe_features import PEFeaturesExtractor

        with patch('redb.settings.ELASTIC_BINARIES_COLLECTION', 'test'):
            extractor = PEFeaturesExtractor(pe_binary_path, mock_logger)

            if extractor.pe is not None:
                # First extract
                extractor.extract()
                result = extractor.prepare_export_data("ClickHouseExporter")

                # Should return tuple of (data, column_names, column_type_names)
                assert isinstance(result, tuple)
                assert len(result) == 3
                data, column_names, column_type_names = result
                assert isinstance(data, list)
                assert isinstance(column_names, list)
                assert isinstance(column_type_names, list)
                # Column names and types should match in length
                assert len(column_names) == len(column_type_names)