Skip to content

Commit fc14480

Browse files
test: cover sync shutdown flush branches
1 parent cbd06a2 commit fc14480

1 file changed

Lines changed: 183 additions & 0 deletions

File tree

‎sdks/python/tests/test_observability.py‎

Lines changed: 183 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
from unittest.mock import AsyncMock, MagicMock, patch
77
from uuid import uuid4
88

9+
import httpx
910
import pytest
1011

1112
from agent_control.observability import (
@@ -448,6 +449,154 @@ async def test_send_batch_without_httpx(self):
448449
assert isinstance(result, bool)
449450

450451

452+
class TestEventBatcherSendBatchSync:
453+
"""Tests for sync HTTP sending used during shutdown fallback."""
454+
455+
def test_send_batch_sync_returns_true_on_202(self):
456+
batcher = EventBatcher(server_url="http://test:8000", api_key="test-key")
457+
response = MagicMock(status_code=202, text="accepted")
458+
client = MagicMock()
459+
client.post.return_value = response
460+
client_context = MagicMock()
461+
client_context.__enter__.return_value = client
462+
463+
with patch("agent_control.observability.httpx.Client", return_value=client_context) as client_ctor:
464+
result = batcher._send_batch_sync([create_mock_event()])
465+
466+
assert result is True
467+
client_ctor.assert_called_once_with(timeout=30.0)
468+
client.post.assert_called_once()
469+
470+
def test_send_batch_sync_returns_false_on_401_without_retry(self):
471+
batcher = EventBatcher()
472+
response = MagicMock(status_code=401, text="unauthorized")
473+
client = MagicMock()
474+
client.post.return_value = response
475+
client_context = MagicMock()
476+
client_context.__enter__.return_value = client
477+
478+
with patch("agent_control.observability.httpx.Client", return_value=client_context) as client_ctor:
479+
result = batcher._send_batch_sync([create_mock_event()])
480+
481+
assert result is False
482+
assert client_ctor.call_count == 1
483+
client.post.assert_called_once()
484+
485+
def test_send_batch_sync_retries_after_server_error_then_succeeds(self):
486+
from agent_control.settings import configure_settings
487+
488+
original = get_settings().model_dump()
489+
configure_settings(max_retries=2, retry_delay=0.25)
490+
batcher = EventBatcher()
491+
492+
first = MagicMock(status_code=500, text="server error")
493+
second = MagicMock(status_code=202, text="accepted")
494+
client = MagicMock()
495+
client.post.side_effect = [first, second]
496+
client_context = MagicMock()
497+
client_context.__enter__.return_value = client
498+
499+
try:
500+
with (
501+
patch(
502+
"agent_control.observability.httpx.Client",
503+
return_value=client_context,
504+
) as client_ctor,
505+
patch("agent_control.observability.time.sleep") as sleep_mock,
506+
):
507+
result = batcher._send_batch_sync([create_mock_event()])
508+
509+
assert result is True
510+
assert client_ctor.call_count == 2
511+
sleep_mock.assert_called_once_with(0.25)
512+
finally:
513+
configure_settings(**original)
514+
515+
def test_send_batch_sync_returns_false_when_deadline_already_expired(self):
516+
batcher = EventBatcher()
517+
518+
with (
519+
patch("agent_control.observability.httpx.Client") as client_ctor,
520+
patch("agent_control.observability.time.monotonic", return_value=2.0),
521+
):
522+
result = batcher._send_batch_sync([create_mock_event()], deadline=1.0)
523+
524+
assert result is False
525+
client_ctor.assert_not_called()
526+
527+
def test_send_batch_sync_returns_false_when_retry_backoff_exceeds_deadline(self):
528+
from agent_control.settings import configure_settings
529+
530+
original = get_settings().model_dump()
531+
configure_settings(max_retries=3, retry_delay=0.25)
532+
batcher = EventBatcher()
533+
534+
client = MagicMock()
535+
client.post.side_effect = httpx.ConnectError("boom")
536+
client_context = MagicMock()
537+
client_context.__enter__.return_value = client
538+
539+
try:
540+
with (
541+
patch(
542+
"agent_control.observability.httpx.Client",
543+
return_value=client_context,
544+
) as client_ctor,
545+
patch(
546+
"agent_control.observability.time.monotonic",
547+
side_effect=[0.0, 1.1],
548+
),
549+
patch("agent_control.observability.time.sleep") as sleep_mock,
550+
):
551+
result = batcher._send_batch_sync([create_mock_event()], deadline=1.0)
552+
553+
assert result is False
554+
assert client_ctor.call_count == 1
555+
sleep_mock.assert_not_called()
556+
finally:
557+
configure_settings(**original)
558+
559+
def test_send_batch_sync_handles_timeout_exception(self):
560+
from agent_control.settings import configure_settings
561+
562+
original = get_settings().model_dump()
563+
configure_settings(max_retries=1)
564+
batcher = EventBatcher()
565+
566+
client = MagicMock()
567+
client.post.side_effect = httpx.TimeoutException("boom")
568+
client_context = MagicMock()
569+
client_context.__enter__.return_value = client
570+
571+
try:
572+
with patch("agent_control.observability.httpx.Client", return_value=client_context):
573+
result = batcher._send_batch_sync([create_mock_event()])
574+
575+
assert result is False
576+
finally:
577+
configure_settings(**original)
578+
579+
def test_send_batch_sync_handles_unexpected_exception(self):
580+
from agent_control.settings import configure_settings
581+
582+
original = get_settings().model_dump()
583+
configure_settings(max_retries=1)
584+
batcher = EventBatcher()
585+
586+
client = MagicMock()
587+
client.post.side_effect = RuntimeError("boom")
588+
client_context = MagicMock()
589+
client_context.__enter__.return_value = client
590+
591+
try:
592+
with patch("agent_control.observability.httpx.Client", return_value=client_context):
593+
result = batcher._send_batch_sync([create_mock_event()])
594+
595+
assert result is False
596+
finally:
597+
configure_settings(**original)
598+
599+
451600
class TestGlobalBatcher:
452601
"""Tests for global batcher functions."""
453602

@@ -602,6 +751,40 @@ def test_shutdown_uses_settings_timeouts(self):
602751
finally:
603752
configure_settings(**original)
604753

754+
def test_sync_shutdown_flush_stops_after_failed_flush_limit(self):
755+
"""Test that sync shutdown fallback exits after configured failed flushes."""
756+
batcher = EventBatcher(batch_size=2)
757+
batcher.shutdown_max_failed_flushes = 2
758+
batcher._client = AsyncMock()
759+
for _ in range(3):
760+
batcher.add_event(create_mock_event())
761+
762+
with patch.object(batcher, "_send_batch_sync", return_value=False) as send_batch_sync:
763+
batcher._flush_all_without_worker(timeout=1.0)
764+
765+
assert send_batch_sync.call_count == 2
766+
assert len(batcher._events) == 3
767+
assert batcher._client is None
768+
769+
def test_sync_shutdown_flush_honors_timeout_before_first_attempt(self):
770+
"""Test that sync shutdown fallback exits if its timeout is already exhausted."""
771+
batcher = EventBatcher()
772+
batcher._client = AsyncMock()
773+
batcher.add_event(create_mock_event())
774+
775+
with (
776+
patch.object(batcher, "_send_batch_sync") as send_batch_sync,
777+
patch(
778+
"agent_control.observability.time.monotonic",
779+
side_effect=[0.0, 0.0],
780+
),
781+
):
782+
batcher._flush_all_without_worker(timeout=0.0)
783+
784+
send_batch_sync.assert_not_called()
785+
assert len(batcher._events) == 1
786+
assert batcher._client is None
787+
605788

606789
class TestSpanLogging:
607790
"""Tests for span start/end logging functions."""

0 commit comments

Comments
 (0)