test_integration.py

"""
Integration tests for bitchat.

This module contains comprehensive integration tests that verify the
complete functionality of the bitchat system, including multi-node
communication, message routing, and network behavior.
"""

import pytest
import pytest_asyncio
import asyncio
from typing import List, Dict, Any, Set

from bitchat.core import BitchatNode
from bitchat.transports import Transport


class MockNetworkTransport(Transport):
    """
    Mock transport that simulates a network between nodes.
    
    This transport implementation creates a virtual network where
    multiple nodes can communicate with each other, making it suitable
    for integration testing without requiring actual network hardware.
    """
    
    def __init__(self, network_id: str = "test") -> None:
        """
        Initialize mock network transport.
        
        :param network_id: Identifier for the virtual network
        """
        self.network_id = network_id
        self._running = False
        self._receive_queue = asyncio.Queue()
        self._peer_timeout = 5.0  # Default peer timeout in seconds
        
        # Initialize network if it doesn't exist
        if network_id not in MockNetworkTransport._networks:
            MockNetworkTransport._networks[network_id] = set()
        
        self._network = MockNetworkTransport._networks[network_id]
        self._network.add(self)
        self._peers = set()  # Track connected peers
        self._peer_leave_events = []  # Store peer leave events
    
    async def start(self) -> None:
        """Start the transport and join the network."""
        if self._running:
            return
            
        self._running = True
        self._peers = set()
        self._peer_leave_events = []
        
        # Start peer monitoring task
        self._monitor_task = asyncio.create_task(self._monitor_peers())
    
    async def stop(self) -> None:
        """Stop the transport and leave the network."""
        if not self._running:
            return
            
        was_running = self._running
        self._running = False
        
        # Cancel the monitoring task
        if hasattr(self, '_monitor_task'):
            self._monitor_task.cancel()
            try:
                await self._monitor_task
            except asyncio.CancelledError:
                pass
        
        # Remove self from network and notify peers
        if self in self._network:
            self._network.remove(self)
            
        # Notify all peers about this node leaving
        if was_running:
            for peer_id, _ in list(self._peers):
                self._notify_peer_leave(peer_id)
        
        # Clear peer tracking
        self._peers.clear()
    
    def send(self, data: bytes) -> None:
        """
        Send data to all other nodes in the network.
        
        :param data: Data to send to other nodes
        """
        if not self._running:
            return
        
        # Send to all other nodes in the network
        for node in self._network:
            if node != self and node._running:
                # Update peer last seen time
                # Extract sender ID from data (first 8 bytes)
                if len(data) >= 8:
                    sender_id = data[:8]
                    node.update_peer_seen(sender_id)
                # Queue the data for the receiving node
                asyncio.create_task(node._receive_queue.put(data))
    
    async def receive(self):
        """
        Receive data from the network.
        
        :yield: Data received from other nodes
        """
        while self._running:
            try:
                data = await asyncio.wait_for(
                    self._receive_queue.get(), timeout=0.1
                )
                yield data
            except asyncio.TimeoutError:
                continue
    
    async def _monitor_peers(self) -> None:
        """Monitor peers and detect timeouts."""
        while self._running:
            try:
                # Check each peer's last seen time
                current_time = asyncio.get_event_loop().time()
                for peer_id, last_seen in list(self._peers):
                    if current_time - last_seen > self._peer_timeout:
                        self._notify_peer_leave(peer_id)
                        self._peers.discard((peer_id, last_seen))
                
                await asyncio.sleep(1)  # Check every second
            except asyncio.CancelledError:
                break
            except Exception:
                # Prevent the task from dying on unexpected errors
                await asyncio.sleep(1)
    
    def _notify_peer_leave(self, peer_id: bytes) -> None:
        """Notify about a peer leaving the network."""
        self._peer_leave_events.append(peer_id)
    
    def update_peer_seen(self, peer_id: bytes) -> None:
        """Update the last seen time for a peer."""
        if not self._running:
            return
            
        current_time = asyncio.get_event_loop().time()
        # Remove any existing entry for this peer
        self._peers = {(p, t) for p, t in self._peers if p != peer_id}
        # Add new entry with current time
        self._peers.add((peer_id, current_time))
    
    # Class variable to track all networks
    _networks: Dict[str, Set['MockNetworkTransport']] = {}


class TestIntegration:
    """
    Integration test cases.
    
    This test class covers end-to-end functionality including
    multi-node communication, message routing, and network behavior.
    """
    
    @pytest.fixture
    def transport_config(self) -> Dict[str, Any]:
        """
        Transport configuration for testing.
        
        :return: Dictionary containing transport configuration
        """
        return {
            "network_id": "test_network"
        }
    
    @pytest_asyncio.fixture
    async def nodes(self, transport_config: Dict[str, Any]) -> List[BitchatNode]:
        """
        Create test nodes for integration testing.
        
        :param transport_config: Configuration for transport instances
        :return: List of BitchatNode instances
        """
        nodes = []
        try:
            # Create two nodes
            for i in range(2):
                transport = MockNetworkTransport(**transport_config)
                node = BitchatNode(transport=transport)
                nodes.append(node)
            
            yield nodes
            
        finally:
            # Cleanup
            for node in nodes:
                if node.is_running:
                    await node.stop()
    
    @pytest_asyncio.fixture
    async def multiple_nodes(self, transport_config: Dict[str, Any]) -> List[BitchatNode]:
        """
        Create multiple test nodes for complex integration testing.
        
        :param transport_config: Configuration for transport instances
        :return: List of BitchatNode instances
        """
        nodes = []
        try:
            # Create multiple nodes
            for i in range(3):
                transport = MockNetworkTransport(**transport_config)
                node = BitchatNode(transport=transport)
                nodes.append(node)
            
            yield nodes
            
        finally:
            # Cleanup
            for node in nodes:
                if node.is_running:
                    await node.stop()
    
    async def test_basic_communication(self, nodes: List[BitchatNode]) -> None:
        """
        Test basic communication between nodes.
        """
        node1, node2 = nodes
        # Register public keys
        for n1 in nodes:
            for n2 in nodes:
                if n1 is not n2:
                    peer_id = bytes(n2.crypto.verify_key)[:8]
                    n1.crypto.add_peer(peer_id, bytes(n2.crypto.verify_key), bytes(n2.crypto.box_public_key))
        received_messages = []
        def on_message(source: bytes, text: str, private: bool) -> None:
            received_messages.append((source, text, private))
        node2.on("message", on_message)
        await node1.start()
        await node2.start()
        await asyncio.sleep(1)
        test_message = "Hello from node1!"
        node1.send_broadcast(test_message)
        await asyncio.sleep(2)
        assert len(received_messages) > 0
    
    async def test_multiple_messages(self, nodes: List[BitchatNode]) -> None:
        """
        Test sending multiple messages.
        
        Verifies that multiple messages can be sent and received
        in sequence.
        """
        for n1 in nodes:
            for n2 in nodes:
                if n1 is not n2:
                    peer_id = bytes(n2.crypto.verify_key)[:8]
                    n1.crypto.add_peer(peer_id, bytes(n2.crypto.verify_key), bytes(n2.crypto.box_public_key))
        node1, node2 = nodes
        received_messages = []
        
        def on_message(source: bytes, text: str, private: bool) -> None:
            received_messages.append(text)
        
        node2.on("message", on_message)
        
        await node1.start()
        await node2.start()
        await asyncio.sleep(1)
        
        # Act - Send multiple messages
        messages = ["Message 1", "Message 2", "Message 3"]
        for msg in messages:
            node1.send_broadcast(msg)
            await asyncio.sleep(0.5)
        
        # Wait for all messages
        await asyncio.sleep(2)
        
        # Assert - Verify all messages were received
        for msg in messages:
            assert msg in received_messages
    
    async def test_peer_events(self, nodes: List[BitchatNode]) -> None:
        """
        Test peer join/leave events.
        
        Verifies that peer join and leave events are properly
        emitted when nodes start and stop.
        """
        # Arrange
        node1, node2 = nodes
        peer_events = []
        
        def on_peer_join(peer_id: bytes) -> None:
            peer_events.append(("join", peer_id))
        
        def on_peer_leave(peer_id: bytes) -> None:
            peer_events.append(("leave", peer_id))
        
        node2.on("peer_join", on_peer_join)
        node2.on("peer_leave", on_peer_leave)
        
        # Act - Start first node
        await node1.start()
        await asyncio.sleep(1)
        
        # Act - Start second node
        await node2.start()
        await asyncio.sleep(2)
        
        # Act - Stop first node
        await node1.stop()
        await asyncio.sleep(1)
        
        # Assert - Verify peer events
        assert len(peer_events) > 0
        assert any(event[0] == "join" for event in peer_events)
    
    async def test_error_handling(self, nodes: List[BitchatNode]) -> None:
        """
        Test error handling in network communication.
        
        Verifies that errors are properly handled and don't crash
        the network communication.
        """
        # Arrange
        node1, node2 = nodes
        errors = []
        
        def on_error(error: Exception) -> None:
            errors.append(error)
        
        node1.on("error", on_error)
        node2.on("error", on_error)
        
        await node1.start()
        await node2.start()
        await asyncio.sleep(1)
        
        # Act - Send invalid data to trigger error
        # This would require access to the transport layer
        # For now, just verify error handler is set up
        assert len(errors) >= 0  # Should not crash
    
    async def test_node_lifecycle(self, nodes: List[BitchatNode]) -> None:
        """
        Test complete node lifecycle.
        
        Verifies that nodes can be started, communicate, and stopped
        without issues.
        """
        for n1 in nodes:
            for n2 in nodes:
                if n1 is not n2:
                    peer_id = bytes(n2.crypto.verify_key)[:8]
                    n1.crypto.add_peer(peer_id, bytes(n2.crypto.verify_key), bytes(n2.crypto.box_public_key))
        node1, node2 = nodes
        
        # Act - Test start
        await node1.start()
        assert node1.is_running
        
        await node2.start()
        assert node2.is_running
        
        # Act - Test communication
        received = []
        node2.on("message", lambda src, text, priv: received.append(text))
        
        node1.send_broadcast("Lifecycle test")
        await asyncio.sleep(2)
        
        # Assert - Verify communication worked
        assert len(received) > 0
        
        # Act - Test stop
        await node1.stop()
        assert not node1.is_running
        
        await node2.stop()
        assert not node2.is_running
    
    async def test_concurrent_nodes(self, multiple_nodes: List[BitchatNode]) -> None:
        """
        Test multiple nodes running concurrently.
        
        Verifies that multiple nodes can run simultaneously and
        communicate with each other.
        """
        for n1 in multiple_nodes:
            for n2 in multiple_nodes:
                if n1 is not n2:
                    peer_id = bytes(n2.crypto.verify_key)[:8]
                    n1.crypto.add_peer(peer_id, bytes(n2.crypto.verify_key), bytes(n2.crypto.box_public_key))
        nodes = multiple_nodes
        
        # Act - Start all nodes
        start_tasks = [node.start() for node in nodes]
        await asyncio.gather(*start_tasks)
        
        # Assert - Verify all are running
        for node in nodes:
            assert node.is_running
        
        # Act - Send messages from each node
        received_messages = {i: [] for i in range(len(nodes))}
        
        for i, node in enumerate(nodes):
            def make_handler(node_id: int):
                def handler(source: bytes, text: str, private: bool) -> None:
                    received_messages[node_id].append(text)
                return handler
            
            node.on("message", make_handler(i))
        
        # Send messages
        for i, node in enumerate(nodes):
            node.send_broadcast(f"Message from node {i}")
        
        # Wait for messages to propagate
        await asyncio.sleep(3)
        
        # Assert - Verify messages were received
        for i in range(len(nodes)):
            assert len(received_messages[i]) > 0
    
    async def test_network_isolation(self, transport_config: Dict[str, Any]) -> None:
        """
        Test network isolation between different networks.
        
        Verifies that nodes in different networks cannot communicate
        with each other.
        """
        # Arrange
        network1_config = {**transport_config, "network_id": "network1"}
        network2_config = {**transport_config, "network_id": "network2"}
        
        node1 = BitchatNode(transport=MockNetworkTransport(**network1_config))
        node2 = BitchatNode(transport=MockNetworkTransport(**network2_config))
        
        received_messages = []
        
        def on_message(source: bytes, text: str, private: bool) -> None:
            received_messages.append(text)
        
        node2.on("message", on_message)
        
        try:
            # Act - Start nodes
            await node1.start()
            await node2.start()
            await asyncio.sleep(1)
            
            # Send message from node1
            node1.send_broadcast("Isolation test")
            await asyncio.sleep(2)
            
            # Assert - Verify message was not received (different networks)
            assert len(received_messages) == 0
            
        finally:
            # Cleanup
            await node1.stop()
            await node2.stop()
    
    async def test_message_ordering(self, nodes: List[BitchatNode]) -> None:
        """
        Test message ordering in the network.
        
        Verifies that messages are received in a reasonable order
        (though exact ordering may not be guaranteed in a distributed system).
        """
        for n1 in nodes:
            for n2 in nodes:
                if n1 is not n2:
                    peer_id = bytes(n2.crypto.verify_key)[:8]
                    n1.crypto.add_peer(peer_id, bytes(n2.crypto.verify_key), bytes(n2.crypto.box_public_key))
        node1, node2 = nodes
        received_messages = []
        
        def on_message(source: bytes, text: str, private: bool) -> None:
            received_messages.append(text)
        
        node2.on("message", on_message)
        
        await node1.start()
        await node2.start()
        await asyncio.sleep(1)
        
        # Act - Send messages in sequence
        messages = [f"Message {i}" for i in range(5)]
        for msg in messages:
            node1.send_broadcast(msg)
            await asyncio.sleep(0.1)
        
        # Wait for all messages
        await asyncio.sleep(2)
        
        # Assert - Verify all messages were received
        assert len(received_messages) >= len(messages)
        for msg in messages:
            assert msg in received_messages
    
    async def test_network_stability(self, multiple_nodes: List[BitchatNode]) -> None:
        """
        Test network stability under load.
        
        Verifies that the network remains stable when multiple nodes
        send messages simultaneously.
        """
        for n1 in multiple_nodes:
            for n2 in multiple_nodes:
                if n1 is not n2:
                    peer_id = bytes(n2.crypto.verify_key)[:8]
                    n1.crypto.add_peer(peer_id, bytes(n2.crypto.verify_key), bytes(n2.crypto.box_public_key))
        nodes = multiple_nodes
        message_counts = {i: 0 for i in range(len(nodes))}
        
        # Set up message handlers
        for i, node in enumerate(nodes):
            def make_handler(node_id: int):
                def handler(source: bytes, text: str, private: bool) -> None:
                    message_counts[node_id] += 1
                return handler
            node.on("message", make_handler(i))
        
        # Start all nodes
        await asyncio.gather(*[node.start() for node in nodes])
        await asyncio.sleep(1)
        
        # Act - Send messages from all nodes simultaneously
        async def send_messages(node, i):
            for j in range(3):
                await asyncio.sleep(j * 0.1)
                node.send_broadcast(f"Msg {i}-{j}")
        send_tasks = [send_messages(node, i) for i, node in enumerate(nodes)]
        await asyncio.gather(*send_tasks)
        await asyncio.sleep(3)  # Wait for message propagation
        
        # Assert - Verify all nodes received some messages
        total_messages = sum(message_counts.values())
        assert total_messages > 0
        
        # Verify network is still stable
        for node in nodes:
            assert node.is_running
    
    async def test_graceful_shutdown(self, nodes: List[BitchatNode]) -> None:
        """
        Test graceful shutdown of nodes.
        
        Verifies that nodes can be shut down gracefully without
        causing issues for other nodes in the network.
        """
        for n1 in nodes:
            for n2 in nodes:
                if n1 is not n2:
                    peer_id = bytes(n2.crypto.verify_key)[:8]
                    n1.crypto.add_peer(peer_id, bytes(n2.crypto.verify_key), bytes(n2.crypto.box_public_key))
        node1, node2 = nodes
        shutdown_events = []
        
        def on_peer_leave(peer_id: bytes) -> None:
            shutdown_events.append(peer_id)
        
        node2.on("peer_leave", on_peer_leave)
        
        await node1.start()
        await node2.start()
        await asyncio.sleep(1)
        
        # Act - Gracefully shutdown node1
        await node1.stop()
        await asyncio.sleep(6)  # Wait longer than peer_timeout (5.0s) for event propagation
        
        # Assert - Verify node2 detected the shutdown
        assert len(shutdown_events) > 0
        assert node2.is_running  # Node2 should still be running
        
        # Act - Shutdown node2
        await node2.stop()
        assert not node2.is_running