1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
from datetime import UTC, datetime
from fastapi.testclient import TestClient
from sqlalchemy import select
from srht_contrib.main import create_app
from srht_contrib.models import ContributionEvent, TrackedActor
def test_read_only_contribution_routes_are_public_and_write_routes_require_api_key(settings, db_engine, session_factory) -> None:
app = create_app(settings, engine=db_engine, session_factory=session_factory)
with TestClient(app) as open_client:
response = open_client.get("/health")
public_contributions = open_client.get("/api/contributions/~ccleberg?from=2026-03-28&to=2026-03-30")
public_stats = open_client.get("/api/contributions/~ccleberg/stats?from=2026-03-28&to=2026-03-30")
assert response.status_code == 200
assert public_contributions.status_code == 200
assert public_stats.status_code == 200
with TestClient(app) as unauthorized:
unauthorized_response = unauthorized.post("/api/contributions/poll?actor=~ccleberg")
assert unauthorized_response.status_code == 401
with TestClient(app) as invalid:
invalid.headers.update({"X-API-Key": "wrong-key"})
invalid_response = invalid.get("/api/repositories")
assert invalid_response.status_code == 401
def test_contributions_api_returns_zero_filled_range(client: TestClient, db_session) -> None:
db_session.add(
ContributionEvent(
service="todo",
event_type="ticket_created",
actor="~ccleberg",
repo_name=None,
resource_id="1",
external_uid="todo:ticket:1:created",
occurred_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC),
weight=1.0,
raw_payload_json=None,
)
)
db_session.commit()
response = client.get("/api/contributions/~ccleberg?from=2026-03-28&to=2026-03-30")
assert response.status_code == 200
assert response.json()["is_indexed"] is True
assert response.json()["indexing_state"] == "indexed"
assert response.json()["is_recent_window_backfilled"] is False
assert response.json()["recent_backfill_state"] == "pending"
assert response.json()["days"] == [
{"date": "2026-03-28", "count": 0, "score": 0.0},
{"date": "2026-03-29", "count": 0, "score": 0.0},
{"date": "2026-03-30", "count": 1, "score": 1.0},
]
def test_public_read_registers_actor_for_lazy_indexing(client: TestClient, db_session) -> None:
response = client.get("/api/contributions/~ccleberg?from=2026-03-28&to=2026-03-30")
tracked_actor = db_session.scalar(select(TrackedActor).where(TrackedActor.actor == "~ccleberg"))
assert response.status_code == 200
assert response.json()["is_indexed"] is False
assert response.json()["indexing_state"] == "pending"
assert response.json()["is_recent_window_backfilled"] is False
assert response.json()["recent_backfill_state"] == "pending"
assert response.json()["recent_backfill_completed_at"] is None
assert response.json()["is_backfilled"] is False
assert response.json()["backfill_state"] == "pending"
assert response.json()["backfill_completed_at"] is None
assert response.json()["last_polled_at"] is None
assert tracked_actor is not None
assert tracked_actor.is_active is True
assert tracked_actor.last_requested_at is not None
def test_contribution_stats_api(client: TestClient, db_session) -> None:
db_session.add_all(
[
ContributionEvent(
service="todo",
event_type="ticket_created",
actor="~ccleberg",
repo_name=None,
resource_id="1",
external_uid="todo:ticket:1:created",
occurred_at=datetime(2026, 3, 29, 10, 0, tzinfo=UTC),
weight=1.0,
raw_payload_json=None,
),
ContributionEvent(
service="todo",
event_type="ticket_comment",
actor="~ccleberg",
repo_name=None,
resource_id="1",
external_uid="todo:ticket:1:comment:2",
occurred_at=datetime(2026, 3, 30, 11, 0, tzinfo=UTC),
weight=0.5,
raw_payload_json=None,
),
]
)
db_session.commit()
response = client.get("/api/contributions/~ccleberg/stats?from=2026-03-28&to=2026-03-30")
assert response.status_code == 200
assert response.json()["total_events"] == 2
assert response.json()["total_score"] == 1.5
assert response.json()["longest_streak"] == 2
assert response.json()["current_streak"] == 2
assert response.json()["is_indexed"] is True
assert response.json()["indexing_state"] == "indexed"
assert response.json()["is_recent_window_backfilled"] is False
assert response.json()["recent_backfill_state"] == "pending"
assert response.json()["is_backfilled"] is False
def test_invalid_date_input_returns_400(client: TestClient) -> None:
response = client.get("/api/contributions/~ccleberg?from=2026-13-01&to=2026-03-30")
assert response.status_code == 400
assert "Invalid date format" in response.json()["detail"]
def test_contribution_routes_use_settings_backed_alias_resolution(settings, db_engine, session_factory) -> None:
alias_settings = settings.model_copy(update={"actor_aliases_json": {"~ccleberg": ["[email protected]"]}})
app = create_app(alias_settings, engine=db_engine, session_factory=session_factory)
with session_factory() as session:
session.add(
ContributionEvent(
service="todo",
event_type="ticket_created",
actor="~ccleberg",
repo_name=None,
resource_id="1",
external_uid="todo:alias:1",
occurred_at=datetime(2026, 3, 30, 10, 0, tzinfo=UTC),
weight=1.0,
raw_payload_json=None,
)
)
session.commit()
with TestClient(app) as client:
client.headers.update({"X-API-Key": alias_settings.api_key})
response = client.get("/api/contributions/[email protected]?from=2026-03-30&to=2026-03-30")
assert response.status_code == 200
assert response.json()["actor"] == "~ccleberg"
|
