# -*- coding: utf-8 -*- """ Tests für Network RPC Correlation. """ import asyncio import pytest from datetime import datetime, timedelta from trixy_core.network.rpc import ( CorrelationManager, CorrelationConfig, PendingRequest, RequestState, ) class TestCorrelationConfig: """Tests für CorrelationConfig.""" def test_default_config(self): """Testet Standard-Konfiguration.""" config = CorrelationConfig() assert config.default_timeout == 30.0 assert config.cleanup_interval == 60.0 assert config.max_pending == 10000 def test_custom_config(self): """Testet benutzerdefinierte Konfiguration.""" config = CorrelationConfig( default_timeout=60.0, max_pending=5000, ) assert config.default_timeout == 60.0 assert config.max_pending == 5000 def test_custom_id_generator(self): """Testet benutzerdefinierten ID-Generator.""" counter = [0] def custom_generator(): counter[0] += 1 return f"custom-{counter[0]}" config = CorrelationConfig(id_generator=custom_generator) assert config.id_generator is not None class TestPendingRequest: """Tests für PendingRequest.""" def test_request_creation(self): """Testet Request-Erstellung.""" request = PendingRequest[str]( correlation_id="test-123", timeout=30.0, ) assert request.correlation_id == "test-123" assert request.state == RequestState.PENDING assert not request.is_complete def test_is_complete_states(self): """Testet is_complete für verschiedene States.""" request = PendingRequest[str](correlation_id="test") request.state = RequestState.PENDING assert not request.is_complete request.state = RequestState.COMPLETED assert request.is_complete request.state = RequestState.TIMEOUT assert request.is_complete request.state = RequestState.CANCELLED assert request.is_complete request.state = RequestState.ERROR assert request.is_complete def test_elapsed(self): """Testet Zeitberechnung.""" request = PendingRequest[str]( correlation_id="test", created_at=datetime.now() - timedelta(seconds=5), ) assert request.elapsed_seconds >= 5 def test_is_expired(self): """Testet Ablaufprüfung.""" request = PendingRequest[str]( correlation_id="test", timeout=1.0, created_at=datetime.now() - timedelta(seconds=2), ) assert request.is_expired() request2 = PendingRequest[str]( correlation_id="test", timeout=10.0, ) assert not request2.is_expired() class TestCorrelationManager: """Tests für CorrelationManager.""" @pytest.fixture def config(self): """Erstellt Test-Konfiguration.""" return CorrelationConfig( default_timeout=1.0, max_pending=100, ) @pytest.fixture def manager(self, config): """Erstellt CorrelationManager.""" return CorrelationManager[str](config=config) @pytest.mark.asyncio async def test_create_request(self, manager): """Testet Request-Erstellung.""" cid = await manager.create_request() assert cid is not None assert len(cid) > 0 assert manager.pending_count == 1 assert cid in manager @pytest.mark.asyncio async def test_create_request_with_metadata(self, manager): """Testet Request-Erstellung mit Metadaten.""" cid = await manager.create_request( timeout=60.0, metadata={"user": "test"}, ) request = manager.get(cid) assert request.timeout == 60.0 assert request.metadata["user"] == "test" @pytest.mark.asyncio async def test_complete_request(self, manager): """Testet Request-Abschluss.""" cid = await manager.create_request() # Task für wait_for_response async def wait_task(): return await manager.wait_for_response(cid) task = asyncio.create_task(wait_task()) # Kurz warten dann abschließen await asyncio.sleep(0.01) assert manager.complete(cid, "result") result = await task assert result == "result" @pytest.mark.asyncio async def test_wait_for_response_timeout(self, manager): """Testet Timeout beim Warten.""" cid = await manager.create_request(timeout=0.1) with pytest.raises(TimeoutError): await manager.wait_for_response(cid) request = manager.get(cid) assert request.state == RequestState.TIMEOUT @pytest.mark.asyncio async def test_fail_request(self, manager): """Testet Request-Fehler.""" cid = await manager.create_request() async def wait_task(): return await manager.wait_for_response(cid) task = asyncio.create_task(wait_task()) await asyncio.sleep(0.01) error = ValueError("Test error") assert manager.fail(cid, error) with pytest.raises(ValueError): await task @pytest.mark.asyncio async def test_cancel_request(self, manager): """Testet Request-Abbruch.""" cid = await manager.create_request() assert manager.cancel(cid) request = manager.get(cid) assert request.state == RequestState.CANCELLED @pytest.mark.asyncio async def test_complete_unknown_request(self, manager): """Testet Abschluss unbekannter Request.""" assert not manager.complete("unknown-id", "result") @pytest.mark.asyncio async def test_complete_already_complete(self, manager): """Testet doppelten Abschluss.""" cid = await manager.create_request() # Simuliere Abschluss request = manager.get(cid) request.state = RequestState.COMPLETED assert not manager.complete(cid, "result") @pytest.mark.asyncio async def test_remove_request(self, manager): """Testet Request-Entfernung.""" cid = await manager.create_request() assert manager.pending_count == 1 removed = await manager.remove(cid) assert removed is not None assert manager.pending_count == 0 @pytest.mark.asyncio async def test_has_pending(self, manager): """Testet Pending-Prüfung.""" cid = await manager.create_request() assert manager.has_pending(cid) assert not manager.has_pending("unknown") @pytest.mark.asyncio async def test_cleanup_expired(self, manager): """Testet Cleanup abgelaufener Requests.""" # Erstelle Request mit kurzem Timeout cid = await manager.create_request(timeout=0.01) # Warte bis abgelaufen await asyncio.sleep(0.05) removed = await manager.cleanup_expired() assert removed == 1 assert manager.pending_count == 0 @pytest.mark.asyncio async def test_cancel_all(self, manager): """Testet Abbruch aller Requests.""" await manager.create_request() await manager.create_request() await manager.create_request() cancelled = await manager.cancel_all() assert cancelled == 3 @pytest.mark.asyncio async def test_get_pending_ids(self, manager): """Testet Abruf aller Pending-IDs.""" cid1 = await manager.create_request() cid2 = await manager.create_request() ids = manager.get_pending_ids() assert cid1 in ids assert cid2 in ids @pytest.mark.asyncio async def test_get_stats(self, manager): """Testet Statistik-Abruf.""" await manager.create_request() await manager.create_request() stats = manager.get_stats() assert stats["total_pending"] == 2 assert "states" in stats @pytest.mark.asyncio async def test_max_pending_limit(self, config): """Testet Maximum-Limit.""" config.max_pending = 3 manager = CorrelationManager[str](config=config) await manager.create_request() await manager.create_request() await manager.create_request() with pytest.raises(RuntimeError): await manager.create_request() @pytest.mark.asyncio async def test_unique_ids(self, manager): """Testet ID-Eindeutigkeit.""" ids = set() for _ in range(100): cid = await manager.create_request() assert cid not in ids ids.add(cid) @pytest.mark.asyncio async def test_custom_id_generator(self): """Testet benutzerdefinierten ID-Generator.""" counter = [0] def generator(): counter[0] += 1 return f"custom-{counter[0]}" config = CorrelationConfig(id_generator=generator) manager = CorrelationManager[str](config=config) cid1 = await manager.create_request() cid2 = await manager.create_request() assert cid1 == "custom-1" assert cid2 == "custom-2" @pytest.mark.asyncio async def test_cleanup_task(self, manager): """Testet Cleanup-Task.""" await manager.start_cleanup() assert manager._cleanup_task is not None await manager.stop_cleanup() assert manager._cleanup_task is None @pytest.mark.asyncio async def test_len_and_contains(self, manager): """Testet __len__ und __contains__.""" assert len(manager) == 0 cid = await manager.create_request() assert len(manager) == 1 assert cid in manager class TestRequestState: """Tests für RequestState.""" def test_state_values(self): """Testet State-Werte.""" assert RequestState.PENDING.name == "PENDING" assert RequestState.COMPLETED.name == "COMPLETED" assert RequestState.TIMEOUT.name == "TIMEOUT" assert RequestState.CANCELLED.name == "CANCELLED" assert RequestState.ERROR.name == "ERROR"