James R. Kelly

16 papers C 1Journal 13Unranked 2
YearRankTypeTitle / Venue / Authors
2026 J jnl
Innovative Reconfiguration Technologies for Antenna Systems: A Comprehensive Review.
IEEE Access
Mirko Magarotto, Alejandro Jiménez-Sáez, James R. Kelly, Antonio-Daniele Capobianco
2025 J jnl
Beam-Steerable Fabry-Pérot Cavity Antenna with Reconfigurable Liquid-Based PRS.
IEEE Access
Syed Tamoor Shah, Muhammad Umar Khan, Rifaqat Hussain, James R. Kelly, Akram Alomainy, Khaled A. Aljaloud
2024 — conf
Circuits and Antennas Reconfigured using Gallium-based Liquid Metal.
NCC
James R. Kelly, Shaker Alkaraki
2023 J jnl
Circuits and Antennas Incorporating Gallium-Based Liquid Metal.
Proc. IEEE
Yi-Wen Wu, Shaker Alkaraki, Shi-Yang Tang, Yi Wang, James R. Kelly
2023 J jnl
Continuous Beam Steering Realized by Tunable Ground in a Patch Antenna.
IEEE Access
Zhishu Qu, Yihua Zhou, Shaker Alkaraki, James R. Kelly, Yue Gao
2023 J jnl
Phased Array Antenna System Enabled by Liquid Metal Phase Shifters.
IEEE Access
Shaker Alkaraki, Quan Wei Lin, James R. Kelly, Zhengpeng Wang, Hang Wong
2023 — conf
The Impact of a Haptic Digital Twin in the Nuclear Industry and Potential Applications.
ARSO
Harun Tugal, Fumiaki Abe, Ipek Caliskanelli, Alice Cryer, Chris Hope, James R. Kelly, Salvador Pacheco Gutierrez, Alexandros Plianos, Masaki Sakamoto, Tomoki Sakaue, Wataru Sato, Shu Shirai, Yolande Smith, Yoshimasa Sugawara, Kaiqiang Zhang, Robert Skilton
2021 J jnl
Hybrid Parasitic Linear Array Antenna for Fine Beamsteering Applications.
IEEE Access
Iyemeh E. Uchendu, James R. Kelly, Raj Mittra, Yue Gao
2021 J jnl
Joint Iterative Optimization-Based Low-Complexity Adaptive Hybrid Beamforming for Massive MU-MIMO Systems.
IEEE Trans. Commun.
Hang Ruan, Pei Xiao, Lixia Xiao, James R. Kelly
2021 J jnl
Secrecy Rate Optimization for Intelligent Reflecting Surface Assisted MIMO System.
IEEE Trans. Inf. Forensics Secur.
Zheng Chu, Wanming Hao, Pei Xiao, De Mi, Zilong Liu, Mohsen Khalily, James R. Kelly, Alexandros P. Feresidis
2019 J jnl
A Technique for MIMO Antenna Design With Flexible Element Number and Pattern Diversity.
IEEE Access
Yasin Kabiri, Alejandro L. Borja, James R. Kelly, Pei Xiao
2017 J jnl
Full-Duplex Wireless-Powered Relay in Two Way Cooperative Networks.
IEEE Access
Gaojie Chen, Pei Xiao, James R. Kelly, Bing Li, Rahim Tafazolli
2017 C conf
Resource allocation and interference management in hybrid millimeter wave networks.
ICCC
Jia Shi, Pei Xiao, James R. Kelly, Jiangbo Si
2017 J jnl
Wide and multi-band reconfigurable Vivaldi antenna with slot-line feed.
Telecommun. Syst.
Sahar Chagharvand, Mohamad Rijal Bin Hamid, Muhammad Ramlee Kamarudin, James R. Kelly
2016 J jnl
Wide-to-narrowband reconfigurable Vivaldi antenna using switched-feed technique.
Telecommun. Syst.
Sahar Chagharvand, Mohamad Rijal Bin Hamid, Muhammad Ramlee Kamarudin, James R. Kelly
2014 J jnl
Reconfigurable Parallel Coupled Band Notch Resonator With Wide Tuning Range.
IEEE Trans. Ind. Electron.
Zhengpeng Wang, James R. Kelly, Peter S. Hall, Alejandro L. Borja, Peter Gardner
tests/unit/test_database_exporters.py
← Index tests/unit/test_database_exporters.py python 
"""
Unit tests for database exporters.
"""
import pytest
import json
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from unittest.mock import Mock, patch, MagicMock

pytestmark = [pytest.mark.unit, pytest.mark.exporters]


# ============================================================================
# Sample Dataclass for Testing
# ============================================================================

@dataclass
class SampleData:
    """Sample dataclass for export testing."""
    sha256: str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    name: str = "test_sample"
    value: int = 42
    _id: str = "test_id"


# ============================================================================
# PrintExporter Tests
# ============================================================================

class TestPrintExporter:
    """Tests for PrintExporter class."""

    def test_initialization(self, mock_logger):
        """Test PrintExporter initialization."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")

        assert exporter.index_prefix == "test_index"
        assert exporter.log is mock_logger

    def test_export_dataclass(self, mock_logger, capsys):
        """Test exporting a dataclass."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")
        sample = SampleData()

        result = exporter.export(sample)

        assert result is True
        captured = capsys.readouterr()
        assert "test_sample" in captured.out

    def test_export_list_of_dataclasses(self, mock_logger, capsys):
        """Test exporting a list of dataclasses."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")
        samples = [SampleData(name="sample1"), SampleData(name="sample2")]

        result = exporter.export(samples)

        assert result is True
        captured = capsys.readouterr()
        assert "sample1" in captured.out
        assert "sample2" in captured.out

    def test_export_dict(self, mock_logger, capsys):
        """Test exporting a dictionary."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")
        data = {"key": "value", "number": 42}

        result = exporter.export(data)

        assert result is True
        captured = capsys.readouterr()
        assert "key" in captured.out
        assert "value" in captured.out

    def test_export_none(self, mock_logger):
        """Test exporting None."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")

        result = exporter.export(None)

        assert result is True

    def test_export_empty_list(self, mock_logger):
        """Test exporting empty list."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")

        result = exporter.export([])

        assert result is True


# ============================================================================
# ElasticsearchExporter Tests
# ============================================================================

class TestElasticsearchExporter:
    """Tests for ElasticsearchExporter class."""

    def test_initialization(self, mock_logger):
        """Test ElasticsearchExporter initialization."""
        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")

            assert exporter.index_prefix == "test_index"
            assert exporter.log is mock_logger
            assert exporter.client is not None

    def test_export_dataclass(self, mock_logger):
        """Test exporting a dataclass to Elasticsearch."""
        with patch('redb.settings.get_elasticsearch_client') as mock_get_client:
            mock_client = Mock()
            mock_get_client.return_value = mock_client

            from redb.extractors.database_exporters import ElasticsearchExporter

            exporter = ElasticsearchExporter(mock_logger, "test_index")
            sample = SampleData()

            result = exporter.export(
                sample,
                index="test_index",
                tag="test_tag",
                hashes={"sha256": sample.sha256}
            )

            assert result is True
            mock_client.index.assert_called()

    def test_export_list(self, mock_logger):
        """Test exporting a list to Elasticsearch."""
        with patch('redb.settings.get_elasticsearch_client') as mock_get_client:
            mock_client = Mock()
            mock_get_client.return_value = mock_client

            from redb.extractors.database_exporters import ElasticsearchExporter

            exporter = ElasticsearchExporter(mock_logger, "test_index")
            samples = [SampleData(name="sample1"), SampleData(name="sample2")]

            result = exporter.export(
                samples,
                index="test_index",
                tag="test_tag",
                hashes={"sha256": "test_hash"}
            )

            assert result is True
            # Should be called twice (once per sample)
            assert mock_client.index.call_count == 2

    def test_export_failure(self, mock_logger):
        """Test handling export failure."""
        with patch('redb.settings.get_elasticsearch_client') as mock_get_client:
            mock_client = Mock()
            mock_client.index.side_effect = Exception("Connection failed")
            mock_get_client.return_value = mock_client

            from redb.extractors.database_exporters import ElasticsearchExporter

            exporter = ElasticsearchExporter(mock_logger, "test_index")
            sample = SampleData()

            result = exporter.export(
                sample,
                index="test_index",
                tag="test_tag",
                hashes={"sha256": sample.sha256}
            )

            assert result is False
            mock_logger.error.assert_called()

    def test_export_with_custom_id(self, mock_logger):
        """Test exporting with custom _id field."""
        with patch('redb.settings.get_elasticsearch_client') as mock_get_client:
            mock_client = Mock()
            mock_get_client.return_value = mock_client

            from redb.extractors.database_exporters import ElasticsearchExporter

            exporter = ElasticsearchExporter(mock_logger, "test_index")
            sample = SampleData(_id="custom_id")

            result = exporter.export(
                sample,
                index="test_index",
                tag="test_tag",
                hashes={"sha256": sample.sha256}
            )

            assert result is True


# ============================================================================
# ClickHouseExporter Tests
# ============================================================================

class TestClickHouseExporter:
    """Tests for ClickHouseExporter class."""

    def test_initialization(self, mock_logger):
        """Test ClickHouseExporter initialization."""
        from redb.extractors.database_exporters import ClickHouseExporter

        mock_client = Mock()
        exporter = ClickHouseExporter(mock_logger, "test_index", client=mock_client)

        assert exporter.index_prefix == "test_index"
        assert exporter.log is mock_logger
        assert exporter.client is mock_client

    def test_export_tuple_data(self, mock_logger):
        """Test exporting tuple data to ClickHouse."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_client.insert.return_value = True
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            data = [["hash123", "sample.exe", 1024]]
            column_names = ["sha256", "name", "size"]
            column_type_names = ["String", "String", "UInt64"]

            result = exporter.export(
                (data, column_names, column_type_names),
                table="test_table"
            )

            assert result is True

    def test_export_multi_table(self, mock_logger):
        """Test exporting multi-table data to ClickHouse."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_client.insert.return_value = True
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            data = {
                'multi_table': True,
                'table1': {
                    'table': 'test_table1',
                    'data': [["hash123", "sample.exe"]],
                    'column_names': ["sha256", "name"],
                    'column_type_names': ["String", "String"]
                }
            }

            result = exporter.export(data)

            assert result is True

    def test_export_failure(self, mock_logger):
        """Test handling export failure."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_client.insert.side_effect = Exception("Connection failed")
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            data = [["hash123", "sample.exe", 1024]]
            column_names = ["sha256", "name", "size"]
            column_type_names = ["String", "String", "UInt64"]

            result = exporter.export(
                (data, column_names, column_type_names),
                table="test_table"
            )

            assert result is False

    def test_export_missing_table(self, mock_logger):
        """Test export without table name raises error."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            data = [["hash123", "sample.exe", 1024]]
            column_names = ["sha256", "name", "size"]
            column_type_names = ["String", "String", "UInt64"]

            result = exporter.export(
                (data, column_names, column_type_names)
                # Missing table parameter
            )

            assert result is False

    def test_get_client(self, mock_logger):
        """Test get_client method."""
        from redb.extractors.database_exporters import ClickHouseExporter

        mock_client = Mock()
        exporter = ClickHouseExporter(mock_logger, "test_index", client=mock_client)

        result = exporter.get_client()

        assert result is mock_client

    def test_get_client_creates_new(self, mock_logger):
        """Test get_client creates new client if none exists."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_new_client = Mock()
            mock_create.return_value = mock_new_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            result = exporter.get_client()

            assert result is mock_new_client


# ============================================================================
# DatabaseExporter Abstract Base Class Tests
# ============================================================================

class TestDatabaseExporter:
    """Tests for DatabaseExporter abstract base class."""

    def test_cannot_instantiate_abstract(self):
        """Test that DatabaseExporter cannot be instantiated directly."""
        from redb.extractors.database_exporters import DatabaseExporter

        with pytest.raises(TypeError):
            DatabaseExporter()

    def test_subclass_must_implement_export(self):
        """Test that subclasses must implement export method."""
        from redb.extractors.database_exporters import DatabaseExporter

        # Define incomplete subclass
        class IncompleteExporter(DatabaseExporter):
            pass

        with pytest.raises(TypeError):
            IncompleteExporter()


# ============================================================================
# Null/Edge Case Tests
# ============================================================================

class TestExporterEdgeCases:
    """Tests for edge cases in exporters."""

    def test_print_exporter_with_complex_nested_data(self, mock_logger, capsys):
        """Test PrintExporter with complex nested data."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")
        data = {
            "level1": {
                "level2": {
                    "level3": [1, 2, 3]
                }
            },
            "array": [{"a": 1}, {"b": 2}]
        }

        result = exporter.export(data)

        assert result is True
        captured = capsys.readouterr()
        assert "level1" in captured.out

    def test_clickhouse_null_handling(self, mock_logger):
        """Test ClickHouse exporter handles null values."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_client.insert.return_value = True
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            # Multi-table data with null values
            data = {
                'multi_table': True,
                'table1': {
                    'table': 'test_table',
                    'data': [[None, "sample.exe", None]],
                    'column_names': ["sha256", "name", "size"],
                    'column_type_names': ["Nullable(String)", "String", "Nullable(UInt64)"]
                }
            }

            result = exporter.export(data)

            # Should handle nulls appropriately
            assert result is True

    def test_clickhouse_array_handling(self, mock_logger):
        """Test ClickHouse exporter handles array types."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_client.insert.return_value = True
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            data = {
                'multi_table': True,
                'table1': {
                    'table': 'test_table',
                    'data': [["hash", ["tag1", "tag2"]]],
                    'column_names': ["sha256", "tags"],
                    'column_type_names': ["String", "Array(String)"]
                }
            }

            result = exporter.export(data)

            assert result is True


# ============================================================================
# Type Conversion Tests
# ============================================================================

class TestExporterTypeConversions:
    """Tests for type conversions in exporters."""

    def test_clickhouse_datetime_handling(self, mock_logger):
        """Test ClickHouse exporter handles datetime values."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_client.insert.return_value = True
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            now = datetime.now(timezone.utc)
            data = {
                'multi_table': True,
                'table1': {
                    'table': 'test_table',
                    'data': [["hash", now]],
                    'column_names': ["sha256", "timestamp"],
                    'column_type_names': ["String", "DateTime64(3, 'UTC')"]
                }
            }

            result = exporter.export(data)

            assert result is True

    def test_print_exporter_datetime_serialization(self, mock_logger, capsys):
        """Test PrintExporter handles datetime serialization."""
        from redb.extractors.database_exporters import PrintExporter

        exporter = PrintExporter(mock_logger, "test_index")
        now = datetime.now(timezone.utc)
        data = {"timestamp": now, "name": "test"}

        result = exporter.export(data)

        assert result is True
        captured = capsys.readouterr()
        assert "timestamp" in captured.out


# ============================================================================
# Connection Context Manager Tests
# ============================================================================

class TestClickHouseConnectionManager:
    """Tests for ClickHouse connection context manager."""

    def test_clickhouse_connection_context(self, mock_logger):
        """Test clickhouse_connection context manager."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            with exporter.clickhouse_connection() as client:
                assert client is mock_client

            # Client should be closed after context exits
            mock_client.close.assert_called()

    def test_clickhouse_connection_close_on_error(self, mock_logger):
        """Test client is closed even if operation fails."""
        with patch('redb.settings.create_clickhouse_client') as mock_create:
            mock_client = Mock()
            mock_create.return_value = mock_client

            from redb.extractors.database_exporters import ClickHouseExporter

            exporter = ClickHouseExporter(mock_logger, "test_index")

            try:
                with exporter.clickhouse_connection() as client:
                    raise ValueError("Test error")
            except ValueError:
                pass

            # Client should still be closed
            mock_client.close.assert_called()