engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

crawler: verbatim lift from agents.ai.unturf.com (off by default)

Browse source 
Lifts the async web fetcher into aborist as an opt-in source. The
implementation comes directly from ~/git/agents.ai.unturf.com/core
(rev 2026-04-28); aborist's adaptations are minimal and documented in
aborist/sources/crawler/__init__.py:

  core/async_web_fetcher.py  ->  aborist/sources/crawler/async_web_fetcher.py
  core/web_fetch.py          ->  aborist/sources/crawler/web_fetch.py

Two source-side changes during the lift:

1. Heavy deps (aiohttp, bs4, miniuri) wrapped in try/except so a bare
   `import aborist.sources.crawler` raises ImportError with the install
   hint instead of leaking AttributeErrors deep in user code.

2. Chat-bot fetch triggers (`has_fresh_fetch_trigger`,
   `has_web_fetch_trigger` from agents.ai.unturf.com/core/keywords)
   replaced with NotImplementedError stubs. Aborist has no chat
   surface — fetch intent is detected at the application layer. The
   two test classes that exercised these triggers are
   `@pytest.mark.skip`'d with the same rationale.

Not lifted: web_cache_manager.py — it backs page caching with
SQLAlchemy. Aborist has its own content-addressed cache via
providence_cache; no need to carry SQLAlchemy as a dep just for
crawled-page memoization.

Off by default:

- `[crawler]` extras section in pyproject.toml carries the heavy
  deps. `[dev]` pulls them in so the crawler tests can run.
- `make test` ignores tests/crawler/ entirely.
- `make bootstrap-crawler` installs the extras into the venv.
- `make test-crawler` runs only the lifted tests after extras land.

Tests: 74 passed, 9 skipped (the chat-bot trigger tests deliberately
dropped). Default `make test` stays at 273 passed, 1 skipped.
This commit is contained in:
russell@unturf.com 2026-04-28 20:23:36 -04:00
parent c92ebad4ed
commit a5e709a470
No known key found for this signature in database
7 changed files with 4206 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

470
tests/crawler/test_async_web_fetcher.py Normal file
View file

@ -0,0 +1,470 @@
"""
Tests for async_web_fetcher module.
Tests URL handling, media extraction, robots.txt compliance,
and async fetching functionality.
"""
import pytest
from unittest.mock import Mock, AsyncMock, patch, MagicMock
import asyncio
import aiohttp
from aborist.sources.crawler.async_web_fetcher import (
CrawlMode,
MediaItem,
strip_uri_fragment,
normalize_link,
get_media_type_from_extension,
get_media_type_from_mime,
extract_media_from_html,
AsyncWebFetcher,
IMAGE_EXTENSIONS,
VIDEO_EXTENSIONS,
AUDIO_EXTENSIONS,
DEFAULT_CRAWL_DELAY,
get_last_fetch_error,
clear_last_fetch_error,
)
class TestCrawlMode:
"""Test CrawlMode enum."""
def test_crawl_mode_values(self):
"""Test all crawl mode values exist."""
assert CrawlMode.TEXT.value == "text"
assert CrawlMode.IMAGES.value == "images"
assert CrawlMode.VIDEOS.value == "videos"
assert CrawlMode.MEDIA.value == "media"
assert CrawlMode.ALL.value == "all"
def test_crawl_mode_from_string(self):
"""Test creating CrawlMode from string."""
assert CrawlMode("text") == CrawlMode.TEXT
assert CrawlMode("images") == CrawlMode.IMAGES
assert CrawlMode("media") == CrawlMode.MEDIA
class TestMediaItem:
"""Test MediaItem dataclass."""
def test_media_item_creation(self):
"""Test creating a MediaItem."""
item = MediaItem(
url="https://example.com/image.jpg",
source_page="https://example.com/",
media_type="image",
mime_type="image/jpeg",
)
assert item.url == "https://example.com/image.jpg"
assert item.source_page == "https://example.com/"
assert item.media_type == "image"
assert item.mime_type == "image/jpeg"
assert item.discovered_at is not None
def test_media_item_optional_fields(self):
"""Test MediaItem with optional fields."""
item = MediaItem(
url="https://example.com/video.mp4",
source_page="https://example.com/",
media_type="video",
alt_text="A video",
width=1920,
height=1080,
)
assert item.alt_text == "A video"
assert item.width == 1920
assert item.height == 1080
class TestURIHandling:
"""Test URI handling functions."""
def test_strip_uri_fragment_basic(self):
"""Test stripping fragment from URI."""
uri = "https://example.com/page#section"
result = strip_uri_fragment(uri)
assert result == "https://example.com/page"
def test_strip_uri_fragment_no_fragment(self):
"""Test URI without fragment."""
uri = "https://example.com/page"
result = strip_uri_fragment(uri)
assert result == "https://example.com/page"
def test_strip_uri_fragment_with_query(self):
"""Test URI with query and fragment."""
uri = "https://example.com/page?q=test#section"
result = strip_uri_fragment(uri)
assert result == "https://example.com/page?q=test"
def test_strip_uri_fragment_empty(self):
"""Test empty URI."""
assert strip_uri_fragment("") == ""
assert strip_uri_fragment(None) is None
def test_normalize_link_dict(self):
"""Test normalizing link from dict."""
link = {"url": "https://example.com/page#section", "anchor_text": "Click here"}
uri, anchor = normalize_link(link)
assert uri == "https://example.com/page"
assert anchor == "Click here"
def test_normalize_link_string(self):
"""Test normalizing link from string."""
uri, anchor = normalize_link("https://example.com/page#section")
assert uri == "https://example.com/page"
assert anchor == ""
class TestMediaTypeDetection:
"""Test media type detection functions."""
def test_get_media_type_from_extension_image(self):
"""Test detecting image from extension."""
assert get_media_type_from_extension("https://example.com/photo.jpg") == "image"
assert get_media_type_from_extension("https://example.com/photo.png") == "image"
assert get_media_type_from_extension("https://example.com/photo.webp") == "image"
assert get_media_type_from_extension("https://example.com/photo.gif") == "image"
def test_get_media_type_from_extension_video(self):
"""Test detecting video from extension."""
assert get_media_type_from_extension("https://example.com/video.mp4") == "video"
assert get_media_type_from_extension("https://example.com/video.webm") == "video"
assert get_media_type_from_extension("https://example.com/video.mov") == "video"
def test_get_media_type_from_extension_audio(self):
"""Test detecting audio from extension."""
assert get_media_type_from_extension("https://example.com/audio.mp3") == "audio"
assert get_media_type_from_extension("https://example.com/audio.wav") == "audio"
assert get_media_type_from_extension("https://example.com/audio.ogg") == "audio"
def test_get_media_type_from_extension_code(self):
"""Test detecting code from extension."""
assert get_media_type_from_extension("https://example.com/file.txt") == "code"
assert get_media_type_from_extension("https://example.com/script.py") == "code"
assert get_media_type_from_extension("https://example.com/app.js") == "code"
def test_get_media_type_from_extension_unknown(self):
"""Test unknown extension."""
assert get_media_type_from_extension("https://example.com/page.html") is None
assert get_media_type_from_extension("https://example.com/file.xyz") is None
def test_get_media_type_from_mime_image(self):
"""Test detecting image from MIME type."""
assert get_media_type_from_mime("image/jpeg") == "image"
assert get_media_type_from_mime("image/png") == "image"
assert get_media_type_from_mime("image/webp") == "image"
def test_get_media_type_from_mime_video(self):
"""Test detecting video from MIME type."""
assert get_media_type_from_mime("video/mp4") == "video"
assert get_media_type_from_mime("video/webm") == "video"
def test_get_media_type_from_mime_audio(self):
"""Test detecting audio from MIME type."""
assert get_media_type_from_mime("audio/mpeg") == "audio"
assert get_media_type_from_mime("audio/wav") == "audio"
def test_get_media_type_from_mime_code(self):
"""Test detecting code from MIME type."""
assert get_media_type_from_mime("application/json") == "code"
assert get_media_type_from_mime("text/plain") == "code"
assert get_media_type_from_mime("application/javascript") == "code"
def test_get_media_type_from_mime_unknown(self):
"""Test unknown MIME type."""
assert get_media_type_from_mime("text/html") is None
assert get_media_type_from_mime("application/octet-stream") is None
class TestMediaExtraction:
"""Test media extraction from HTML."""
def test_extract_images_from_html(self):
"""Test extracting images from HTML."""
html = '''
<html>
<body>
<img src="/images/photo.jpg" alt="A photo">
<img src="https://cdn.example.com/logo.png" alt="Logo">
</body>
</html>
'''
media = extract_media_from_html(html, "https://example.com/", CrawlMode.IMAGES)
assert len(media) >= 1
urls = [m['url'] for m in media]
# Should have resolved relative URL
assert any('photo.jpg' in url for url in urls)
def test_extract_videos_from_html(self):
"""Test extracting videos from HTML."""
html = '''
<html>
<body>
<video src="/videos/movie.mp4"></video>
<source src="/videos/clip.webm" type="video/webm">
</body>
</html>
'''
media = extract_media_from_html(html, "https://example.com/", CrawlMode.VIDEOS)
urls = [m['url'] for m in media]
assert any('movie.mp4' in url or 'clip.webm' in url for url in urls)
def test_extract_media_filters_by_mode(self):
"""Test that extraction respects crawl mode."""
html = '''
<html>
<body>
<img src="/photo.jpg">
<video src="/video.mp4"></video>
<audio src="/audio.mp3"></audio>
</body>
</html>
'''
# Images only
images = extract_media_from_html(html, "https://example.com/", CrawlMode.IMAGES)
image_urls = [m['url'] for m in images]
assert any('photo.jpg' in url for url in image_urls)
assert not any('video.mp4' in url for url in image_urls)
# Videos only
videos = extract_media_from_html(html, "https://example.com/", CrawlMode.VIDEOS)
video_urls = [m['url'] for m in videos]
assert any('video.mp4' in url for url in video_urls)
class TestAsyncWebFetcher:
"""Test AsyncWebFetcher class."""
def test_fetcher_initialization(self):
"""Test AsyncWebFetcher initialization."""
fetcher = AsyncWebFetcher(
user_agent="TestBot/1.0",
default_crawl_delay=1.0,
)
assert fetcher.user_agent == "TestBot/1.0"
assert fetcher.default_crawl_delay == 1.0
def test_fetcher_default_values(self):
"""Test AsyncWebFetcher default values."""
fetcher = AsyncWebFetcher()
assert fetcher.default_crawl_delay == DEFAULT_CRAWL_DELAY
def test_get_domain(self):
"""Test domain extraction."""
fetcher = AsyncWebFetcher()
assert fetcher._get_domain("https://example.com/page") == "example.com"
assert fetcher._get_domain("https://sub.example.com/page") == "sub.example.com"
def test_simple_stem(self):
"""Test simple word stemming."""
fetcher = AsyncWebFetcher()
# Test stemming - actual behavior may vary
stem = fetcher._simple_stem("running")
assert "run" in stem # Should contain "run"
# Short words should not be stemmed
assert fetcher._simple_stem("go") == "go"
def test_extract_text_content(self):
"""Test text extraction from HTML."""
fetcher = AsyncWebFetcher()
html = '''
<html>
<head><title>Test Page</title></head>
<body>
<h1>Hello World</h1>
<p>This is a test paragraph.</p>
<script>alert('ignored');</script>
<style>.hidden { display: none; }</style>
</body>
</html>
'''
text = fetcher.extract_text_content(html)
assert "Hello World" in text
assert "test paragraph" in text
assert "alert" not in text # Script content should be removed
assert ".hidden" not in text # Style content should be removed
def test_extract_title(self):
"""Test title extraction from HTML."""
fetcher = AsyncWebFetcher()
html = '<html><head><title>Page Title</title></head><body></body></html>'
title = fetcher.extract_title(html)
assert title == "Page Title"
def test_extract_title_missing(self):
"""Test title extraction when missing."""
fetcher = AsyncWebFetcher()
html = '<html><head></head><body></body></html>'
title = fetcher.extract_title(html)
# Returns "Untitled" when no title present
assert title == "Untitled" or title == ""
class TestAsyncWebFetcherScoring:
"""Test link and page scoring functionality."""
def test_score_link_keyword_match(self):
"""Test link scoring with keyword matches."""
fetcher = AsyncWebFetcher()
# Link with keyword in URL
score = fetcher._score_link(
"https://example.com/pricing",
"View Pricing",
query_keywords=["pricing", "cost"],
)
assert score > 0 # Should have positive score
# Link without keywords
score_none = fetcher._score_link(
"https://example.com/about",
"About Us",
query_keywords=["pricing", "cost"],
)
assert score > score_none # Keyword match should score higher
def test_score_link_anchor_text(self):
"""Test link scoring with anchor text."""
fetcher = AsyncWebFetcher()
# Anchor text with keyword
score = fetcher._score_link(
"https://example.com/page",
"Our Pricing Plans",
query_keywords=["pricing"],
)
assert score > 0
def test_score_page_keyword_match(self):
"""Test page scoring with keyword matches."""
fetcher = AsyncWebFetcher()
page_data = {
"url": "https://example.com/pricing",
"title": "Pricing Plans",
"extracted_text": "Our pricing is competitive. See our cost breakdown.",
}
score = fetcher._score_page(
page_data,
query_keywords=["pricing", "cost"],
)
assert score > 0
class TestAsyncWebFetcherAsync:
"""Test async methods of AsyncWebFetcher."""
@pytest.mark.asyncio
async def test_can_fetch_respects_robots(self):
"""Test that _can_fetch respects robots.txt."""
fetcher = AsyncWebFetcher()
# Mock the robots.txt fetch to return a parser that disallows
with patch.object(fetcher, '_fetch_robots_txt', new_callable=AsyncMock) as mock_robots:
mock_parser = MagicMock()
mock_parser.can_fetch.return_value = False
mock_robots.return_value = mock_parser
result = await fetcher._can_fetch("https://example.com/blocked")
assert result is False
@pytest.mark.asyncio
async def test_can_fetch_allows_when_parser_allows(self):
"""Test _can_fetch when robots parser allows."""
fetcher = AsyncWebFetcher()
with patch.object(fetcher, '_fetch_robots_txt', new_callable=AsyncMock) as mock_robots:
mock_parser = MagicMock()
mock_parser.can_fetch.return_value = True
mock_robots.return_value = mock_parser
result = await fetcher._can_fetch("https://example.com/allowed")
assert result is True
@pytest.mark.asyncio
async def test_enforce_crawl_delay(self):
"""Test crawl delay enforcement."""
fetcher = AsyncWebFetcher(default_crawl_delay=0.1)
domain = "example.com"
# First request should not delay
start = asyncio.get_event_loop().time()
await fetcher._enforce_crawl_delay(domain)
first_time = asyncio.get_event_loop().time() - start
# Second request should delay
start = asyncio.get_event_loop().time()
await fetcher._enforce_crawl_delay(domain)
second_time = asyncio.get_event_loop().time() - start
# Second call should have some delay (at least part of 0.1s)
assert second_time >= 0.05 # Allow some tolerance
@pytest.mark.asyncio
async def test_check_media_url_signature(self):
"""Test check_media_url function signature."""
fetcher = AsyncWebFetcher()
# Verify the method exists and has correct signature
import inspect
sig = inspect.signature(fetcher.check_media_url)
params = list(sig.parameters.keys())
# Should have url parameter
assert 'url' in params or len(params) >= 1
class TestErrorHandling:
"""Test error handling functionality."""
def test_get_last_fetch_error_initially_none(self):
"""Test that last fetch error is None initially."""
clear_last_fetch_error()
error = get_last_fetch_error()
assert error is None
def test_clear_last_fetch_error(self):
"""Test clearing the last fetch error."""
clear_last_fetch_error()
assert get_last_fetch_error() is None
class TestExtensionSets:
"""Test extension and MIME type sets."""
def test_image_extensions(self):
"""Test image extension set."""
assert '.jpg' in IMAGE_EXTENSIONS
assert '.jpeg' in IMAGE_EXTENSIONS
assert '.png' in IMAGE_EXTENSIONS
assert '.gif' in IMAGE_EXTENSIONS
assert '.webp' in IMAGE_EXTENSIONS
assert '.svg' in IMAGE_EXTENSIONS
def test_video_extensions(self):
"""Test video extension set."""
assert '.mp4' in VIDEO_EXTENSIONS
assert '.webm' in VIDEO_EXTENSIONS
assert '.mov' in VIDEO_EXTENSIONS
assert '.avi' in VIDEO_EXTENSIONS
def test_audio_extensions(self):
"""Test audio extension set."""
assert '.mp3' in AUDIO_EXTENSIONS
assert '.wav' in AUDIO_EXTENSIONS
assert '.ogg' in AUDIO_EXTENSIONS
assert '.flac' in AUDIO_EXTENSIONS
if __name__ == '__main__':
pytest.main([__file__, '-v'])

343
tests/crawler/test_web_fetch.py Normal file
View file

@ -0,0 +1,343 @@
"""
Tests for core.web_fetch module.
Tests URL extraction, sources footer building, and web fetch helpers.
"""
import pytest
from unittest.mock import Mock, AsyncMock, patch
from aborist.sources.crawler.web_fetch import (
extract_url_from_content,
is_web_fetch_request,
build_sources_footer,
detect_fresh_fetch_intent,
extract_domain,
format_progress_message,
format_answering_message,
generate_keyword_stems,
)
class TestURLExtraction:
"""Test URL extraction from message content."""
def test_extract_url_with_protocol(self):
"""Test extracting URL with http:// or https://"""
content = "Check out https://example.com for more info"
url = extract_url_from_content(content)
assert url == "https://example.com"
def test_extract_url_http_protocol(self):
"""Test extracting URL with http:// protocol"""
content = "Visit http://insecure.example.com"
url = extract_url_from_content(content)
assert url == "http://insecure.example.com"
def test_extract_url_without_protocol(self):
"""Test extracting URL without protocol (adds https://)"""
content = "Go to example.com"
url = extract_url_from_content(content)
assert url == "https://example.com"
def test_extract_subdomain_url(self):
"""Test extracting URL with subdomain"""
content = "Check media.unturf.com/test"
url = extract_url_from_content(content)
assert url == "https://media.unturf.com/test"
def test_extract_url_with_path(self):
"""Test extracting URL with path"""
content = "Read https://docs.example.com/api/v1/reference"
url = extract_url_from_content(content)
assert url == "https://docs.example.com/api/v1/reference"
def test_extract_url_with_query(self):
"""Test extracting URL with query parameters"""
content = "See https://example.com/search?q=test&lang=en"
url = extract_url_from_content(content)
assert url == "https://example.com/search?q=test&lang=en"
def test_no_url_in_content(self):
"""Test when no URL is present"""
content = "This is just a regular message without any URLs"
url = extract_url_from_content(content)
assert url is None
def test_extract_first_url_when_multiple(self):
"""Test that first URL is extracted when multiple are present"""
content = "Check https://first.com and https://second.com"
url = extract_url_from_content(content)
assert url == "https://first.com"
def test_extract_url_with_www(self):
"""Test extracting URL with www prefix"""
content = "Visit www.example.com"
url = extract_url_from_content(content)
assert url == "https://www.example.com"
def test_extract_url_ignores_discord_embeds(self):
"""Test that angle brackets don't break URL extraction"""
content = "Check out <https://example.com>"
url = extract_url_from_content(content)
# Angle brackets should be part of the URL match
assert "example.com" in url
@pytest.mark.skip(
reason="chat-bot fetch triggers (has_web_fetch_trigger) were "
"dropped during the aborist lift — aborist has no chat surface; "
"see aborist/sources/crawler/__init__.py for rationale."
)
class TestWebFetchDetection:
"""Test web fetch request detection."""
def test_is_web_fetch_request_with_url(self):
"""Test detection when URL is present"""
assert is_web_fetch_request("Fetch https://example.com") is True
def test_is_web_fetch_request_without_url(self):
"""Test detection when no URL is present"""
assert is_web_fetch_request("Tell me about Python") is False
def test_is_web_fetch_request_domain_only(self):
"""Test detection with domain-only URL"""
assert is_web_fetch_request("What's on example.com?") is True
@pytest.mark.skip(
reason="chat-bot fresh-fetch triggers (has_fresh_fetch_trigger) were "
"dropped during the aborist lift — aborist has no chat surface; "
"see aborist/sources/crawler/__init__.py for rationale."
)
class TestFreshFetchDetection:
"""Test fresh fetch intent detection."""
def test_detect_fresh_with_keyword(self):
"""Test detection with 'fresh' keyword"""
assert detect_fresh_fetch_intent("Get fresh data from example.com") is True
def test_detect_refresh_with_keyword(self):
"""Test detection with 'refresh' keyword"""
assert detect_fresh_fetch_intent("Refresh the content") is True
def test_detect_reload_with_keyword(self):
"""Test detection with 'reload' keyword"""
assert detect_fresh_fetch_intent("Reload example.com") is True
def test_detect_latest_with_keyword(self):
"""Test detection with 'latest' keyword"""
assert detect_fresh_fetch_intent("Get the latest from example.com") is True
def test_no_fresh_intent(self):
"""Test when no fresh fetch keywords present"""
assert detect_fresh_fetch_intent("Fetch example.com") is False
def test_fresh_detection_case_insensitive(self):
"""Test that fresh detection is case insensitive"""
assert detect_fresh_fetch_intent("FRESH data please") is True
class TestSourcesFooter:
"""Test sources footer building."""
def test_build_sources_footer_single_source(self):
"""Test building footer with single source"""
sources = [{'url': 'https://example.com', 'title': 'Example Page'}]
footer = build_sources_footer(sources)
assert 'Sources:' in footer
assert 'Example Page' in footer
assert '<https://example.com>' in footer
def test_build_sources_footer_multiple_sources(self):
"""Test building footer with multiple sources"""
sources = [
{'url': 'https://example.com', 'title': 'Example Page'},
{'url': 'https://test.com', 'title': 'Test Page'}
]
footer = build_sources_footer(sources)
assert 'Example Page' in footer
assert 'Test Page' in footer
assert footer.count('<') == 2 # Two URLs with angle brackets
def test_build_sources_footer_empty_list(self):
"""Test building footer with empty source list"""
footer = build_sources_footer([])
assert footer == ""
def test_build_sources_footer_missing_title(self):
"""Test building footer when title is missing"""
sources = [{'url': 'https://example.com'}]
footer = build_sources_footer(sources)
assert 'Untitled' in footer
assert '<https://example.com>' in footer
def test_build_sources_footer_missing_url(self):
"""Test building footer when URL is missing"""
sources = [{'title': 'Example Page'}]
footer = build_sources_footer(sources)
# Should not include entry if URL is missing
assert footer == '**Sources:**\n'
def test_build_sources_footer_angle_brackets_suppress_embeds(self):
"""Test that angle brackets are used to suppress Discord embeds"""
sources = [{'url': 'https://example.com', 'title': 'Test'}]
footer = build_sources_footer(sources)
# URLs should be in angle brackets
assert '<https://example.com>' in footer
assert 'https://example.com>' not in footer.replace('<https://example.com>', '')
class TestDomainExtraction:
"""Test domain extraction from URLs."""
def test_extract_domain_basic(self):
"""Test extracting domain from basic URL"""
domain = extract_domain("https://example.com/path")
assert domain == "example.com"
def test_extract_domain_with_subdomain(self):
"""Test extracting domain with subdomain"""
domain = extract_domain("https://api.example.com")
assert domain == "api.example.com"
def test_extract_domain_with_port(self):
"""Test extracting domain with port"""
domain = extract_domain("https://example.com:8080/path")
assert domain == "example.com:8080"
def test_extract_domain_invalid_url(self):
"""Test extracting domain from invalid URL"""
domain = extract_domain("not a url")
# Should return None or handle gracefully
assert domain is None or isinstance(domain, str)
class TestProgressFormatting:
"""Test progress message formatting."""
def test_format_progress_message_basic(self):
"""Test basic progress message formatting"""
msg = format_progress_message(
url="https://example.com",
task="Find pricing",
keywords=["pricing", "cost"],
depth=2
)
assert "example.com" in msg
assert "pricing" in msg
assert "cost" in msg
assert "**Depth:** 2" in msg
def test_format_progress_message_with_crawl_delay(self):
"""Test progress message with crawl delay"""
msg = format_progress_message(
url="https://example.com",
crawl_delay=5.0
)
assert "**Crawl delay:** 5.0s" in msg
def test_format_progress_message_no_crawl_delay_if_default(self):
"""Test that default crawl delay is not shown"""
msg = format_progress_message(
url="https://example.com",
crawl_delay=2.0
)
assert "Crawl delay" not in msg
def test_format_progress_message_fresh_mode(self):
"""Test progress message with fresh mode"""
msg = format_progress_message(
url="https://example.com",
cache_mode="fresh"
)
assert "**Mode:** fresh" in msg
def test_format_answering_message(self):
"""Test answering phase message formatting"""
msg = format_answering_message(5)
assert "5 page(s)" in msg
assert "Analyzing" in msg
class TestKeywordStemming:
"""Test keyword stemming functionality."""
def test_generate_stems_basic(self):
"""Test generating stems from keywords"""
stems = generate_keyword_stems(["running", "tests"], [])
assert "run" in stems or "test" in stems
def test_generate_stems_no_duplicates(self):
"""Test that stems don't duplicate keywords"""
stems = generate_keyword_stems(["run"], ["running"])
# "run" is already in keywords, so stem shouldn't be added
assert "run" not in stems
def test_generate_stems_variations(self):
"""Test stem generation from variations"""
stems = generate_keyword_stems([], ["pricing", "prices"])
# Should generate stems if they're different from variations
assert isinstance(stems, list)
def test_generate_stems_no_stemming_needed(self):
"""Test when no stemming is needed"""
stems = generate_keyword_stems(["api", "cli"], [])
# Short words shouldn't be stemmed
assert len(stems) == 0
def test_generate_stems_phrase(self):
"""Test stem generation from phrase"""
stems = generate_keyword_stems(["pull requests"], [])
# Should split phrase and stem individual words
assert isinstance(stems, list)
class TestWebFetchIntegration:
"""Test web fetch integration with mock web cache manager."""
@pytest.mark.asyncio
async def test_fetch_and_cache_success(self):
"""Test successful fetch and cache operation"""
# This would require mocking web_cache_manager
# For now, just verify the function exists and has correct signature
from aborist.sources.crawler.web_fetch import fetch_and_cache
import inspect
sig = inspect.signature(fetch_and_cache)
assert 'url' in sig.parameters
assert 'web_cache_manager' in sig.parameters
assert 'keywords' in sig.parameters
assert 'depth' in sig.parameters
assert 'fresh' in sig.parameters
@pytest.mark.asyncio
async def test_fetch_and_cache_with_progress_callback(self):
"""Test fetch with progress callback"""
from aborist.sources.crawler.web_fetch import fetch_and_cache
# Mock web cache manager
mock_manager = Mock()
mock_manager.get_or_fetch = AsyncMock(return_value={
'url': 'https://example.com',
'title': 'Test',
'extracted_text': 'Content',
'source_urls': []
})
# Mock progress callback
progress_calls = []
async def progress_callback(msg):
progress_calls.append(msg)
result = await fetch_and_cache(
url='https://example.com',
web_cache_manager=mock_manager,
progress_callback=progress_callback
)
assert result is not None
assert result['url'] == 'https://example.com'
mock_manager.get_or_fetch.assert_called_once()
if __name__ == '__main__':
pytest.main([__file__, '-v'])