jpmschweitzer/tatlock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity
Files
014afc960c52c0c5fffd1f679d829ffe79d3bd1f
tatlock/tests/agents/librarian/test_tools.py
T
jpmschweitzerandClaude 78066fab1b style: apply ruff's automatic fixes and formatter 
Mechanical only, and separated from the judgment calls that follow so the
reviewable changes are not buried in a 98-file whitespace diff.

227 automatic fixes: 60 blank lines carrying whitespace, 60 unsorted import
blocks, 34 Optional[X] to X | None, 28 unused imports, 16 deprecated typing
imports, 12 datetime.timezone.utc to datetime.UTC, and assorted smaller
modernisations. Then `ruff format` over src and tests: 98 files reformatted,
35 already conforming.

No file among the unused-import findings defines __all__ or is an __init__.py,
so nothing here removes a re-export.

`make test`: 658 passed, unchanged from HEAD.

Two things observed while verifying, neither addressed here:

`pytest tests/` cannot collect — tests/e2e/test_orchestration_e2e.py uses an
`e2e` marker that is not registered, and the config is strict about markers.
This fails identically at HEAD, so it predates this change; `make test` passes
because it ignores tests/e2e, tests/integration and tests/contracts.

test_tatlock_tool_call_logging_calculator is flaky. It failed once in a full run
with these changes and passed on the next, passes in isolation with them, and
fails in isolation at HEAD. It is order- or timing-dependent, not a regression
from this commit — established by running the full suite both ways rather than
by reasoning about which change could have caused it.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-11 17:25:18 +02:00

480 lines
17 KiB
Python
Raw Blame History

"""
Tests for Librarian tools.
Tests the tool functions that wrap the Library-Desk API,
including the new web search and content extraction tools.
"""
from unittest.mock import AsyncMock, patch
import pytest
from src.agents.librarian.client import (
BatchExtractionResponse,
ContentExtractionResult,
WebSearchResponse,
WebSearchResult,
WikiPage,
)
from src.agents.librarian.tools import (
CLEAR_TAGS_SENTINEL,
read_url,
read_urls_batch,
search_web,
update_wiki_page,
)
@pytest.fixture
def mock_client():
"""Create a mock LibraryDeskClient."""
client = AsyncMock()
return client
# ============================================================================
# Web Search Tests
# ============================================================================
@pytest.mark.unit
class TestSearchWeb:
"""Tests for search_web tool."""
@pytest.mark.asyncio
async def test_search_web_success(self, mock_client):
"""Test successful web search."""
mock_response = WebSearchResponse(
query="Python async programming",
search_type="web",
results=[
WebSearchResult(
title="Async Python Tutorial",
url="https://example.com/async",
content="Full content about async programming...",
snippet="Learn async programming in Python",
source="example.com",
),
WebSearchResult(
title="AsyncIO Documentation",
url="https://docs.python.org/asyncio",
content="Official asyncio docs content...",
snippet="Python asyncio library reference",
source="docs.python.org",
),
],
total_results=2,
search_time_ms=150,
sources_summary="**Sources:**\n- example.com\n- docs.python.org",
)
mock_client.search_web.return_value = mock_response
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await search_web("Python async programming")
assert "Python async programming" in result
assert "Async Python Tutorial" in result
assert "https://example.com/async" in result
assert "example.com" in result
assert "150ms" in result or "2 results" in result
@pytest.mark.asyncio
async def test_search_web_no_results(self, mock_client):
"""Test web search with no results."""
mock_response = WebSearchResponse(
query="nonexistent query xyz123",
search_type="web",
results=[],
total_results=0,
search_time_ms=50,
)
mock_client.search_web.return_value = mock_response
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await search_web("nonexistent query xyz123")
assert "No results found" in result
@pytest.mark.asyncio
async def test_search_web_error_handling(self, mock_client):
"""Test web search errors return a user-safe message without internals."""
mock_client.search_web.side_effect = Exception(
"Connection failed to http://internal-host:8089"
)
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await search_web("test query")
assert "unable to search the web" in result
# Exception detail (internal URLs etc.) must not leak
assert "Connection failed" not in result
assert "internal-host" not in result
@pytest.mark.asyncio
async def test_search_web_with_news_type(self, mock_client):
"""Test web search with news search type."""
mock_response = WebSearchResponse(
query="latest tech news",
search_type="news",
results=[
WebSearchResult(
title="Tech News Today",
url="https://news.example.com/tech",
snippet="Breaking tech news",
source="news.example.com",
published_date="2024-01-15",
),
],
total_results=1,
search_time_ms=100,
)
mock_client.search_web.return_value = mock_response
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await search_web("latest tech news", search_type="news")
assert "Tech News Today" in result
mock_client.search_web.assert_called_with(
query="latest tech news",
limit=10,
search_type="news",
)
# ============================================================================
# Read URL Tests
# ============================================================================
@pytest.mark.unit
class TestReadUrl:
"""Tests for read_url tool."""
@pytest.mark.asyncio
async def test_read_url_success(self, mock_client):
"""Test successful URL content extraction."""
mock_result = ContentExtractionResult(
url="https://example.com/article",
title="Great Article Title",
content="This is the full article content extracted from the page.",
author="John Doe",
date="2024-01-10",
language="en",
success=True,
)
mock_client.extract_content.return_value = mock_result
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await read_url("https://example.com/article")
assert "Great Article Title" in result
assert "https://example.com/article" in result
assert "John Doe" in result
assert "full article content" in result
@pytest.mark.asyncio
async def test_read_url_failure(self, mock_client):
"""Test URL extraction failure."""
mock_result = ContentExtractionResult(
url="https://example.com/blocked",
success=False,
error="403 Forbidden",
)
mock_client.extract_content.return_value = mock_result
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await read_url("https://example.com/blocked")
assert "Could not read page" in result
assert "403 Forbidden" in result
@pytest.mark.asyncio
async def test_read_url_with_max_length(self, mock_client):
"""Test URL extraction with custom max length."""
mock_result = ContentExtractionResult(
url="https://example.com/long",
title="Long Article",
content="X" * 10000,
success=True,
)
mock_client.extract_content.return_value = mock_result
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
await read_url("https://example.com/long", max_length=2000)
mock_client.extract_content.assert_called_with(
url="https://example.com/long",
include_metadata=True,
max_length=2000,
)
# ============================================================================
# Batch URL Tests
# ============================================================================
@pytest.mark.unit
class TestReadUrlsBatch:
"""Tests for read_urls_batch tool."""
@pytest.mark.asyncio
async def test_batch_success(self, mock_client):
"""Test successful batch extraction."""
mock_response = BatchExtractionResponse(
results=[
ContentExtractionResult(
url="https://example.com/1",
title="Article 1",
content="Content from article 1",
success=True,
),
ContentExtractionResult(
url="https://example.com/2",
title="Article 2",
content="Content from article 2",
success=True,
),
],
total_urls=2,
successful=2,
failed=0,
extraction_time_ms=300,
)
mock_client.extract_content_batch.return_value = mock_response
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await read_urls_batch(
[
"https://example.com/1",
"https://example.com/2",
]
)
assert "Article 1" in result
assert "Article 2" in result
assert "2/2" in result or "Extracted 2" in result
@pytest.mark.asyncio
async def test_batch_partial_failure(self, mock_client):
"""Test batch extraction with some failures."""
mock_response = BatchExtractionResponse(
results=[
ContentExtractionResult(
url="https://example.com/good",
title="Good Article",
content="Content extracted successfully",
success=True,
),
ContentExtractionResult(
url="https://example.com/bad",
success=False,
error="Connection timeout",
),
],
total_urls=2,
successful=1,
failed=1,
extraction_time_ms=500,
)
mock_client.extract_content_batch.return_value = mock_response
with patch("src.agents.librarian.tools.LibraryDeskClient") as mock_client_class:
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
result = await read_urls_batch(
[
"https://example.com/good",
"https://example.com/bad",
]
)
# Should contain successful result
assert "Good Article" in result
# Should report failure
assert "Failed" in result
assert "Connection timeout" in result
# ============================================================================
# Response Model Tests
# ============================================================================
@pytest.mark.unit
class TestWebSearchModels:
"""Tests for web search response models."""
def test_web_search_result_model(self):
"""Test WebSearchResult model."""
result = WebSearchResult(
title="Test Title",
url="https://example.com",
content="Full content here",
snippet="Short snippet",
source="example.com",
published_date="2024-01-15",
)
assert result.title == "Test Title"
assert result.url == "https://example.com"
assert result.content == "Full content here"
assert result.source == "example.com"
def test_web_search_result_defaults(self):
"""Test WebSearchResult default values."""
result = WebSearchResult(
title="Title",
url="https://example.com",
)
assert result.content == ""
assert result.snippet == ""
assert result.source == ""
assert result.published_date is None
def test_web_search_response_model(self):
"""Test WebSearchResponse model."""
response = WebSearchResponse(
query="test query",
search_type="web",
results=[
WebSearchResult(title="R1", url="https://example.com/1"),
WebSearchResult(title="R2", url="https://example.com/2"),
],
total_results=2,
search_time_ms=100,
sources_summary="**Sources:** example.com",
)
assert response.query == "test query"
assert len(response.results) == 2
assert response.total_results == 2
def test_content_extraction_result_model(self):
"""Test ContentExtractionResult model."""
result = ContentExtractionResult(
url="https://example.com",
title="Title",
content="Content",
author="Author",
date="2024-01-01",
language="en",
success=True,
)
assert result.url == "https://example.com"
assert result.success is True
assert result.author == "Author"
def test_content_extraction_failure(self):
"""Test ContentExtractionResult for failed extraction."""
result = ContentExtractionResult(
url="https://example.com",
success=False,
error="404 Not Found",
)
assert result.success is False
assert result.error == "404 Not Found"
assert result.content == ""
def test_batch_extraction_response_model(self):
"""Test BatchExtractionResponse model."""
response = BatchExtractionResponse(
results=[
ContentExtractionResult(url="https://1.com", success=True),
ContentExtractionResult(url="https://2.com", success=False),
],
total_urls=2,
successful=1,
failed=1,
extraction_time_ms=500,
)
assert response.total_urls == 2
assert response.successful == 1
assert response.failed == 1
# ============================================================================
# Wiki Update Tests (tag sentinel behavior)
# ============================================================================
@pytest.mark.unit
class TestUpdateWikiPageTagSentinels:
"""Empty list leaves tags unchanged; the clear sentinel empties them."""
def _page(self, tags: list[str]) -> WikiPage:
return WikiPage(id=42, path="test/page", title="Test Page", tags=tags)
def _patched_client(self, mock_client):
patcher = patch("src.agents.librarian.tools.LibraryDeskClient")
mock_client_class = patcher.start()
mock_client_class.return_value.__aenter__.return_value = mock_client
mock_client_class.return_value.__aexit__.return_value = None
return patcher
@pytest.mark.asyncio
async def test_empty_tags_means_leave_unchanged(self, mock_client):
mock_client.update_wiki_page.return_value = self._page(["existing"])
patcher = self._patched_client(mock_client)
try:
await update_wiki_page(42, content="new content")
finally:
patcher.stop()
_, kwargs = mock_client.update_wiki_page.call_args
assert kwargs["tags"] is None
@pytest.mark.asyncio
async def test_clear_sentinel_sends_empty_tag_list(self, mock_client):
mock_client.update_wiki_page.return_value = self._page([])
patcher = self._patched_client(mock_client)
try:
result = await update_wiki_page(42, tags=[CLEAR_TAGS_SENTINEL])
finally:
patcher.stop()
_, kwargs = mock_client.update_wiki_page.call_args
assert kwargs["tags"] == []
assert "tags (cleared)" in result
@pytest.mark.asyncio
async def test_real_tags_are_passed_through(self, mock_client):
mock_client.update_wiki_page.return_value = self._page(["a", "b"])
patcher = self._patched_client(mock_client)
try:
await update_wiki_page(42, tags=["a", "b"])
finally:
patcher.stop()
_, kwargs = mock_client.update_wiki_page.call_args
assert kwargs["tags"] == ["a", "b"]
Reference in New Issue View Git Blame Copy Permalink