diff --git a/pyproject.toml b/pyproject.toml index 2bba0d65..152eaca5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -94,6 +94,11 @@ asyncio_default_fixture_loop_scope = "session" addopts = "--strict-markers -n auto --ignore=tests/alembic" testpaths = ["tests"] pythonpath = ["src"] +filterwarnings = [ + "ignore:Call to deprecated close\\. \\(Use aclose\\(\\) instead\\).*:DeprecationWarning", + "ignore:websockets\\.legacy is deprecated; see .* for upgrade instructions:DeprecationWarning", + "ignore:websockets\\.server\\.WebSocketServerProtocol is deprecated:DeprecationWarning", +] [tool.coverage.report] exclude_lines = [ diff --git a/src/deriver/__main__.py b/src/deriver/__main__.py index 778f1803..e7934678 100644 --- a/src/deriver/__main__.py +++ b/src/deriver/__main__.py @@ -10,11 +10,13 @@ from src.telemetry import initialize_telemetry_async, shutdown_telemetry from .queue_manager import main +logger = logging.getLogger(__name__) + def start_metrics_server() -> None: """Start the Prometheus metrics HTTP server on port 9090.""" start_http_server(9090) - print("[DERIVER] Prometheus metrics server started on port 9090") + logger.info("Prometheus metrics server started on port 9090") def setup_logging(): @@ -63,10 +65,9 @@ async def run_deriver(): if __name__ == "__main__": - print("[DERIVER] Starting deriver queue processor") - # Setup logging before starting the main loop setup_logging() + logger.info("Starting deriver queue processor") asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) try: @@ -74,11 +75,11 @@ if __name__ == "__main__": if settings.METRICS.ENABLED: start_metrics_server() - print("[DERIVER] Running main loop") + logger.info("Running main loop") asyncio.run(run_deriver()) except KeyboardInterrupt: - print("[DERIVER] Shutdown initiated via KeyboardInterrupt") + logger.info("Shutdown initiated via KeyboardInterrupt") except Exception as e: - print(f"[DERIVER] Error in main process: {str(e)}") + logger.exception("Error in main process: %s", e) finally: - print("[DERIVER] Deriver process exiting") + logger.info("Deriver process exiting") diff --git a/src/main.py b/src/main.py index 4d5ba401..c5c2aa24 100644 --- a/src/main.py +++ b/src/main.py @@ -28,7 +28,6 @@ from src.routers import ( webhooks, workspaces, ) -from src.security import create_admin_jwt from src.telemetry import ( initialize_telemetry_async, metrics_endpoint, @@ -75,12 +74,6 @@ logger = logging.getLogger(__name__) logging.getLogger("cashews.backends.redis.client").setLevel(logging.CRITICAL) -# JWT Setup -async def setup_admin_jwt(): - token = create_admin_jwt() - print(f"\n ADMIN JWT: {token}\n") - - def before_send(event: "Event", hint: "Hint | None") -> "Event | None": """Filter out events raised from known non-actionable exceptions before Sentry sees them.""" if not hint: diff --git a/src/security.py b/src/security.py index 46dd2dce..6988919e 100644 --- a/src/security.py +++ b/src/security.py @@ -5,7 +5,7 @@ from typing import Annotated import jwt from fastapi import Depends, Request from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer -from pydantic import BaseModel +from pydantic import BaseModel, Field from src.config import settings from src.utils.formatting import parse_datetime_iso, utc_now_iso @@ -55,7 +55,7 @@ class JWTParams(BaseModel): `s`: (string) session name """ - t: str = utc_now_iso() + t: str = Field(default_factory=utc_now_iso) exp: str | None = None ad: bool | None = None w: str | None = None diff --git a/src/telemetry/metrics_collector.py b/src/telemetry/metrics_collector.py index 8c7cb35e..0b25e126 100644 --- a/src/telemetry/metrics_collector.py +++ b/src/telemetry/metrics_collector.py @@ -6,6 +6,7 @@ metrics from deriver and dialectic operations during benchmarking. """ import json +import logging import statistics from datetime import datetime from pathlib import Path @@ -14,6 +15,8 @@ from typing_extensions import TypedDict from src.config import settings +logger = logging.getLogger(__name__) + class MetricStats(TypedDict): """Statistics for a single metric type.""" @@ -70,7 +73,7 @@ class MetricsCollector: self.metrics_by_type.clear() self.task_count = 0 self.is_collecting = True - print(f"📊 Started metrics collection for run: {run_id}") + logger.info("Started metrics collection for run: %s", run_id) def collect_metrics( self, metrics_list: list[tuple[str, str | int | float, str]] @@ -146,10 +149,10 @@ class MetricsCollector: if self.end_time and self.start_time else 0 ) - print(f"📊 Finalized metrics collection for run: {self.run_id}") - print(f" Tasks processed: {self.task_count}") - print(f" Collection duration: {duration:.2f}s") - print(f" Metric types collected: {len(self.metrics_by_type)}") + logger.info("Finalized metrics collection for run: %s", self.run_id) + logger.info("Tasks processed: %s", self.task_count) + logger.info("Collection duration: %.2fs", duration) + logger.info("Metric types collected: %s", len(self.metrics_by_type)) def get_aggregated_stats(self) -> dict[str, MetricStats]: """ @@ -218,28 +221,28 @@ class MetricsCollector: with open(filepath, "w") as f: json.dump(export_data, f, indent=2, default=str) - print(f"📊 Exported metrics to: {filepath}") + logger.info("Exported metrics to: %s", filepath) def print_summary(self) -> None: """ Print a summary of collected metrics to the console. """ if not self.metrics_by_type: - print("📊 No metrics collected") + print("No metrics collected") return stats = self.get_aggregated_stats() - print(f"\n{'=' * 80}") - print(f"📊 PERFORMANCE METRICS SUMMARY - {self.run_id}") - print(f"{'=' * 80}") + print("=" * 80) + print(f"PERFORMANCE METRICS SUMMARY - {self.run_id}") + print("=" * 80) print(f"Tasks processed: {self.task_count}") if self.start_time and self.end_time: duration = (self.end_time - self.start_time).total_seconds() print(f"Collection duration: {duration:.2f}s") - print("\nAggregated Performance Metrics:") + print("Aggregated Performance Metrics:") print( f"{'Metric':<40} {'Count':<8} {'Mean':<12} {'Median':<12} {'Min':<12} {'Max':<12} {'Unit'}" ) @@ -273,7 +276,7 @@ class MetricsCollector: f"{metric_name:<40} {stat['count']:<8} {mean_str:<12} {median_str:<12} {min_str:<12} {max_str:<12} {unit_display}" ) - print(f"{'=' * 80}") + print("=" * 80) def cleanup_collection(self) -> None: """ @@ -283,7 +286,7 @@ class MetricsCollector: self.end_time = datetime.now() # delete the metrics file metrics_file = get_metrics_file_path() - if metrics_file: + if metrics_file and metrics_file.exists(): metrics_file.unlink() diff --git a/tests/conftest.py b/tests/conftest.py index ebb2db68..8ff205ff 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -67,6 +67,25 @@ DB_URI = ( ) CONNECTION_URI = make_url(DB_URI) +_RUNTIME_MOCK_TEST_BLOCKLIST_PREFIXES = ( + # Benchmarks and migration tests have their own execution/runtime constraints. + "tests/bench/", + "tests/alembic/", + "tests/unified/", +) + + +def _requires_runtime_mocks(nodeid: str) -> bool: + return not any( + nodeid.startswith(prefix) for prefix in _RUNTIME_MOCK_TEST_BLOCKLIST_PREFIXES + ) + + +def _get_nodeid(request: pytest.FixtureRequest) -> str: + node = getattr(request, "node", None) + nodeid = getattr(node, "nodeid", "") + return nodeid if isinstance(nodeid, str) else "" + def _get_test_db_url(worker_id: str) -> URL: """Get a worker-specific test database URL for pytest-xdist parallelism.""" @@ -404,8 +423,12 @@ def _content_to_embedding(content: str) -> list[float]: @pytest.fixture(autouse=True) -def mock_openai_embeddings(): +def mock_openai_embeddings(request: pytest.FixtureRequest): """Mock OpenAI embeddings API calls for testing""" + if not _requires_runtime_mocks(_get_nodeid(request)): + yield + return + with ( patch("src.embedding_client.embedding_client.embed") as mock_embed, patch("src.embedding_client.embedding_client.batch_embed") as mock_batch_embed, @@ -431,8 +454,12 @@ def mock_openai_embeddings(): @pytest.fixture(autouse=True) -def mock_vector_store(): +def mock_vector_store(request: pytest.FixtureRequest): """Mock vector store operations for testing""" + if not _requires_runtime_mocks(_get_nodeid(request)): + yield + return + from unittest.mock import AsyncMock, MagicMock from src.vector_store import ( @@ -531,8 +558,11 @@ def mock_vector_store(): @pytest.fixture(autouse=True) -def mock_llm_call_functions(): +def mock_llm_call_functions(request: pytest.FixtureRequest): """Mock LLM functions to avoid needing API keys during tests""" + if not _requires_runtime_mocks(_get_nodeid(request)): + yield + return # Create an async generator for streaming responses async def mock_stream(*args, **kwargs): # pyright: ignore[reportUnusedParameter, reportMissingParameterType, reportUnknownParameterType] @@ -574,8 +604,12 @@ def mock_llm_call_functions(): @pytest.fixture(autouse=True) -def mock_honcho_llm_call(): +def mock_honcho_llm_call(request: pytest.FixtureRequest): """Generic mock for the honcho_llm_call decorator to avoid actual LLM calls during tests""" + if not _requires_runtime_mocks(_get_nodeid(request)): + yield + return + from unittest.mock import AsyncMock from src.utils.representation import ( @@ -687,12 +721,16 @@ def mock_honcho_llm_call(): @pytest.fixture(autouse=True) -def mock_tracked_db(db_engine: AsyncEngine): +def mock_tracked_db(db_engine: AsyncEngine, request: pytest.FixtureRequest): """Mock tracked_db to create fresh sessions per call. Using a session factory instead of a shared session avoids asyncio lock errors when multiple tracked_db calls run concurrently via asyncio.gather. """ + if not _requires_runtime_mocks(_get_nodeid(request)): + yield + return + from contextlib import asynccontextmanager session_factory = async_sessionmaker(bind=db_engine, expire_on_commit=False) @@ -719,8 +757,12 @@ def mock_tracked_db(db_engine: AsyncEngine): @pytest.fixture(autouse=True) -def enable_deriver_for_tests(): +def enable_deriver_for_tests(request: pytest.FixtureRequest): """Enable deriver globally for tests that need queue processing""" + if not _requires_runtime_mocks(_get_nodeid(request)): + yield + return + from src.config import settings original_value = settings.DERIVER.ENABLED @@ -730,8 +772,12 @@ def enable_deriver_for_tests(): @pytest.fixture(autouse=True) -def mock_crud_collection_operations(): +def mock_crud_collection_operations(request: pytest.FixtureRequest): """Mock CRUD operations that try to commit to database during tests""" + if not _requires_runtime_mocks(_get_nodeid(request)): + yield + return + from nanoid import generate as generate_nanoid from src import models diff --git a/tests/deriver/test_queue_processing.py b/tests/deriver/test_queue_processing.py index 409630bd..9538c871 100644 --- a/tests/deriver/test_queue_processing.py +++ b/tests/deriver/test_queue_processing.py @@ -19,14 +19,16 @@ class TestQueueProcessing: async def test_get_and_claim_work_units( self, + db_session: AsyncSession, sample_queue_items: list[models.QueueItem], sample_session_with_peers: tuple[models.Session, list[models.Peer]], ) -> None: """Test that get_and_claim_work_units correctly identifies unprocessed work""" - session, _peers = sample_session_with_peers # pyright: ignore[reportUnusedVariable] + session, _peers = sample_session_with_peers # Verify we have queue items from our test setup assert len(sample_queue_items) == 9 # 6 representation + 3 summary + expected_work_units = {item.work_unit_key for item in sample_queue_items} # Create a queue manager instance queue_manager = QueueManager() @@ -34,16 +36,24 @@ class TestQueueProcessing: # Get available work units work_units = await queue_manager.get_and_claim_work_units() - # Should have some work units available (may include items from other tests) + # Should return claimed work units from this test's seeded queue data assert len(work_units) > 0 + assert set(work_units).issubset(expected_work_units) # Check that all work units have the expected structure for work_unit in work_units: assert isinstance(work_unit, str) assert work_unit.split(":")[0] in ["representation", "summary"] + assert f":{session.workspace_name}:" in work_unit - # The test is mainly verifying that get_and_claim_work_units works without errors - # and returns properly structured work unit key strings + tracked_keys = ( + await db_session.execute( + select(models.ActiveQueueSession.work_unit_key).where( + models.ActiveQueueSession.work_unit_key.in_(list(work_units.keys())) + ) + ) + ).scalars() + assert set(tracked_keys) == set(work_units.keys()) async def test_work_unit_claiming( self, diff --git a/tests/deriver/test_vector_reconciliation.py b/tests/deriver/test_vector_reconciliation.py index 857a0fda..3b8e0e92 100644 --- a/tests/deriver/test_vector_reconciliation.py +++ b/tests/deriver/test_vector_reconciliation.py @@ -778,38 +778,38 @@ class TestMessageEmbeddings: class TestEndToEndReconciliation: """Test full reconciliation cycle.""" - async def test_reconciliation_cycle_completes( - self, - db_session: AsyncSession, - ) -> None: - """Test full reconciliation cycle processes documents and embeddings.""" - # This would be an integration test with the full cycle - # For now, we verify the function signature and return type + async def test_reconciliation_cycle_completes_when_no_work_found(self) -> None: + """Reconciliation should run each stage once and stop when no work is found.""" + mock_vector_store = MagicMock(spec=VectorStore) with ( - patch("src.reconciler.sync_vectors.tracked_db") as mock_tracked_db, - patch("src.reconciler.sync_vectors.get_external_vector_store"), patch( - "src.reconciler.sync_vectors._get_documents_needing_sync" - ) as mock_get_docs, + "src.reconciler.sync_vectors.get_external_vector_store", + return_value=mock_vector_store, + ), patch( - "src.reconciler.sync_vectors._get_message_embeddings_needing_sync" - ) as mock_get_embs, - patch("src.crud.document.cleanup_soft_deleted_documents") as mock_cleanup, + "src.reconciler.sync_vectors._reconcile_documents_batch", + new_callable=AsyncMock, + ) as mock_reconcile_docs, + patch( + "src.reconciler.sync_vectors._reconcile_message_embeddings_batch", + new_callable=AsyncMock, + ) as mock_reconcile_embs, + patch( + "src.reconciler.sync_vectors._cleanup_documents_batch", + new_callable=AsyncMock, + ) as mock_cleanup_docs, ): - # Mock to return empty results (no work to do) - mock_get_docs.return_value = [] - mock_get_embs.return_value = [] - mock_cleanup.return_value = 0 + mock_reconcile_docs.return_value = False + mock_reconcile_embs.return_value = False + mock_cleanup_docs.return_value = False - # Mock context manager - mock_db_context = MagicMock() - mock_db_context.__aenter__ = AsyncMock(return_value=db_session) - mock_db_context.__aexit__ = AsyncMock(return_value=None) - mock_tracked_db.return_value = mock_db_context - - # Run cycle metrics = await run_vector_reconciliation_cycle() - # Verify metrics returned - assert isinstance(metrics, ReconciliationMetrics) - assert metrics.total_synced == 0 # No work done + assert isinstance(metrics, ReconciliationMetrics) + assert metrics.total_synced == 0 + assert metrics.total_failed == 0 + assert metrics.total_cleaned == 0 + + mock_reconcile_docs.assert_awaited_once() + mock_reconcile_embs.assert_awaited_once() + mock_cleanup_docs.assert_awaited_once() diff --git a/tests/reconciler/test_queue_cleanup.py b/tests/reconciler/test_queue_cleanup.py new file mode 100644 index 00000000..b9c3a9eb --- /dev/null +++ b/tests/reconciler/test_queue_cleanup.py @@ -0,0 +1,32 @@ +from contextlib import asynccontextmanager +from unittest.mock import AsyncMock + +import pytest + +from src import models +from src.reconciler import queue_cleanup + + +@pytest.mark.asyncio +async def test_cleanup_queue_items_executes_delete_and_commit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_db = AsyncMock() + called_operation_names: list[str | None] = [] + + @asynccontextmanager + async def fake_tracked_db(operation_name: str | None = None): + called_operation_names.append(operation_name) + yield fake_db + + monkeypatch.setattr(queue_cleanup, "tracked_db", fake_tracked_db) + + await queue_cleanup.cleanup_queue_items() + + assert called_operation_names == ["cleanup_queue_items"] + fake_db.execute.assert_awaited_once() + fake_db.commit.assert_awaited_once() + + delete_stmt = fake_db.execute.await_args.args[0] + assert delete_stmt.table.name == models.QueueItem.__tablename__ + assert delete_stmt.whereclause is not None diff --git a/tests/sdk/test_client.py b/tests/sdk/test_client.py index 897ef09d..89371149 100644 --- a/tests/sdk/test_client.py +++ b/tests/sdk/test_client.py @@ -187,64 +187,85 @@ async def test_get_queue_status(client_fixture: tuple[Honcho, str]): """ honcho_client, client_type = client_fixture + def assert_queue_status(status: QueueStatusResponse) -> None: + assert status.total_work_units >= 0 + assert status.completed_work_units >= 0 + assert status.in_progress_work_units >= 0 + assert status.pending_work_units >= 0 + assert status.total_work_units == ( + status.completed_work_units + + status.in_progress_work_units + + status.pending_work_units + ) + + if status.sessions is not None: + for session_status in status.sessions.values(): + assert session_status.total_work_units == ( + session_status.completed_work_units + + session_status.in_progress_work_units + + session_status.pending_work_units + ) + if client_type == "async": # Test with no parameters - this should work in the SDK even though API requires at least one status = await honcho_client.aio.queue_status() assert isinstance(status, QueueStatusResponse) - assert hasattr(status, "total_work_units") - assert hasattr(status, "completed_work_units") - assert hasattr(status, "in_progress_work_units") - assert hasattr(status, "pending_work_units") + assert_queue_status(status) # Test with peer_id only peer = await honcho_client.aio.peer(id="test-peer-queue-status") await peer.aio.get_metadata() # Create the peer status = await honcho_client.aio.queue_status(observer=peer.id) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) # Test with session_id only session = await honcho_client.aio.session(id="test-session-queue-status") await session.aio.get_metadata() # Create the session status = await honcho_client.aio.queue_status(session=session.id) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) # Test with both peer and session status = await honcho_client.aio.queue_status( observer=peer.id, session=session.id ) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) # Test with sender status = await honcho_client.aio.queue_status(observer=peer.id, sender=peer.id) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) else: # Test with no parameters status = honcho_client.queue_status() assert isinstance(status, QueueStatusResponse) - assert hasattr(status, "total_work_units") - assert hasattr(status, "completed_work_units") - assert hasattr(status, "in_progress_work_units") - assert hasattr(status, "pending_work_units") + assert_queue_status(status) # Test with peer_id only peer = honcho_client.peer(id="test-peer-queue-status") peer.get_metadata() # Create the peer status = honcho_client.queue_status(observer=peer.id) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) # Test with session_id only session = honcho_client.session(id="test-session-queue-status") session.get_metadata() # Create the session status = honcho_client.queue_status(session=session.id) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) # Test with both peer and session status = honcho_client.queue_status(observer=peer.id, session=session.id) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) # Test with sender status = honcho_client.queue_status(observer=peer.id, sender=peer.id) assert isinstance(status, QueueStatusResponse) + assert_queue_status(status) @pytest.mark.asyncio diff --git a/tests/sdk/test_session.py b/tests/sdk/test_session.py index 04102d66..264a5c5c 100644 --- a/tests/sdk/test_session.py +++ b/tests/sdk/test_session.py @@ -430,31 +430,43 @@ async def test_session_queue_status( """ honcho_client, client_type = client_fixture + def assert_session_queue_status(status: QueueStatusResponse) -> None: + assert status.total_work_units >= 0 + assert status.completed_work_units >= 0 + assert status.in_progress_work_units >= 0 + assert status.pending_work_units >= 0 + assert status.total_work_units == ( + status.completed_work_units + + status.in_progress_work_units + + status.pending_work_units + ) + # Session queue status should be flattened (no per-session map) + assert status.sessions is None + if client_type == "async": session = await honcho_client.aio.session(id="test-session-deriver-status") assert isinstance(session, Session) status = await session.aio.queue_status() assert isinstance(status, QueueStatusResponse) - assert hasattr(status, "total_work_units") - assert hasattr(status, "completed_work_units") - assert hasattr(status, "in_progress_work_units") - assert hasattr(status, "pending_work_units") - assert status.sessions is None + assert_session_queue_status(status) # Test with observer only peer = await honcho_client.aio.peer(id="test-peer-session-deriver") await peer.aio.get_metadata() # Create the peer status = await session.aio.queue_status(observer=peer.id) assert isinstance(status, QueueStatusResponse) + assert_session_queue_status(status) # Test with sender only status = await session.aio.queue_status(sender=peer.id) assert isinstance(status, QueueStatusResponse) + assert_session_queue_status(status) # Test with both observer and sender status = await session.aio.queue_status(observer=peer.id, sender=peer.id) assert isinstance(status, QueueStatusResponse) + assert_session_queue_status(status) else: session = honcho_client.session(id="test-session-deriver-status") assert isinstance(session, Session) @@ -462,25 +474,24 @@ async def test_session_queue_status( # Test with no parameters status = session.queue_status() assert isinstance(status, QueueStatusResponse) - assert hasattr(status, "total_work_units") - assert hasattr(status, "completed_work_units") - assert hasattr(status, "in_progress_work_units") - assert hasattr(status, "pending_work_units") - assert status.sessions is None + assert_session_queue_status(status) # Test with observer only peer = honcho_client.peer(id="test-peer-session-deriver") peer.get_metadata() # Create the peer status = session.queue_status(observer=peer.id) assert isinstance(status, QueueStatusResponse) + assert_session_queue_status(status) # Test with sender only status = session.queue_status(sender=peer.id) assert isinstance(status, QueueStatusResponse) + assert_session_queue_status(status) # Test with both observer and sender status = session.queue_status(observer=peer.id, sender=peer.id) assert isinstance(status, QueueStatusResponse) + assert_session_queue_status(status) @pytest.mark.asyncio diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py new file mode 100644 index 00000000..057808c7 --- /dev/null +++ b/tests/test_dependencies.py @@ -0,0 +1,155 @@ +import uuid +from typing import Any + +import pytest + +import src.dependencies as dependencies_module +from src.config import settings +from src.db import request_context +from src.dependencies import get_db as real_get_db +from src.dependencies import tracked_db as real_tracked_db + + +class FakeSession: + def __init__(self, *, in_transaction: bool = False): + self._in_transaction: bool = in_transaction + self.execute_calls: list[Any] = [] + self.rollback_calls: int = 0 + self.close_calls: int = 0 + + async def execute(self, statement: Any) -> None: + self.execute_calls.append(statement) + + async def rollback(self) -> None: + self.rollback_calls += 1 + + async def close(self) -> None: + self.close_calls += 1 + + def in_transaction(self) -> bool: + return self._in_transaction + + +@pytest.mark.asyncio +async def test_get_db_sets_application_name_when_tracing_enabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_db = FakeSession() + monkeypatch.setattr(dependencies_module, "SessionLocal", lambda: fake_db) + monkeypatch.setattr(settings.DB, "TRACING", True) + + context_token = request_context.set("request:test-ctx") + dep_gen = real_get_db() + + try: + db = await anext(dep_gen) + assert db is fake_db + assert len(fake_db.execute_calls) == 1 + assert "SET application_name = 'request:test-ctx'" in str( + fake_db.execute_calls[0] + ) + finally: + await dep_gen.aclose() + request_context.reset(context_token) + + assert fake_db.rollback_calls == 0 + assert fake_db.close_calls == 1 + + +@pytest.mark.asyncio +async def test_get_db_rolls_back_and_closes_when_consumer_raises( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_db = FakeSession() + monkeypatch.setattr(dependencies_module, "SessionLocal", lambda: fake_db) + monkeypatch.setattr(settings.DB, "TRACING", False) + + dep_gen = real_get_db() + await anext(dep_gen) + + with pytest.raises(RuntimeError, match="boom"): + await dep_gen.athrow(RuntimeError("boom")) + + assert fake_db.rollback_calls == 1 + assert fake_db.close_calls == 1 + + +@pytest.mark.asyncio +async def test_tracked_db_creates_and_resets_task_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_db = FakeSession() + monkeypatch.setattr(dependencies_module, "SessionLocal", lambda: fake_db) + monkeypatch.setattr(settings.DB, "TRACING", True) + monkeypatch.setattr( + uuid, + "uuid4", + lambda: uuid.UUID("12345678-1234-5678-1234-567812345678"), + ) + + clear_token = request_context.set(None) + try: + async with real_tracked_db("cleanup_job"): + assert request_context.get() == "task:cleanup_job:12345678" + finally: + request_context.reset(clear_token) + + assert request_context.get() is None + assert len(fake_db.execute_calls) == 1 + assert "SET application_name = 'task:cleanup_job:12345678'" in str( + fake_db.execute_calls[0] + ) + assert fake_db.rollback_calls == 0 + assert fake_db.close_calls == 1 + + +@pytest.mark.asyncio +async def test_tracked_db_preserves_existing_request_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_db = FakeSession() + monkeypatch.setattr(dependencies_module, "SessionLocal", lambda: fake_db) + monkeypatch.setattr(settings.DB, "TRACING", True) + + context_token = request_context.set("request:existing") + try: + async with real_tracked_db("ignored_op"): + assert request_context.get() == "request:existing" + finally: + request_context.reset(context_token) + + assert len(fake_db.execute_calls) == 1 + assert "SET application_name = 'request:existing'" in str(fake_db.execute_calls[0]) + assert fake_db.rollback_calls == 0 + assert fake_db.close_calls == 1 + + +@pytest.mark.asyncio +async def test_tracked_db_rolls_back_on_error_and_closes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_db = FakeSession() + monkeypatch.setattr(dependencies_module, "SessionLocal", lambda: fake_db) + monkeypatch.setattr(settings.DB, "TRACING", False) + + with pytest.raises(ValueError, match="failed operation"): + async with real_tracked_db("operation"): + raise ValueError("failed operation") + + assert fake_db.rollback_calls == 1 + assert fake_db.close_calls == 1 + + +@pytest.mark.asyncio +async def test_tracked_db_rolls_back_open_transaction_on_exit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_db = FakeSession(in_transaction=True) + monkeypatch.setattr(dependencies_module, "SessionLocal", lambda: fake_db) + monkeypatch.setattr(settings.DB, "TRACING", False) + + async with real_tracked_db("operation"): + pass + + assert fake_db.rollback_calls == 1 + assert fake_db.close_calls == 1 diff --git a/tests/webhooks/test_webhook_delivery.py b/tests/webhooks/test_webhook_delivery.py new file mode 100644 index 00000000..3c4235f2 --- /dev/null +++ b/tests/webhooks/test_webhook_delivery.py @@ -0,0 +1,236 @@ +import hashlib +import hmac +import json +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest + +from src.config import settings +from src.utils.queue_payload import WebhookPayload +from src.webhooks import webhook_delivery + + +class FakeAsyncClient: + def __init__(self, responses: dict[str, httpx.Response | Exception]): + self._responses: dict[str, httpx.Response | Exception] = responses + self.calls: list[dict[str, Any]] = [] + + async def __aenter__(self) -> "FakeAsyncClient": + return self + + async def __aexit__( + self, exc_type: object, exc: object, tb: object + ) -> None: # pragma: no cover - required async CM signature + _ = (exc_type, exc, tb) + + async def post( + self, *, url: str, content: str, headers: dict[str, str] + ) -> httpx.Response: + self.calls.append({"url": url, "content": content, "headers": headers}) + result = self._responses[url] + if isinstance(result, Exception): + raise result + return result + + +def test_generate_webhook_signature_uses_secret( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings.WEBHOOK, "SECRET", "unit-test-secret") + payload = '{"key":"value"}' + + expected = hmac.new( + b"unit-test-secret", payload.encode("utf-8"), hashlib.sha256 + ).hexdigest() + actual = webhook_delivery._generate_webhook_signature(payload) # pyright: ignore[reportPrivateUsage] + + assert actual == expected + + +def test_generate_webhook_signature_raises_without_secret( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings.WEBHOOK, "SECRET", "") + + with pytest.raises(ValueError, match="WEBHOOK_SECRET not found"): + webhook_delivery._generate_webhook_signature("{}") # pyright: ignore[reportPrivateUsage] + + +@pytest.mark.asyncio +async def test_get_webhook_urls_returns_all_endpoint_urls( + monkeypatch: pytest.MonkeyPatch, +) -> None: + query = object() + list_endpoints = AsyncMock(return_value=query) + monkeypatch.setattr(webhook_delivery, "list_webhook_endpoints", list_endpoints) + + endpoint_a = SimpleNamespace(url="https://a.example.com/hook") + endpoint_b = SimpleNamespace(url="https://b.example.com/hook") + execute_result = MagicMock() + execute_result.scalars.return_value.all.return_value = [endpoint_a, endpoint_b] + + db = AsyncMock() + db.execute.return_value = execute_result + + urls = await webhook_delivery._get_webhook_urls( # pyright: ignore[reportPrivateUsage] + db, "workspace-a" + ) + + assert urls == [endpoint_a.url, endpoint_b.url] + list_endpoints.assert_awaited_once_with("workspace-a") + db.execute.assert_awaited_once_with(query) + + +@pytest.mark.asyncio +async def test_get_webhook_urls_returns_empty_list_on_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + webhook_delivery, + "list_webhook_endpoints", + AsyncMock(side_effect=RuntimeError("boom")), + ) + db = AsyncMock() + + urls = await webhook_delivery._get_webhook_urls( # pyright: ignore[reportPrivateUsage] + db, "workspace-a" + ) + + assert urls == [] + db.execute.assert_not_called() + + +@pytest.mark.asyncio +async def test_deliver_webhook_skips_when_no_urls( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_client = FakeAsyncClient({}) + + def async_client_factory(*args: Any, **kwargs: Any) -> FakeAsyncClient: + _ = (args, kwargs) + return fake_client + + monkeypatch.setattr(httpx, "AsyncClient", async_client_factory) + monkeypatch.setattr( + webhook_delivery, + "_get_webhook_urls", + AsyncMock(return_value=[]), + ) + + payload = WebhookPayload(event_type="peer.created", data={"id": "p_123"}) + await webhook_delivery.deliver_webhook(AsyncMock(), payload, "workspace-a") + + assert fake_client.calls == [] + + +@pytest.mark.asyncio +async def test_deliver_webhook_posts_signed_payload_to_each_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings.WEBHOOK, "SECRET", "delivery-secret") + monkeypatch.setattr(webhook_delivery, "utc_now_iso", lambda: "2026-02-13T00:00:00Z") + + urls = [ + "https://a.example.com/hook", + "https://b.example.com/hook", + ] + monkeypatch.setattr( + webhook_delivery, + "_get_webhook_urls", + AsyncMock(return_value=urls), + ) + + fake_client = FakeAsyncClient( + { + urls[0]: httpx.Response( + status_code=202, request=httpx.Request("POST", urls[0]) + ), + urls[1]: httpx.ConnectError("connection failed"), + } + ) + + def async_client_factory(*args: Any, **kwargs: Any) -> FakeAsyncClient: + _ = (args, kwargs) + return fake_client + + monkeypatch.setattr(httpx, "AsyncClient", async_client_factory) + + payload = WebhookPayload( + event_type="message.created", + data={"id": "m_1", "workspace": "workspace-a"}, + ) + await webhook_delivery.deliver_webhook(AsyncMock(), payload, "workspace-a") + + expected_event_json = json.dumps( + { + "type": payload.event_type, + "data": payload.data, + "timestamp": "2026-02-13T00:00:00Z", + }, + separators=(",", ":"), + sort_keys=True, + ) + expected_signature = webhook_delivery._generate_webhook_signature( # pyright: ignore[reportPrivateUsage] + expected_event_json + ) + + assert len(fake_client.calls) == 2 + for call in fake_client.calls: + assert call["content"] == expected_event_json + assert call["headers"]["Content-Type"] == "application/json" + assert call["headers"]["X-Honcho-Signature"] == expected_signature + + +@pytest.mark.asyncio +async def test_deliver_webhook_handles_signature_generation_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings.WEBHOOK, "SECRET", "") + monkeypatch.setattr( + webhook_delivery, + "_get_webhook_urls", + AsyncMock(return_value=["https://a.example.com/hook"]), + ) + fake_client = FakeAsyncClient( + { + "https://a.example.com/hook": httpx.Response( + status_code=200, + request=httpx.Request("POST", "https://a.example.com/hook"), + ) + } + ) + + def async_client_factory(*args: Any, **kwargs: Any) -> FakeAsyncClient: + _ = (args, kwargs) + return fake_client + + monkeypatch.setattr(httpx, "AsyncClient", async_client_factory) + + payload = WebhookPayload(event_type="workspace.updated", data={"id": "ws_1"}) + await webhook_delivery.deliver_webhook(AsyncMock(), payload, "workspace-a") + + assert fake_client.calls == [] + + +@pytest.mark.asyncio +async def test_deliver_webhook_catches_request_errors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings.WEBHOOK, "SECRET", "delivery-secret") + monkeypatch.setattr( + webhook_delivery, + "_get_webhook_urls", + AsyncMock(side_effect=httpx.RequestError("network issue")), + ) + + def async_client_factory(*args: Any, **kwargs: Any) -> FakeAsyncClient: + _ = (args, kwargs) + return FakeAsyncClient({}) + + monkeypatch.setattr(httpx, "AsyncClient", async_client_factory) + + payload = WebhookPayload(event_type="workspace.updated", data={"id": "ws_1"}) + await webhook_delivery.deliver_webhook(AsyncMock(), payload, "workspace-a")