saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
140 lines
4.8 KiB
Python
140 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.collectors.base import CompanyContext, SourceConfig
|
|
from app.collectors.website import WebsiteCollector
|
|
from app.models.enums import SourceStatus, SourceType
|
|
|
|
ARTICLE_HTML = """
|
|
<html><body>
|
|
<nav>Home | About</nav>
|
|
<article>
|
|
<h1>{title}</h1>
|
|
<p>{body}</p>
|
|
</article>
|
|
</body></html>
|
|
"""
|
|
|
|
COMPANY = CompanyContext(
|
|
id="c1", name="Acme Corp", official_website="https://example.com", monitoring_focus=None
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_discover_uses_sitemap_when_available():
|
|
with respx.mock:
|
|
respx.get("https://example.com/robots.txt").mock(return_value=httpx.Response(404))
|
|
respx.get("https://example.com/sitemap.xml").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
content=(
|
|
'<?xml version="1.0"?>'
|
|
'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'
|
|
"<url><loc>https://example.com/about</loc></url>"
|
|
"<url><loc>https://example.com/products</loc></url>"
|
|
"</urlset>"
|
|
),
|
|
)
|
|
)
|
|
collector = WebsiteCollector()
|
|
discovered = await collector.discover(COMPANY)
|
|
|
|
assert len(discovered) == 1
|
|
pages = discovered[0].configuration_metadata["pages"]
|
|
assert "https://example.com/about" in pages
|
|
assert "https://example.com/products" in pages
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_discover_falls_back_to_heuristic_paths_without_sitemap():
|
|
with respx.mock:
|
|
respx.get("https://example.com/robots.txt").mock(return_value=httpx.Response(404))
|
|
respx.get("https://example.com/sitemap.xml").mock(return_value=httpx.Response(404))
|
|
collector = WebsiteCollector()
|
|
discovered = await collector.discover(COMPANY)
|
|
|
|
pages = discovered[0].configuration_metadata["pages"]
|
|
assert any(p.endswith("/about") for p in pages)
|
|
assert any(p.endswith("/careers") for p in pages)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_extracts_documents_from_configured_pages():
|
|
source = SourceConfig(
|
|
id="s1",
|
|
source_type=SourceType.WEBSITE,
|
|
name="Acme website",
|
|
base_url="https://example.com",
|
|
configuration_metadata={
|
|
"pages": ["https://example.com/about", "https://example.com/products"]
|
|
},
|
|
)
|
|
with respx.mock:
|
|
respx.get("https://example.com/robots.txt").mock(return_value=httpx.Response(404))
|
|
respx.get("https://example.com/about").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
html=ARTICLE_HTML.format(
|
|
title="About Acme", body="Acme Corp builds electric utility vehicles."
|
|
),
|
|
)
|
|
)
|
|
respx.get("https://example.com/products").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
html=ARTICLE_HTML.format(
|
|
title="Products", body="Our product line includes trucks and vans."
|
|
),
|
|
)
|
|
)
|
|
collector = WebsiteCollector()
|
|
result = await collector.collect(source, COMPANY)
|
|
|
|
assert result.status == SourceStatus.ACTIVE
|
|
assert len(result.documents) == 2
|
|
assert any("electric utility vehicles" in d.content_text for d in result.documents)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_skips_pages_disallowed_by_robots_txt():
|
|
source = SourceConfig(
|
|
id="s1",
|
|
source_type=SourceType.WEBSITE,
|
|
name="Acme website",
|
|
base_url="https://example.com",
|
|
configuration_metadata={"pages": ["https://example.com/private"]},
|
|
)
|
|
with respx.mock:
|
|
respx.get("https://example.com/robots.txt").mock(
|
|
return_value=httpx.Response(200, text="User-agent: *\nDisallow: /private\n")
|
|
)
|
|
collector = WebsiteCollector()
|
|
result = await collector.collect(source, COMPANY)
|
|
|
|
assert result.documents == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_deduplicates_identical_content_across_pages():
|
|
source = SourceConfig(
|
|
id="s1",
|
|
source_type=SourceType.WEBSITE,
|
|
name="Acme website",
|
|
base_url="https://example.com",
|
|
configuration_metadata={"pages": ["https://example.com/a", "https://example.com/a-mirror"]},
|
|
)
|
|
same_html = ARTICLE_HTML.format(title="Same", body="Identical content on both URLs.")
|
|
with respx.mock:
|
|
respx.get("https://example.com/robots.txt").mock(return_value=httpx.Response(404))
|
|
respx.get("https://example.com/a").mock(return_value=httpx.Response(200, html=same_html))
|
|
respx.get("https://example.com/a-mirror").mock(
|
|
return_value=httpx.Response(200, html=same_html)
|
|
)
|
|
collector = WebsiteCollector()
|
|
result = await collector.collect(source, COMPANY)
|
|
|
|
assert len(result.documents) == 1
|