Space-Banane
126 lines
3.7 KiB
Python
126 lines
3.7 KiB
Python
import tempfile
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.main import QueryOptions, collect_merged_activity, _fetch_source_usernames
|
|
from app.cache import FileCache
|
|
from app.settings import Settings
|
|
from app.sources import github as github_source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_merged_activity_combines_multiple_usernames(monkeypatch) -> None:
|
|
settings = Settings(
|
|
_env_file=None,
|
|
GITHUB_USERNAME="octocat, hubot",
|
|
GITEA_BASE_URL="https://gitea.example.com",
|
|
GITEA_USERNAME="alice, bob",
|
|
)
|
|
|
|
github_calls: list[str] = []
|
|
gitea_calls: list[str] = []
|
|
|
|
async def fake_fetch_github_activity(username, token, from_date, to_date, timeout_seconds=20.0):
|
|
github_calls.append(username)
|
|
return {"2026-01-01": 1 if username == "octocat" else 2}
|
|
|
|
async def fake_fetch_gitea_activity(base_url, username, token, from_date, to_date, timeout_seconds=20.0):
|
|
gitea_calls.append(username)
|
|
return {"2026-01-01": 3 if username == "alice" else 4}
|
|
|
|
monkeypatch.setattr("app.main.fetch_github_activity", fake_fetch_github_activity)
|
|
monkeypatch.setattr("app.main.fetch_gitea_activity", fake_fetch_gitea_activity)
|
|
|
|
result = await collect_merged_activity(
|
|
settings=settings,
|
|
cache=None,
|
|
options=QueryOptions(year=None, days=1, theme="light", source="all"),
|
|
)
|
|
|
|
assert sorted(github_calls) == ["hubot", "octocat"]
|
|
assert sorted(gitea_calls) == ["alice", "bob"]
|
|
assert result.merged["2026-01-01"] == {"github": 3, "gitea": 7, "total": 10}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_source_usernames_skips_failed_accounts() -> None:
|
|
async def ok_fetch():
|
|
return {"2026-01-01": 2}
|
|
|
|
async def bad_fetch():
|
|
raise github_source.GitHubSourceError("boom")
|
|
|
|
result, stale = await _fetch_source_usernames(
|
|
None,
|
|
"github",
|
|
["octocat", "hubot"],
|
|
date(2026, 1, 1),
|
|
date(2026, 1, 1),
|
|
lambda username: ok_fetch if username == "octocat" else bad_fetch,
|
|
)
|
|
|
|
assert result == {"2026-01-01": 2}
|
|
assert stale is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_source_usernames_does_not_swallow_programming_errors() -> None:
|
|
def broken_fetcher(username):
|
|
async def fetch():
|
|
raise TypeError("bad shape")
|
|
|
|
return fetch
|
|
|
|
with pytest.raises(TypeError, match="bad shape"):
|
|
await _fetch_source_usernames(
|
|
None,
|
|
"github",
|
|
["octocat", "hubot"],
|
|
date(2026, 1, 1),
|
|
date(2026, 1, 1),
|
|
broken_fetcher,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_source_usernames_keeps_cache_after_username_list_changes() -> None:
|
|
with tempfile.TemporaryDirectory(dir=Path.cwd()) as temp_dir:
|
|
cache = FileCache(cache_dir=temp_dir, default_ttl_seconds=0)
|
|
|
|
def prime_fetcher(username):
|
|
async def fetch():
|
|
return {"2026-01-01": 1}
|
|
|
|
return fetch
|
|
|
|
await _fetch_source_usernames(
|
|
cache,
|
|
"github",
|
|
["octocat"],
|
|
date(2026, 1, 1),
|
|
date(2026, 1, 1),
|
|
prime_fetcher,
|
|
)
|
|
|
|
def fetch_after_change(username):
|
|
async def fetch():
|
|
if username == "octocat":
|
|
raise github_source.GitHubSourceError("down")
|
|
return {"2026-01-01": 2}
|
|
|
|
return fetch
|
|
|
|
result, stale = await _fetch_source_usernames(
|
|
cache,
|
|
"github",
|
|
["octocat", "hubot"],
|
|
date(2026, 1, 1),
|
|
date(2026, 1, 1),
|
|
fetch_after_change,
|
|
)
|
|
|
|
assert result == {"2026-01-01": 3}
|
|
assert stale is True
|