Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 54ea1d6057 | |||
| 0b7345f79b | |||
| 894186d615 | |||
| 4e02b74f07 | |||
| 6421c3542f | |||
| 5614bbef83 | |||
| dd7555f293 | |||
| 63bae92dee |
@@ -1,9 +0,0 @@
|
|||||||
[run]
|
|
||||||
branch = True
|
|
||||||
source = .
|
|
||||||
|
|
||||||
[report]
|
|
||||||
omit =
|
|
||||||
bot_bottle/egress_addon.py
|
|
||||||
bot_bottle/cli/tui.py
|
|
||||||
tests/*
|
|
||||||
@@ -39,14 +39,8 @@ jobs:
|
|||||||
with:
|
with:
|
||||||
python-version: "3.12"
|
python-version: "3.12"
|
||||||
|
|
||||||
- name: Install dev requirements
|
|
||||||
run: python3 -m pip install -r requirements-dev.txt
|
|
||||||
|
|
||||||
- name: Run unit tests
|
- name: Run unit tests
|
||||||
run: python3 -m coverage run -m unittest discover -t . -s tests/unit -v
|
run: python3 -m unittest discover -t . -s tests/unit -v
|
||||||
|
|
||||||
- name: Report unit coverage
|
|
||||||
run: python3 -m coverage report -m
|
|
||||||
|
|
||||||
integration:
|
integration:
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
|
|||||||
@@ -8,7 +8,6 @@ on:
|
|||||||
- '**.py'
|
- '**.py'
|
||||||
- '.pylintrc'
|
- '.pylintrc'
|
||||||
- 'pyrightconfig.json'
|
- 'pyrightconfig.json'
|
||||||
- '.coveragerc'
|
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
@@ -46,19 +45,10 @@ jobs:
|
|||||||
echo "errors=$ERRORS" >> $GITHUB_OUTPUT
|
echo "errors=$ERRORS" >> $GITHUB_OUTPUT
|
||||||
echo "Pyright errors: $ERRORS"
|
echo "Pyright errors: $ERRORS"
|
||||||
|
|
||||||
- name: Run coverage and extract percentage
|
|
||||||
id: coverage
|
|
||||||
run: |
|
|
||||||
python -m coverage run -m unittest discover -t . -s tests/unit > /dev/null 2>&1 || true
|
|
||||||
PERCENT=$(python -m coverage report 2>/dev/null | grep '^TOTAL' | grep -oP '\d+(?=%)' | tail -1)
|
|
||||||
echo "percent=$PERCENT" >> $GITHUB_OUTPUT
|
|
||||||
echo "Coverage: $PERCENT%"
|
|
||||||
|
|
||||||
- name: Update badges in README
|
- name: Update badges in README
|
||||||
run: |
|
run: |
|
||||||
PYLINT_SCORE="${{ steps.pylint.outputs.score }}"
|
PYLINT_SCORE="${{ steps.pylint.outputs.score }}"
|
||||||
PYRIGHT_ERRORS="${{ steps.pyright.outputs.errors }}"
|
PYRIGHT_ERRORS="${{ steps.pyright.outputs.errors }}"
|
||||||
COVERAGE_PERCENT="${{ steps.coverage.outputs.percent }}"
|
|
||||||
|
|
||||||
PYLINT_SCORE_ENCODED=$(echo "$PYLINT_SCORE" | sed 's|/|%2F|g')
|
PYLINT_SCORE_ENCODED=$(echo "$PYLINT_SCORE" | sed 's|/|%2F|g')
|
||||||
|
|
||||||
@@ -68,12 +58,9 @@ jobs:
|
|||||||
if [ -n "$PYRIGHT_ERRORS" ]; then
|
if [ -n "$PYRIGHT_ERRORS" ]; then
|
||||||
sed -i "s|/badge/pyright-[^)]*|/badge/pyright-${PYRIGHT_ERRORS}%20errors-brightgreen|" README.md
|
sed -i "s|/badge/pyright-[^)]*|/badge/pyright-${PYRIGHT_ERRORS}%20errors-brightgreen|" README.md
|
||||||
fi
|
fi
|
||||||
if [ -n "$COVERAGE_PERCENT" ]; then
|
|
||||||
sed -i "s|/badge/coverage-[^)]*|/badge/coverage-${COVERAGE_PERCENT}%25-brightgreen|" README.md
|
|
||||||
fi
|
|
||||||
|
|
||||||
echo "Updated badges:"
|
echo "Updated badges:"
|
||||||
grep -E "pylint|pyright|coverage" README.md | head -3
|
grep -E "pylint|pyright" README.md | head -2
|
||||||
|
|
||||||
- name: Commit and push badge updates
|
- name: Commit and push badge updates
|
||||||
run: |
|
run: |
|
||||||
@@ -86,7 +73,7 @@ jobs:
|
|||||||
else
|
else
|
||||||
echo "Badge changes detected, committing..."
|
echo "Badge changes detected, committing..."
|
||||||
git add README.md
|
git add README.md
|
||||||
MSG="chore: update quality badges"$'\n\n'"- Pylint: ${{ steps.pylint.outputs.score }}"$'\n'"- Pyright: ${{ steps.pyright.outputs.errors }} errors"$'\n'"- Coverage: ${{ steps.coverage.outputs.percent }}%"$'\n\n'"[skip ci]"
|
MSG="chore: update quality badges"$'\n\n'"- Pylint: ${{ steps.pylint.outputs.score }}"$'\n'"- Pyright: ${{ steps.pyright.outputs.errors }} errors"$'\n\n'"[skip ci]"
|
||||||
git commit -m "$MSG"
|
git commit -m "$MSG"
|
||||||
git push
|
git push
|
||||||
fi
|
fi
|
||||||
|
|||||||
@@ -22,4 +22,3 @@ venv/
|
|||||||
.pytest_cache/
|
.pytest_cache/
|
||||||
.mypy_cache/
|
.mypy_cache/
|
||||||
.ruff_cache/
|
.ruff_cache/
|
||||||
.coverage
|
|
||||||
|
|||||||
@@ -6,8 +6,7 @@
|
|||||||
|
|
||||||
[](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
|
[](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
|
||||||
[](https://github.com/PyCQA/pylint)
|
[](https://github.com/PyCQA/pylint)
|
||||||
[](https://github.com/microsoft/pyright)
|
[](https://github.com/microsoft/pyright)
|
||||||
[](https://coverage.readthedocs.io/)
|
|
||||||
|
|
||||||
**Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.
|
**Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.
|
||||||
|
|
||||||
|
|||||||
@@ -319,7 +319,7 @@ def _list_once() -> int:
|
|||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
def _try_init_green() -> int: # pragma: no cover
|
def _try_init_green() -> int:
|
||||||
"""Initialise a green color pair and return its attr, or 0."""
|
"""Initialise a green color pair and return its attr, or 0."""
|
||||||
try:
|
try:
|
||||||
curses.start_color()
|
curses.start_color()
|
||||||
@@ -330,7 +330,7 @@ def _try_init_green() -> int: # pragma: no cover
|
|||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
def _main_loop(stdscr: "curses._CursesWindow") -> None: # type: ignore # pragma: no cover
|
def _main_loop(stdscr: "curses._CursesWindow") -> None: # type: ignore
|
||||||
curses.curs_set(0)
|
curses.curs_set(0)
|
||||||
stdscr.timeout(_REFRESH_INTERVAL_MS)
|
stdscr.timeout(_REFRESH_INTERVAL_MS)
|
||||||
green_attr = _try_init_green()
|
green_attr = _try_init_green()
|
||||||
@@ -420,7 +420,7 @@ def _render(
|
|||||||
status_line: str,
|
status_line: str,
|
||||||
*,
|
*,
|
||||||
green_attr: int = 0, # noqa: F841 — unused, but required by interface
|
green_attr: int = 0, # noqa: F841 — unused, but required by interface
|
||||||
) -> None: # pragma: no cover
|
) -> None:
|
||||||
stdscr.erase()
|
stdscr.erase()
|
||||||
h, w = stdscr.getmaxyx()
|
h, w = stdscr.getmaxyx()
|
||||||
header = f"bot-bottle supervise ({len(pending)} pending)"
|
header = f"bot-bottle supervise ({len(pending)} pending)"
|
||||||
@@ -471,7 +471,7 @@ def _detail_view(
|
|||||||
qp: QueuedProposal,
|
qp: QueuedProposal,
|
||||||
*,
|
*,
|
||||||
green_attr: int = 0,
|
green_attr: int = 0,
|
||||||
) -> None: # pragma: no cover
|
) -> None:
|
||||||
"""Render the full proposal. Scrollable. Press q to return."""
|
"""Render the full proposal. Scrollable. Press q to return."""
|
||||||
lines = _detail_lines(qp, green_attr=green_attr)
|
lines = _detail_lines(qp, green_attr=green_attr)
|
||||||
offset = 0
|
offset = 0
|
||||||
@@ -523,7 +523,7 @@ def _detail_view(
|
|||||||
return
|
return
|
||||||
|
|
||||||
|
|
||||||
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None: # type: ignore # pragma: no cover
|
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None: # type: ignore
|
||||||
"""Suspend curses, open $EDITOR on the proposed file, return edited content."""
|
"""Suspend curses, open $EDITOR on the proposed file, return edited content."""
|
||||||
suffix = _suffix_for_tool(qp.proposal.tool)
|
suffix = _suffix_for_tool(qp.proposal.tool)
|
||||||
curses.endwin()
|
curses.endwin()
|
||||||
@@ -534,7 +534,7 @@ def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:
|
|||||||
return edited
|
return edited
|
||||||
|
|
||||||
|
|
||||||
def _prompt(stdscr: "curses._CursesWindow", label: str) -> str: # type: ignore # pragma: no cover
|
def _prompt(stdscr: "curses._CursesWindow", label: str) -> str: # type: ignore
|
||||||
"""One-line input at the bottom of the screen."""
|
"""One-line input at the bottom of the screen."""
|
||||||
curses.curs_set(1)
|
curses.curs_set(1)
|
||||||
h, _ = stdscr.getmaxyx()
|
h, _ = stdscr.getmaxyx()
|
||||||
|
|||||||
+1
-1
@@ -1,4 +1,4 @@
|
|||||||
# PRD 0066: Separate agent and bottle selection
|
# PRD prd-new: Separate agent and bottle selection
|
||||||
|
|
||||||
- **Status:** Active
|
- **Status:** Active
|
||||||
- **Author:** claude
|
- **Author:** claude
|
||||||
@@ -4,4 +4,3 @@
|
|||||||
|
|
||||||
pylint>=3.0.0
|
pylint>=3.0.0
|
||||||
pyright>=1.1.300
|
pyright>=1.1.300
|
||||||
coverage>=7.0.0
|
|
||||||
|
|||||||
@@ -92,9 +92,9 @@ class TestSandboxEscape(unittest.TestCase):
|
|||||||
"on PATH: curl -sSL https://smolmachines.com/install.sh | sh"
|
"on PATH: curl -sSL https://smolmachines.com/install.sh | sh"
|
||||||
)
|
)
|
||||||
|
|
||||||
# Throwaway static key for the git-gate fixture. It need not
|
# Throwaway "identity file" for the git-gate's `identity` field.
|
||||||
# be a real SSH key: test 5 reaches gitleaks before any SSH
|
# It need not be a real SSH key: test 5 reaches gitleaks before
|
||||||
# attempt anyway.
|
# any SSH attempt anyway.
|
||||||
fd, kp = tempfile.mkstemp(prefix="sandbox-test-key.")
|
fd, kp = tempfile.mkstemp(prefix="sandbox-test-key.")
|
||||||
os.close(fd)
|
os.close(fd)
|
||||||
cls._key_path = Path(kp)
|
cls._key_path = Path(kp)
|
||||||
@@ -123,10 +123,7 @@ class TestSandboxEscape(unittest.TestCase):
|
|||||||
"git-gate": {"repos": {
|
"git-gate": {"repos": {
|
||||||
"throwaway": {
|
"throwaway": {
|
||||||
"url": "ssh://git@unreachable.invalid:22/throwaway.git",
|
"url": "ssh://git@unreachable.invalid:22/throwaway.git",
|
||||||
"key": {
|
"identity": str(cls._key_path),
|
||||||
"provider": "static",
|
|
||||||
"path": str(cls._key_path),
|
|
||||||
},
|
|
||||||
},
|
},
|
||||||
}},
|
}},
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -198,7 +198,6 @@ class TestSmolmachinesLaunch(unittest.TestCase):
|
|||||||
# connect fails, which is the property chunk 3 will
|
# connect fails, which is the property chunk 3 will
|
||||||
# preserve once egress is actually running.
|
# preserve once egress is actually running.
|
||||||
r = self.bottle.exec(
|
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 "
|
f"curl -s --show-error --max-time 3 http://{self.plan.bundle_ip}:9099 "
|
||||||
"2>&1 || true"
|
"2>&1 || true"
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -24,36 +24,61 @@ from bot_bottle.dlp_detectors import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
# (case id, sample body carrying the token, substring expected in the reason).
|
|
||||||
# One row per known token shape; all are block-severity credential matches.
|
|
||||||
# `# gitleaks:allow` marks the synthetic tokens so a source scan won't flag them.
|
|
||||||
_TOKEN_PATTERN_CASES: list[tuple[str, str, str]] = [
|
|
||||||
("aws_access_key", "key=AKIAIOSFODNN7EXAMPLE", "AWS access key"),
|
|
||||||
("github_classic", "token: ghp_" + "A" * 36, "GitHub token"), # gitleaks:allow
|
|
||||||
("github_fine_grained", "pat=github_pat_" + "A" * 82, "fine-grained"), # gitleaks:allow
|
|
||||||
("anthropic", "auth: sk-ant-" + "A" * 93, "Anthropic"), # gitleaks:allow
|
|
||||||
("openai", "key=sk-" + "A" * 48, "OpenAI"), # gitleaks:allow
|
|
||||||
("stripe_live", "stripe: sk_live_" + "A" * 24, "Stripe"), # gitleaks:allow
|
|
||||||
("bearer_jwt", "Authorization: Bearer " + "A" * 60, "Bearer JWT"), # gitleaks:allow
|
|
||||||
("openai_project", "key=sk-proj-" + "A" * 48, "OpenAI project"), # gitleaks:allow
|
|
||||||
("huggingface", "token=hf_" + "A" * 34, "HuggingFace"), # gitleaks:allow
|
|
||||||
("databricks", "dapi" + "a" * 32, "Databricks"), # gitleaks:allow
|
|
||||||
("slack_bot", "xoxb-00000000000-00000000000-" + "A" * 24, "Slack"), # gitleaks:allow
|
|
||||||
("npm", "npm_" + "A" * 36, "npm"), # gitleaks:allow
|
|
||||||
("sendgrid", "SG." + "A" * 22 + "." + "B" * 43, "SendGrid"), # gitleaks:allow
|
|
||||||
("pypi", "pypi-" + "A" * 80, "PyPI"), # gitleaks:allow
|
|
||||||
("vault", "hvs." + "A" * 24, "Vault"), # gitleaks:allow
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
class TestScanTokenPatterns(unittest.TestCase):
|
class TestScanTokenPatterns(unittest.TestCase):
|
||||||
def test_detects_each_token_pattern(self):
|
def test_aws_access_key(self):
|
||||||
for case_id, sample, expected in _TOKEN_PATTERN_CASES:
|
result = scan_token_patterns("key=AKIAIOSFODNN7EXAMPLE")
|
||||||
with self.subTest(case_id):
|
|
||||||
result = scan_token_patterns(sample)
|
|
||||||
assert result is not None
|
assert result is not None
|
||||||
self.assertEqual("block", result.severity)
|
self.assertEqual("block", result.severity)
|
||||||
self.assertIn(expected, result.reason)
|
self.assertIn("AWS access key", result.reason)
|
||||||
|
|
||||||
|
def test_github_classic_token(self):
|
||||||
|
result = scan_token_patterns(
|
||||||
|
"token: ghp_" + "A" * 36,
|
||||||
|
)
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("GitHub token", result.reason)
|
||||||
|
|
||||||
|
def test_github_fine_grained_token(self):
|
||||||
|
result = scan_token_patterns(
|
||||||
|
"pat=github_pat_" + "A" * 82,
|
||||||
|
)
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("fine-grained", result.reason)
|
||||||
|
|
||||||
|
def test_anthropic_api_key(self):
|
||||||
|
result = scan_token_patterns(
|
||||||
|
"auth: sk-ant-" + "A" * 93,
|
||||||
|
)
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("Anthropic", result.reason)
|
||||||
|
|
||||||
|
def test_openai_api_key(self):
|
||||||
|
result = scan_token_patterns(
|
||||||
|
"key=sk-" + "A" * 48,
|
||||||
|
)
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("OpenAI", result.reason)
|
||||||
|
|
||||||
|
def test_stripe_live_key(self):
|
||||||
|
result = scan_token_patterns(
|
||||||
|
"stripe: sk_live_" + "A" * 24,
|
||||||
|
)
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("Stripe", result.reason)
|
||||||
|
|
||||||
|
def test_bearer_jwt(self):
|
||||||
|
result = scan_token_patterns(
|
||||||
|
"Authorization: Bearer " + "A" * 60,
|
||||||
|
)
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("Bearer JWT", result.reason)
|
||||||
|
|
||||||
|
def test_openai_project_key(self):
|
||||||
|
result = scan_token_patterns(
|
||||||
|
"key=sk-proj-" + "A" * 48,
|
||||||
|
)
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("OpenAI project", result.reason)
|
||||||
|
|
||||||
def test_clean_text_returns_none(self):
|
def test_clean_text_returns_none(self):
|
||||||
self.assertIsNone(scan_token_patterns("hello world"))
|
self.assertIsNone(scan_token_patterns("hello world"))
|
||||||
@@ -282,6 +307,44 @@ class TestEncodedVariants(unittest.TestCase):
|
|||||||
self.assertEqual(len(v), len(set(v)))
|
self.assertEqual(len(v), len(set(v)))
|
||||||
|
|
||||||
|
|
||||||
|
class TestScanTokenPatternsExtended(unittest.TestCase):
|
||||||
|
def test_huggingface_token(self):
|
||||||
|
result = scan_token_patterns("token=hf_" + "A" * 34) # gitleaks:allow
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("HuggingFace", result.reason)
|
||||||
|
|
||||||
|
def test_databricks_token(self):
|
||||||
|
result = scan_token_patterns("dapi" + "a" * 32) # gitleaks:allow
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("Databricks", result.reason)
|
||||||
|
|
||||||
|
def test_slack_bot_token(self):
|
||||||
|
# Use all-zero numeric segments to keep entropy low
|
||||||
|
result = scan_token_patterns("xoxb-00000000000-00000000000-" + "A" * 24) # gitleaks:allow
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("Slack", result.reason)
|
||||||
|
|
||||||
|
def test_npm_token(self):
|
||||||
|
result = scan_token_patterns("npm_" + "A" * 36) # gitleaks:allow
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("npm", result.reason)
|
||||||
|
|
||||||
|
def test_sendgrid_key(self):
|
||||||
|
result = scan_token_patterns("SG." + "A" * 22 + "." + "B" * 43) # gitleaks:allow
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("SendGrid", result.reason)
|
||||||
|
|
||||||
|
def test_pypi_token(self):
|
||||||
|
result = scan_token_patterns("pypi-" + "A" * 80) # gitleaks:allow
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("PyPI", result.reason)
|
||||||
|
|
||||||
|
def test_vault_token(self):
|
||||||
|
result = scan_token_patterns("hvs." + "A" * 24) # gitleaks:allow
|
||||||
|
assert result is not None
|
||||||
|
self.assertIn("Vault", result.reason)
|
||||||
|
|
||||||
|
|
||||||
class TestUnicodeNormalization(unittest.TestCase):
|
class TestUnicodeNormalization(unittest.TestCase):
|
||||||
def test_fullwidth_chars_normalized(self):
|
def test_fullwidth_chars_normalized(self):
|
||||||
# Fullwidth ASCII chars (U+FF21..U+FF3A) should map to ASCII
|
# Fullwidth ASCII chars (U+FF21..U+FF3A) should map to ASCII
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import os
|
|||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import patch
|
|
||||||
|
|
||||||
from bot_bottle.git_gate import (
|
from bot_bottle.git_gate import (
|
||||||
GitGate,
|
GitGate,
|
||||||
@@ -14,8 +13,6 @@ from bot_bottle.git_gate import (
|
|||||||
git_gate_render_access_hook,
|
git_gate_render_access_hook,
|
||||||
git_gate_render_entrypoint,
|
git_gate_render_entrypoint,
|
||||||
git_gate_render_hook,
|
git_gate_render_hook,
|
||||||
revoke_git_gate_provisioned_keys,
|
|
||||||
_resolve_identity_file,
|
|
||||||
git_gate_upstreams_for_bottle,
|
git_gate_upstreams_for_bottle,
|
||||||
)
|
)
|
||||||
from bot_bottle.manifest import ManifestIndex
|
from bot_bottle.manifest import ManifestIndex
|
||||||
@@ -331,68 +328,6 @@ class TestPrepare(unittest.TestCase):
|
|||||||
self.assertIn("exec git daemon", content)
|
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):
|
class TestShellEscaping(unittest.TestCase):
|
||||||
"""Regression tests: all three render functions must produce syntactically
|
"""Regression tests: all three render functions must produce syntactically
|
||||||
valid sh code even when names and upstream URLs contain shell-special
|
valid sh code even when names and upstream URLs contain shell-special
|
||||||
|
|||||||
@@ -364,23 +364,6 @@ class TestHandleToolsCall(unittest.TestCase):
|
|||||||
self.config,
|
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):
|
def test_capability_block_call_raises_unknown_tool(self):
|
||||||
with self.assertRaises(_RpcError) as cm:
|
with self.assertRaises(_RpcError) as cm:
|
||||||
handle_tools_call(
|
handle_tools_call(
|
||||||
@@ -443,31 +426,6 @@ class TestHandleToolsCall(unittest.TestCase):
|
|||||||
|
|
||||||
|
|
||||||
class TestHandleListEgressRoutes(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):
|
def test_url_error_returns_tool_error(self):
|
||||||
class _Opener:
|
class _Opener:
|
||||||
def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore
|
def open(self, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 # type: ignore
|
||||||
@@ -527,13 +485,6 @@ class TestFormatResponseText(unittest.TestCase):
|
|||||||
self.assertIn("the operator modified", text.lower())
|
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 ------------------------------------------------
|
# --- End-to-end HTTP sanity ------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user