Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f9ee2740c4 | ||
|
|
ced481e06b |
@@ -833,6 +833,49 @@ def clone_one(cfg, full_name, dest, mode=None):
|
|||||||
return meta
|
return meta
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_granted(cfg, want, repos=None):
|
||||||
|
"""Turn what the user typed into the repo the forge actually grants them.
|
||||||
|
|
||||||
|
`get notes` used to mean `<your-login>/notes` and nothing else. On a real
|
||||||
|
account that is wrong for almost everything: of 172 repos granted to this
|
||||||
|
estate's own admin, 157 are owned by an ORG, so the bare name failed for
|
||||||
|
91% of what the user could see and the error said "Repository not found"
|
||||||
|
-- which reads like a permissions problem rather than a naming one.
|
||||||
|
|
||||||
|
So a bare name is resolved against the granted list, which is the forge's
|
||||||
|
own answer about what this account may have. An owner-qualified name is
|
||||||
|
taken as given. An ambiguous bare name is REFUSED with the candidates
|
||||||
|
rather than guessed: picking one of two repos called `notes` owned by
|
||||||
|
different teams is not a guess worth making for someone.
|
||||||
|
"""
|
||||||
|
if "/" in want:
|
||||||
|
return parse_repo_arg(want, cfg["login"])
|
||||||
|
if repos is None:
|
||||||
|
try:
|
||||||
|
repos, _truncated = list_repos(cfg)
|
||||||
|
except (SystemExit, ValueError, OSError) as e:
|
||||||
|
# Best effort. If the listing is unreachable, a name the user
|
||||||
|
# typed in full must still clone, and a bare name should degrade
|
||||||
|
# to their own namespace rather than refusing outright -- the
|
||||||
|
# clone that follows reports the real problem precisely.
|
||||||
|
log(f"could not read your repo list ({e}); assuming "
|
||||||
|
f"{cfg['login']}/{want}")
|
||||||
|
return parse_repo_arg(want, cfg["login"])
|
||||||
|
matches = [r.get("full_name") for r in repos
|
||||||
|
if (r.get("full_name") or "").rsplit("/", 1)[-1] == want]
|
||||||
|
if len(matches) == 1:
|
||||||
|
return matches[0]
|
||||||
|
if len(matches) > 1:
|
||||||
|
listed = "\n".join(f" {m}" for m in sorted(matches))
|
||||||
|
raise SystemExit(
|
||||||
|
f"{want!r} is ambiguous — {len(matches)} repos you can see have "
|
||||||
|
f"that name:\n{listed}\nRe-run with the owner, e.g. "
|
||||||
|
f"granthi-sync get {sorted(matches)[0]}")
|
||||||
|
# Nothing granted by that name. Fall back to the caller's own namespace so
|
||||||
|
# the message names a concrete repo instead of a guess.
|
||||||
|
return parse_repo_arg(want, cfg["login"])
|
||||||
|
|
||||||
|
|
||||||
def cmd_get(args):
|
def cmd_get(args):
|
||||||
cfg = require_linked(load_config())
|
cfg = require_linked(load_config())
|
||||||
device_id(cfg)
|
device_id(cfg)
|
||||||
@@ -840,7 +883,7 @@ def cmd_get(args):
|
|||||||
return _get_all(cfg, args)
|
return _get_all(cfg, args)
|
||||||
if not args.repo:
|
if not args.repo:
|
||||||
raise SystemExit("give a repo name, or --all")
|
raise SystemExit("give a repo name, or --all")
|
||||||
full_name = parse_repo_arg(args.repo, cfg["login"])
|
full_name = resolve_granted(cfg, args.repo)
|
||||||
name = full_name.rsplit("/", 1)[-1]
|
name = full_name.rsplit("/", 1)[-1]
|
||||||
clone_one(cfg, full_name, os.path.abspath(args.into or name), args.mode)
|
clone_one(cfg, full_name, os.path.abspath(args.into or name), args.mode)
|
||||||
return 0
|
return 0
|
||||||
@@ -1293,12 +1336,17 @@ def cmd_bootstrap(args):
|
|||||||
into = os.path.abspath(args.into or ".")
|
into = os.path.abspath(args.into or ".")
|
||||||
print(f"workspace: {name}\n")
|
print(f"workspace: {name}\n")
|
||||||
|
|
||||||
granted = {r.get("full_name") for r in list_repos(cfg)[0]}
|
granted_repos, _trunc = list_repos(cfg)
|
||||||
granted |= {(r.get("full_name") or "").rsplit("/", 1)[-1] for r in []}
|
granted = {r.get("full_name") for r in granted_repos}
|
||||||
pulled = skipped = denied = 0
|
pulled = skipped = denied = 0
|
||||||
for repo in ws["repos"]:
|
for repo in ws["repos"]:
|
||||||
want = repo["name"]
|
want = repo["name"]
|
||||||
full = want if "/" in want else f"{cfg['login']}/{want}"
|
try:
|
||||||
|
full = resolve_granted(cfg, want, repos=granted_repos)
|
||||||
|
except SystemExit as e:
|
||||||
|
log(f"AMBIGUOUS {want}: {e}")
|
||||||
|
denied += 1
|
||||||
|
continue
|
||||||
if full not in granted:
|
if full not in granted:
|
||||||
# The forge decides. A manifest asking for a repo this account was
|
# The forge decides. A manifest asking for a repo this account was
|
||||||
# not granted is a permissions answer, not an error to route round.
|
# not granted is a permissions answer, not an error to route round.
|
||||||
|
|||||||
@@ -1229,5 +1229,60 @@ class TestWorkspaceBootstrap(GitScenarioBase):
|
|||||||
client.cmd_bootstrap(self.ns())
|
client.cmd_bootstrap(self.ns())
|
||||||
self.assertIn("no workspace.json", str(e.exception))
|
self.assertIn("no workspace.json", str(e.exception))
|
||||||
|
|
||||||
|
|
||||||
|
class TestResolveGranted(unittest.TestCase):
|
||||||
|
"""`get <name>` has to find the repo the forge actually grants.
|
||||||
|
|
||||||
|
Found by running the real flow against production: of 172 repos granted
|
||||||
|
to this estate's own admin, 157 are owned by an ORG, so a bare name failed
|
||||||
|
for 91% of what the user could see -- and the error read "Repository not
|
||||||
|
found", which sounds like a permissions problem rather than a naming one.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.cfg = {"login": "alice", "gitea_base": "http://forge.example",
|
||||||
|
"token": "t"}
|
||||||
|
self.repos = [{"full_name": "Nirlabinc/Ai-Assistant"},
|
||||||
|
{"full_name": "alice/notes"},
|
||||||
|
{"full_name": "Shreai/notes"}]
|
||||||
|
|
||||||
|
def test_a_bare_name_finds_an_org_owned_repo(self):
|
||||||
|
self.assertEqual(
|
||||||
|
client.resolve_granted(self.cfg, "Ai-Assistant", self.repos),
|
||||||
|
"Nirlabinc/Ai-Assistant")
|
||||||
|
|
||||||
|
def test_an_owner_qualified_name_is_taken_as_given(self):
|
||||||
|
self.assertEqual(
|
||||||
|
client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant",
|
||||||
|
self.repos),
|
||||||
|
"Nirlabinc/Ai-Assistant")
|
||||||
|
|
||||||
|
def test_an_ambiguous_bare_name_is_refused_with_the_candidates(self):
|
||||||
|
"""Two teams with a repo called `notes` is not a guess worth making
|
||||||
|
on someone's behalf."""
|
||||||
|
with self.assertRaises(SystemExit) as e:
|
||||||
|
client.resolve_granted(self.cfg, "notes", self.repos)
|
||||||
|
msg = str(e.exception)
|
||||||
|
self.assertIn("ambiguous", msg)
|
||||||
|
self.assertIn("Shreai/notes", msg)
|
||||||
|
self.assertIn("alice/notes", msg)
|
||||||
|
|
||||||
|
def test_an_unknown_name_falls_back_to_your_own_namespace(self):
|
||||||
|
self.assertEqual(
|
||||||
|
client.resolve_granted(self.cfg, "brand-new", self.repos),
|
||||||
|
"alice/brand-new")
|
||||||
|
|
||||||
|
def test_an_unreachable_listing_does_not_block_a_clone(self):
|
||||||
|
"""A network blip must not stop someone cloning a repo they named."""
|
||||||
|
def boom(cfg):
|
||||||
|
raise SystemExit("listing repos failed (HTTP 502)")
|
||||||
|
with mock.patch.object(client, "list_repos", boom), \
|
||||||
|
mock.patch("sys.stdout", new_callable=io.StringIO):
|
||||||
|
self.assertEqual(client.resolve_granted(self.cfg, "notes"),
|
||||||
|
"alice/notes")
|
||||||
|
self.assertEqual(
|
||||||
|
client.resolve_granted(self.cfg, "Nirlabinc/Ai-Assistant"),
|
||||||
|
"Nirlabinc/Ai-Assistant")
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user