From 84286a7eaa31bdc5737cbbb54ad5957dc2e446f2 Mon Sep 17 00:00:00 2001 From: arsam Date: Fri, 18 Sep 2026 23:06:51 +0330 Subject: [PATCH] Import tests: tests/test_scraper Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality. --- tests/test_scraper.py | 69 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 tests/test_scraper.py diff --git a/tests/test_scraper.py b/tests/test_scraper.py new file mode 100644 index 0000000..76419d7 --- /dev/null +++ b/tests/test_scraper.py @@ -0,0 +1,69 @@ +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