import base64
import hashlib
import io
import json
import os
import subprocess
import tempfile
import unittest
import xml.etree.ElementTree as ET
from contextlib import redirect_stderr, redirect_stdout
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import patch
from scripts import star_history
UTC = timezone.utc
TOKEN_SENTINEL = "TOKEN_TEST_DO_NOT_LEAK_7z9"
class FixedClock:
def __init__(self, value: str):
self.value = datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(
tzinfo=UTC
)
def now(self):
return self.value
class FakeGitHub:
def __init__(self, pages=None):
self.pages = pages or {}
self.page_calls = []
def fetch_stargazer_page(self, after):
self.page_calls.append(after)
return self.pages[after]
class FakeRunner:
def __init__(self, completed):
self.completed = completed
self.arguments = None
def run(self, arguments):
self.arguments = list(arguments)
return self.completed
def edge(cursor, timestamp):
return star_history.StargazerEdge(
cursor,
datetime.fromisoformat(timestamp.replace("Z", "+00:00")),
)
def page(total, edges, has_next=False, end_cursor=None, remaining=1_000):
return star_history.StargazerPage(
total_count=total,
edges=tuple(edges),
has_next_page=has_next,
end_cursor=end_cursor,
rate_remaining=remaining,
)
def state_with_snapshots(snapshots=None):
return {
"schema_version": 1,
"repository": "666ghj/MiroFish",
"timezone": "UTC",
"ongoing_interval_days": 13,
"reconstruction": {
"method": "current_stargazers_starred_at",
"generated_at": "2026-07-01T00:00:00Z",
"daily": [],
},
"snapshots": snapshots or [],
}
def seed_workspace(workspace, state):
state_path = workspace / ".github/star-history/history.json"
light_path = workspace / "static/image/star-history-light.svg"
dark_path = workspace / "static/image/star-history-dark.svg"
state_path.parent.mkdir(parents=True, exist_ok=True)
light_path.parent.mkdir(parents=True, exist_ok=True)
state_path.write_bytes(star_history.canonical_state_bytes(state))
light_path.write_bytes(star_history.render_svg(state, "light"))
dark_path.write_bytes(star_history.render_svg(state, "dark"))
return state_path, light_path, dark_path
class StarHistoryBehaviorTests(unittest.TestCase):
def test_backfill_writes_hand_computed_completed_daily_history(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
edges = [
edge("c6", "2026-03-02T00:01:00Z"),
edge("c5", "2026-03-01T00:00:00Z"),
edge("c4", "2026-02-28T23:59:59Z"),
edge("c3", "2026-02-28T23:59:59Z"),
edge("c2", "2026-02-27T10:00:00Z"),
edge("c1", "2026-02-25T12:00:00Z"),
]
github = FakeGitHub({None: page(6, edges)})
result = star_history.execute(
"backfill",
github=github,
clock=FixedClock("2026-03-02T12:00:00Z"),
workspace=workspace,
)
self.assertTrue(result.changed)
state = json.loads(
(workspace / ".github/star-history/history.json").read_text()
)
self.assertEqual(
state["reconstruction"]["daily"],
[
{"date": "2026-02-25", "stars": 1},
{"date": "2026-02-27", "stars": 2},
{"date": "2026-02-28", "stars": 4},
{"date": "2026-03-01", "stars": 5},
],
)
self.assertEqual(state["snapshots"], [])
star_history.check_workspace(workspace)
def test_backfill_reads_101_edges_across_pages(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
january_first = [
edge(f"jan1-{index}", f"2026-01-01T00:00:{index % 60:02d}Z")
for index in range(100)
]
first_edges = [edge("jan2", "2026-01-02T00:00:00Z")] + january_first[:99]
github = FakeGitHub(
{
None: page(101, first_edges, True, "next-page"),
"next-page": page(101, january_first[99:]),
}
)
star_history.execute(
"backfill",
github=github,
clock=FixedClock("2026-01-03T00:00:00Z"),
workspace=workspace,
)
state = json.loads(
(workspace / ".github/star-history/history.json").read_text()
)
self.assertEqual(github.page_calls, [None, "next-page"])
self.assertEqual(
state["reconstruction"]["daily"],
[
{"date": "2026-01-01", "stars": 100},
{"date": "2026-01-02", "stars": 101},
],
)
def test_backfill_fails_closed_on_cursor_or_count_inconsistency(self):
cases = {
"duplicate edge": page(
2,
[edge("same", "2026-01-01T00:00:00Z"), edge("same", "2026-01-02T00:00:00Z")],
),
"count mismatch": page(2, [edge("only", "2026-01-01T00:00:00Z")]),
}
for label, first_page in cases.items():
with self.subTest(label=label), tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
with self.assertRaises(star_history.StarHistoryError):
star_history.execute(
"backfill",
github=FakeGitHub({None: first_page}),
clock=FixedClock("2026-01-03T00:00:00Z"),
workspace=workspace,
)
self.assertFalse(
(workspace / ".github/star-history/history.json").exists()
)
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
github = FakeGitHub(
{
None: page(
2,
[edge("first", "2026-01-02T00:00:00Z")],
True,
"repeated-page",
),
"repeated-page": page(
2,
[edge("second", "2026-01-01T00:00:00Z")],
False,
"repeated-page",
),
}
)
with self.assertRaises(star_history.StarHistoryError):
star_history.execute(
"backfill",
github=github,
clock=FixedClock("2026-01-03T00:00:00Z"),
workspace=workspace,
)
self.assertFalse(
(workspace / ".github/star-history/history.json").exists()
)
def test_backfill_accepts_exact_rate_limit_reserve_after_first_page(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
github = FakeGitHub(
{
None: page(
1,
[edge("only", "2026-01-01T00:00:00Z")],
remaining=star_history.RATE_LIMIT_RESERVE,
)
}
)
result = star_history.execute(
"backfill",
github=github,
clock=FixedClock("2026-01-02T00:00:00Z"),
workspace=workspace,
)
self.assertTrue(result.changed)
self.assertEqual(github.page_calls, [None])
def test_backfill_refuses_dangling_output_symlink(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
state_path = workspace / ".github/star-history/history.json"
state_path.parent.mkdir(parents=True)
state_path.symlink_to(workspace / "missing-history.json")
with self.assertRaises(star_history.StarHistoryError):
star_history.execute(
"backfill",
github=FakeGitHub({None: page(0, [])}),
clock=FixedClock("2026-01-02T00:00:00Z"),
workspace=workspace,
)
self.assertTrue(state_path.is_symlink())
def test_initialize_starts_from_one_honest_aggregate_snapshot(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
result = star_history.execute(
"initialize",
github=None,
clock=FixedClock("2026-07-20T05:00:00Z"),
workspace=workspace,
star_count=42,
)
self.assertTrue(result.changed)
state = star_history.load_state(workspace)
self.assertEqual(
state["reconstruction"]["method"], "aggregate_snapshot_only"
)
self.assertEqual(state["reconstruction"]["daily"], [])
self.assertEqual(
state["snapshots"],
[{"at": "2026-07-20T05:00:00Z", "stars": 42}],
)
star_history.check_workspace(workspace)
def test_due_boundary_matches_configured_interval(self):
baseline = state_with_snapshots(
[{"at": "2026-07-20T05:00:00Z", "stars": 100}]
)
latest = datetime(2026, 7, 20, 5, 0, 0, tzinfo=UTC)
boundary = latest + star_history.timedelta(
days=star_history.INTERVAL_DAYS
)
cases = [
(
(boundary - star_history.timedelta(seconds=1)).strftime(
"%Y-%m-%dT%H:%M:%SZ"
),
False,
),
(boundary.strftime("%Y-%m-%dT%H:%M:%SZ"), True),
]
for now, expected in cases:
with self.subTest(now=now), tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
seed_workspace(workspace, baseline)
result = star_history.execute(
"due",
github=None,
clock=FixedClock(now),
workspace=workspace,
)
self.assertIs(result.due, expected)
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
seed_workspace(workspace, state_with_snapshots())
result = star_history.execute(
"due",
github=None,
clock=FixedClock("2026-07-20T05:00:00Z"),
workspace=workspace,
)
self.assertTrue(result.due)
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
future_state = state_with_snapshots()
future_state["reconstruction"]["generated_at"] = "2026-08-01T00:00:00Z"
seed_workspace(workspace, future_state)
with self.assertRaises(star_history.StarHistoryError):
star_history.execute(
"due",
github=None,
clock=FixedClock("2026-07-20T05:00:00Z"),
workspace=workspace,
)
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
seed_workspace(workspace, baseline)
with self.assertRaises(star_history.StarHistoryError):
star_history.execute(
"due",
github=None,
clock=FixedClock("2026-07-20T04:59:59Z"),
workspace=workspace,
)
def test_record_before_due_does_not_write(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
state_path, light_path, dark_path = seed_workspace(
workspace,
state_with_snapshots(
[{"at": "2026-07-20T05:00:00Z", "stars": 100}]
),
)
before = tuple(path.read_bytes() for path in (state_path, light_path, dark_path))
latest = datetime(2026, 7, 20, 5, 0, 0, tzinfo=UTC)
before_due = latest + star_history.timedelta(
days=star_history.INTERVAL_DAYS
) - star_history.timedelta(seconds=1)
result = star_history.execute(
"record",
github=None,
clock=FixedClock(before_due.strftime("%Y-%m-%dT%H:%M:%SZ")),
workspace=workspace,
star_count=101,
)
self.assertFalse(result.changed)
self.assertEqual(
before,
tuple(path.read_bytes() for path in (state_path, light_path, dark_path)),
)
def test_record_applies_count_file_without_github_credentials(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
state_path, light_path, dark_path = seed_workspace(
workspace, state_with_snapshots()
)
count_file = workspace / "fetched-count"
count_file.write_bytes(b"123\n")
stdout = io.StringIO()
stderr = io.StringIO()
with (
patch.dict(
os.environ,
{"GITHUB_TOKEN": TOKEN_SENTINEL, "GH_TOKEN": TOKEN_SENTINEL},
clear=False,
),
patch.object(star_history, "_repository_root", return_value=workspace),
patch.object(
star_history,
"SystemClock",
return_value=FixedClock("2026-07-20T05:00:00Z"),
),
redirect_stdout(stdout),
redirect_stderr(stderr),
):
exit_code = star_history.main(
["record", "--count-file", str(count_file), "--force"]
)
self.assertEqual(exit_code, 0)
self.assertEqual(stderr.getvalue(), "")
self.assertNotIn(TOKEN_SENTINEL, stdout.getvalue())
self.assertEqual(
star_history.load_state(workspace)["snapshots"],
[{"at": "2026-07-20T05:00:00Z", "stars": 123}],
)
for output in (state_path, light_path, dark_path):
self.assertNotIn(TOKEN_SENTINEL.encode(), output.read_bytes())
star_history.check_workspace(workspace)
def test_count_file_rejects_symlinks_malformed_and_extreme_values(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
valid = root / "valid"
valid.write_bytes(b"0\n")
self.assertEqual(star_history.load_star_count_file(valid), 0)
cases = {
"empty": b"",
"negative": b"-1\n",
"leading-zero": b"01\n",
"json": b'{"stargazers_count": 1}\n',
"too-large": b"9" * (star_history.MAX_COUNT_FILE_BYTES + 1),
"out-of-range": str(star_history.MAX_STAR_COUNT + 1).encode() + b"\n",
}
for name, payload in cases.items():
path = root / name
path.write_bytes(payload)
with self.subTest(name=name), self.assertRaises(
star_history.StarHistoryError
):
star_history.load_star_count_file(path)
link = root / "link"
link.symlink_to(valid)
with self.assertRaises(star_history.StarHistoryError):
star_history.load_star_count_file(link)
def test_force_same_day_same_count_is_idempotent(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
state_path, light_path, dark_path = seed_workspace(
workspace,
state_with_snapshots(
[{"at": "2026-07-20T05:00:00Z", "stars": 100}]
),
)
before = tuple(path.read_bytes() for path in (state_path, light_path, dark_path))
result = star_history.execute(
"record",
github=None,
clock=FixedClock("2026-07-20T06:00:00Z"),
workspace=workspace,
force=True,
star_count=100,
)
self.assertFalse(result.changed)
self.assertEqual(
before,
tuple(path.read_bytes() for path in (state_path, light_path, dark_path)),
)
def test_force_same_day_changed_count_replaces_snapshot(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
seed_workspace(
workspace,
state_with_snapshots(
[{"at": "2026-07-20T05:00:00Z", "stars": 100}]
),
)
result = star_history.execute(
"record",
github=None,
clock=FixedClock("2026-07-20T06:00:00Z"),
workspace=workspace,
force=True,
star_count=101,
)
self.assertTrue(result.changed)
state = star_history.load_state(workspace)
self.assertEqual(
state["snapshots"],
[{"at": "2026-07-20T06:00:00Z", "stars": 101}],
)
def test_new_date_appends_even_when_count_is_unchanged(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
seed_workspace(
workspace,
state_with_snapshots(
[{"at": "2026-07-20T05:00:00Z", "stars": 100}]
),
)
result = star_history.execute(
"record",
github=None,
clock=FixedClock("2026-08-04T05:00:00Z"),
workspace=workspace,
star_count=100,
)
self.assertTrue(result.changed)
self.assertEqual(
star_history.load_state(workspace)["snapshots"],
[
{"at": "2026-07-20T05:00:00Z", "stars": 100},
{"at": "2026-08-04T05:00:00Z", "stars": 100},
],
)
def test_schema_rejects_unknown_identity_fields_and_boolean_counts(self):
baseline = state_with_snapshots()
cases = []
top = deepcopy(baseline)
top["login"] = "secret-user"
cases.append(top)
reconstruction = deepcopy(baseline)
reconstruction["reconstruction"]["avatar"] = "secret-avatar"
cases.append(reconstruction)
daily = deepcopy(baseline)
daily["reconstruction"]["daily"] = [
{"date": "2026-06-30", "stars": 1, "user": "secret-user"}
]
cases.append(daily)
snapshot = deepcopy(baseline)
snapshot["snapshots"] = [
{"at": "2026-07-20T05:00:00Z", "stars": 1, "profile_url": "secret"}
]
cases.append(snapshot)
boolean_count = deepcopy(baseline)
boolean_count["snapshots"] = [
{"at": "2026-07-20T05:00:00Z", "stars": True}
]
cases.append(boolean_count)
for state in cases:
with self.subTest(state=state), self.assertRaises(
star_history.StarHistoryError
):
star_history.validate_state(state)
def test_svg_is_accessible_self_contained_and_deterministic(self):
state = {
"schema_version": 1,
"repository": "666ghj/MiroFish",
"timezone": "UTC",
"ongoing_interval_days": 13,
"reconstruction": {
"method": "current_stargazers_starred_at",
"generated_at": "2026-01-04T00:00:00Z",
"daily": [
{"date": "2026-01-01", "stars": 1},
{"date": "2026-01-03", "stars": 3},
],
},
"snapshots": [
{"at": "2026-01-20T10:00:00Z", "stars": 3},
{"at": "2026-02-04T10:00:00Z", "stars": 2},
],
}
light = star_history.render_svg(state, "light")
dark = star_history.render_svg(state, "dark")
self.assertEqual(light, star_history.render_svg(state, "light"))
self.assertNotEqual(light, dark)
self.assertIn(b"viewBox=\"0 0 800 533.333\"", light)
self.assertIn(b"Star History", light)
self.assertIn(b"666ghj/MiroFish", light)
self.assertIn(b"star-history.com", light)
self.assertIn(b"feTurbulence", light)
self.assertIn(b"feDisplacementMap", light)
self.assertIn(b"filter=\"url(#xkcdify)\"", light)
self.assertNotIn(b"stroke-dasharray", light)
self.assertNotIn(b"'
)
reviewed_watermark = (
''
)
valid_shell = (
'"
)
processing_instruction = (
''
+ valid_shell
)
unsafe_payloads = [
b'',
b'',
b'',
b'',
b'',
(
''
).encode(),
(
'"
).encode(),
processing_instruction.encode(),
processing_instruction.encode("utf-16"),
processing_instruction.encode("utf-16-be"),
b"\xef\xbb\xbf" + valid_shell.encode(),
(
''
).encode(),
valid_shell.replace(
"",
'',
).encode(),
valid_shell.replace(
"",
'',
).encode(),
valid_shell.replace(
"",
'',
).encode(),
valid_shell.replace(
"",
'',
).encode(),
]
for payload in unsafe_payloads:
with self.subTest(payload=payload), self.assertRaises(
star_history.StarHistoryError
):
star_history._validate_svg(payload)
def test_check_detects_svg_tampering(self):
with tempfile.TemporaryDirectory() as temporary:
workspace = Path(temporary)
_, light_path, _ = seed_workspace(workspace, state_with_snapshots())
light_path.write_bytes(
light_path.read_bytes().replace(b"Star History", b"Star Historx", 1)
)
with self.assertRaises(star_history.StarHistoryError):
star_history.execute(
"check",
github=None,
clock=FixedClock("2026-07-20T05:00:00Z"),
workspace=workspace,
)
class GitHubAdapterTests(unittest.TestCase):
def test_graphql_gateway_uses_identity_free_paginated_query(self):
payload = {
"data": {
"repository": {
"stargazers": {
"totalCount": 1,
"edges": [
{"cursor": "edge-cursor", "starredAt": "2026-01-01T00:00:00Z"}
],
"pageInfo": {"hasNextPage": False, "endCursor": "edge-cursor"},
}
},
"rateLimit": {"cost": 1, "remaining": 4999, "resetAt": "2026-01-01T01:00:00Z"},
}
}
runner = FakeRunner(
subprocess.CompletedProcess([], 0, json.dumps(payload), "")
)
gateway = star_history.GhGraphQLGateway(runner)
result = gateway.fetch_stargazer_page("previous-page")
arguments = "\n".join(runner.arguments)
self.assertEqual(result.total_count, 1)
self.assertIn("first: 100", arguments)
self.assertIn("after: $after", arguments)
self.assertIn("after=previous-page", arguments)
expected_query = """\
query StarTimes($owner: String!, $name: String!, $after: String) {
repository(owner: $owner, name: $name) {
stargazers(
first: 100
after: $after
orderBy: {field: STARRED_AT, direction: DESC}
) {
totalCount
edges { cursor starredAt }
pageInfo { hasNextPage endCursor }
}
}
rateLimit { cost remaining resetAt }
}
"""
self.assertEqual(star_history.GRAPHQL_QUERY, expected_query)
for forbidden in (
"nodes",
" node ",
" login ",
" avatar ",
" databaseId ",
" email ",
" url ",
TOKEN_SENTINEL,
):
self.assertNotIn(forbidden, arguments)
def test_graphql_gateway_rejects_malformed_nested_shape(self):
payload = {
"data": {
"repository": {
"stargazers": {
"totalCount": 0,
"edges": [],
"pageInfo": [],
}
},
"rateLimit": {"remaining": 4999},
}
}
gateway = star_history.GhGraphQLGateway(
FakeRunner(subprocess.CompletedProcess([], 0, json.dumps(payload), ""))
)
with self.assertRaises(star_history.StarHistoryError):
gateway.fetch_stargazer_page(None)
def test_graphql_gateway_does_not_echo_failed_command_stderr(self):
runner = FakeRunner(
subprocess.CompletedProcess([], 1, "", f"failure {TOKEN_SENTINEL}")
)
gateway = star_history.GhGraphQLGateway(runner)
with self.assertRaises(star_history.StarHistoryError) as captured:
gateway.fetch_stargazer_page(None)
self.assertNotIn(TOKEN_SENTINEL, str(captured.exception))
if __name__ == "__main__":
unittest.main()