fix: date-order tags and scope the write guard

check_version read latest_tag as the first entry of GitHub's REST
listing, which sorts by name (nesoid: 2.5 before 2.5-4, gargoyle:
stable-20090825 before 2026.1.1), and found the declared tag only on that
first page of 100. The newest tag now comes from GraphQL by commit date,
and the declared one is asked for by name on GitHub, GitLab and Forgejo.

The clean-tree guard of the write modes looked at the whole emulators
directory: a loop of --emulator X --rebase-refs refused every profile
after the first it wrote. It now looks at the selected profiles only.
This commit is contained in:
Abdessamad Derraz committed 2026-10-23 15:40:56 +02:00
1 parent 18fc5ce53f
commit 9dd2a3d95f
4 files changed
+169 -21

No files matched your search

+22 -20
View File
@@ -2179,20 +2179,16 @@ def check_version(
declared = str(profile.get("core_version") or "")
if not declared:
return None
tags = upstream.list_tags(repo, cache_dir, offline)
release = upstream.latest_release(repo, cache_dir, offline)
matches = declared in tags
commit = (
upstream.resolve_tag_commit(repo, declared, cache_dir, offline)
if matches
else None
)
# Asked by name: the first page of tags misses a release that a project
# publishing nightlies has pushed past a hundred newer tags.
commit = upstream.tag_commit(repo, declared, cache_dir, offline)
return VersionReport(
declared,
tags[0] if tags else None,
upstream.newest_tag(repo, cache_dir, offline),
release.tag if release else None,
release.date if release else None,
matches,
commit is not None,
commit,
)
@@ -3462,31 +3458,37 @@ def realign_prose(
return messages
def _refuse_dirty_tree(emulators_dir: str) -> None:
"""Writes need a clean profile tree to roll back to."""
def _refuse_dirty_tree(emulators_dir: str, names: list[str]) -> None:
"""Writes need a clean copy of each profile they touch to roll back to.
Only the selected profiles count: a pass writing one profile after
another used to refuse every profile after the first it wrote.
"""
try:
dirty = emulators_dir_is_dirty(emulators_dir)
dirty = emulators_dir_is_dirty(emulators_dir, names)
except RuntimeError as e:
print(f"{e}. Pass --force to write anyway.", file=sys.stderr)
raise SystemExit(1) from e
if dirty:
print(
f"{emulators_dir} carries uncommitted changes. "
"Commit them first or pass --force.",
f"{emulators_dir} carries uncommitted changes to the selected "
"profiles. Commit them first or pass --force.",
file=sys.stderr,
)
raise SystemExit(1)
def emulators_dir_is_dirty(emulators_dir: str) -> bool:
def emulators_dir_is_dirty(emulators_dir: str, names: list[str] | None = None) -> bool:
"""True when the profile directory carries uncommitted changes.
Git is asked from inside the directory, so a profile tree that lives in
another repository answers for itself. A directory git cannot read is
never reported clean: the writes have nothing to roll back to.
``names`` narrows the question to those profiles. Git is asked from
inside the directory, so a profile tree that lives in another
repository answers for itself. A directory git cannot read is never
reported clean: the writes have nothing to roll back to.
"""
paths = [f"{name}.yml" for name in names] if names is not None else ["."]
result = subprocess.run(
["git", "-C", emulators_dir, "status", "--porcelain", "--", "."],
["git", "-C", emulators_dir, "status", "--porcelain", "--", *paths],
capture_output=True,
text=True,
check=False,
@@ -4227,7 +4229,7 @@ def main() -> None:
or args.realign_prose or args.attribute_refs
)
if writes and not args.dry_run and not args.force:
_refuse_dirty_tree(args.emulators_dir)
_refuse_dirty_tree(args.emulators_dir, list(selected))
# A profile that could not be compared or written is not a profile with
# nothing to review: it stays in every output and fails the run.
+86 -1
View File
@@ -614,6 +614,73 @@ def list_tags(repo: Repo, cache_dir: str, offline: bool = False) -> list[str]:
return [t["name"] for t in payload if isinstance(t, dict) and t.get("name")]
GITHUB_GRAPHQL = "https://api.github.com/graphql"
_NEWEST_TAG_QUERY = (
"query($owner:String!,$name:String!){repository(owner:$owner,name:$name)"
"{refs(refPrefix:\"refs/tags/\",first:1,"
"orderBy:{field:TAG_COMMIT_DATE,direction:DESC}){nodes{name}}}}"
)
def _github_graphql(
query: str, variables: dict, cache_dir: str, offline: bool
) -> object | None:
"""A cached GitHub GraphQL answer, or None without a token.
GraphQL refuses anonymous calls, so an untokened run keeps the REST
answer it already had.
"""
from common import github_token # noqa: PLC0415
body = json.dumps({"query": query, "variables": variables}, sort_keys=True)
target = Path(cache_dir) / "_api" / f"{hashlib.sha256(body.encode()).hexdigest()}.json"
if target.is_file():
fresh = time.time() - target.stat().st_mtime <= MOVING_TTL
if fresh or offline:
return json.loads(target.read_text(encoding="utf-8"))
token = github_token()
if offline or not token:
return None
headers = _headers(GITHUB_GRAPHQL, accept_json=True)
headers["Content-Type"] = "application/json"
req = urllib.request.Request(
GITHUB_GRAPHQL, data=body.encode(), headers=headers, method="POST"
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
payload = json.loads(resp.read().decode())
except urllib.error.HTTPError as exc:
raise _http_failure(GITHUB_GRAPHQL, exc) from exc
except (urllib.error.URLError, http.client.HTTPException, OSError) as exc:
raise UpstreamError(f"{GITHUB_GRAPHQL}: {exc}") from exc
if not isinstance(payload, dict) or payload.get("errors"):
return None
write_cache(target, json.dumps(payload))
return payload
def newest_tag(repo: Repo, cache_dir: str, offline: bool = False) -> str | None:
"""The tag on the newest commit.
GitHub's REST listing sorts tags by name: nesoid lists 2.5 before 2.5-4,
gargoyle stable-20090825 before 2026.1.1. GraphQL sorts by commit date.
Elsewhere, or without a token, the forge's own first entry.
"""
if repo.family == "github":
payload = _github_graphql(
_NEWEST_TAG_QUERY, {"owner": repo.owner, "name": repo.name},
cache_dir, offline,
)
try:
nodes = payload["data"]["repository"]["refs"]["nodes"]
except (TypeError, KeyError):
nodes = None
if isinstance(nodes, list):
return nodes[0].get("name") if nodes else None
tags = list_tags(repo, cache_dir, offline)
return tags[0] if tags else None
def resolve_tag_commit(
repo: Repo, tag: str, cache_dir: str, offline: bool = False
) -> str | None:
@@ -630,6 +697,22 @@ def resolve_tag_commit(
return None
def _named_tag_commit(
repo: Repo, tag: str, cache_dir: str, offline: bool
) -> str | None:
"""A GitLab or Forgejo tag's commit, asked for by name."""
quoted = urllib.parse.quote(tag, safe="")
if repo.family == "gitlab":
url = f"{repo.api_base}/projects/{_project(repo)}/repository/tags/{quoted}"
else:
url = f"{repo.api_base}/repos/{repo.slug}/tags/{quoted}"
payload = _api(url, cache_dir, offline, absent_max_age=MOVING_TTL)
commit = payload.get("commit") if isinstance(payload, dict) else None
if not isinstance(commit, dict):
return None
return commit.get("id") or commit.get("sha")
def tag_commit(
repo: Repo, tag: str, cache_dir: str, offline: bool = False
) -> str | None:
@@ -640,7 +723,9 @@ def tag_commit(
An annotated tag points at a tag object, which is dereferenced.
"""
if repo.family != "github":
return resolve_tag_commit(repo, tag, cache_dir, offline)
return _named_tag_commit(repo, tag, cache_dir, offline) or resolve_tag_commit(
repo, tag, cache_dir, offline
)
quoted = urllib.parse.quote(tag)
# A tag pushed after the first lookup must be seen: the 404 expires,
# the commit a tag names does not.
+14
View File
@@ -3152,6 +3152,7 @@ class TestCheckVersion(unittest.TestCase):
"tagsha" if tag == "v3.0.0" else None
)
)
profile_sync.upstream.tag_commit = profile_sync.upstream.resolve_tag_commit
def tearDown(self):
(
@@ -4530,6 +4531,19 @@ class TestEmulatorsDirStatus(unittest.TestCase):
Path(root, "a.yml").write_text("emulator: a\n")
self.assertTrue(profile_sync.emulators_dir_is_dirty(root))
def test_only_the_selected_profiles_count(self):
# A loop of --emulator X --rebase-refs refused every profile after
# the first one it wrote.
with tempfile.TemporaryDirectory() as root:
self._git("init", "-q", cwd=root)
Path(root, "a.yml").write_text("emulator: a\n")
Path(root, "b.yml").write_text("emulator: b\n")
self._git("add", "a.yml", "b.yml", cwd=root)
self._git("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "p", cwd=root)
Path(root, "a.yml").write_text("emulator: a2\n")
self.assertTrue(profile_sync.emulators_dir_is_dirty(root, ["a"]))
self.assertFalse(profile_sync.emulators_dir_is_dirty(root, ["b"]))
def test_a_tree_outside_any_repository_is_not_clean(self):
with tempfile.TemporaryDirectory() as root, mock.patch.dict(
os.environ, {"GIT_CEILING_DIRECTORIES": str(Path(root).parent)}
+47
View File
@@ -910,5 +910,52 @@ class TestCurrentSlug(unittest.TestCase):
)
class TestNewestTag(unittest.TestCase):
"""GitHub's REST tag list is sorted by name: nesoid lists 2.5 before
2.5-4, so the first entry was reported as the latest tag."""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.repo = make_repo("github.com", "o", "n")
self.saved = (upstream._github_graphql, upstream.list_tags)
upstream.list_tags = lambda repo, cache, offline=False: ["2.5", "2.5-4"]
def tearDown(self):
upstream._github_graphql, upstream.list_tags = self.saved
self.tmp.cleanup()
def test_commit_date_order_wins(self):
upstream._github_graphql = lambda *a: {
"data": {"repository": {"refs": {"nodes": [{"name": "2.5-4"}]}}}
}
self.assertEqual(upstream.newest_tag(self.repo, self.tmp.name), "2.5-4")
def test_without_a_token_the_forge_order_stays(self):
upstream._github_graphql = lambda *a: None
self.assertEqual(upstream.newest_tag(self.repo, self.tmp.name), "2.5")
class TestNamedTagCommit(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.saved = upstream._http_json
def tearDown(self):
upstream._http_json = self.saved
self.tmp.cleanup()
def test_gitlab_tag_is_asked_by_name(self):
asked = []
def answer(url):
asked.append(url)
return {"name": "1.2.35", "commit": {"id": "c74fc36"}}
upstream._http_json = answer
repo = parse_repo("https://gitlab.com/jengun/sdltrs")
self.assertEqual(upstream.tag_commit(repo, "1.2.35", self.tmp.name), "c74fc36")
self.assertTrue(asked[0].endswith("/repository/tags/1.2.35"))
if __name__ == "__main__":
unittest.main()