RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: Various Codex Audits (#386) 

* fix: Various Codex Audits

* fix: Address Comments
This commit is contained in:
Vineeth Voruganti 2026-02-13 12:00:15 -05:00 committed by GitHub
parent 97df0a80cd
commit beb282bfbc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 599 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
pyproject.toml
View File

@ -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 = [

15
src/deriver/__main__.py
View File

@ -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")

7
src/main.py
View File

@ -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:

4
src/security.py
View File

@ -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

29
src/telemetry/metrics_collector.py
View File

@ -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()

60
tests/conftest.py
View File

@ -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

18
tests/deriver/test_queue_processing.py
View File

@ -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,

56
tests/deriver/test_vector_reconciliation.py
View File

@ -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()

32
tests/reconciler/test_queue_cleanup.py Normal file
View File

@ -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

37
tests/sdk/test_client.py
View File

@ -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

31
tests/sdk/test_session.py
View File

@ -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

155
tests/test_dependencies.py Normal file
View File

@ -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

236
tests/webhooks/test_webhook_delivery.py Normal file
View File

@ -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")