test: reorganize suite into unit/integration/canaries directories
Replace the hand-maintained INTEGRATION_NAMES classifier (and the
bespoke run_tests.py around it) with a directory-driven split:
tests/unit/ unit tests, always run
tests/integration/ Docker-dependent, skip cleanly without Docker
tests/canaries/ upstream-regression checks, opt-in via
CLAUDE_BOTTLE_RUN_CANARIES=1
The pinned-pipelock-image check moves to the canary suite — it tests
upstream packaging, not our code, so it shouldn't gate every dev push.
A scheduled canaries.yml workflow runs it weekly.
The manifest-runtime tests collapse the four assertRaises cases for
distinct 'runtime' values into one subTest loop and drop the
error-message-wording assertions; the contract is "any value is
rejected", not "the error literally contains 'auto-detect'".
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,39 @@
|
||||
"""Unit: bottle 'runtime' field is no longer supported (PRD 0003).
|
||||
|
||||
gVisor is now auto-detected by the Docker factory. A manifest carrying
|
||||
the legacy 'runtime' field must fail, regardless of value, rather than
|
||||
silently ignoring."""
|
||||
|
||||
import unittest
|
||||
|
||||
from claude_bottle.log import Die
|
||||
from claude_bottle.manifest import Bottle, Manifest
|
||||
|
||||
|
||||
def _manifest_with_runtime(value: object) -> dict:
|
||||
return {
|
||||
"bottles": {"dev": {"runtime": value}},
|
||||
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
||||
}
|
||||
|
||||
|
||||
class TestManifestRuntimeRemoved(unittest.TestCase):
|
||||
def test_loads_when_runtime_absent(self):
|
||||
m = Manifest.from_json_obj({
|
||||
"bottles": {"dev": {}},
|
||||
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
||||
})
|
||||
self.assertIn("dev", m.bottles)
|
||||
|
||||
def test_bottle_dataclass_has_no_runtime_attribute(self):
|
||||
self.assertFalse(hasattr(Bottle(), "runtime"))
|
||||
|
||||
def test_any_runtime_value_is_rejected(self):
|
||||
for value in ("runsc", "runc", "kata-runtime", "", 42, None):
|
||||
with self.subTest(value=value):
|
||||
with self.assertRaises(Die):
|
||||
Manifest.from_json_obj(_manifest_with_runtime(value))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,82 @@
|
||||
"""Unit: allowlist resolution — pipelock_bottle_allowlist,
|
||||
pipelock_bottle_ssh_hostnames, pipelock_bottle_ssh_ip_cidrs,
|
||||
pipelock_bottle_ssh_trusted_domains, pipelock_effective_allowlist."""
|
||||
|
||||
import unittest
|
||||
|
||||
from claude_bottle.log import Die
|
||||
from claude_bottle.manifest import Manifest
|
||||
from claude_bottle.pipelock import (
|
||||
pipelock_bottle_allowlist,
|
||||
pipelock_bottle_ssh_hostnames,
|
||||
pipelock_bottle_ssh_ip_cidrs,
|
||||
pipelock_bottle_ssh_trusted_domains,
|
||||
pipelock_effective_allowlist,
|
||||
)
|
||||
from tests.fixtures import fixture_minimal, fixture_with_egress, fixture_with_ssh
|
||||
|
||||
|
||||
class TestBottleAllowlist(unittest.TestCase):
|
||||
def test_egress_allowlist_present(self):
|
||||
out = pipelock_bottle_allowlist(fixture_with_egress().bottles["dev"])
|
||||
self.assertIn("github.com", out)
|
||||
self.assertIn("gitlab.com", out)
|
||||
self.assertIn("registry.npmjs.org", out)
|
||||
|
||||
def test_empty_when_no_egress_block(self):
|
||||
out = pipelock_bottle_allowlist(fixture_minimal().bottles["dev"])
|
||||
self.assertEqual([], out)
|
||||
|
||||
def test_rejects_non_string_entry(self):
|
||||
bad = {
|
||||
"bottles": {"dev": {"egress": {"allowlist": ["github.com", 42]}}},
|
||||
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
||||
}
|
||||
with self.assertRaises(Die):
|
||||
Manifest.from_json_obj(bad)
|
||||
|
||||
|
||||
class TestSSHHostnames(unittest.TestCase):
|
||||
def test_hostnames_include_both(self):
|
||||
hosts = pipelock_bottle_ssh_hostnames(fixture_with_ssh().bottles["dev"])
|
||||
self.assertIn("100.78.141.42", hosts)
|
||||
self.assertIn("github.com", hosts)
|
||||
|
||||
def test_ip_cidrs_only_ipv4(self):
|
||||
cidrs = pipelock_bottle_ssh_ip_cidrs(fixture_with_ssh().bottles["dev"])
|
||||
self.assertIn("100.78.141.42/32", cidrs)
|
||||
self.assertNotIn("github.com", cidrs)
|
||||
|
||||
def test_trusted_domains_only_hostnames(self):
|
||||
trusted = pipelock_bottle_ssh_trusted_domains(fixture_with_ssh().bottles["dev"])
|
||||
self.assertIn("github.com", trusted)
|
||||
self.assertNotIn("100.78.141.42", trusted)
|
||||
|
||||
|
||||
class TestEffectiveAllowlist(unittest.TestCase):
|
||||
def test_union_and_dedup(self):
|
||||
manifest = Manifest.from_json_obj({
|
||||
"bottles": {
|
||||
"dev": {
|
||||
"egress": {"allowlist": ["registry.npmjs.org"]},
|
||||
"ssh": [
|
||||
{"Host": "ts", "IdentityFile": "/dev/null",
|
||||
"Hostname": "100.78.141.42", "User": "git", "Port": 30009},
|
||||
{"Host": "gh", "IdentityFile": "/dev/null",
|
||||
"Hostname": "github.com", "User": "git", "Port": 22},
|
||||
],
|
||||
}
|
||||
},
|
||||
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
||||
})
|
||||
eff = pipelock_effective_allowlist(manifest.bottles["dev"])
|
||||
self.assertIn("api.anthropic.com", eff)
|
||||
self.assertIn("registry.npmjs.org", eff)
|
||||
self.assertIn("100.78.141.42", eff)
|
||||
self.assertIn("github.com", eff)
|
||||
self.assertEqual(len(eff), len(set(eff)), "deduplicated")
|
||||
self.assertEqual(eff, sorted(eff), "sorted")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Unit: is_ipv4_literal — the classifier that decides whether
|
||||
bottle.ssh[].Hostname goes into pipelock's ssrf.ip_allowlist (IPv4
|
||||
literal) or trusted_domains (hostname)."""
|
||||
|
||||
import unittest
|
||||
|
||||
from claude_bottle.util import is_ipv4_literal
|
||||
|
||||
|
||||
class TestIPv4Classify(unittest.TestCase):
|
||||
def test_positive(self):
|
||||
for ip in ("127.0.0.1", "10.0.0.5", "100.78.141.42", "0.0.0.0", "255.255.255.255"):
|
||||
with self.subTest(ip=ip):
|
||||
self.assertTrue(is_ipv4_literal(ip), ip)
|
||||
|
||||
def test_negative(self):
|
||||
for hn in (
|
||||
"github.com",
|
||||
"gitea.dideric.is",
|
||||
"100.78.141",
|
||||
"100.78.141.42.5",
|
||||
"::1",
|
||||
"fe80::1",
|
||||
"localhost",
|
||||
"",
|
||||
"1.2.3.4.example.com",
|
||||
):
|
||||
with self.subTest(hn=hn):
|
||||
self.assertFalse(is_ipv4_literal(hn), hn)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,82 @@
|
||||
"""Unit: PipelockProxy.prepare — produces a pipelock YAML config
|
||||
containing the expected top-level keys and per-bottle entries. We
|
||||
don't fully parse YAML; we grep for content shape."""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from claude_bottle.backend.docker.pipelock import DockerPipelockProxy
|
||||
from claude_bottle.manifest import Manifest
|
||||
from tests.fixtures import fixture_minimal, fixture_with_ssh
|
||||
|
||||
|
||||
class TestPipelockProxyPrepare(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.out_dir = Path(tempfile.mkdtemp())
|
||||
self.proxy = DockerPipelockProxy()
|
||||
|
||||
def tearDown(self):
|
||||
import shutil
|
||||
shutil.rmtree(self.out_dir, ignore_errors=True)
|
||||
|
||||
def test_minimal(self):
|
||||
yaml_path = self.out_dir / "min.yaml"
|
||||
self.proxy.prepare(fixture_minimal().bottles["dev"], "demo", yaml_path)
|
||||
content = yaml_path.read_text()
|
||||
self.assertIn("mode: strict", content)
|
||||
self.assertIn("enforce: true", content)
|
||||
self.assertIn("api_allowlist:", content)
|
||||
self.assertIn("api.anthropic.com", content)
|
||||
self.assertIn("raw.githubusercontent.com", content)
|
||||
self.assertIn("forward_proxy:", content)
|
||||
self.assertIn("enabled: true", content)
|
||||
self.assertIn("dlp:", content)
|
||||
self.assertIn("include_defaults: true", content)
|
||||
self.assertIn("scan_env: true", content)
|
||||
# No ssh entries → no trusted_domains nor ssrf block.
|
||||
self.assertNotIn("trusted_domains:", content)
|
||||
self.assertNotIn("ssrf:", content)
|
||||
|
||||
def test_ssh_blocks(self):
|
||||
yaml_path = self.out_dir / "ssh.yaml"
|
||||
self.proxy.prepare(fixture_with_ssh().bottles["dev"], "demo", yaml_path)
|
||||
content = yaml_path.read_text()
|
||||
self.assertIn("trusted_domains:", content)
|
||||
self.assertIn("github.com", content)
|
||||
self.assertIn("ssrf:", content)
|
||||
self.assertIn("ip_allowlist:", content)
|
||||
self.assertIn("100.78.141.42/32", content)
|
||||
# ipv4 host should also be in api_allowlist (strict mode requires both).
|
||||
self.assertIn("100.78.141.42", content)
|
||||
|
||||
def test_secret_hygiene(self):
|
||||
manifest = Manifest.from_json_obj({
|
||||
"bottles": {
|
||||
"dev": {
|
||||
"env": {
|
||||
"MY_SECRET": "literal-value-should-not-appear",
|
||||
"ANOTHER": "?prompt-message",
|
||||
},
|
||||
"egress": {"allowlist": ["github.com"]},
|
||||
}
|
||||
},
|
||||
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
|
||||
})
|
||||
yaml_path = self.out_dir / "secret.yaml"
|
||||
self.proxy.prepare(manifest.bottles["dev"], "demo", yaml_path)
|
||||
content = yaml_path.read_text()
|
||||
self.assertNotIn("literal-value-should-not-appear", content)
|
||||
self.assertNotIn("MY_SECRET", content)
|
||||
self.assertNotIn("prompt-message", content)
|
||||
|
||||
def test_file_mode_is_600(self):
|
||||
yaml_path = self.out_dir / "min.yaml"
|
||||
self.proxy.prepare(fixture_minimal().bottles["dev"], "demo", yaml_path)
|
||||
mode = os.stat(yaml_path).st_mode & 0o777
|
||||
self.assertEqual(0o600, mode)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user