Carlos Viegas

24 papers A* 1A 5C 1Journal 15Unranked 2
YearRankTypeTitle / Venue / Authors
2026 J jnl
Eng. Appl. Artif. Intell.
Antonio Santo, Juan José Cabrera, Carlos Viegas, David Valiente, Arturo Gil
2026 J jnl
Robotics
Tiago Pereira, Carlos Viegas, Salviano Soares, Nuno M. Fonseca Ferreira
2025 J jnl
Algorithms
Tiago Gameiro, Tiago Pereira, Hamid Moghadaspour, Francesco Di Giorgio, Carlos Viegas, Nuno M. Fonseca Ferreira, João Paulo Ferreira, Salviano Soares, António Valente
2025 J jnl
Array
Juan José Cabrera, Antonio Santo, Arturo Gil, Carlos Viegas, Luis Payá
2024 J jnl
Eng. Appl. Artif. Intell.
Jorge Pereira, Jérôme Mendes, Jorge Sampaio Silveira Junior, Carlos Viegas, João Ruivo Paulo
2024 J jnl
CoRR
Juan José Cabrera, Antonio Santo, Arturo Gil, Carlos Viegas, Luis Payá
2024 J jnl
CoRR
Antonio Santo, Juan José Cabrera, David Valiente, Carlos Viegas, Arturo Gil
2024 J jnl
Sensors
Tiago Pereira, Tiago Gameiro, José Pedro Pereira, Carlos Viegas, Nuno M. Fonseca Ferreira
2023 J jnl
Remote. Sens.
Babak Chehreh, Alexandra Moutinho, Carlos Viegas
2023 J jnl
Sensors
Tiago Pereira, Tiago Gameiro, Carlos Viegas, Victor D. N. Santos, Nuno M. F. Ferreira
2022 J jnl
Expert Syst. Appl.
Jorge Sampaio Silveira Junior, João Ruivo Paulo, Jérôme Mendes, Daniela Alves, Luís Mário Ribeiro, Carlos Viegas
2022 J jnl
J. Intell. Robotic Syst.
Carlos Viegas, Babak Chehreh, José Andrade, João Lourenço
2022 conf
ICPR Workshops (2)
Ricardo A. Ribeiro, Alexandre Bernardino, Gonçalo Cruz, Diogo Silva, Luís Félix, João Caetano, Duarte Folgado, João Francisco, Nuno Simões, Carlos Viegas, Domingos Xavier Viegas, Houda Harkat, José M. P. Nascimento
2022 J jnl
Remote. Sens.
Pegah Mohammadpour, Domingos Xavier Viegas, Carlos Viegas
2022 C conf
IECON
Jorge Pereira, Jérôme Mendes, Jorge Sampaio Silveira Junior, Carlos Viegas, João Ruivo Paulo
2019 conf
WF-IoT
Miguel Antunes, Luís Miguel Ferreira, Carlos Viegas, A. Paulo Coimbra, Aníbal T. de Almeida
2018 J jnl
J. Intell. Robotic Syst.
Mahmoud Tavakoli, Carlos Viegas, Lucio Sgrigna, Aníbal T. de Almeida
2015 A conf
IROS
Jose Carlos Romao, Mahmoud Tavakoli, Carlos Viegas, Pedro Neto, Aníbal T. de Almeida
2015 A conf
IROS
Mahmoud Tavakoli, Lucio Sgrigna, Carlos Viegas, Aníbal T. de Almeida
2015 A conf
IROS
Mahmoud Tavakoli, Carlos Viegas, Jose Carlos Romao, Pedro Neto, Aníbal T. de Almeida
2014 A conf
IROS
Carlos Viegas, Mahmoud Tavakoli
2013 A conf
IROS
Mahmoud Tavakoli, Carlos Viegas, Lino Marques, J. Norberto Pires, Aníbal T. de Almeida
2013 A* conf
ICRA
Mahmoud Tavakoli, Carlos Viegas, Lino Marques, J. Norberto Pires, Aníbal T. de Almeida
2013 J jnl
Robotics Auton. Syst.
Mahmoud Tavakoli, Carlos Viegas, Lino Marques, J. Norberto Pires, Aníbal T. de Almeida
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()