Compare commits

...

11 Commits

Author SHA1 Message Date
didericis-claude 45a096413f fix: add type annotations to __exit__ context manager (pyright)
lint / lint (push) Successful in 1m47s
test / unit (pull_request) Successful in 43s
test / integration (pull_request) Successful in 16s
2026-06-25 15:03:06 -04:00
didericis c6479d62e4 test: add coverage for git gate and supervise server 2026-06-25 15:03:06 -04:00
didericis d0cad3a559 chore: ignore coverage data 2026-06-25 15:03:06 -04:00
didericis c2ddac1be5 test: fix integration coverage failures 2026-06-25 15:03:06 -04:00
didericis 446414144e test: tune coverage exclusions 2026-06-25 15:03:06 -04:00
didericis 8188d6304e ci: add coverage.py reporting 2026-06-25 15:03:06 -04:00
github-actions[bot] 9f7c067e85 ci(prd): assign sequential numbers to new PRDs 2026-06-25 11:42:07 +00:00
didericis-codex 90e84a52e6 fix: remove unused supervise import for pyright
test / unit (pull_request) Successful in 37s
test / integration (pull_request) Successful in 17s
lint / lint (push) Successful in 1m48s
Update Quality Badges / update-badges (push) Failing after 1m18s
prd-number / assign-numbers (push) Successful in 23s
test / unit (push) Successful in 33s
test / integration (push) Successful in 17s
2026-06-25 05:45:55 -04:00
didericis-claude 75755a472f refactor: drop redundant single-parent fast path in _resolve_one_bottle
lint / lint (push) Failing after 1m50s
test / unit (pull_request) Successful in 36s
test / integration (pull_request) Successful in 18s
_fold_parents with one name returns after the first resolve; the
single-element branch was a verbatim copy of the general path.
2026-06-25 05:10:03 -04:00
didericis-claude 2f3dc57fa9 fix: resolve pyright reportUnnecessaryIsInstance in _resolve_one_bottle
Validate list entries against object-typed raw_list before narrowing to
list[str], so the isinstance(pname, str) check is not redundant.
2026-06-25 05:10:03 -04:00
didericis-claude 302920e290 feat: support multiple parents in bottle extends:
Allow extends: to accept a list of bottle names in addition to a plain
string. Parents are resolved independently and folded left-to-right
into a single combined parent before the child is merged on top, so
orthogonal concerns (base env, networking, agent provider) can live in
separate bottles without forcing a linear chain.

Merge rules for the parent fold: env dict-merge with later winning on
collision; git-gate.user per-field overlay; git-gate.repos union by
name with later winning per-field on same name; egress.routes
concatenated; all scalar fields (supervise, agent_provider, egress.log)
use last-wins. The existing child-wins-over-all-parents rule is
unchanged. Cycle detection, diamond deduplication, and missing/invalid
parent errors all work across multi-parent graphs.

Closes #268
2026-06-25 05:10:03 -04:00
13 changed files with 600 additions and 33 deletions
+9
View File
@@ -0,0 +1,9 @@
[run]
branch = True
source = .
[report]
omit =
bot_bottle/egress_addon.py
bot_bottle/cli/tui.py
tests/*
+7 -1
View File
@@ -39,8 +39,14 @@ jobs:
with:
python-version: "3.12"
- name: Install dev requirements
run: python3 -m pip install -r requirements-dev.txt
- name: Run unit tests
run: python3 -m unittest discover -t . -s tests/unit -v
run: python3 -m coverage run -m unittest discover -t . -s tests/unit -v
- name: Report unit coverage
run: python3 -m coverage report -m
integration:
runs-on: ubuntu-latest
+1
View File
@@ -22,3 +22,4 @@ venv/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.coverage
+6 -7
View File
@@ -45,7 +45,6 @@ from ..supervise import (
TOOL_EGRESS_BLOCK,
TOOL_GITLEAKS_ALLOW,
TOOL_EGRESS_TOKEN_ALLOW,
archive_proposal,
list_pending_proposals,
render_diff,
write_audit_entry,
@@ -320,7 +319,7 @@ def _list_once() -> int:
return 0
def _try_init_green() -> int:
def _try_init_green() -> int: # pragma: no cover
"""Initialise a green color pair and return its attr, or 0."""
try:
curses.start_color()
@@ -331,7 +330,7 @@ def _try_init_green() -> int:
return 0
def _main_loop(stdscr: "curses._CursesWindow") -> None: # type: ignore
def _main_loop(stdscr: "curses._CursesWindow") -> None: # type: ignore # pragma: no cover
curses.curs_set(0)
stdscr.timeout(_REFRESH_INTERVAL_MS)
green_attr = _try_init_green()
@@ -421,7 +420,7 @@ def _render(
status_line: str,
*,
green_attr: int = 0, # noqa: F841 — unused, but required by interface
) -> None:
) -> None: # pragma: no cover
stdscr.erase()
h, w = stdscr.getmaxyx()
header = f"bot-bottle supervise ({len(pending)} pending)"
@@ -472,7 +471,7 @@ def _detail_view(
qp: QueuedProposal,
*,
green_attr: int = 0,
) -> None:
) -> None: # pragma: no cover
"""Render the full proposal. Scrollable. Press q to return."""
lines = _detail_lines(qp, green_attr=green_attr)
offset = 0
@@ -524,7 +523,7 @@ def _detail_view(
return
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None: # type: ignore
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None: # type: ignore # pragma: no cover
"""Suspend curses, open $EDITOR on the proposed file, return edited content."""
suffix = _suffix_for_tool(qp.proposal.tool)
curses.endwin()
@@ -535,7 +534,7 @@ def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:
return edited
def _prompt(stdscr: "curses._CursesWindow", label: str) -> str: # type: ignore
def _prompt(stdscr: "curses._CursesWindow", label: str) -> str: # type: ignore # pragma: no cover
"""One-line input at the bottom of the screen."""
curses.curs_set(1)
h, _ = stdscr.getmaxyx()
+110 -18
View File
@@ -49,33 +49,125 @@ def _resolve_one_bottle(
repos_cache[name] = _resolve_repos_raw({}, child_raw)
return bottle
if not isinstance(parent_name_raw, str):
# Normalize to list, accepting both str and list[str].
raw_list: list[object]
if isinstance(parent_name_raw, str):
raw_list = [parent_name_raw]
elif isinstance(parent_name_raw, list):
raw_list = parent_name_raw
else:
raise ManifestError(
f"bottle '{name}' extends must be a string "
f"bottle '{name}' extends must be a string or list of strings "
f"(was {type(parent_name_raw).__name__})"
)
parent_name: str = parent_name_raw
if parent_name == name:
raise ManifestError(
f"bottle '{name}' extends itself; remove the "
f"self-reference"
)
if parent_name not in raws:
avail = ", ".join(sorted(raws.keys())) or "(none)"
raise ManifestError(
f"bottle '{name}' extends '{parent_name}' which is not "
f"defined. Available bottles: {avail}"
)
parent = _resolve_one_bottle(
parent_name, raws, cache, repos_cache, seen + (name,)
# Validate each entry before resolving any of them.
parent_names: list[str] = []
for i, pname in enumerate(raw_list):
if not isinstance(pname, str):
raise ManifestError(
f"bottle '{name}' extends[{i}] must be a string "
f"(was {type(pname).__name__})"
)
parent_names.append(pname)
if pname == name:
raise ManifestError(
f"bottle '{name}' extends itself; remove the self-reference"
)
if pname not in raws:
avail = ", ".join(sorted(raws.keys())) or "(none)"
raise ManifestError(
f"bottle '{name}' extends '{pname}' which is not "
f"defined. Available bottles: {avail}"
)
combined_parent, combined_repos_raw = _fold_parents(
parent_names, raws, cache, repos_cache, seen + (name,)
)
merged_repos_raw = _resolve_repos_raw(repos_cache[parent_name], child_raw)
bottle = _merge_bottles(parent, child_raw, merged_repos_raw, name)
merged_repos_raw = _resolve_repos_raw(combined_repos_raw, child_raw)
bottle = _merge_bottles(combined_parent, child_raw, merged_repos_raw, name)
cache[name] = bottle
repos_cache[name] = merged_repos_raw
return bottle
def _fold_parents(
parent_names: list[str],
raws: dict[str, dict[str, object]],
cache: dict[str, ManifestBottle],
repos_cache: dict[str, dict[str, object]],
seen: tuple[str, ...],
) -> tuple[ManifestBottle, dict[str, object]]:
"""Resolve each parent and fold them left-to-right.
Later parents win over earlier ones on conflict. The `seen` tuple
carries the current bottle's name so cycle detection works across
every parent edge in the multi-parent graph."""
first = parent_names[0]
effective = _resolve_one_bottle(first, raws, cache, repos_cache, seen)
effective_repos_raw = repos_cache[first]
for pname in parent_names[1:]:
later = _resolve_one_bottle(pname, raws, cache, repos_cache, seen)
later_repos_raw = repos_cache[pname]
effective, effective_repos_raw = _fold_two_bottles(
effective, effective_repos_raw, later, later_repos_raw
)
return effective, effective_repos_raw
def _fold_two_bottles(
earlier: ManifestBottle,
earlier_repos_raw: dict[str, object],
later: ManifestBottle,
later_repos_raw: dict[str, object],
) -> tuple[ManifestBottle, dict[str, object]]:
"""Combine two resolved parent bottles; later wins over earlier."""
from .manifest import ManifestBottle, ManifestGitUser
from .manifest_egress import ManifestEgressConfig
from .manifest_git import parse_git_gate_config
from .manifest_util import as_json_object
merged_env = {**earlier.env, **later.env}
merged_git_user = ManifestGitUser(
name=later.git_user.name or earlier.git_user.name,
email=later.git_user.email or earlier.git_user.email,
)
# Repos: union by name; for same-name entries, later wins per-field.
# Unlike _resolve_repos_raw, an empty later_repos_raw means "no repos
# declared" — it does NOT clear the earlier parent's repos.
names = list(earlier_repos_raw) + [
n for n in later_repos_raw if n not in earlier_repos_raw
]
merged_repos_raw: dict[str, object] = {
n: {
**as_json_object(earlier_repos_raw.get(n, {}), "earlier parent repo"),
**as_json_object(later_repos_raw.get(n, {}), "later parent repo"),
}
for n in names
}
if merged_repos_raw:
merged_git, _ = parse_git_gate_config("_fold", {"repos": merged_repos_raw})
else:
merged_git = ()
# Egress: routes concatenate; scalar fields use last-wins.
merged_egress = ManifestEgressConfig(
routes=earlier.egress.routes + later.egress.routes,
Log=later.egress.Log,
)
return ManifestBottle(
env=merged_env,
agent_provider=later.agent_provider,
git=merged_git,
git_user=merged_git_user,
egress=merged_egress,
supervise=later.supervise,
), merged_repos_raw
def _merge_bottles(
parent: ManifestBottle,
child_raw: dict[str, object],
+2
View File
@@ -87,5 +87,7 @@ def load_bottle_chain_from_dir(
parent = fm.get("extends")
if isinstance(parent, str):
to_load.append(parent)
elif isinstance(parent, list):
to_load.extend(p for p in parent if isinstance(p, str))
return resolve_bottles(raws)[bottle_name]
+166
View File
@@ -0,0 +1,166 @@
# PRD 0065: Multi-parent `extends:` for bottles
- **Status:** Active
- **Author:** didericis
- **Created:** 2026-06-25
- **Issue:** #268
- **Extends:** PRD 0025 (`0025-bottle-extends.md`)
## Summary
Allow a bottle's `extends:` field to accept either a single bottle name (existing
behavior) or a list of bottle names (new). Multiple parents are resolved
independently and folded left-to-right into a single effective parent before the
child is merged on top. This lets orthogonal concerns (base env, networking/egress,
agent provider) live in separate bottles and be composed without forcing them into a
linear chain.
## Problem
PRD 0025 shipped single-parent `extends:` and listed "No multi-parent inheritance"
as a non-goal. In practice, users want to compose multiple orthogonal bottles — a
base environment, a networking profile, and an agent-provider override — without
creating a three-level linear chain that couples unrelated parents to each other.
The linear chain workaround has two problems:
1. **Ordering constraint.** `networking extends base` works, but then
`agent extends networking` can't also pick up `base` without going through
`networking`, coupling two unrelated concerns.
2. **Quadratic duplication.** N orthogonal bottles require O(N²) chain variants
(one chain per permutation of applied concerns).
Multi-parent `extends:` removes both constraints: each orthogonal concern stays in
its own bottle, and the child bottle is the only place that names the combination.
## Goals / Success Criteria
- `extends:` accepts a list of strings in addition to a plain string.
- Backward compat: existing single-string `extends:` is unchanged.
- Parents are resolved left-to-right; later entries win on conflict.
- Child wins over all parents (unchanged from PRD 0025).
- Cycle detection covers multi-parent graphs, not just linear chains.
- Diamond inheritance: a shared ancestor is resolved once (via the existing cache).
- Invalid list entries (non-string, undefined bottle, self-reference) die at parse
with clear messages.
- `manifest_loader.py`'s `load_bottle_chain_from_dir` enqueues all parents from a
list `extends:` so the resolver sees every bottle in the graph.
## Non-goals
- No change to the agent-vs-bottle trust boundary (PRD 0025 "Alternatives
considered" option 2 stays rejected).
- No MRO / C3 linearization. Left-to-right fold is sufficient for the expected use
cases.
- No preflight display of per-field provenance across multiple parents (same open
question as PRD 0025; remains a follow-up).
## Design
### Schema
`extends:` now accepts either form:
```yaml
# single parent (unchanged)
extends: base
# multiple parents (new)
extends: [base, networking]
```
Both forms are normalized to a list internally. A list with one element behaves
identically to the string form.
### Merge rules for multi-parent fold
Parents are folded pairwise left-to-right before the child merge. For each step in
the fold, the "earlier" bottle is the running accumulator and the "later" bottle is
the next parent. Rules per field:
| Field | Fold rule |
|--------------------|--------------------------------------------------------------|
| `env` | dict merge; later wins on key collision |
| `git-gate.user` | per-field overlay; later's non-empty fields win |
| `git-gate.repos` | union by name; for same-name entries, later wins per-field |
| `egress.routes` | concatenate (earlier first, later appended) |
| `egress.log` | later wins (last-wins) |
| `agent_provider` | later wins (last-wins) |
| `supervise` | later wins (last-wins) |
After the fold, the combined parent is merged against the child using the existing
PRD 0025 rules (child always wins). The child's `egress.routes` appends to the
combined parent's concatenated routes; `validate_egress_routes` runs once on the
final merged set and catches duplicate hosts.
### Algorithm
```
extends: [p1, p2, p3]
fold:
combined = resolve(p1)
combined = fold_two(combined, resolve(p2))
combined = fold_two(combined, resolve(p3))
merge:
result = _merge_bottles(combined, child_raw, name)
```
`fold_two(earlier, later)` applies the rules in the table above. Cycle detection
(the `seen` tuple) is passed to each parent resolution call unchanged — if any
parent's chain circles back to the current bottle, it is caught. The `cache` dict
ensures a shared ancestor is only resolved once across all parents.
### Error cases
| Condition | Error message shape |
|----------------------------------------|------------------------------------------------------------------|
| `extends` is not a string or list | `extends must be a string or list of strings (was <type>)` |
| A list entry is not a string | `extends[<i>] must be a string (was <type>)` |
| A list entry names an undefined bottle | `extends '<name>' which is not defined. Available bottles: ...` |
| A list entry is the bottle itself | `extends itself; remove the self-reference` |
| Cycle through any parent edge | `is in an extends cycle: <chain>` |
## Implementation
### `bot_bottle/manifest_extends.py`
- `_resolve_one_bottle`: accept `str | list[str]` for `extends`; normalize to list;
validate each entry; for a single-entry list fall through to the existing
single-parent path; for multiple entries call `_fold_parents` then
`_merge_bottles`.
- `_fold_parents(parent_names, raws, cache, repos_cache, seen)`: resolve each
parent and fold pairwise left-to-right; return `(effective_bottle,
effective_repos_raw)`.
- `_fold_two_bottles(earlier, earlier_repos_raw, later, later_repos_raw)`: apply
the fold rules above; return `(folded_bottle, folded_repos_raw)`.
### `bot_bottle/manifest_loader.py`
- `load_bottle_chain_from_dir`: when `extends` is a list, enqueue all parent names
for loading (previously only `isinstance(parent, str)` was handled).
### `tests/unit/test_manifest_extends.py`
- `TestExtendsErrors.test_non_string_extends_dies`: update to use an integer
`extends` value (a list is now valid).
- New class `TestExtendsMultiParent` covering all cases listed in the issue.
## Testing strategy
Unit tests via `ManifestIndex.from_json_obj` (same resolver surface used by all
paths). No integration test changes needed — downstream code consumes the already-
merged bottle and is unchanged.
Test cases:
- Two-parent list: env union, egress routes concat, git repos union
- Last-parent-wins on scalar (supervise, agent_provider)
- Child wins over all parents on conflict
- Diamond: two parents share an ancestor; ancestor resolved once
- Single-element list: identical to string form
- Non-string extends value → ManifestError
- Non-string list entry → ManifestError
- Undefined bottle in list → ManifestError
- Self-reference in list → ManifestError
- Cycle through multi-parent edge → ManifestError
+1
View File
@@ -4,3 +4,4 @@
pylint>=3.0.0
pyright>=1.1.300
coverage>=7.0.0
+7 -4
View File
@@ -92,9 +92,9 @@ class TestSandboxEscape(unittest.TestCase):
"on PATH: curl -sSL https://smolmachines.com/install.sh | sh"
)
# Throwaway "identity file" for the git-gate's `identity` field.
# It need not be a real SSH key: test 5 reaches gitleaks before
# any SSH attempt anyway.
# Throwaway static key for the git-gate fixture. It need not
# be a real SSH key: test 5 reaches gitleaks before any SSH
# attempt anyway.
fd, kp = tempfile.mkstemp(prefix="sandbox-test-key.")
os.close(fd)
cls._key_path = Path(kp)
@@ -123,7 +123,10 @@ class TestSandboxEscape(unittest.TestCase):
"git-gate": {"repos": {
"throwaway": {
"url": "ssh://git@unreachable.invalid:22/throwaway.git",
"identity": str(cls._key_path),
"key": {
"provider": "static",
"path": str(cls._key_path),
},
},
}},
},
@@ -198,6 +198,7 @@ class TestSmolmachinesLaunch(unittest.TestCase):
# connect fails, which is the property chunk 3 will
# preserve once egress is actually running.
r = self.bottle.exec(
"env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
f"curl -s --show-error --max-time 3 http://{self.plan.bundle_ip}:9099 "
"2>&1 || true"
)
+65
View File
@@ -4,6 +4,7 @@ import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from bot_bottle.git_gate import (
GitGate,
@@ -13,6 +14,8 @@ from bot_bottle.git_gate import (
git_gate_render_access_hook,
git_gate_render_entrypoint,
git_gate_render_hook,
revoke_git_gate_provisioned_keys,
_resolve_identity_file,
git_gate_upstreams_for_bottle,
)
from bot_bottle.manifest import ManifestIndex
@@ -328,6 +331,68 @@ class TestPrepare(unittest.TestCase):
self.assertIn("exec git daemon", content)
class TestDynamicKeyProvisioning(unittest.TestCase):
def setUp(self):
self.stage = Path(tempfile.mkdtemp())
def tearDown(self):
import shutil
shutil.rmtree(self.stage, ignore_errors=True)
def _gitea_manifest(self):
return ManifestIndex.from_json_obj({
"bottles": {
"dev": {
"git-gate": {
"repos": {
"repo": {
"url": "ssh://git@gitea.example.com/org/repo.git",
"key": {
"provider": "gitea",
"forge_token_env": "GITEA_TOKEN",
},
"host_key": "ssh-ed25519 AAAA...",
},
},
}
}
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
def test_resolve_identity_file_static_uses_entry_path(self):
entry = fixture_with_git().bottles["dev"].git[0]
self.assertEqual(entry.IdentityFile, _resolve_identity_file(entry, "demo", self.stage))
def test_resolve_identity_file_gitea_provisions_key(self):
entry = self._gitea_manifest().bottles["dev"].git[0]
with patch("bot_bottle.git_gate._provision_dynamic_key", return_value="/tmp/provisioned-key") as mock_provision:
self.assertEqual("/tmp/provisioned-key", _resolve_identity_file(entry, "demo", self.stage))
mock_provision.assert_called_once()
def test_revoke_skips_non_gitea_and_missing_id_file(self):
revoke_git_gate_provisioned_keys(fixture_with_git().bottles["dev"], self.stage)
def test_revoke_calls_delete_for_gitea_entry(self):
bottle = self._gitea_manifest().bottles["dev"]
(self.stage / "repo-deploy-key-id").write_text("123\n")
with patch.dict("os.environ", {"GITEA_TOKEN": "token"}), patch(
"bot_bottle.deploy_key_provisioner.get_provisioner"
) as mock_get_provisioner:
provisioner = mock_get_provisioner.return_value
revoke_git_gate_provisioned_keys(bottle, self.stage)
mock_get_provisioner.assert_called_once()
provisioner.delete.assert_called_once_with("org/repo", "123")
def test_revoke_missing_token_raises(self):
bottle = self._gitea_manifest().bottles["dev"]
(self.stage / "repo-deploy-key-id").write_text("123\n")
with patch.dict("os.environ", {}, clear=True), self.assertRaises(RuntimeError) as cm:
revoke_git_gate_provisioned_keys(bottle, self.stage)
self.assertIn("env var is not set", str(cm.exception))
class TestShellEscaping(unittest.TestCase):
"""Regression tests: all three render functions must produce syntactically
valid sh code even when names and upstream URLs contain shell-special
+176 -3
View File
@@ -423,9 +423,182 @@ class TestExtendsErrors(unittest.TestCase):
)
self.assertIn("extends cycle", msg)
def test_non_string_extends_dies(self):
msg = _error_message(_build, child={"extends": ["base"]})
self.assertIn("extends must be a string", msg)
def test_non_string_non_list_extends_dies(self):
msg = _error_message(_build, child={"extends": 123})
self.assertIn("extends must be a string or list of strings", msg)
def test_list_entry_non_string_dies(self):
msg = _error_message(_build, child={"extends": [123]})
self.assertIn("extends[0] must be a string", msg)
class TestExtendsMultiParent(unittest.TestCase):
"""extends: [p1, p2, ...] — multi-parent composition (issue #268)."""
_GIT_A = {"url": "ssh://git@host-a/a.git", "key": {"provider": "static", "path": "/k"}}
_GIT_B = {"url": "ssh://git@host-b/b.git", "key": {"provider": "static", "path": "/k"}}
def test_single_element_list_same_as_string(self):
m = _build(
base={"env": {"X": "1"}},
child={"extends": ["base"]},
)
self.assertEqual({"X": "1"}, dict(m.bottles["child"].env))
def test_two_parents_env_union(self):
m = _build(
p1={"env": {"A": "1"}},
p2={"env": {"B": "2"}},
child={"extends": ["p1", "p2"]},
)
self.assertEqual({"A": "1", "B": "2"}, dict(m.bottles["child"].env))
def test_two_parents_env_last_wins_on_collision(self):
m = _build(
p1={"env": {"X": "from-p1"}},
p2={"env": {"X": "from-p2"}},
child={"extends": ["p1", "p2"]},
)
self.assertEqual("from-p2", m.bottles["child"].env["X"])
def test_child_wins_over_all_parents(self):
m = _build(
p1={"env": {"X": "from-p1"}},
p2={"env": {"X": "from-p2"}},
child={"extends": ["p1", "p2"], "env": {"X": "from-child"}},
)
self.assertEqual("from-child", m.bottles["child"].env["X"])
def test_two_parents_supervise_last_wins(self):
m = _build(
p1={"supervise": False},
p2={"supervise": True},
child={"extends": ["p1", "p2"]},
)
self.assertTrue(m.bottles["child"].supervise)
def test_child_supervise_overrides_all_parents(self):
m = _build(
p1={"supervise": True},
p2={"supervise": True},
child={"extends": ["p1", "p2"], "supervise": False},
)
self.assertFalse(m.bottles["child"].supervise)
def test_two_parents_egress_routes_concatenated(self):
m = _build(
p1={"egress": {"routes": [{"host": "a.example.com"}]}},
p2={"egress": {"routes": [{"host": "b.example.com"}]}},
child={"extends": ["p1", "p2"]},
)
hosts = [r.Host for r in m.bottles["child"].egress.routes]
self.assertEqual(["a.example.com", "b.example.com"], hosts)
def test_child_egress_appends_after_combined_parents(self):
m = _build(
p1={"egress": {"routes": [{"host": "a.example.com"}]}},
p2={"egress": {"routes": [{"host": "b.example.com"}]}},
child={
"extends": ["p1", "p2"],
"egress": {"routes": [{"host": "c.example.com"}]},
},
)
hosts = [r.Host for r in m.bottles["child"].egress.routes]
self.assertEqual(["a.example.com", "b.example.com", "c.example.com"], hosts)
def test_two_parents_git_repos_union(self):
m = _build(
p1={"git-gate": {"repos": {"a": self._GIT_A}}},
p2={"git-gate": {"repos": {"b": self._GIT_B}}},
child={"extends": ["p1", "p2"]},
)
names = {e.Name for e in m.bottles["child"].git}
self.assertEqual({"a", "b"}, names)
def test_two_parents_git_same_name_later_wins_per_field(self):
# Both parents declare the same repo name. p2's `key` wins; p1's
# `host_key` is preserved because p2 doesn't override it.
p1_entry = {
"url": "ssh://git@host-a/repo.git",
"host_key": "ecdsa AAAA",
"key": {"provider": "static", "path": "/k1"},
}
p2_entry = {
"url": "ssh://git@host-a/repo.git", # required, same url
"key": {"provider": "gitea", "forge_token_env": "TOK"},
}
m = _build(
p1={"git-gate": {"repos": {"repo": p1_entry}}},
p2={"git-gate": {"repos": {"repo": p2_entry}}},
child={"extends": ["p1", "p2"]},
)
entries = m.bottles["child"].git
self.assertEqual(1, len(entries))
e = entries[0]
self.assertEqual("ssh://git@host-a/repo.git", e.Upstream)
self.assertEqual("ecdsa AAAA", e.KnownHostKey)
self.assertEqual("gitea", e.Key.provider)
def test_p1_repos_preserved_when_p2_has_none(self):
m = _build(
p1={"git-gate": {"repos": {"a": self._GIT_A}}},
p2={"env": {"X": "1"}},
child={"extends": ["p1", "p2"]},
)
names = [e.Name for e in m.bottles["child"].git]
self.assertEqual(["a"], names)
def test_diamond_shared_ancestor_resolved_once(self):
# a <- b, a <- c; child extends [b, c]
# `a` must be resolved once and cached.
m = _build(
a={"env": {"FROM_A": "1"}, "supervise": False},
b={"extends": "a", "env": {"FROM_B": "1"}},
c={"extends": "a", "env": {"FROM_C": "1"}},
child={"extends": ["b", "c"]},
)
child = m.bottles["child"]
self.assertEqual("1", child.env["FROM_A"])
self.assertEqual("1", child.env["FROM_B"])
self.assertEqual("1", child.env["FROM_C"])
# supervise=False from `a` threads through both b and c; c is the
# later parent so its effective supervise (False) wins.
self.assertFalse(child.supervise)
def test_three_parents_env_fold_order(self):
m = _build(
p1={"env": {"X": "1", "A": "a"}},
p2={"env": {"X": "2", "B": "b"}},
p3={"env": {"X": "3", "C": "c"}},
child={"extends": ["p1", "p2", "p3"]},
)
env = dict(m.bottles["child"].env)
self.assertEqual("3", env["X"])
self.assertEqual("a", env["A"])
self.assertEqual("b", env["B"])
self.assertEqual("c", env["C"])
def test_undefined_bottle_in_list_dies(self):
msg = _error_message(
_build,
base={"env": {}},
child={"extends": ["base", "ghost"]},
)
self.assertIn("extends 'ghost'", msg)
self.assertIn("not defined", msg)
def test_self_reference_in_list_dies(self):
msg = _error_message(_build, child={"extends": ["child"]})
self.assertIn("extends itself", msg)
def test_cycle_through_multi_parent_edge_dies(self):
msg = _error_message(
_build,
a={"extends": ["b", "c"]},
b={},
c={"extends": "a"},
)
self.assertIn("extends cycle", msg)
class TestExtendsAvailableInBottleKeys(unittest.TestCase):
+49
View File
@@ -364,6 +364,23 @@ class TestHandleToolsCall(unittest.TestCase):
self.config,
)
def test_missing_name_raises(self):
with self.assertRaises(_RpcError) as cm:
handle_tools_call({"arguments": {}}, self.config)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
def test_arguments_must_be_object(self):
with self.assertRaises(_RpcError) as cm:
handle_tools_call(
{
"name": _sv.TOOL_EGRESS_ALLOW,
"arguments": [],
},
self.config,
)
self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
self.assertIn("must be an object", cm.exception.message)
def test_capability_block_call_raises_unknown_tool(self):
with self.assertRaises(_RpcError) as cm:
handle_tools_call(
@@ -426,6 +443,31 @@ class TestHandleToolsCall(unittest.TestCase):
class TestHandleListEgressRoutes(unittest.TestCase):
def test_success_returns_body_text(self):
class _Resp:
def __enter__(self):
return self
def __exit__(self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: object) -> bool:
return False
def read(self):
return b"[{\"host\": \"example.com\"}]"
class _Opener:
def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore
return _Resp()
with patch.object(supervise_server.urllib.request, "build_opener", return_value=_Opener()):
result = handle_list_egress_routes(
{},
ServerConfig(bottle_slug="dev", queue_dir=Path("/unused")),
)
self.assertFalse(result["isError"]) # type: ignore[index]
text = result["content"][0]["text"] # type: ignore[index]
self.assertIn("example.com", text)
def test_url_error_returns_tool_error(self):
class _Opener:
def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore
@@ -485,6 +527,13 @@ class TestFormatResponseText(unittest.TestCase):
self.assertIn("the operator modified", text.lower())
class TestFormatPendingResponseText(unittest.TestCase):
def test_formats_timeout_message(self):
text = supervise_server.format_pending_response_text(12.5)
self.assertIn("status: pending", text)
self.assertIn("12.5s", text)
# --- End-to-end HTTP sanity ------------------------------------------------