"""Daemon settings persistence, concurrent saves, and failed reload recovery."""
from concurrent.futures import ThreadPoolExecutor
import json
import stat
import sys
import threading
import time
import unittest
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from harness import Albedo, Provider, exclusive, text
SERVER = Path(__file__).with_name("fake_mcp_server.py")
@exclusive
class SettingsTest(unittest.TestCase):
def setUp(self):
self.provider = Provider(lambda _request: text("ok"))
self.addCleanup(self.provider.close)
self.app = Albedo(self.provider)
self.app.__enter__()
self.addCleanup(self.app.__exit__, None, None, None)
self.session = self.app.session()
self.route = f"/sessions/{self.session}/settings"
def restore(self, path, prior):
if prior is None:
path.unlink(missing_ok=True)
else:
path.write_bytes(prior)
def request(self, path, body=None, method=None):
with self.app.api(path, body, method=method) as response:
return json.load(response)
def snapshot(self):
return self.request("/settings")
def profile(self, **fields):
return {
"extension": "openai",
"baseUrl": self.provider.url,
"model": "fixture-model",
"protocol": "responses",
**fields,
}
def save_mcp(self, name, server, **secrets):
return self.request(
self.route + "/mcp/" + name,
{"server": server, "secrets": secrets},
"PUT",
)
def capability(
self,
kind="instructions",
name="project:AGENTS.md",
scope="session",
enabled=False,
):
return self.request(
self.route + "/capabilities",
{"kind": kind, "name": name, "scope": scope, "enabled": enabled},
)
def test_profiles_preserve_keys_and_unrelated_fields_and_redact_secrets(self):
path = self.app.home / "config.json"
config = json.loads(path.read_text())
config["unrelated"] = {"kept": True}
config["providers"]["settings-one"] = self.profile(custom="keep")
path.write_text(json.dumps(config))
self.request(
"/settings/providers/settings-one",
self.profile(apiKey="settings-secret"),
"PUT",
)
self.request("/settings/providers/settings-one", self.profile(), "PUT")
snapshot = self.snapshot()
self.assertEqual(snapshot["profiles"]["active"], "settings-one")
self.assertTrue(snapshot["profiles"]["providers"]["settings-one"]["hasKey"])
self.assertNotIn("settings-secret", json.dumps(snapshot))
stored = json.loads(path.read_text())
self.assertNotIn("apiKey", stored["providers"]["settings-one"])
self.assertEqual(stored["providers"]["settings-one"]["custom"], "keep")
self.assertEqual(stored["unrelated"], {"kept": True})
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
self.request("/settings/providers/settings-two", self.profile(), "PUT")
self.assertEqual(self.snapshot()["profiles"]["active"], "settings-two")
self.request("/settings/providers/settings-one", method="DELETE")
self.assertNotIn("settings-one", self.snapshot()["credentials"]["providers"])
self.request("/settings/providers/settings-two", method="DELETE")
snapshot = self.snapshot()
self.assertEqual(
snapshot["profiles"]["active"], min(snapshot["profiles"]["providers"])
)
def test_legacy_flat_provider_remains_available_after_save(self):
path = self.app.home / "config.json"
path.write_text(json.dumps(self.profile(apiKey="legacy-secret", custom="keep")))
self.assertTrue(self.snapshot()["profiles"]["providers"]["default"]["hasKey"])
self.request("/settings/providers/new-profile", self.profile(), "PUT")
self.assertIn("default", self.snapshot()["profiles"]["providers"])
self.assertEqual(json.loads(path.read_text())["custom"], "keep")
self.assertNotIn("legacy-secret", json.dumps(self.snapshot()))
def test_provider_validation_normalizes_profiles_and_preserves_key_omission(self):
profile = self.profile(
model=" \t" + "é" * 256 + "\n",
baseUrl=" \t" + self.provider.url + "/v1/// \n",
apiKey="profile-secret",
ignored="drop",
)
del profile["extension"]
name = "A" + "a" * 60 + "._-"
self.request("/settings/providers/" + name, profile, "PUT")
stored = json.loads((self.app.home / "config.json").read_text())
self.assertEqual(
stored["providers"][name],
{
"extension": "openai",
"model": "é" * 256,
"baseUrl": self.provider.url + "/v1",
"protocol": "responses",
},
)
self.assertTrue(self.snapshot()["profiles"]["providers"][name]["hasKey"])
del profile["apiKey"]
self.request("/settings/providers/" + name, profile, "PUT")
self.assertTrue(self.snapshot()["profiles"]["providers"][name]["hasKey"])
profile["apiKey"] = ""
self.request("/settings/providers/" + name, profile, "PUT")
self.assertFalse(self.snapshot()["profiles"]["providers"][name]["hasKey"])
subscription = self.profile(extension="codex", protocol="chat_completions")
del subscription["baseUrl"]
self.request("/settings/providers/subscription", subscription, "PUT")
self.assertEqual(
self.snapshot()["profiles"]["providers"]["subscription"]["baseUrl"], ""
)
def test_invalid_provider_profiles_leave_settings_and_credentials_untouched(self):
paths = [self.app.home / name for name in ("config.json", "creds.json")]
before = [path.read_bytes() if path.exists() else None for path in paths]
invalid_profiles = [
[],
{},
self.profile(model=" \t\n"),
self.profile(model="é" * 257),
self.profile(model="fixture\x00model"),
self.profile(model="fixture\x7fmodel"),
self.profile(protocol="wrong"),
self.profile(extension=""),
self.profile(extension=None),
self.profile(baseUrl=None),
self.profile(baseUrl=""),
self.profile(baseUrl="file:///tmp/model"),
self.profile(baseUrl="HTTPS://example.com"),
self.profile(baseUrl="https://"),
self.profile(baseUrl="https://user@example.com"),
self.profile(baseUrl="https://example.com?"),
self.profile(baseUrl="https://example.com#"),
self.profile(baseUrl="https://example.com:invalid"),
self.profile(apiKey=None),
self.profile(apiKey="secret key"),
self.profile(apiKey="secret\nkey"),
self.profile(apiKey="secret\x7fkey"),
]
cases = [("invalid-profile", profile) for profile in invalid_profiles]
cases += [
(name, self.profile())
for name in ("_bad", "a" * 65, "é", "a b", "name\n", "a" * 64 + "\n")
]
for name, profile in cases:
with self.subTest(name=name, profile=profile):
route = "/settings/providers/" + urllib.parse.quote(name, safe="")
with self.assertRaises(urllib.error.HTTPError) as failure:
self.request(route, profile, "PUT")
self.assertEqual(failure.exception.code, 400)
self.assertNotIn("secret", failure.exception.read().decode())
self.assertEqual(
[path.read_bytes() if path.exists() else None for path in paths],
before,
)
def test_ui_mutations_return_complete_preferences_before_any_picker_use(self):
fixture = json.loads(
(Path(__file__).parents[1] / "fixtures/mutation_responses.json").read_text()
)["ui"]
with self.app.api(
"/settings/ui", {"thinking": False}, method="PATCH"
) as response:
self.assertEqual(response.status, fixture["status"])
self.assertEqual(json.load(response), fixture["body"])
session_path = f"/settings/ui/sessions/{self.session}"
self.assertEqual(
self.request(session_path, {"pinned": False}, "PATCH"), fixture["body"]
)
prefs = self.request(session_path + "/open", method="POST")
self.assertEqual(prefs, dict(fixture["body"], opens={self.session: 1}))
def test_ui_updates_and_open_counts_survive_restart_and_session_deletion(self):
session_path = f"/settings/ui/sessions/{self.session}"
(self.app.home / "picker.json").write_text(
json.dumps({"unrelated": {"keep": True}})
)
(self.app.home / "capabilities.json").write_text(
json.dumps({"unrelated": {"keep": True}})
)
self.request("/settings/ui", {"thinking": True}, "PATCH")
self.request(session_path, {"pinned": True, "archived": True}, "PATCH")
for _ in range(3):
self.request(session_path + "/open", method="POST")
self.capability()
self.app.restart()
for name in ("picker.json", "capabilities.json"):
self.assertEqual(
json.loads((self.app.home / name).read_text())["unrelated"],
{"keep": True},
)
prefs = self.snapshot()["ui"]
self.assertTrue(prefs["thinking"])
self.assertIn(self.session, prefs["pinned"])
self.assertIn(self.session, prefs["archived"])
self.assertEqual(prefs["opens"][self.session], 3)
self.request(f"/sessions/{self.session}", method="DELETE")
snapshot = self.snapshot()
self.assertNotIn(self.session, snapshot["ui"]["pinned"])
self.assertNotIn(self.session, snapshot["ui"]["archived"])
self.assertNotIn(self.session, snapshot["ui"]["opens"])
self.assertNotIn(self.session, snapshot["capabilities"]["sessions"])
def test_concurrent_clients_preserve_independent_updates(self):
ids = [self.session, self.app.session()]
jobs = [
lambda: self.request("/settings/ui", {"thinking": True}, "PATCH"),
lambda: self.request("/settings/ui", {"tools": True}, "PATCH"),
]
for session in ids:
jobs.append(
lambda session=session: self.request(
f"/settings/ui/sessions/{session}", {"pinned": True}, "PATCH"
)
)
for _ in range(12):
jobs.append(
lambda: self.request(
f"/settings/ui/sessions/{ids[0]}/open", method="POST"
)
)
with ThreadPoolExecutor(max_workers=6) as pool:
list(pool.map(lambda job: job(), jobs))
prefs = self.snapshot()["ui"]
self.assertTrue(prefs["thinking"])
self.assertTrue(prefs["tools"])
self.assertTrue(set(ids) <= set(prefs["pinned"]))
self.assertEqual(prefs["opens"][ids[0]], 12)
def test_capability_inheritance_and_reload_failure_restore_previous_choice(self):
(self.app.workspace / "AGENTS.md").write_text("unique settings instruction")
self.capability(scope="global", enabled=False)
self.capability(enabled=True)
self.capability(enabled=None)
caps = self.snapshot()["capabilities"]
self.assertFalse(caps["global"]["instructions"]["project:AGENTS.md"])
self.assertNotIn(
"project:AGENTS.md", caps["sessions"][self.session]["instructions"]
)
self.app.prompt(self.session, "hello").close()
self.app.idle(self.session)
self.assertNotIn(
"unique settings instruction",
json.dumps(self.provider.requests[-1]["request"]),
)
self.request(
f"/sessions/{self.session}/extensions", {"name": "mcp", "enabled": True}
)
self.capability(kind="mcp", name="broken", enabled=False)
self.save_mcp("broken", {"type": "stdio", "command": "/missing/settings-mcp"})
before = self.snapshot()
with self.assertRaises(urllib.error.HTTPError) as failure:
self.capability(kind="mcp", name="broken", enabled=True)
self.assertEqual(failure.exception.code, 409)
after = self.snapshot()
self.assertEqual(after["mcp"], before["mcp"])
self.assertEqual(after["capabilities"], before["capabilities"])
def test_capability_defaults_and_session_overrides_control_context_and_commands(
self,
):
instruction = "CAPABILITY_INSTRUCTION_MARKER"
description = "CAPABILITY_SKILL_MARKER"
(self.app.workspace / "AGENTS.md").write_text(instruction)
skill = self.app.workspace / ".albedo/skills/preference-demo/SKILL.md"
skill.parent.mkdir(parents=True)
skill.write_text(
"---\nname: preference-demo\ndescription: "
+ description
+ "\n---\nSkill body.\n"
)
def prompt_context(session):
self.app.prompt(session, "inspect capability preferences").close()
self.app.idle(session)
return json.dumps(self.provider.requests[-1]["request"])
def catalog_choice(session, name):
catalog = self.request(f"/sessions/{session}/catalog")
return next(
row for row in catalog["candidates"] if row["preference_key"] == name
)
default_session = self.app.session()
choice = catalog_choice(default_session, "preference-demo")
self.assertIsNone(choice["global_preference"])
self.assertIsNone(choice["session_override"])
self.assertTrue(choice["effective_enabled"])
self.assertTrue(choice["eligible"])
context = prompt_context(default_session)
self.assertIn(instruction, context)
self.assertIn(description, context)
commands_route = f"/sessions/{self.session}/commands"
self.capability(kind="skills", name="preference-demo", enabled=False)
choice = catalog_choice(self.session, "preference-demo")
self.assertIsNone(choice["global_preference"])
self.assertFalse(choice["session_override"])
self.assertFalse(choice["effective_enabled"])
self.assertFalse(choice["eligible"])
self.assertNotIn(
"/preference-demo", {item["name"] for item in self.request(commands_route)}
)
self.capability(kind="skills", name="preference-demo", enabled=None)
choice = catalog_choice(self.session, "preference-demo")
self.assertIsNone(choice["global_preference"])
self.assertIsNone(choice["session_override"])
self.assertTrue(choice["effective_enabled"])
self.assertTrue(choice["eligible"])
self.assertIn(
"/preference-demo", {item["name"] for item in self.request(commands_route)}
)
self.capability(scope="global", enabled=False)
self.capability(
kind="skills", name="preference-demo", scope="global", enabled=False
)
disabled_session = self.app.session()
for name in ("preference-demo", "project:AGENTS.md"):
choice = catalog_choice(disabled_session, name)
self.assertFalse(choice["global_preference"])
self.assertIsNone(choice["session_override"])
self.assertFalse(choice["effective_enabled"])
self.assertFalse(choice["eligible"])
self.assertTrue(choice["valid"])
context = prompt_context(disabled_session)
self.assertNotIn(instruction, context)
self.assertNotIn(description, context)
self.assertNotIn(
"/preference-demo",
{
item["name"]
for item in self.request(f"/sessions/{disabled_session}/commands")
},
)
self.capability(enabled=True)
self.capability(kind="skills", name="preference-demo", enabled=True)
choice = catalog_choice(self.session, "preference-demo")
self.assertFalse(choice["global_preference"])
self.assertTrue(choice["session_override"])
self.assertTrue(choice["effective_enabled"])
self.assertFalse(
catalog_choice(disabled_session, "preference-demo")["effective_enabled"]
)
context = prompt_context(self.session)
self.assertIn(instruction, context)
self.assertIn(description, context)
self.assertIn(
"/preference-demo", {item["name"] for item in self.request(commands_route)}
)
self.capability(kind="skills", name="preference-demo", enabled=None)
choice = catalog_choice(self.session, "preference-demo")
self.assertIsNone(choice["session_override"])
self.assertFalse(choice["effective_enabled"])
self.assertNotIn(
"/preference-demo", {item["name"] for item in self.request(commands_route)}
)
def test_session_override_does_not_hide_malformed_selected_global_preference(self):
skill = self.app.workspace / ".albedo/skills/validated-choice/SKILL.md"
skill.parent.mkdir(parents=True)
skill.write_text(
"---\nname: validated-choice\ndescription: PREPARED_CHOICE\n---\nbody\n"
)
self.capability(kind="skills", name="validated-choice", enabled=True)
commands_route = f"/sessions/{self.session}/commands"
prepared = self.request(commands_route)
self.assertIn("/validated-choice", {row["name"] for row in prepared})
settings_file = self.app.home / "capabilities.json"
preferences = json.loads(settings_file.read_text())
preferences["global"] = {"skills": {"validated-choice": "invalid"}}
settings_file.write_text(json.dumps(preferences))
malformed = settings_file.read_bytes()
skill.write_text(
skill.read_text().replace("PREPARED_CHOICE", "REPAIRED_CHOICE")
)
reload_request = {"name": "/reload", "args": {"target": "session"}}
with self.assertRaises(urllib.error.HTTPError) as failure:
self.request(commands_route, reload_request)
self.assertEqual(failure.exception.code, 409)
self.assertEqual(settings_file.read_bytes(), malformed)
self.assertEqual(self.request(commands_route), prepared)
preferences["global"]["skills"]["validated-choice"] = False
settings_file.write_text(json.dumps(preferences))
repaired = settings_file.read_bytes()
self.request(commands_route, reload_request)
self.assertEqual(settings_file.read_bytes(), repaired)
command = next(
row
for row in self.request(commands_route)
if row["name"] == "/validated-choice"
)
self.assertEqual(command["description"], "REPAIRED_CHOICE")
catalog = self.request(f"/sessions/{self.session}/catalog")
choice = next(
row
for row in catalog["candidates"]
if row["preference_key"] == "validated-choice"
)
self.assertFalse(choice["global_preference"])
self.assertTrue(choice["session_override"])
self.assertTrue(choice["eligible"])
def test_empty_capability_selections_ignore_malformed_preferences(self):
(self.app.home / "capabilities.json").write_text("invalid json")
self.request(
f"/sessions/{self.session}/extensions", {"name": "mcp", "enabled": True}
)
self.app.prompt(self.session, "empty selections remain usable").close()
self.app.idle(self.session)
self.assertEqual(
self.provider.requests[-1]["request"]["model"], "fixture-model"
)
def test_disabled_instruction_is_not_read_until_enabled(self):
(self.app.workspace / "AGENTS.md").write_bytes(b"\xff")
(self.app.workspace / "CLAUDE.md").write_text("READABLE_INSTRUCTION")
self.capability(scope="global", enabled=False)
before = self.snapshot()["capabilities"]
catalog = self.request(f"/sessions/{self.session}/catalog")
invalid = next(
row
for row in catalog["candidates"]
if row["preference_key"] == "project:AGENTS.md"
)
self.assertFalse(invalid["valid"])
self.assertTrue(invalid["diagnostic"])
self.assertFalse(invalid["effective_enabled"])
self.assertFalse(invalid["eligible"])
self.assertEqual(self.snapshot()["capabilities"], before)
self.app.prompt(self.session, "skip disabled invalid UTF-8").close()
self.app.idle(self.session)
self.assertIn(
"READABLE_INSTRUCTION",
json.dumps(self.provider.requests[-1]["request"]),
)
with self.assertRaises(urllib.error.HTTPError) as failure:
self.capability(enabled=True)
self.assertEqual(failure.exception.code, 409)
self.assertIn("AGENTS.md must be UTF-8 text", failure.exception.read().decode())
self.assertEqual(self.snapshot()["capabilities"], before)
def test_capability_size_limit_accepts_boundary_and_rejects_growth(self):
path = self.app.home / "capabilities.json"
limit = 1048576
for size in (limit - 1, limit):
document = {
"padding": "",
"global": {"instructions": {"project:AGENTS.md": False}},
}
overhead = len(json.dumps(document, separators=(",", ":")).encode())
document["padding"] = "a" * (size - overhead)
content = json.dumps(document, separators=(",", ":")).encode()
self.assertEqual(len(content), size)
path.write_bytes(content)
self.assertEqual(
self.snapshot()["capabilities"], {"global": document["global"]}
)
self.capability(scope="global", enabled=False)
self.assertEqual(path.stat().st_size, size)
with self.assertRaises(urllib.error.HTTPError) as failure:
self.capability(name="new-choice")
self.assertIn("exceeds 1 MiB", failure.exception.read().decode())
self.assertEqual(json.loads(path.read_bytes()), document)
self.assertLessEqual(path.stat().st_size, limit)
content += b" "
path.write_bytes(content)
with self.assertRaises(urllib.error.HTTPError):
self.snapshot()
with self.assertRaises(urllib.error.HTTPError):
self.capability()
self.assertEqual(path.read_bytes(), content)
def test_mcp_save_delete_and_failed_connection_restore_settings_and_credentials(
self,
):
self.request(
f"/sessions/{self.session}/extensions", {"name": "mcp", "enabled": True}
)
server = {"type": "stdio", "command": sys.executable, "args": [str(SERVER)]}
self.save_mcp(
"settings-mcp", server, env={"FAKE_SECRET": "settings-mcp-secret"}
)
before = self.snapshot()
self.assertNotIn("settings-mcp-secret", json.dumps(before))
self.assertEqual(
before["credentials"]["mcp"]["settings-mcp"]["env"], ["FAKE_SECRET"]
)
creds = (self.app.home / "creds.json").read_bytes()
with self.assertRaises(urllib.error.HTTPError):
self.save_mcp(
"settings-mcp",
dict(server, command="/missing/mcp"),
env={"FAKE_SECRET": "changed-secret"},
)
self.assertEqual(self.snapshot()["mcp"], before["mcp"])
self.assertEqual(
json.loads((self.app.home / "creds.json").read_bytes()), json.loads(creds)
)
self.app.prompt(self.session, "still usable").close()
self.app.idle(self.session)
tools = self.provider.requests[-1]["request"]["tools"]
self.assertTrue(
any(
tool.get("name", tool.get("function", {}).get("name", "")).startswith(
"mcp_settings_mcp_"
)
for tool in tools
)
)
self.request(self.route + "/mcp/settings-mcp", method="DELETE")
self.assertNotIn("settings-mcp", self.snapshot()["mcp"])
self.assertNotIn("settings-mcp", self.snapshot()["credentials"]["mcp"])
def test_malformed_documents_and_inputs_are_rejected_without_overwriting(self):
cases = [
(
"picker.json",
{"tools": "yes"},
lambda: self.request("/settings/ui", {"tools": True}, "PATCH"),
),
("capabilities.json", {"global": {"skills": []}}, self.capability),
("capabilities.json", {"sessions": []}, self.capability),
(
"capabilities.json",
{"sessions": {"other-session": {"mcp": {"broken": "yes"}}}},
self.capability,
),
(
"extensions.json",
{"mcp": {"servers": []}},
lambda: self.save_mcp(
"sample", {"type": "http", "url": "http://localhost/mcp"}
),
),
(
"config.json",
{"providers": []},
lambda: self.request(
"/settings/providers/sample", self.profile(), "PUT"
),
),
(
"config.json",
{"providers": {"broken": self.profile(protocol="wrong")}},
lambda: self.request(
"/settings/providers/sample", self.profile(), "PUT"
),
),
(
"config.json",
{"providers": {"broken\n": self.profile()}},
lambda: self.request(
"/settings/providers/sample", self.profile(), "PUT"
),
),
]
for name, malformed, save in cases:
path = self.app.home / name
prior = path.read_bytes() if path.exists() else None
try:
for content in (b'{"broken":', json.dumps(malformed).encode()):
path.write_bytes(content)
with self.assertRaises(urllib.error.HTTPError):
save()
self.assertEqual(path.read_bytes(), content)
with self.assertRaises(urllib.error.HTTPError):
self.snapshot()
finally:
self.restore(path, prior)
invalid = [
lambda: self.request("/settings/ui", {"tools": "yes"}, "PATCH"),
lambda: self.request(
"/settings/providers/sample", self.profile(protocol="wrong"), "PUT"
),
lambda: self.request(
self.route + "/capabilities",
{"kind": "skills", "name": "sample", "scope": "global"},
),
lambda: self.save_mcp(
"sample", {"type": "http", "url": "file:///tmp/server"}
),
]
for save in invalid:
with self.assertRaises(urllib.error.HTTPError):
save()
def test_failed_persistence_reports_restoration_failure_and_keeps_old_files(self):
before = {
name: (self.app.home / name).read_bytes()
if (self.app.home / name).exists()
else None
for name in ("config.json", "creds.json")
}
mode = stat.S_IMODE(self.app.home.stat().st_mode)
try:
self.app.home.chmod(0o500)
with self.assertRaises(urllib.error.HTTPError) as failure:
self.request(
"/settings/providers/failing",
self.profile(apiKey="must-not-save"),
"PUT",
)
self.assertIn("restoration failed", str(failure.exception))
finally:
self.app.home.chmod(mode)
for name, content in before.items():
path = self.app.home / name
self.assertEqual(path.read_bytes() if path.exists() else None, content)
def test_reload_and_global_extension_save_complete_when_they_overlap(self):
other = self.app.session()
marker = self.app.workspace / "mcp-started"
script = (
"import pathlib,time; "
f"pathlib.Path({str(marker)!r}).touch(); "
"time.sleep(1); "
f"exec(pathlib.Path({str(SERVER)!r}).read_text())"
)
self.save_mcp(
"slow-start",
{"type": "stdio", "command": sys.executable, "args": ["-c", script]},
)
with ThreadPoolExecutor(max_workers=2) as pool:
extension = pool.submit(
self.request,
f"/sessions/{other}/extensions",
{"name": "mcp", "scope": "global", "enabled": True},
)
deadline = time.monotonic() + 10
while not marker.exists() and time.monotonic() < deadline:
time.sleep(0.01)
self.assertTrue(marker.exists(), "MCP initialization did not start")
capability = pool.submit(self.capability, name="overlapping-settings")
extension.result(timeout=10)
capability.result(timeout=10)
def test_every_new_route_rejects_missing_authentication_and_browser_origins(self):
routes = [
("GET", "/settings"),
("PUT", "/settings/providers/auth-check"),
("DELETE", "/settings/providers/auth-check"),
("PATCH", "/settings/ui"),
("PATCH", f"/settings/ui/sessions/{self.session}"),
("POST", f"/settings/ui/sessions/{self.session}/open"),
("POST", self.route + "/capabilities"),
("PUT", self.route + "/mcp/auth-check"),
("DELETE", self.route + "/mcp/auth-check"),
]
for method, path in routes:
for headers in (
{},
{"Origin": "http://localhost"},
{
"Authorization": "Bearer " + self.app.connection["token"],
"Origin": "http://localhost",
},
):
request = urllib.request.Request(
self.app.base + path,
method=method,
headers=headers,
)
with self.assertRaises(urllib.error.HTTPError) as failure:
urllib.request.urlopen(request, timeout=10)
self.assertEqual(failure.exception.code, 403, (method, path, headers))
with failure.exception as response:
rejected = json.load(response)
if "Origin" in headers:
self.assertNotIn("code", rejected)
self.assertIsNone(response.headers.get("Albedo-Error-Code"))
else:
self.assertEqual(rejected["code"], "authentication_required")
self.assertEqual(
response.headers["Albedo-Error-Code"], "authentication_required"
)
def test_busy_session_rejects_settings_before_persistence(self):
entered = threading.Event()
release = threading.Event()
def script(_request):
entered.set()
release.wait(10)
return text("finished")
self.provider.script = script
self.addCleanup(release.set)
self.app.prompt(self.session, "wait").close()
self.assertTrue(entered.wait(10))
before = self.snapshot()["capabilities"]
try:
with self.assertRaises(urllib.error.HTTPError) as failure:
self.capability()
self.assertEqual(failure.exception.code, 409)
self.assertEqual(self.snapshot()["capabilities"], before)
finally:
release.set()
self.app.idle(self.session)