mirror of
https://github.com/onyx-dot-app/onyx.git
synced 2026-10-23 11:53:08 -05:00
Wenxi
119418c07c
## Description Make local Craft startup recover from stale Telepresence agents and use the deployed cluster configuration. Zed reuses compatible intercepts, preserves intercepts it does not own, and cleans up newly created intercepts after failed readiness checks. Allow image builds behind a trusted TLS inspection proxy through an optional BuildKit CA secret. Sandbox package downloads use the secret without persisting it. Add LibreOffice Writer and Calc for document and spreadsheet conversions. Readiness waits for API health and leaves services running if its check fails. The optional build CA secret also covers backend dependencies and Chromium downloads. ## How Has This Been Tested? - Follow-up: all 29 local-development tests and Dockerfile build checks pass. - 29 local development workflow tests passed, including intercept compatibility, ownership, and failed rollout cleanup. - Zed shell syntax and pre-commit checks passed. - No running Zed services or sandbox pods were restarted during review fixes. ## Additional Options - [ ] [Optional] Please cherry-pick this PR to the latest release version. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> <!-- coderabbit-ai-generated:v1 --> ## Summary by CodeRabbit * **Documentation** * Updated local Kubernetes setup guidance for personal configuration overrides, startup retries, and API/frontend readiness checks. * Clarified that intercepts are reused only when they forward port 8080 correctly, and that Zed preserves reused intercepts while cleaning up ones it creates if rollout checks fail. * Added guidance for launching the API and worker together and building images behind TLS inspection proxies with an optional certificate bundle. * **New Features** * Sandbox images built with skills enabled now include LibreOffice Writer and Calc. * Image downloads can use an optional trusted certificate bundle. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
458 lines
19 KiB
Python
458 lines
19 KiB
Python
"""Contracts for shared Craft configuration and staged readiness."""
|
|
|
|
import base64
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).parents[1]))
|
|
from runtime import Cluster, application_ready # noqa: E402
|
|
|
|
|
|
class RuntimeTests(unittest.TestCase):
|
|
def test_command_errors_report_stderr_without_disclosing_configuration(
|
|
self,
|
|
) -> None:
|
|
"""Retain actionable errors while captured configuration stays private."""
|
|
for failure in (
|
|
subprocess.CalledProcessError(
|
|
1, ["kubectl"], output="private-config", stderr="permission denied"
|
|
),
|
|
subprocess.TimeoutExpired(
|
|
["kubectl"], 30, output=b"private-config", stderr=b"request timed out"
|
|
),
|
|
):
|
|
output, errors = io.StringIO(), io.StringIO()
|
|
with (
|
|
self.subTest(failure=type(failure).__name__),
|
|
patch("runtime.subprocess.run", side_effect=failure),
|
|
contextlib.redirect_stdout(output),
|
|
contextlib.redirect_stderr(errors),
|
|
):
|
|
with self.assertRaises(type(failure)) as raised:
|
|
Cluster(Path.cwd()).command("kubectl")
|
|
self.assertIs(raised.exception, failure)
|
|
self.assertIn(
|
|
"permission denied"
|
|
if isinstance(failure, subprocess.CalledProcessError)
|
|
else "request timed out",
|
|
errors.getvalue(),
|
|
)
|
|
self.assertNotIn("private-config", output.getvalue() + errors.getvalue())
|
|
|
|
def test_deployed_sources_and_local_overrides_keep_secrets_in_memory(self) -> None:
|
|
"""Match Kubernetes precedence without writing resolved credentials to local files."""
|
|
workload = {
|
|
"spec": {
|
|
"template": {
|
|
"spec": {
|
|
"containers": [
|
|
{
|
|
"name": "api-server",
|
|
"envFrom": [
|
|
{"configMapRef": {"name": "settings"}},
|
|
{"secretRef": {"name": "credentials"}},
|
|
],
|
|
"env": [
|
|
{"name": "EMPTY"},
|
|
{"name": "POSTGRES_HOST", "value": "database"},
|
|
{"name": "COPY", "value": "$(SHARED)-$$literal"},
|
|
{
|
|
"name": "SELECTED",
|
|
"valueFrom": {
|
|
"secretKeyRef": {
|
|
"name": "credentials",
|
|
"key": "SHARED",
|
|
}
|
|
},
|
|
},
|
|
{
|
|
"name": "UNSET",
|
|
"valueFrom": {
|
|
"secretKeyRef": {
|
|
"name": "absent",
|
|
"key": "value",
|
|
"optional": True,
|
|
}
|
|
},
|
|
},
|
|
],
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
def kube(*args: str, **_kwargs: object) -> str:
|
|
if args[1:3] == ("configmap", "settings"):
|
|
return json.dumps(
|
|
{
|
|
"data": {
|
|
"SHARED": "config",
|
|
"S3_ENDPOINT_URL": "http://store:8333",
|
|
"LOG_LEVEL": "INFO",
|
|
}
|
|
}
|
|
)
|
|
if args[1:3] == ("secret", "credentials"):
|
|
return json.dumps(
|
|
{
|
|
"data": {
|
|
"SHARED": base64.b64encode(b"private-credential").decode()
|
|
}
|
|
}
|
|
)
|
|
return ""
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
override = root / "personal.env"
|
|
contents = (
|
|
"LOG_LEVEL=DEBUG\nK8S_CONTEXT=wrong\nGEN_AI_API_KEY=<REPLACE THIS>\n"
|
|
)
|
|
override.write_text(contents)
|
|
cluster = Cluster(root, "local", "team", "sandboxes")
|
|
with (
|
|
patch.object(Cluster, "api_workload", return_value=workload),
|
|
patch.object(Cluster, "kube", side_effect=kube),
|
|
):
|
|
env = cluster.environment(override)
|
|
self.assertEqual(env["SELECTED"], "private-credential")
|
|
self.assertEqual(env["COPY"], "private-credential-$literal")
|
|
self.assertEqual(env["POSTGRES_HOST"], "database.team.svc.cluster.local")
|
|
self.assertEqual(
|
|
env["S3_ENDPOINT_URL"], "http://store.team.svc.cluster.local:8333"
|
|
)
|
|
self.assertEqual(env["LOG_LEVEL"], "DEBUG")
|
|
self.assertEqual(env["K8S_CONTEXT"], "kind-local")
|
|
self.assertEqual(env["SANDBOX_NAMESPACE"], "sandboxes")
|
|
self.assertNotIn("UNSET", env)
|
|
self.assertEqual(env["EMPTY"], "")
|
|
self.assertEqual(override.read_text(), contents)
|
|
self.assertEqual(list(root.iterdir()), [override])
|
|
|
|
def test_missing_required_secret_fails_before_launch(self) -> None:
|
|
"""Reject incomplete deployed configuration before starting a local application."""
|
|
workload = {
|
|
"spec": {
|
|
"template": {
|
|
"spec": {
|
|
"containers": [
|
|
{
|
|
"name": "api-server",
|
|
"envFrom": [{"secretRef": {"name": "missing"}}],
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
}
|
|
with (
|
|
patch.object(Cluster, "api_workload", return_value=workload),
|
|
patch.object(Cluster, "kube", return_value=""),
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "Missing secret"):
|
|
Cluster(Path.cwd()).deployment_env()
|
|
|
|
def test_failed_rollout_never_reports_infrastructure_ready(self) -> None:
|
|
"""Do not report usable infrastructure after a failed workload rollout."""
|
|
workload = {
|
|
"items": [
|
|
{
|
|
"kind": "Deployment",
|
|
"metadata": {
|
|
"name": "api",
|
|
"labels": {"app.kubernetes.io/instance": "onyx"},
|
|
},
|
|
"spec": {"replicas": 1},
|
|
}
|
|
]
|
|
}
|
|
output = io.StringIO()
|
|
with (
|
|
patch.object(
|
|
Cluster,
|
|
"kube",
|
|
side_effect=[
|
|
json.dumps(workload),
|
|
subprocess.CalledProcessError(1, ["kubectl", "rollout"]),
|
|
],
|
|
),
|
|
contextlib.redirect_stdout(output),
|
|
):
|
|
with self.assertRaises(subprocess.CalledProcessError):
|
|
Cluster(Path.cwd()).infrastructure_ready()
|
|
self.assertNotIn("Infrastructure ready.", output.getvalue())
|
|
|
|
def test_application_readiness_rejects_html_health_and_requires_real_payload(
|
|
self,
|
|
) -> None:
|
|
"""Reject login-page responses as API health while accepting a ready API and frontend."""
|
|
health = {"ready": False}
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
api = self.path.endswith("/api/health/ready")
|
|
body = (
|
|
b'{"success":true}'
|
|
if api and health["ready"]
|
|
else b"<html>login page</html>"
|
|
)
|
|
self.send_response(200)
|
|
self.send_header(
|
|
"Content-Type",
|
|
"application/json" if api and health["ready"] else "text/html",
|
|
)
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, _format: str, *args: object) -> None:
|
|
pass
|
|
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
url = f"http://127.0.0.1:{server.server_port}"
|
|
with self.assertRaisesRegex(ValueError, "Timed out"):
|
|
application_ready(url, timeout=0.05)
|
|
health["ready"] = True
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
application_ready(url, timeout=1)
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join()
|
|
|
|
def test_application_readiness_rejects_non_http_urls(self) -> None:
|
|
"""Restrict readiness probes to HTTP and HTTPS endpoints."""
|
|
with self.assertRaisesRegex(ValueError, "HTTP"):
|
|
application_ready("file:///etc/passwd")
|
|
|
|
def test_intercept_recovers_known_stale_agents_after_failure_or_timeout(
|
|
self,
|
|
) -> None:
|
|
"""Recover stale agent identities for both failures and timeouts without restarting unrelated failures."""
|
|
for failure in (
|
|
subprocess.CalledProcessError(1, ["telepresence", "intercept"]),
|
|
subprocess.TimeoutExpired(["telepresence", "intercept"], 150),
|
|
):
|
|
for stale in (False, True):
|
|
logs = (
|
|
'rpc error: code = NotFound desc = Session "agent:stale" not found'
|
|
if stale
|
|
else "unrelated network failure"
|
|
)
|
|
with (
|
|
self.subTest(failure=type(failure).__name__, stale=stale),
|
|
patch.object(Cluster, "connect"),
|
|
patch.object(
|
|
Cluster,
|
|
"api_workload",
|
|
return_value={"metadata": {"name": "api"}},
|
|
),
|
|
patch.object(Cluster, "intercept_active", return_value=False),
|
|
patch.object(
|
|
Cluster, "command", side_effect=[failure, ""]
|
|
) as command,
|
|
patch.object(Cluster, "kube", return_value=logs) as kube,
|
|
contextlib.redirect_stdout(io.StringIO()),
|
|
):
|
|
if stale:
|
|
Cluster(Path.cwd()).intercept()
|
|
else:
|
|
with self.assertRaises(type(failure)):
|
|
Cluster(Path.cwd()).intercept()
|
|
self.assertEqual(command.call_count, 2 if stale else 1)
|
|
restarts = [
|
|
call
|
|
for call in kube.call_args_list
|
|
if call.args[:2] == ("rollout", "restart")
|
|
]
|
|
self.assertEqual(len(restarts), int(stale))
|
|
|
|
def test_stale_intercept_recovery_retries_only_once(self) -> None:
|
|
"""Stop after one recovery attempt if intercept creation still fails."""
|
|
with (
|
|
patch.object(Cluster, "connect"),
|
|
patch.object(
|
|
Cluster, "api_workload", return_value={"metadata": {"name": "api"}}
|
|
),
|
|
patch.object(Cluster, "intercept_active", return_value=False),
|
|
patch.object(
|
|
Cluster,
|
|
"command",
|
|
side_effect=subprocess.TimeoutExpired(
|
|
["telepresence", "intercept"], 150
|
|
),
|
|
) as command,
|
|
patch.object(
|
|
Cluster,
|
|
"kube",
|
|
return_value='code = NotFound desc = Session "agent:stale" not found',
|
|
),
|
|
):
|
|
with self.assertRaises(subprocess.TimeoutExpired):
|
|
Cluster(Path.cwd()).intercept()
|
|
self.assertEqual(command.call_count, 2)
|
|
|
|
def test_failed_stale_agent_log_probe_preserves_the_intercept_failure(self) -> None:
|
|
"""Do not mask intercept failures when the diagnostic container is unavailable."""
|
|
for probe_failure in (
|
|
subprocess.CalledProcessError(1, ["kubectl", "logs"]),
|
|
subprocess.TimeoutExpired(["kubectl", "logs"], 30),
|
|
):
|
|
original = subprocess.TimeoutExpired(["telepresence", "intercept"], 150)
|
|
with (
|
|
self.subTest(probe=type(probe_failure).__name__),
|
|
patch.object(Cluster, "connect"),
|
|
patch.object(
|
|
Cluster, "api_workload", return_value={"metadata": {"name": "api"}}
|
|
),
|
|
patch.object(Cluster, "intercept_active", return_value=False),
|
|
patch.object(Cluster, "command", side_effect=original) as command,
|
|
patch.object(Cluster, "kube", side_effect=probe_failure) as kube,
|
|
):
|
|
with self.assertRaises(subprocess.TimeoutExpired) as raised:
|
|
Cluster(Path.cwd()).intercept()
|
|
self.assertIs(raised.exception, original)
|
|
self.assertEqual(command.call_count, 1)
|
|
self.assertEqual(kube.call_count, 1)
|
|
|
|
def test_migrations_reject_overrides_to_another_database(self) -> None:
|
|
"""Prevent personal overrides from redirecting migrations away from the selected database."""
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "override.env"
|
|
path.write_text("POSTGRES_HOST=other.example.com\n")
|
|
with patch.object(
|
|
Cluster,
|
|
"deployment_env",
|
|
return_value={"POSTGRES_HOST": "postgres.external.example.com"},
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "POSTGRES_HOST"):
|
|
Cluster(Path(directory)).database_environment(path)
|
|
path.write_text("POSTGRES_DB=another_database\n")
|
|
with patch.object(
|
|
Cluster, "deployment_env", return_value={"POSTGRES_HOST": "database"}
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "POSTGRES_DB"):
|
|
Cluster(Path(directory)).database_environment(path)
|
|
|
|
def test_intercept_ownership_and_rollout_failure(self) -> None:
|
|
"""Only newly acquired intercepts belong to this launcher and get failure cleanup."""
|
|
for reused in (False, True):
|
|
for failed in (False, True):
|
|
with (
|
|
self.subTest(reused=reused, failed=failed),
|
|
tempfile.TemporaryDirectory() as directory,
|
|
patch.object(Cluster, "connect"),
|
|
patch.object(
|
|
Cluster,
|
|
"api_workload",
|
|
return_value={"metadata": {"name": "api"}},
|
|
),
|
|
patch.object(Cluster, "intercept_active", return_value=reused),
|
|
patch.object(Cluster, "command") as command,
|
|
patch.object(
|
|
Cluster,
|
|
"kube",
|
|
side_effect=ValueError("rollout failed") if failed else None,
|
|
),
|
|
contextlib.redirect_stdout(io.StringIO()),
|
|
):
|
|
marker = Path(directory) / "owner"
|
|
cluster = Cluster(Path(directory))
|
|
if failed:
|
|
with self.assertRaisesRegex(ValueError, "rollout failed"):
|
|
cluster.intercept(marker)
|
|
else:
|
|
cluster.intercept(marker)
|
|
self.assertEqual(marker.exists(), not reused)
|
|
if not reused:
|
|
self.assertEqual(marker.read_text(), "" if failed else "api\n")
|
|
leaves = [
|
|
call
|
|
for call in command.call_args_list
|
|
if call.args[:2] == ("telepresence", "leave")
|
|
]
|
|
self.assertEqual(len(leaves), int(failed and not reused))
|
|
|
|
def test_readiness_requires_active_intercept_in_the_selected_context(self) -> None:
|
|
"""Require the selected local API intercept instead of accepting another cluster or waiting intercept."""
|
|
cluster = Cluster(Path.cwd(), "selected")
|
|
active = [
|
|
{
|
|
"name": "api",
|
|
"intercept_info": [
|
|
{
|
|
"disposition": 1,
|
|
"spec": {
|
|
"name": "api",
|
|
"target_host": "127.0.0.1",
|
|
"target_port": 8080,
|
|
"container_port": 8080,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
]
|
|
correct = {
|
|
"user_daemon": {"kubernetes_context": "kind-selected", "namespace": "onyx"}
|
|
}
|
|
with patch.object(
|
|
Cluster, "command", side_effect=[json.dumps(correct), json.dumps(active)]
|
|
):
|
|
self.assertTrue(cluster.intercept_active("api"))
|
|
for field, value in (
|
|
("target_port", 8081),
|
|
("container_port", 9000),
|
|
("target_host", "other-host"),
|
|
):
|
|
with self.subTest(field=field):
|
|
spec = active[0]["intercept_info"][0]["spec"]
|
|
original = spec[field]
|
|
spec[field] = value
|
|
with patch.object(
|
|
Cluster,
|
|
"command",
|
|
side_effect=[json.dumps(correct), json.dumps(active)],
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "incompatible intercept"):
|
|
cluster.intercept_active("api")
|
|
spec[field] = original
|
|
active[0]["intercept_info"][0]["disposition"] = 2
|
|
with patch.object(
|
|
Cluster, "command", side_effect=[json.dumps(correct), json.dumps(active)]
|
|
):
|
|
self.assertFalse(cluster.intercept_active("api"))
|
|
with patch.object(
|
|
Cluster,
|
|
"command",
|
|
return_value=json.dumps(
|
|
{
|
|
"user_daemon": {
|
|
"kubernetes_context": "kind-other",
|
|
"namespace": "onyx",
|
|
}
|
|
}
|
|
),
|
|
):
|
|
self.assertFalse(cluster.intercept_active("api"))
|
|
with (
|
|
patch.object(
|
|
Cluster, "api_workload", return_value={"metadata": {"name": "api"}}
|
|
),
|
|
patch.object(Cluster, "intercept_active", return_value=False),
|
|
):
|
|
with self.assertRaisesRegex(ValueError, "No active API intercept"):
|
|
cluster.application_ready("http://localhost:3000", timeout=1)
|