refactor: unify identity/provisioned_key into key block
lint / lint (push) Failing after 1m45s
test / unit (pull_request) Successful in 35s
test / integration (pull_request) Successful in 17s

Replace the two mutually-exclusive repo keys (identity and
provisioned_key) with a single required key block. key.provider
is "static" (path to host SSH key) or "gitea" (deploy-key lifecycle
via provisioner_token env var, replacing token_env).

Internal fields: ManifestProvisionedKeyConfig → ManifestKeyConfig;
ProvisionedKey field removed from ManifestGitEntry; Key field added.
git_gate.py checks entry.Key.provider == "gitea" instead of
entry.ProvisionedKey is not None.
This commit is contained in:
2026-06-19 22:01:43 +00:00
parent c1c225aa05
commit 1a8718ca9d
11 changed files with 202 additions and 163 deletions
+11 -12
View File
@@ -389,13 +389,12 @@ def _provision_dynamic_key(
Returns the host-side path to the private key file so the caller
can inject it into the GitGateUpstream as `identity_file`."""
from .deploy_key_provisioner import get_provisioner
pk = entry.ProvisionedKey
assert pk is not None
token = os.environ.get(pk.token_env)
pk = entry.Key
token = os.environ.get(pk.provisioner_token)
if token is None:
raise RuntimeError(
f"git-gate.repos[{entry.Name!r}] provisioned_key.token_env"
f" = {pk.token_env!r}: env var is not set"
f"git-gate.repos[{entry.Name!r}] key.provisioner_token"
f" = {pk.provisioner_token!r}: env var is not set"
)
api_url = pk.api_url or f"https://{entry.UpstreamHost}"
provisioner = get_provisioner(pk.provider, token, api_url)
@@ -428,18 +427,18 @@ def revoke_git_gate_provisioned_keys(bottle: ManifestBottle, stage_dir: Path) ->
address manually."""
from .deploy_key_provisioner import get_provisioner
for entry in bottle.git:
if entry.ProvisionedKey is None:
if entry.Key.provider != "gitea":
continue
pk = entry.ProvisionedKey
pk = entry.Key
id_file = stage_dir / f"{entry.Name}-deploy-key-id"
if not id_file.exists():
continue
key_id = id_file.read_text().strip()
token = os.environ.get(pk.token_env)
token = os.environ.get(pk.provisioner_token)
if token is None:
raise RuntimeError(
f"git-gate.repos[{entry.Name!r}] provisioned_key.token_env"
f" = {pk.token_env!r}: env var is not set;"
f"git-gate.repos[{entry.Name!r}] key.provisioner_token"
f" = {pk.provisioner_token!r}: env var is not set;"
f" cannot revoke deploy key {key_id}"
)
api_url = pk.api_url or f"https://{entry.UpstreamHost}"
@@ -463,7 +462,7 @@ class GitGate(ABC):
entrypoint, pre-receive hook, and access-hook scripts (mode
600) under `stage_dir`. Pure host-side, no docker subprocess.
For `provisioned_key` entries, also generates and registers
For `gitea` key entries, also generates and registers
a fresh deploy key via the forge API and writes the private key
+ key ID to `stage_dir`.
@@ -472,7 +471,7 @@ class GitGate(ABC):
before passing the plan to `.start`."""
upstreams_list = list(git_gate_upstreams_for_bottle(bottle))
for i, entry in enumerate(bottle.git):
if entry.ProvisionedKey is not None:
if entry.Key.provider == "gitea":
key_file = _provision_dynamic_key(entry, slug, stage_dir)
upstreams_list[i] = dataclasses.replace(
upstreams_list[i], identity_file=key_file
+2 -1
View File
@@ -56,7 +56,7 @@ from .manifest_egress import (
ManifestEgressConfig,
ManifestEgressRoute,
)
from .manifest_git import ManifestGitEntry, ManifestGitUser, parse_git_gate_config
from .manifest_git import ManifestGitEntry, ManifestGitUser, ManifestKeyConfig, parse_git_gate_config
from .manifest_schema import BOTTLE_KEYS
# Re-export everything that callers currently import from this module.
@@ -64,6 +64,7 @@ __all__ = [
"ManifestError",
"ManifestGitEntry",
"ManifestGitUser",
"ManifestKeyConfig",
"ManifestAgentProvider",
"EGRESS_AUTH_SCHEMES",
"ManifestEgressRoute",
+68 -60
View File
@@ -13,6 +13,8 @@ from .manifest_util import ManifestError, as_json_object
# defence; this regex is belt-and-suspenders and documents intent).
_GIT_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$")
_KEY_PROVIDERS = {"static", "gitea"}
def _opt_str(value: object, label: str) -> str:
if value is None:
@@ -69,20 +71,22 @@ def validate_unique_git_names(bottle_name: str, git: tuple[ManifestGitEntry, ...
@dataclass(frozen=True)
class ManifestProvisionedKeyConfig:
"""Configuration for automatic deploy-key lifecycle management
(PRD 0048). Used when a git-gate.repos entry opts out of a
static identity file and instead wants a fresh SSH keypair
generated at spin-up and revoked at teardown.
class ManifestKeyConfig:
"""Configuration for a repo's SSH key in git-gate.repos.
`provider` names the contrib sub-package to load (e.g. `gitea`).
`token_env` is the name of a host-side env var carrying the API
token; the value is read at provision time, never stored on the
plan. `api_url` is the forge's HTTP API root; if empty, it is
derived from the upstream URL's host at provision time."""
`provider` is either `"static"` (a pre-existing key on the host) or
`"gitea"` (automatic deploy-key lifecycle via the Gitea API).
For `static`: `path` is the host-side absolute path to the SSH private key.
For `gitea`: `provisioner_token` is the name of a host-side env var
carrying the Gitea API token; the value is read at provision time,
never stored on the plan. `api_url` is the forge's HTTP API root; if
empty, it is derived from the upstream URL's host at provision time."""
provider: str
token_env: str
path: str = ""
provisioner_token: str = ""
api_url: str = ""
@@ -99,15 +103,16 @@ class ManifestGitEntry:
stashed in the `Upstream*` fields so the git-gate render step
doesn't have to re-parse.
Manifest source: `git-gate.repos.<Name>` (PRD 0047/0048). Exactly
one of `identity` (static key path) or `provisioned_key` (automatic
lifecycle) must be present. The internal field names are stable."""
Manifest source: `git-gate.repos.<Name>` (PRD 0047/0048). A `key`
block is required; `key.provider` is `"static"` or `"gitea"`. For
`static`, `IdentityFile` is populated at parse time from `key.path`.
For `gitea`, `IdentityFile` is populated at provision time."""
Name: str
Upstream: str
Key: ManifestKeyConfig = ManifestKeyConfig(provider="")
IdentityFile: str = ""
KnownHostKey: str = ""
ProvisionedKey: Optional[ManifestProvisionedKeyConfig] = None
RemoteKey: str = ""
UpstreamUser: str = ""
UpstreamHost: str = ""
@@ -120,8 +125,8 @@ class ManifestGitEntry:
) -> "ManifestGitEntry":
"""Parse one entry from `git-gate.repos.<repo_name>`.
YAML keys: `url` (required), exactly one of `identity` or
`provisioned_key` (required), `host_key` (optional).
YAML keys: `url` (required), `key` (required object with
`provider`, and provider-specific fields), `host_key` (optional).
The repo_name becomes `Name`."""
if not repo_name:
raise ManifestError(
@@ -135,10 +140,10 @@ class ManifestGitEntry:
label = f"git-gate.repos[{repo_name!r}]"
d = as_json_object(raw, f"bottle '{bottle_name}' {label}")
for k in d:
if k not in {"url", "identity", "provisioned_key", "host_key"}:
if k not in {"url", "key", "host_key"}:
raise ManifestError(
f"bottle '{bottle_name}' {label} has unknown key {k!r}; "
f"allowed: url, identity, provisioned_key, host_key"
f"allowed: url, key, host_key"
)
upstream = d.get("url")
if not isinstance(upstream, str) or not upstream:
@@ -146,32 +151,13 @@ class ManifestGitEntry:
f"bottle '{bottle_name}' {label} missing required string field 'url'"
)
has_identity = "identity" in d
has_provisioned = "provisioned_key" in d
if has_identity and has_provisioned:
if "key" not in d:
raise ManifestError(
f"bottle '{bottle_name}' {label} must set exactly one of "
f"'identity' or 'provisioned_key'; got both."
)
if not has_identity and not has_provisioned:
raise ManifestError(
f"bottle '{bottle_name}' {label} must set exactly one of "
f"'identity' or 'provisioned_key'; got neither."
f"bottle '{bottle_name}' {label} missing required 'key' block"
)
key_config = _parse_key_config(bottle_name, label, d["key"])
ident = ""
provisioned_key: Optional[ManifestProvisionedKeyConfig] = None
if has_identity:
raw_ident = d.get("identity")
if not isinstance(raw_ident, str) or not raw_ident:
raise ManifestError(
f"bottle '{bottle_name}' {label} 'identity' must be a non-empty string"
)
ident = raw_ident
else:
provisioned_key = _parse_provisioned_key_config(
bottle_name, label, d["provisioned_key"]
)
ident = key_config.path if key_config.provider == "static" else ""
khk = _opt_str(
d.get("host_key"),
@@ -183,9 +169,9 @@ class ManifestGitEntry:
return cls(
Name=repo_name,
Upstream=upstream,
Key=key_config,
IdentityFile=ident,
KnownHostKey=khk,
ProvisionedKey=provisioned_key,
RemoteKey=host,
UpstreamUser=user,
UpstreamHost=host,
@@ -194,36 +180,58 @@ class ManifestGitEntry:
)
def _parse_provisioned_key_config(
def _parse_key_config(
bottle_name: str, label: str, raw: object
) -> ManifestProvisionedKeyConfig:
d = as_json_object(raw, f"bottle '{bottle_name}' {label}.provisioned_key")
for k in d:
if k not in {"provider", "token_env", "api_url"}:
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key has unknown key {k!r}; "
f"allowed: provider, token_env, api_url"
)
) -> ManifestKeyConfig:
d = as_json_object(raw, f"bottle '{bottle_name}' {label}.key")
provider = d.get("provider")
if not isinstance(provider, str) or not provider:
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key missing required "
f"bottle '{bottle_name}' {label}.key missing required "
f"string field 'provider'"
)
token_env = d.get("token_env")
if not isinstance(token_env, str) or not token_env:
if provider not in _KEY_PROVIDERS:
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key missing required "
f"string field 'token_env'"
f"bottle '{bottle_name}' {label}.key provider {provider!r} is unknown; "
f"allowed: {', '.join(sorted(_KEY_PROVIDERS))}"
)
if provider == "static":
for k in d:
if k not in {"provider", "path"}:
raise ManifestError(
f"bottle '{bottle_name}' {label}.key has unknown key {k!r} "
f"for provider 'static'; allowed: provider, path"
)
path = d.get("path")
if not isinstance(path, str) or not path:
raise ManifestError(
f"bottle '{bottle_name}' {label}.key missing required "
f"string field 'path' for provider 'static'"
)
return ManifestKeyConfig(provider=provider, path=path)
# provider == "gitea"
for k in d:
if k not in {"provider", "provisioner_token", "api_url"}:
raise ManifestError(
f"bottle '{bottle_name}' {label}.key has unknown key {k!r} "
f"for provider 'gitea'; allowed: provider, provisioner_token, api_url"
)
provisioner_token = d.get("provisioner_token")
if not isinstance(provisioner_token, str) or not provisioner_token:
raise ManifestError(
f"bottle '{bottle_name}' {label}.key missing required "
f"string field 'provisioner_token' for provider 'gitea'"
)
api_url_raw = d.get("api_url", "")
if not isinstance(api_url_raw, str):
raise ManifestError(
f"bottle '{bottle_name}' {label}.provisioned_key 'api_url' must be a string"
f"bottle '{bottle_name}' {label}.key 'api_url' must be a string"
)
return ManifestProvisionedKeyConfig(
return ManifestKeyConfig(
provider=provider,
token_env=token_env,
provisioner_token=provisioner_token,
api_url=api_url_raw,
)
+2 -2
View File
@@ -46,12 +46,12 @@ def fixture_with_git_dict() -> dict[str, Any]:
"repos": {
"bot-bottle": {
"url": "ssh://git@gitea.dideric.is:30009/didericis/bot-bottle.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
"host_key": "ssh-ed25519 AAAA...",
},
"foo": {
"url": "ssh://git@github.com/didericis/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
"host_key": "ssh-ed25519 BBBB...",
},
},
+1 -1
View File
@@ -51,7 +51,7 @@ def _manifest(*, supervise: bool, with_git: bool, with_egress: bool) -> Manifest
bottle["git-gate"] = {"repos": {
"upstream": {
"url": "ssh://git@example.com:22/x/y.git",
"identity": "/etc/hostname", # any existing file
"key": {"provider": "static", "path": "/etc/hostname"},
},
}}
if with_egress:
+1 -1
View File
@@ -284,7 +284,7 @@ class TestPrepare(unittest.TestCase):
"bottles": {"dev": {"git-gate": {"repos": {
"foo": {
"url": "ssh://git@github.com/didericis/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+1 -1
View File
@@ -112,7 +112,7 @@ class TestAgentGitUserOverlay(unittest.TestCase):
class TestAgentGitUserRejections(unittest.TestCase):
def test_agent_repos_dies_bottle_only(self):
msg = _error_message(_manifest, agent_git={
"repos": {"r": {"url": "ssh://git@x/y.git", "identity": "/dev/null"}},
"repos": {"r": {"url": "ssh://git@x/y.git", "key": {"provider": "static", "path": "/dev/null"}}},
})
self.assertIn("git-gate.repos", msg)
self.assertIn("bottle-only", msg)
+3 -3
View File
@@ -116,8 +116,8 @@ class TestExtendsGitMerge(unittest.TestCase):
"""git-gate.user overlays by field; git-gate.repos merges by upstream
host, with child entries replacing duplicate hosts."""
_GIT_ENTRY_A = {"url": "ssh://git@host-a/a.git", "identity": "/dev/null"}
_GIT_ENTRY_B = {"url": "ssh://git@host-b/b.git", "identity": "/dev/null"}
_GIT_ENTRY_A = {"url": "ssh://git@host-a/a.git", "key": {"provider": "static", "path": "/dev/null"}}
_GIT_ENTRY_B = {"url": "ssh://git@host-b/b.git", "key": {"provider": "static", "path": "/dev/null"}}
def test_child_git_repos_merge_with_parent(self):
m = _build(
@@ -131,7 +131,7 @@ class TestExtendsGitMerge(unittest.TestCase):
self.assertEqual(["a", "b"], names)
def test_child_git_repo_replaces_same_host(self):
replacement = {"url": "ssh://git@host-a/replacement.git", "identity": "/dev/null"}
replacement = {"url": "ssh://git@host-a/replacement.git", "key": {"provider": "static", "path": "/dev/null"}}
m = _build(
base={"git-gate": {"repos": {"a": self._GIT_ENTRY_A}}},
child={
+109 -79
View File
@@ -17,7 +17,7 @@ class TestGitEntryParsing(unittest.TestCase):
m = Manifest.from_json_obj(_manifest({
"bot-bottle": {
"url": "ssh://git@gitea.dideric.is:30009/didericis/bot-bottle.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
entries = m.bottles["dev"].git
@@ -33,7 +33,7 @@ class TestGitEntryParsing(unittest.TestCase):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/didericis/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
e = m.bottles["dev"].git[0]
@@ -44,7 +44,7 @@ class TestGitEntryParsing(unittest.TestCase):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
self.assertEqual("", m.bottles["dev"].git[0].KnownHostKey)
@@ -53,7 +53,7 @@ class TestGitEntryParsing(unittest.TestCase):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
"host_key": "ssh-ed25519 AAAA",
},
}))
@@ -63,7 +63,7 @@ class TestGitEntryParsing(unittest.TestCase):
m = Manifest.from_json_obj(_manifest({
"my-repo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
self.assertEqual("my-repo", m.bottles["dev"].git[0].Name)
@@ -71,10 +71,10 @@ class TestGitEntryParsing(unittest.TestCase):
def test_missing_url_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {"identity": "/dev/null"},
"foo": {"key": {"provider": "static", "path": "/dev/null"}},
}))
def test_missing_identity_dies(self):
def test_missing_key_block_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {"url": "ssh://git@github.com/foo.git"},
@@ -85,7 +85,7 @@ class TestGitEntryParsing(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
"IdentityFile": "/dev/null", # old PascalCase key
},
}))
@@ -95,7 +95,7 @@ class TestGitEntryParsing(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "https://github.com/didericis/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -104,7 +104,7 @@ class TestGitEntryParsing(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "git@github.com:didericis/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -113,7 +113,7 @@ class TestGitEntryParsing(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -122,7 +122,7 @@ class TestGitEntryParsing(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -131,7 +131,7 @@ class TestGitEntryParsing(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com:notaport/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -139,7 +139,7 @@ class TestGitEntryParsing(unittest.TestCase):
m = Manifest.from_json_obj(_manifest({
"bot-bottle": {
"url": "ssh://git@100.78.141.42:30009/didericis/bot-bottle.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
e = m.bottles["dev"].git[0]
@@ -156,11 +156,11 @@ class TestGitEntryCrossValidation(unittest.TestCase):
"bottles": {"dev": {"git-gate": {"repos": {
"foo": {
"url": "ssh://git@a.example/x.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
"bar": {
"url": "ssh://git@b.example/y.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
@@ -190,7 +190,7 @@ class TestGitEntryCrossValidation(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"o'reilly": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -199,7 +199,7 @@ class TestGitEntryCrossValidation(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"my repo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -208,7 +208,7 @@ class TestGitEntryCrossValidation(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo;bar": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -217,7 +217,7 @@ class TestGitEntryCrossValidation(unittest.TestCase):
Manifest.from_json_obj(_manifest({
"foo$bar": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
@@ -225,7 +225,7 @@ class TestGitEntryCrossValidation(unittest.TestCase):
m = Manifest.from_json_obj(_manifest({
"my.repo-name_1": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
self.assertEqual("my.repo-name_1", m.bottles["dev"].git[0].Name)
@@ -243,111 +243,141 @@ class TestGitEntryCrossValidation(unittest.TestCase):
self.assertIn("PRD 0047", msg)
class TestProvisionedKey(unittest.TestCase):
"""git-gate.repos entries that use provisioned_key (PRD 0048)."""
class TestStaticKey(unittest.TestCase):
"""git-gate.repos entries with key.provider = "static"."""
def test_provisioned_key_minimal(self):
def test_static_key_minimal(self):
m = Manifest.from_json_obj(_manifest({
"bot-bottle": {
"url": "ssh://git@gitea.dideric.is:30009/didericis/bot-bottle.git",
"provisioned_key": {
"key": {"provider": "static", "path": "/home/user/.ssh/id_ed25519"},
},
}))
e = m.bottles["dev"].git[0]
self.assertEqual("bot-bottle", e.Name)
self.assertEqual("static", e.Key.provider)
self.assertEqual("/home/user/.ssh/id_ed25519", e.Key.path)
self.assertEqual("/home/user/.ssh/id_ed25519", e.IdentityFile)
def test_static_key_sets_identity_file_at_parse_time(self):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"key": {"provider": "static", "path": "/dev/null"},
},
}))
self.assertEqual("/dev/null", m.bottles["dev"].git[0].IdentityFile)
def test_static_key_missing_path_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"key": {"provider": "static"},
},
}))
def test_static_key_unknown_field_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"key": {"provider": "static", "path": "/dev/null", "api_url": "x"},
},
}))
class TestGiteaKey(unittest.TestCase):
"""git-gate.repos entries with key.provider = "gitea"."""
def test_gitea_key_minimal(self):
m = Manifest.from_json_obj(_manifest({
"bot-bottle": {
"url": "ssh://git@gitea.dideric.is:30009/didericis/bot-bottle.git",
"key": {
"provider": "gitea",
"token_env": "GITEA_TOKEN",
"provisioner_token": "GITEA_TOKEN",
},
},
}))
e = m.bottles["dev"].git[0]
self.assertEqual("bot-bottle", e.Name)
self.assertIsNotNone(e.ProvisionedKey)
assert e.ProvisionedKey is not None
self.assertEqual("gitea", e.ProvisionedKey.provider)
self.assertEqual("GITEA_TOKEN", e.ProvisionedKey.token_env)
self.assertEqual("", e.ProvisionedKey.api_url)
self.assertEqual("gitea", e.Key.provider)
self.assertEqual("GITEA_TOKEN", e.Key.provisioner_token)
self.assertEqual("", e.Key.api_url)
self.assertEqual("", e.IdentityFile)
def test_provisioned_key_with_api_url(self):
def test_gitea_key_with_api_url(self):
m = Manifest.from_json_obj(_manifest({
"repo": {
"url": "ssh://git@gitea.example.com/org/repo.git",
"provisioned_key": {
"key": {
"provider": "gitea",
"token_env": "MY_TOKEN",
"provisioner_token": "MY_TOKEN",
"api_url": "https://gitea.example.com",
},
},
}))
pk = m.bottles["dev"].git[0].ProvisionedKey
assert pk is not None
self.assertEqual("https://gitea.example.com", pk.api_url)
self.assertEqual("https://gitea.example.com", m.bottles["dev"].git[0].Key.api_url)
def test_both_identity_and_provisioned_key_dies(self):
with self.assertRaises(ManifestError) as ctx:
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
"provisioned_key": {"provider": "gitea", "token_env": "T"},
},
}))
self.assertIn("exactly one of", str(ctx.exception))
self.assertIn("got both", str(ctx.exception))
def test_gitea_key_has_no_identity_file_at_parse_time(self):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/didericis/foo.git",
"key": {"provider": "gitea", "provisioner_token": "T"},
},
}))
self.assertEqual("", m.bottles["dev"].git[0].IdentityFile)
def test_neither_identity_nor_provisioned_key_dies(self):
with self.assertRaises(ManifestError) as ctx:
Manifest.from_json_obj(_manifest({
"foo": {"url": "ssh://git@github.com/foo.git"},
}))
self.assertIn("exactly one of", str(ctx.exception))
self.assertIn("got neither", str(ctx.exception))
def test_unknown_key_in_provisioned_key_block_dies(self):
def test_gitea_key_missing_provisioner_token_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"provisioned_key": {
"key": {"provider": "gitea"},
},
}))
def test_gitea_key_unknown_field_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"key": {
"provider": "gitea",
"token_env": "T",
"provisioner_token": "T",
"key_type": "rsa", # not allowed
},
},
}))
class TestKeyBlockValidation(unittest.TestCase):
"""Validation rules on the key block shared across providers."""
def test_missing_provider_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"provisioned_key": {"token_env": "T"},
"key": {"path": "/dev/null"},
},
}))
def test_missing_token_env_dies(self):
def test_unknown_provider_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"provisioned_key": {"provider": "gitea"},
"key": {"provider": "github"},
},
}))
def test_provisioned_key_entry_has_no_identity_file(self):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/didericis/foo.git",
"provisioned_key": {"provider": "gitea", "token_env": "T"},
},
}))
self.assertEqual("", m.bottles["dev"].git[0].IdentityFile)
def test_identity_entry_has_no_provisioned_key(self):
m = Manifest.from_json_obj(_manifest({
"foo": {
"url": "ssh://git@github.com/foo.git",
"identity": "/dev/null",
},
}))
self.assertIsNone(m.bottles["dev"].git[0].ProvisionedKey)
def test_missing_key_block_dies(self):
with self.assertRaises(ManifestError):
Manifest.from_json_obj(_manifest({
"foo": {"url": "ssh://git@github.com/foo.git"},
}))
class TestEmptyGitGateField(unittest.TestCase):
+1 -1
View File
@@ -76,7 +76,7 @@ class TestGitGateGitconfigRender(unittest.TestCase):
"bottles": {"dev": {"git-gate": {"repos": {
"bot-bottle": {
"url": "ssh://git@100.78.141.42:30009/didericis/bot-bottle.git",
"identity": "/dev/null",
"key": {"provider": "static", "path": "/dev/null"},
},
}}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+3 -2
View File
@@ -33,7 +33,7 @@ from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
from bot_bottle.backend.util import AGENT_CA_PATH
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
from bot_bottle.manifest import ManifestGitEntry, Manifest
from bot_bottle.manifest import ManifestGitEntry, ManifestKeyConfig, Manifest
from bot_bottle.supervise import SupervisePlan
@@ -100,7 +100,7 @@ def _plan(
git_gate_json["repos"] = {
g.Name: {
"url": g.Upstream,
"identity": g.IdentityFile,
"key": {"provider": g.Key.provider or "static", "path": g.Key.path or g.IdentityFile},
}
for g in git
}
@@ -360,6 +360,7 @@ class TestProvisionGit(unittest.TestCase):
git=[ManifestGitEntry(
Name="bot-bottle",
Upstream="ssh://git@host/repo.git",
Key=ManifestKeyConfig(provider="static", path="~/.ssh/id_ed25519"),
IdentityFile="~/.ssh/id_ed25519",
)],
stage_dir=self.stage,