arsam
84286a7eaa
Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
import pytest
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scrape_url_returns_none_on_connection_error(monkeypatch):
|
|
import httpx
|
|
|
|
class FailClient:
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *_): pass
|
|
async def get(self, *args, **kwargs):
|
|
raise httpx.ConnectError("refused")
|
|
|
|
monkeypatch.setattr("axon.services.scraper.httpx.AsyncClient", FailClient)
|
|
|
|
from axon.services.scraper import scrape_url
|
|
result = await scrape_url("https://unreachable.example.com")
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scrape_url_returns_none_on_4xx(monkeypatch):
|
|
import httpx
|
|
|
|
class BadResponse:
|
|
def raise_for_status(self):
|
|
raise httpx.HTTPStatusError("404", request=None, response=None)
|
|
@property
|
|
def text(self): return ""
|
|
|
|
class FakeClient:
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *_): pass
|
|
async def get(self, *args, **kwargs): return BadResponse()
|
|
|
|
monkeypatch.setattr("axon.services.scraper.httpx.AsyncClient", FakeClient)
|
|
|
|
from axon.services.scraper import scrape_url
|
|
result = await scrape_url("https://example.com/404")
|
|
assert result is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scrape_urls_returns_dict_keyed_by_url(monkeypatch):
|
|
async def fake_scrape(url, timeout=5):
|
|
return f"content of {url}"
|
|
|
|
monkeypatch.setattr("axon.services.scraper.scrape_url", fake_scrape)
|
|
|
|
from axon.services.scraper import scrape_urls
|
|
urls = ["https://a.com", "https://b.com"]
|
|
result = await scrape_urls(urls, max_concurrent=2, timeout=5)
|
|
assert result["https://a.com"] == "content of https://a.com"
|
|
assert result["https://b.com"] == "content of https://b.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scrape_urls_skips_individual_failures(monkeypatch):
|
|
async def fake_scrape(url, timeout=5):
|
|
if "bad" in url:
|
|
return None
|
|
return "good content"
|
|
|
|
monkeypatch.setattr("axon.services.scraper.scrape_url", fake_scrape)
|
|
|
|
from axon.services.scraper import scrape_urls
|
|
result = await scrape_urls(["https://good.com", "https://bad.com"])
|
|
assert result["https://good.com"] == "good content"
|
|
assert result["https://bad.com"] is None
|