1
+import io
2
+import json
3
+import tempfile
4
+import unittest
5
+from argparse import Namespace
6
+from datetime import UTC, datetime
7
+from pathlib import Path
8
+from unittest import mock
9
+
10
+import scripts.analysis_gate as analysis_gate
11
+import scripts.analyze_fallback as analyze_fallback
12
+import scripts.crawl as crawl
13
+import scripts.generate_content as generate_content
14
+
15
+
16
+class _FakeHTTPResponse(io.BytesIO):
17
+ def __enter__(self):
18
+ return self
19
+
20
+ def __exit__(self, exc_type, exc, tb):
21
+ self.close()
22
+ return False
23
+
24
+
25
+FIXED_RUN_DATETIME = "2026-05-18T08:00:00Z"
26
+FIXED_RUN_TIME = datetime(2026, 5, 18, 8, 0, 0, tzinfo=UTC)
27
+
28
+
29
+def make_api_repo(full_name: str, *, stars: int, created_at: str, topics: list[str]) -> dict:
30
+ owner, name = full_name.split("/", 1)
31
+ return {
32
+ "name": name,
33
+ "full_name": full_name,
34
+ "description": f"{name} helps teams ship reliable automation.",
35
+ "language": "Python",
36
+ "stargazers_count": stars,
37
+ "forks_count": max(1, stars // 10),
38
+ "created_at": created_at,
39
+ "topics": topics,
40
+ "license": {"spdx_id": "MIT"},
41
+ "html_url": f"https://github.com/{full_name}",
42
+ "owner": {"login": owner},
43
+ "fork": False,
44
+ "is_template": False,
45
+ }
46
+
47
+
48
+def make_raw_payload() -> dict:
49
+ return {
50
+ "week": "2026-W21",
51
+ "crawled_at": FIXED_RUN_DATETIME,
52
+ "new_repos": [
53
+ {
54
+ "name": "signal-kit",
55
+ "owner": "octo",
56
+ "full_name": "octo/signal-kit",
57
+ "description": "Signal extraction for release teams.",
58
+ "language": "Python",
59
+ "stars": 120,
60
+ "forks": 12,
61
+ "created_at": "2026-05-12T09:00:00Z",
62
+ "topics": ["ai", "automation", "developer-tooling"],
63
+ "license": "MIT",
64
+ "url": "https://github.com/octo/signal-kit",
65
+ }
66
+ ],
67
+ "trending_repos": [
68
+ {
69
+ "name": "momentum-watch",
70
+ "owner": "octo",
71
+ "full_name": "octo/momentum-watch",
72
+ "description": "Observability for weekly launches.",
73
+ "language": "Go",
74
+ "stars": 180,
75
+ "forks": 18,
76
+ "created_at": "2026-05-10T12:00:00Z",
77
+ "topics": ["observability", "analytics", "platform"],
78
+ "license": "Apache-2.0",
79
+ "url": "https://github.com/octo/momentum-watch",
80
+ "stars_gained": 35,
81
+ }
82
+ ],
83
+ "signals": {
84
+ "top_topics": [
85
+ {"topic": "automation", "count": 2},
86
+ {"topic": "observability", "count": 1},
87
+ ]
88
+ },
89
+ "metadata": {
90
+ "api_calls_used": 2,
91
+ "cache_hits": 1,
92
+ "stale_cache_hits": 0,
93
+ "rate_limit_limit": 5000,
94
+ "rate_limit_remaining": 4990,
95
+ "rate_limit_reset": 1747567200,
96
+ "rate_limit_resource": "search",
97
+ "partial_failures": [],
98
+ "snapshot_path": "data/snapshots/2026-W21-stars.json",
99
+ },
100
+ }
101
+
102
+
103
+def make_analysis_markdown() -> str:
104
+ return f'''---
105
+title: "Week 21, 2026 Analysis"
106
+date: {FIXED_RUN_DATETIME}
107
+week: "2026-W21"
108
+year: 2026
109
+tags: [ai, automation, developer-tooling]
110
+categories: [weekly]
111
+repos_featured: 2
112
+stars_tracked: 300
113
+top_repo: "octo/signal-kit"
114
+quality_score: 86
115
+summary: "Reliable automation and observability projects set the tone for the week."
116
+---
117
+
118
+## Notable New Repositories
119
+
120
+[octo/signal-kit](https://github.com/octo/signal-kit) stood out because it solves release coordination without pretending to be a full platform rewrite. The project packages practical automation, readable defaults, and evidence of disciplined engineering. Teams watching shipping velocity can understand why it matters in one pass, which is a stronger signal than yet another thin wrapper around generic assistants. The repo reads like operational software built for repeat use instead of launch-day theater.
121
+
122
+## Trending This Week
123
+
124
+[octo/momentum-watch](https://github.com/octo/momentum-watch) captured attention because the work is grounded in observability and run health rather than novelty claims. The weekly delta is directionally useful here, and the trend matters because more teams are prioritizing measurement, incident feedback loops, and durable visibility into developer workflows instead of vanity dashboards.
125
+
126
+## Trend Analysis
127
+
128
+### Signal
129
+
130
+The durable signal is a return to automation that lowers toil and gives teams more confidence in repeatable delivery. [octo/signal-kit](https://github.com/octo/signal-kit) and [octo/momentum-watch](https://github.com/octo/momentum-watch) both point toward software that reduces coordination overhead, improves trust in pipelines, and respects how operators actually work. That pattern is more convincing than broad claims about agents replacing engineering judgment.
131
+
132
+### Noise
133
+
134
+The weak signal is the usual rush of products that market autonomy without proving fit, maintenance discipline, or measurable outcomes. This week was healthier than most, but the broader ecosystem still produces wrappers that borrow the language of automation while skipping the hard parts of observability, testing, and operational ownership.
135
+
136
+## What's Missing
137
+
138
+### Gaps
139
+
140
+The biggest gap is stronger investment in security review, test ergonomics, and smaller-team operations tooling that can be adopted without a platform migration. The ecosystem is getting better at coordination, but it still underserves practical defensive tooling and deployment confidence for teams that need reliability before they need spectacle.
141
+
142
+## Conclusion
143
+
144
+The week matters because practical automation won attention on merit. If this pattern holds, the next wave of winners will be tools that save teams time, expose real operating signals, and make release quality easier to trust.
145
+'''
146
+
147
+
148
+class PipelineIntegrationTests(unittest.TestCase):
149
+ def test_crawl_script_produces_valid_json_output_schema(self) -> None:
150
+ tests_root = Path(__file__).resolve().parent
151
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
152
+ base = Path(tmpdir)
153
+ output_path = base / "data" / "raw" / "2026-W21.json"
154
+ snapshot_dir = base / "data" / "snapshots"
155
+ snapshot_dir.mkdir(parents=True)
156
+
157
+ new_repo = make_api_repo(
158
+ "octo/signal-kit",
159
+ stars=120,
160
+ created_at="2026-05-12T09:00:00Z",
161
+ topics=["ai", "automation", "developer-tooling"],
162
+ )
163
+ trending_repo = make_api_repo(
164
+ "octo/momentum-watch",
165
+ stars=180,
166
+ created_at="2026-05-10T12:00:00Z",
167
+ topics=["observability", "analytics", "platform"],
168
+ )
169
+
170
+ class FakeClient:
171
+ def __init__(self, token: str) -> None:
172
+ self.token = token
173
+ self.api_calls_used = 2
174
+ self.cache_hits = 1
175
+ self.stale_cache_hits = 0
176
+ self.rate_limit_limit = 5000
177
+ self.rate_limit_remaining = 4990
178
+ self.rate_limit_reset = 1747567200
179
+ self.rate_limit_resource = "search"
180
+ self.errors = []
181
+
182
+ def search_repositories(self, query: str, *, max_results: int = 1000):
183
+ if query.startswith("created:"):
184
+ return [new_repo]
185
+ if query.startswith("pushed:"):
186
+ return [trending_repo]
187
+ raise AssertionError(f"Unexpected query: {query}")
188
+
189
+ def has_readme(self, full_name: str) -> bool:
190
+ return True
191
+
192
+ args = Namespace(
193
+ since="2026-05-11",
194
+ as_of="2026-05-18",
195
+ max_results=10,
196
+ output=str(output_path),
197
+ )
198
+
199
+ with mock.patch.object(crawl, "parse_args", return_value=args), mock.patch.dict(
200
+ "os.environ", {"GITHUB_TOKEN": "token"}, clear=False
201
+ ), mock.patch.object(crawl, "GitHubClient", FakeClient), mock.patch.object(
202
+ crawl, "load_previous_star_snapshot", return_value={"octo/momentum-watch": 145}
203
+ ), mock.patch.object(crawl, "utc_now", return_value=FIXED_RUN_TIME), mock.patch.object(
204
+ crawl, "SNAPSHOT_ROOT", snapshot_dir
205
+ ):
206
+ exit_code = crawl.main()
207
+
208
+ self.assertEqual(exit_code, 0)
209
+ payload = json.loads(output_path.read_text(encoding="utf-8"))
210
+ crawl.validate_payload(payload)
211
+ self.assertEqual(payload["week"], "2026-W21")
212
+ self.assertEqual(payload["trending_repos"][0]["stars_gained"], 35)
213
+ self.assertTrue((snapshot_dir / "2026-W21-stars.json").exists())
214
+
215
+ def test_generate_content_produces_valid_hugo_content(self) -> None:
216
+ tests_root = Path(__file__).resolve().parent
217
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
218
+ base = Path(tmpdir)
219
+ summary_path = base / "data" / "analyzed" / "2026-W21-summary.md"
220
+ summary_path.parent.mkdir(parents=True)
221
+ summary_path.write_text(make_analysis_markdown(), encoding="utf-8")
222
+
223
+ previous_cwd = Path.cwd()
224
+ try:
225
+ import os
226
+
227
+ os.chdir(base)
228
+ output_path = generate_content.generate_content(summary_path)
229
+ finally:
230
+ os.chdir(previous_cwd)
231
+
232
+ self.assertEqual(output_path, base / "content" / "weekly" / "2026" / "W21.md")
233
+ rendered = output_path.read_text(encoding="utf-8")
234
+ self.assertIn('title: "Week 21, 2026"', rendered)
235
+ self.assertIn('week: "2026-W21"', rendered)
236
+ self.assertIn("draft: false", rendered)
237
+ self.assertNotIn("quality_score", rendered)
238
+ self.assertIn("## Notable New Repositories", rendered)
239
+
240
+ def test_analyze_fallback_can_process_raw_data(self) -> None:
241
+ tests_root = Path(__file__).resolve().parent
242
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
243
+ base = Path(tmpdir)
244
+ raw_path = base / "data" / "raw" / "2026-W21.json"
245
+ output_path = base / "data" / "analyzed" / "2026-W21-summary.md"
246
+ raw_path.parent.mkdir(parents=True)
247
+ output_path.parent.mkdir(parents=True)
248
+ raw_path.write_text(json.dumps(make_raw_payload()), encoding="utf-8")
249
+
250
+ response = _FakeHTTPResponse(
251
+ json.dumps({"choices": [{"message": {"content": make_analysis_markdown()}}]}).encode("utf-8")
252
+ )
253
+
254
+ with mock.patch.dict("os.environ", {"GITHUB_TOKEN": "token"}, clear=False), mock.patch.object(
255
+ analyze_fallback.request, "urlopen", return_value=response
256
+ ):
257
+ exit_code = analyze_fallback.main(
258
+ [
259
+ "--raw-json",
260
+ str(raw_path),
261
+ "--output",
262
+ str(output_path),
263
+ "--current-datetime",
264
+ FIXED_RUN_DATETIME,
265
+ "--analyzed-dir",
266
+ str(output_path.parent),
267
+ ]
268
+ )
269
+
270
+ self.assertEqual(exit_code, 0)
271
+ written = output_path.read_text(encoding="utf-8")
272
+ self.assertIn("Week 21, 2026 Analysis", written)
273
+ self.assertIn("## Trend Analysis", written)
274
+
275
+ def test_analysis_gate_validates_analysis_output_correctly(self) -> None:
276
+ tests_root = Path(__file__).resolve().parent
277
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
278
+ base = Path(tmpdir)
279
+ raw_path = base / "data" / "raw" / "2026-W21.json"
280
+ raw_path.parent.mkdir(parents=True)
281
+ raw_path.write_text(json.dumps(make_raw_payload()), encoding="utf-8")
282
+
283
+ valid_path = base / "data" / "analyzed" / "2026-W21-summary.md"
284
+ valid_path.parent.mkdir(parents=True)
285
+ valid_path.write_text(make_analysis_markdown(), encoding="utf-8")
286
+
287
+ self.assertEqual(
288
+ analysis_gate.main(
289
+ [
290
+ "--analysis-file",
291
+ str(valid_path),
292
+ "--raw-json",
293
+ str(raw_path),
294
+ "--current-datetime",
295
+ FIXED_RUN_DATETIME,
296
+ "--source",
297
+ "integration-test",
298
+ ]
299
+ ),
300
+ 0,
301
+ )
302
+
303
+ invalid_path = base / "data" / "analyzed" / "invalid-summary.md"
304
+ invalid_path.write_text(make_analysis_markdown().replace("quality_score: 86", "quality_score: 40"), encoding="utf-8")
305
+
306
+ with self.assertRaises(SystemExit) as exc:
307
+ analysis_gate.main(
308
+ [
309
+ "--analysis-file",
310
+ str(invalid_path),
311
+ "--raw-json",
312
+ str(raw_path),
313
+ "--current-datetime",
314
+ FIXED_RUN_DATETIME,
315
+ "--source",
316
+ "integration-test",
317
+ ]
318
+ )
319
+
320
+ self.assertEqual(exc.exception.code, 1)
321
+
322
+
323
+if __name__ == "__main__":
324
+ unittest.main()