"""Unit: trusted agent forge identity & guidance (PRD 0082). Covers the net-new surface: agent `author`, agent `forge-accounts` (Gitea API URL validation + host token reference), the repo→forge association resolved at composition, the synthesized proxy-held egress route, and the generated, non-secret prompt guidance. Legacy git-gate.user / cwd-agent behavior lives in the manifest test modules. """ from __future__ import annotations import os import tempfile import unittest from pathlib import Path from typing import Callable from unittest.mock import patch from bot_bottle.egress import ( egress_forge_routes, egress_render_routes, egress_resolve_token_values, egress_routes_for_bottle, egress_token_env_map, ) from bot_bottle.manifest import ManifestError, ManifestIndex from bot_bottle.manifest.forge import ( ManifestForgeAccount, canonicalize_forge_url, render_forge_guidance, ) GITEA = { "url": "https://gitea.dideric.is/api/v1", "auth": {"type": "token", "token_secret": "GITEA_CLAUDE_TOKEN"}, } def _repo(*, forge: str | None = None) -> dict[str, object]: entry: dict[str, object] = { "url": "ssh://git@100.78.141.42:30009/didericis/bot-bottle.git", "key": {"provider": "static", "path": "/k"}, } if forge is not None: entry["forge"] = forge return entry def _index( *, author: dict[str, object] | None = None, forge_accounts: dict[str, object] | None = None, repo_forge: str | None = None, ) -> ManifestIndex: """Build an eager index with one agent 'claude' + bottle 'bb'.""" agent: dict[str, object] = {"bottle": "bb", "prompt": ""} if author is not None: agent["author"] = author if forge_accounts is not None: agent["forge-accounts"] = forge_accounts bottle: dict[str, object] = { "git-gate": {"repos": {"bot-bottle": _repo(forge=repo_forge)}} } return ManifestIndex.from_json_obj( {"bottles": {"bb": bottle}, "agents": {"claude": agent}} ) def _error( callable_: Callable[..., object], *args: object, **kwargs: object, ) -> str: try: callable_(*args, **kwargs) except ManifestError as e: return str(e) raise AssertionError("expected ManifestError was not raised") # --------------------------------------------------------------------------- # author # --------------------------------------------------------------------------- class TestAuthor(unittest.TestCase): def test_populates_git_identity(self) -> None: idx = _index(author={"name": "didericis-claude", "email": "e+c@x.is"}) m = idx.load_for_agent("claude", ()) self.assertEqual("didericis-claude", m.bottle.git_user.name) self.assertEqual("e+c@x.is", m.bottle.git_user.email) self.assertEqual( "name=didericis-claude, email=e+c@x.is", m.git_identity_summary(), ) def test_absent_author_no_identity(self) -> None: m = _index().load_for_agent("claude", ()) self.assertTrue(m.bottle.git_user.is_empty()) self.assertIsNone(m.git_identity_summary()) def test_name_required(self) -> None: msg = _error(_index, author={"email": "e@x.is"}) self.assertIn("author.name must be a non-empty string", msg) def test_email_required(self) -> None: msg = _error(_index, author={"name": "n"}) self.assertIn("author.email must be a non-empty string", msg) def test_email_rejects_whitespace(self) -> None: msg = _error(_index, author={"name": "n", "email": "a b@x.is"}) self.assertIn("must not contain whitespace", msg) def test_name_rejects_newline(self) -> None: msg = _error(_index, author={"name": "a\nb", "email": "e@x.is"}) self.assertIn("author.name must not contain newlines", msg) def test_unknown_key(self) -> None: msg = _error( _index, author={"name": "n", "email": "e@x.is", "role": "x"} ) self.assertIn("author has unknown key", msg) # --------------------------------------------------------------------------- # forge-accounts URL validation # --------------------------------------------------------------------------- class TestForgeUrl(unittest.TestCase): def test_canonical_ok(self) -> None: canonical, origin, host, prefix = canonicalize_forge_url( "a", "g", "https://Gitea.Dideric.is/api/v1/" ) # trailing slash normalized; host lowercased. self.assertEqual("https://gitea.dideric.is/api/v1", canonical) self.assertEqual("https://gitea.dideric.is", origin) self.assertEqual("gitea.dideric.is", host) self.assertEqual("/api/v1", prefix) def test_port_preserved(self) -> None: canonical, origin, _host, _ = canonicalize_forge_url( "a", "g", "https://gitea.local:3000/api/v1" ) self.assertEqual("https://gitea.local:3000/api/v1", canonical) self.assertEqual("https://gitea.local:3000", origin) def test_http_fails(self) -> None: self.assertIn("must use https", _error( canonicalize_forge_url, "a", "g", "http://gitea/api/v1")) def test_userinfo_fails(self) -> None: self.assertIn("userinfo", _error( canonicalize_forge_url, "a", "g", "https://u:p@gitea/api/v1")) def test_query_fails(self) -> None: self.assertIn("query string", _error( canonicalize_forge_url, "a", "g", "https://gitea/api/v1?x=1")) def test_fragment_fails(self) -> None: self.assertIn("fragment", _error( canonicalize_forge_url, "a", "g", "https://gitea/api/v1#x")) def test_missing_host_fails(self) -> None: self.assertIn("hostname", _error( canonicalize_forge_url, "a", "g", "https:///api/v1")) def test_bad_path_fails(self) -> None: self.assertIn("Gitea API base", _error( canonicalize_forge_url, "a", "g", "https://gitea/api/v2")) def test_non_string_fails(self) -> None: self.assertIn("required", _error( canonicalize_forge_url, "a", "g", None)) def test_malformed_url_fails(self) -> None: # An unparseable URL (invalid IPv6 literal) trips urlsplit's ValueError. self.assertIn("not a valid URL", _error( canonicalize_forge_url, "a", "g", "https://[/api/v1")) class TestForgeAccount(unittest.TestCase): def test_parses(self) -> None: acct = ManifestForgeAccount.from_dict("a", "didericis-gitea", GITEA) self.assertEqual("didericis-gitea", acct.alias) self.assertEqual("gitea.dideric.is", acct.host) self.assertEqual("token", acct.auth_type) self.assertEqual("GITEA_CLAUDE_TOKEN", acct.token_secret) def test_non_kebab_alias_fails(self) -> None: self.assertIn("valid alias", _error( ManifestForgeAccount.from_dict, "a", "Bad_Alias", GITEA)) def test_auth_required(self) -> None: self.assertIn("missing required 'auth'", _error( ManifestForgeAccount.from_dict, "a", "g", {"url": "https://gitea/api/v1"})) def test_bad_auth_type_fails(self) -> None: self.assertIn("auth.type must be", _error( ManifestForgeAccount.from_dict, "a", "g", {"url": "https://gitea/api/v1", "auth": {"type": "basic", "token_secret": "T"}})) def test_token_secret_required(self) -> None: self.assertIn("token_secret must be", _error( ManifestForgeAccount.from_dict, "a", "g", {"url": "https://gitea/api/v1", "auth": {"type": "token"}})) def test_unknown_key_fails(self) -> None: self.assertIn("unknown key", _error( ManifestForgeAccount.from_dict, "a", "g", {**GITEA, "bogus": 1})) def test_unknown_auth_key_fails(self) -> None: self.assertIn("auth has unknown key", _error( ManifestForgeAccount.from_dict, "a", "g", {"url": "https://gitea/api/v1", "auth": {"type": "token", "token_secret": "T", "bogus": 1}})) # --------------------------------------------------------------------------- # repo -> forge association (composition) # --------------------------------------------------------------------------- class TestForgeAssociations(unittest.TestCase): def test_resolves_when_repo_references_alias(self) -> None: idx = _index( forge_accounts={"didericis-gitea": GITEA}, repo_forge="didericis-gitea", ) m = idx.load_for_agent("claude", ()) self.assertEqual(1, len(m.forge_associations)) assoc = m.forge_associations[0] self.assertEqual("didericis-gitea", assoc.alias) self.assertEqual(("bot-bottle",), assoc.repo_names) def test_unreferenced_account_yields_no_association(self) -> None: idx = _index(forge_accounts={"didericis-gitea": GITEA}) # repo has no forge m = idx.load_for_agent("claude", ()) self.assertEqual((), m.forge_associations) def test_unknown_alias_fails_closed(self) -> None: idx = _index( forge_accounts={"didericis-gitea": GITEA}, repo_forge="nope" ) msg = _error(idx.load_for_agent, "claude", ()) self.assertIn("references forge alias 'nope'", msg) self.assertIn("not defined on agent", msg) def test_forge_without_account_fails_closed(self) -> None: idx = _index(repo_forge="didericis-gitea") # no forge-accounts at all self.assertIn("(none)", _error(idx.load_for_agent, "claude", ())) def _two_alias_index(self, t1: str, t2: str) -> ManifestIndex: """Two aliases on the SAME host, each referenced by a distinct repo.""" def acct(token: str) -> dict[str, object]: return { "url": "https://same.example/api/v1", "auth": {"type": "token", "token_secret": token}, } return ManifestIndex.from_json_obj({ "bottles": {"bb": {"git-gate": {"repos": { "r1": {"url": "ssh://git@h/x/r1.git", "key": {"provider": "static", "path": "/k"}, "forge": "g1"}, "r2": {"url": "ssh://git@h/x/r2.git", "key": {"provider": "static", "path": "/k"}, "forge": "g2"}, }}}}, "agents": {"claude": {"bottle": "bb", "prompt": "", "forge-accounts": {"g1": acct(t1), "g2": acct(t2)}}}, }) def test_conflicting_aliases_same_host_fail_closed(self) -> None: # Two aliases on the same host with DIFFERENT tokens is ambiguous — # the proxy routes by host, so it must fail before launch rather than # silently authenticate every call as one account. idx = self._two_alias_index("FIRST_TOKEN", "SECOND_TOKEN") msg = _error(idx.load_for_agent, "claude", ()) self.assertIn("both resolve to host 'same.example'", msg) self.assertIn("ambiguous", msg) def test_matching_aliases_same_host_ok(self) -> None: # Identical url/auth under two aliases is unambiguous: one route. idx = self._two_alias_index("SAME_TOKEN", "SAME_TOKEN") m = idx.load_for_agent("claude", ()) self.assertEqual(2, len(m.forge_associations)) # both referenced routes = egress_forge_routes(m.forge_associations) self.assertEqual(1, len(routes)) # deduped to a single proxy route self.assertEqual("SAME_TOKEN", routes[0].token_ref) # --------------------------------------------------------------------------- # synthesized egress route (proxy-held credential) # --------------------------------------------------------------------------- class TestForgeEgressRoutes(unittest.TestCase): def _assoc(self): idx = _index( forge_accounts={"didericis-gitea": GITEA}, repo_forge="didericis-gitea", ) return idx.load_for_agent("claude", ()) def test_route_shape(self) -> None: routes = egress_forge_routes(self._assoc().forge_associations) self.assertEqual(1, len(routes)) r = routes[0] self.assertEqual("gitea.dideric.is", r.host) self.assertEqual("token", r.auth_scheme) self.assertEqual("GITEA_CLAUDE_TOKEN", r.token_ref) self.assertTrue(r.inspect) # scoped to the API prefix. self.assertEqual("/api/v1", r.matches[0].paths[0].value) def test_token_slot_and_resolution(self) -> None: m = self._assoc() forge_routes = egress_forge_routes(m.forge_associations) routes = egress_routes_for_bottle(m.bottle, (), forge_routes) token_map = egress_token_env_map(routes) # one slot mapping the proxy env slot -> host env var name. self.assertEqual({"EGRESS_TOKEN_0": "GITEA_CLAUDE_TOKEN"}, token_map) resolved = egress_resolve_token_values( token_map, {"GITEA_CLAUDE_TOKEN": "sekret-value"} ) self.assertEqual({"EGRESS_TOKEN_0": "sekret-value"}, resolved) def test_rendered_routes_hide_secret_name_and_value(self) -> None: m = self._assoc() routes = egress_routes_for_bottle( m.bottle, (), egress_forge_routes(m.forge_associations) ) rendered = egress_render_routes(routes) self.assertIn("gitea.dideric.is", rendered) self.assertIn("EGRESS_TOKEN_0", rendered) # slot, not the host var name self.assertNotIn("GITEA_CLAUDE_TOKEN", rendered) self.assertNotIn("sekret-value", rendered) def test_dedup_by_host(self) -> None: # Two repos referencing the same alias share one route/credential. idx = ManifestIndex.from_json_obj({ "bottles": {"bb": {"git-gate": {"repos": { "r1": _repo(forge="g"), "r2": { "url": "ssh://git@h/x/other.git", "key": {"provider": "static", "path": "/k"}, "forge": "g", }, }}}}, "agents": {"claude": {"bottle": "bb", "prompt": "", "forge-accounts": {"g": GITEA}}}, }) m = idx.load_for_agent("claude", ()) routes = egress_forge_routes(m.forge_associations) self.assertEqual(1, len(routes)) # --------------------------------------------------------------------------- # generated prompt guidance # --------------------------------------------------------------------------- class TestForgeGuidance(unittest.TestCase): def _assoc(self): idx = _index( forge_accounts={"didericis-gitea": GITEA}, repo_forge="didericis-gitea", ) return idx.load_for_agent("claude", ()).forge_associations def test_empty_when_no_associations(self) -> None: self.assertEqual("", render_forge_guidance(())) def test_names_account_repo_and_workflow(self) -> None: text = render_forge_guidance(self._assoc()) self.assertIn("didericis-gitea", text) self.assertIn("https://gitea.dideric.is/api/v1", text) self.assertIn("bot-bottle", text) # branch-backed PR workflow + AGit prohibition. self.assertIn("refs/heads/", text) self.assertIn("refs/for/*", text) self.assertIn("pull request", text) def test_no_token_secret_name_or_value(self) -> None: text = render_forge_guidance(self._assoc()) self.assertNotIn("GITEA_CLAUDE_TOKEN", text) def test_prompt_file_appends_guidance(self) -> None: from bot_bottle.backend.resolve_common import prepare_agent_state_dir idx = _index( author={"name": "n", "email": "e@x.is"}, forge_accounts={"didericis-gitea": GITEA}, repo_forge="didericis-gitea", ) # Give the agent a real prompt body. m = idx.load_for_agent("claude", ()) from dataclasses import replace m = replace(m, agent=replace(m.agent, prompt="BASE PROMPT")) with tempfile.TemporaryDirectory() as td: with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": td}): _dir, prompt_file = prepare_agent_state_dir("slug1", m) body = Path(prompt_file).read_text() self.assertIn("BASE PROMPT", body) self.assertIn("Forge access", body) self.assertIn("https://gitea.dideric.is/api/v1", body) def test_prompt_file_no_guidance_without_forge(self) -> None: from bot_bottle.backend.resolve_common import prepare_agent_state_dir m = _index(author={"name": "n", "email": "e@x.is"}).load_for_agent( "claude", () ) with tempfile.TemporaryDirectory() as td: with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": td}): _dir, prompt_file = prepare_agent_state_dir("slug2", m) body = Path(prompt_file).read_text() self.assertNotIn("Forge access", body) if __name__ == "__main__": unittest.main()