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).
74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.collectors.base import CompanyContext, SourceConfig
|
|
from app.collectors.github import GithubCollector
|
|
from app.models.enums import SourceStatus, SourceType
|
|
|
|
COMPANY = CompanyContext(id="c1", name="Acme Corp", official_website=None, monitoring_focus=None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_discover_finds_org_login():
|
|
with respx.mock:
|
|
respx.get("https://api.github.com/search/users").mock(
|
|
return_value=httpx.Response(200, text=json.dumps({"items": [{"login": "acme-corp"}]}))
|
|
)
|
|
discovered = await GithubCollector().discover(COMPANY)
|
|
|
|
assert discovered[0].configuration_metadata["org"] == "acme-corp"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_returns_repo_documents():
|
|
source = SourceConfig(
|
|
id="s1",
|
|
source_type=SourceType.GITHUB,
|
|
name="Acme GitHub",
|
|
base_url="https://github.com/acme-corp",
|
|
configuration_metadata={"org": "acme-corp"},
|
|
)
|
|
repos = [
|
|
{
|
|
"full_name": "acme-corp/battery-sdk",
|
|
"description": "SDK for battery management systems",
|
|
"language": "Python",
|
|
"stargazers_count": 42,
|
|
"pushed_at": "2026-05-01T12:00:00Z",
|
|
"html_url": "https://github.com/acme-corp/battery-sdk",
|
|
"archived": False,
|
|
}
|
|
]
|
|
with respx.mock:
|
|
respx.get("https://api.github.com/orgs/acme-corp/repos").mock(
|
|
return_value=httpx.Response(200, text=json.dumps(repos))
|
|
)
|
|
result = await GithubCollector().collect(source, COMPANY)
|
|
|
|
assert result.status == SourceStatus.ACTIVE
|
|
assert len(result.documents) == 1
|
|
assert "battery management" in result.documents[0].content_text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_reports_rate_limited_on_403():
|
|
source = SourceConfig(
|
|
id="s1",
|
|
source_type=SourceType.GITHUB,
|
|
name="Acme GitHub",
|
|
base_url=None,
|
|
configuration_metadata={"org": "acme-corp"},
|
|
)
|
|
with respx.mock:
|
|
respx.get("https://api.github.com/orgs/acme-corp/repos").mock(
|
|
return_value=httpx.Response(403)
|
|
)
|
|
result = await GithubCollector().collect(source, COMPANY)
|
|
|
|
assert result.status == SourceStatus.RATE_LIMITED
|