Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions reclaim.py
Original file line number Diff line number Diff line change
Expand Up @@ -721,6 +721,22 @@ def find_manifests(log_dir):
return out


def manifest_for_project(log_dir, project):
"""
The size-routing manifest for `project`, or None if it has never routed (no
manifest yet). Mirrors latest_log_for_project()'s role for the whole-target
log path — a deployment wrapper driving BOTH routed and non-routed targets
(e.g. a per-target size_routing config flag) needs a single lookup that
works either way without re-deriving the badge (badge_of lives in archive.py,
not reclaim.py; matching on the manifest's own recorded `project` field avoids
that cross-module dependency).
"""
for m in find_manifests(log_dir):
if m.get("project") == project:
return m
return None


def verify_target(manifest, holds, canonical_roots=(), allowlist=None, opt_in=False):
"""
Per-target AGGREGATE structural double-check for a size-routed target (Phase 2,
Expand Down
28 changes: 28 additions & 0 deletions tests/test_routing_reclaim.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,34 @@ def test_empty_when_no_vault(self):
self.assertEqual(reclaim.find_manifests(d), [])


class TestManifestForProject(unittest.TestCase):
def test_finds_matching_project(self):
with tempfile.TemporaryDirectory() as d:
vault = os.path.join(d, "_vault")
os.makedirs(vault)
for badge, proj in (("aaa", "repository_X1D_2_spectral-standoff"),
("bbb", "repository_X1D_3_transcriptomics")):
with open(os.path.join(vault, "%s.manifest.json" % badge), "w") as fh:
json.dump({"badge": badge, "project": proj, "objects": []}, fh)
m = reclaim.manifest_for_project(d, "repository_X1D_3_transcriptomics")
self.assertEqual(m["badge"], "bbb")

def test_none_when_no_match(self):
with tempfile.TemporaryDirectory() as d:
self.assertIsNone(reclaim.manifest_for_project(d, "repository_X1D_3_transcriptomics"))

def test_none_when_project_field_absent(self):
# A manifest that predates the 'project' field (shouldn't happen in
# practice — schema_version 1 always writes it — but a lookup must not
# crash or false-match on a bare .get(None) == None comparison).
with tempfile.TemporaryDirectory() as d:
vault = os.path.join(d, "_vault")
os.makedirs(vault)
with open(os.path.join(vault, "aaa.manifest.json"), "w") as fh:
json.dump({"badge": "aaa", "objects": []}, fh)
self.assertIsNone(reclaim.manifest_for_project(d, "repository_X1D_3_transcriptomics"))


class TestDeleteMultiTar(unittest.TestCase):
def setUp(self):
self._hsi = reclaim.hsi_exists
Expand Down