Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336"""Old implicit pools survive startup without expanding future plain profiles.
These scenarios boot real daemons from old config and credential files: accountIDs, cross-file publication, and restart behavior cannot be checked by a freshprofile or a transport-only test."""
from contextlib import closingimport jsonimport statimport unittest
from harness import Albedo, Provider, exclusive, text
def values(document, store): value = document.get("accounts", {}).get(store, []) return value if isinstance(value, list) else [value]
def read(app, name): return json.loads((app.home / name).read_text())
class AccountMigrationTest(unittest.TestCase): @exclusive def test_fresh_home_does_not_create_config_or_credentials(self): with Albedo(providers={}) as app: for _ in range(2): self.assertFalse((app.home / "config.json").exists()) self.assertFalse((app.home / "creds.json").exists()) with app.api("/settings?group=providers") as response: self.assertEqual(json.load(response)["profiles"], {}) app.restart() self.assertFalse((app.home / "config.json").exists()) self.assertFalse((app.home / "creds.json").exists())
@exclusive def test_marker_only_config_keeps_an_empty_named_provider_map(self): with Albedo(providers={}) as app:
def marker_only(fixture): (fixture.home / "config.json").write_text( json.dumps({"accountsVersion": 2}) )
app.restart(prepare=marker_only) self.assertEqual(read(app, "config.json")["providers"], {}) with app.api("/settings?group=providers") as response: self.assertEqual(json.load(response)["profiles"], {})
@exclusive def test_existing_pools_become_explicit_and_keys_are_deduplicated(self): with closing(Provider(lambda _request: text("ok"))) as provider: profiles = { "fixture": { "extension": "openai", "baseUrl": provider.url, "apiKey": "same-key", "model": "fixture-model", "protocol": "chat_completions", }, "same": { "extension": "openai", "baseUrl": provider.url, "apiKey": "same-key", }, "codex": {"extension": "codex"}, "claude": {"extension": "claude"}, "empty": {"extension": "antigravity"}, "pinned": {"extension": "codex", "accountId": "codex-two"}, "explicit": {"extension": "codex", "accountIds": ["codex-two"]}, "go-a": {"extension": "opencode-go", "apiKey": "go-a"}, "go-b": {"extension": "opencode-go", "apiKey": "go-b"}, "go-oauth": {"extension": "opencode-go"}, }
def prepare(app): app.env.pop("OPENCODE_GO_API_KEY", None) app.env.pop("OPENCODE_API_KEY", None) app.store_secrets( "accounts", { "openai-codex": [ {"type": "oauth", "access": "one"}, { "type": "oauth", "access": "two", "_albedo_account_id": "codex-two", }, ], "anthropic": {"type": "oauth", "access": "claude"}, "opencode-go": [ {"type": "oauth", "access": "go-oauth"}, {"key": "legacy-go", "limitedUntil": 123}, ], }, ) app.store_secrets("providers", {"fixture": {"apiKey": "older-key"}})
with Albedo(provider, providers=profiles, prepare=prepare) as app: config = read(app, "config.json") creds = read(app, "creds.json") migrated = config["providers"] self.assertEqual(config["accountsVersion"], 2) codex_ids = [ account["_albedo_account_id"] for account in values(creds, "openai-codex") ] self.assertEqual(migrated["codex"]["accountIds"], codex_ids) self.assertEqual(migrated["pinned"]["accountId"], "codex-two") self.assertEqual(migrated["explicit"]["accountIds"], ["codex-two"]) self.assertEqual( migrated["claude"]["accountId"], values(creds, "anthropic")[0]["_albedo_account_id"], ) self.assertNotIn("accountId", migrated["empty"]) self.assertNotIn("accountIds", migrated["empty"]) keys = values(creds, "openai") self.assertEqual(len(keys), 1) self.assertEqual(keys[0]["apiKey"], "same-key") self.assertEqual(keys[0]["label"], "fixture") for name in ("fixture", "same"): self.assertEqual( migrated[name]["accountId"], keys[0]["_albedo_account_id"] ) self.assertNotIn(name, creds["providers"]) self.assertNotIn("apiKey", migrated[name]) go_accounts = values(creds, "opencode-go") go_ids = [account["_albedo_account_id"] for account in go_accounts] self.assertEqual(len(go_ids), 4) for name in ("go-a", "go-b"): own = next( account["_albedo_account_id"] for account in go_accounts if account.get("apiKey") == name ) self.assertEqual( migrated[name]["accountIds"], [own] + [account_id for account_id in go_ids if account_id != own], ) self.assertEqual(migrated["go-oauth"]["accountIds"], go_ids) legacy = next( account for account in go_accounts if account.get("apiKey") == "legacy-go" ) self.assertEqual(legacy["limitedUntil"], 123) self.assertEqual( stat.S_IMODE((app.home / "creds.json").stat().st_mode), 0o600 ) session = app.session() app.prompt(session, "check migrated key").close() app.idle(session) self.assertEqual( provider.requests[-1]["authorization"], "Bearer same-key" )
@exclusive def test_alibaba_profiles_keep_the_old_pool_and_account_endpoints(self): with closing(Provider(lambda _request: text("ok"))) as provider: profiles = { "fixture": { "extension": "openai", "baseUrl": provider.url, "apiKey": "fixture-key", }, "ali-a": {"extension": "alibaba", "apiKey": "ali-a", "baseUrl": ""}, "ali-b": { "extension": "alibaba", "apiKey": "ali-b", "baseUrl": provider.url, }, }
def prepare(app): app.env["ALIBABA_API_KEY"] = "environment-only-secret" app.store_secrets( "accounts", { "alibaba": { "key": "legacy-ali", "baseUrl": provider.url + "/legacy", "limitedUntil": 456, } }, )
with Albedo(provider, providers=profiles, prepare=prepare) as app: config = read(app, "config.json") creds = read(app, "creds.json") self.assertNotIn("alibaba", creds["accounts"]) self.assertNotIn( "environment-only-secret", (app.home / "creds.json").read_text() ) keys = { account["apiKey"]: account for account in values(creds, "openai") } self.assertEqual( set(keys), {"fixture-key", "ali-a", "ali-b", "legacy-ali"} ) self.assertEqual( keys["legacy-ali"]["endpoint"], provider.url + "/legacy" ) self.assertEqual(keys["legacy-ali"]["limitedUntil"], 456) expected = [ keys[key]["_albedo_account_id"] for key in ("ali-a", "ali-b", "legacy-ali") ] expected.append("env:ALIBABA_API_KEY") for name in ("ali-a", "ali-b"): profile = config["providers"][name] self.assertEqual(profile["extension"], "openai") self.assertEqual(profile["protocol"], "chat_completions") self.assertEqual(profile["accountIds"], expected) self.assertNotIn(name, creds["providers"]) self.assertEqual( config["providers"]["ali-a"]["baseUrl"], keys["ali-a"]["endpoint"] ) self.assertEqual( keys["ali-a"]["endpoint"], "https://token-plan.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1", )
@exclusive def test_restart_keeps_deliberate_plain_profiles_plain(self): def prepare(app): app.store_secrets( "accounts", { "openai-codex": [ {"type": "oauth", "access": "one"}, {"type": "oauth", "access": "two"}, ] }, )
with Albedo(prepare=prepare) as app:
def plain(fixture): config = read(fixture, "config.json") config["providers"]["plain"] = {"extension": "codex"} (fixture.home / "config.json").write_text(json.dumps(config))
before = read(app, "creds.json") app.restart(prepare=plain) self.assertEqual( read(app, "config.json")["providers"]["plain"], {"extension": "codex"} ) self.assertEqual(read(app, "creds.json"), before) app.restart() self.assertEqual(read(app, "creds.json"), before) self.assertEqual( read(app, "config.json")["providers"]["plain"], {"extension": "codex"} )
@exclusive def test_handwritten_key_replaces_an_existing_pin_or_rotation_on_restart(self): with Albedo() as app: for field in ("accountId", "accountIds"):
def replace_key(fixture): config = read(fixture, "config.json") profile = config["providers"][fixture.profile] previous = profile["accountId"] profile.pop("accountId") profile[field] = previous if field == "accountId" else [previous] profile["apiKey"] = field + "-secret" (fixture.home / "config.json").write_text(json.dumps(config))
app.restart(prepare=replace_key) profile = read(app, "config.json")["providers"][app.profile] self.assertNotIn("accountIds", profile) self.assertNotIn("apiKey", profile) key = next( account for account in values(read(app, "creds.json"), "openai") if account["_albedo_account_id"] == profile["accountId"] ) self.assertEqual(key["apiKey"], field + "-secret") session = app.session() app.prompt(session, "use the handwritten key").close() app.idle(session) self.assertEqual( app.provider.requests[-1]["authorization"], "Bearer " + field + "-secret", )
@exclusive def test_alibaba_environment_only_profile_keeps_its_custom_endpoint(self): with closing(Provider(lambda _request: text("ok"))) as provider:
def prepare(app): app.env["ALIBABA_API_KEY"] = "environment-only-secret"
profiles = { "ali": { "extension": "alibaba", "baseUrl": provider.url, "model": "fixture-model", } } with Albedo(provider, providers=profiles, prepare=prepare) as app: profile = read(app, "config.json")["providers"]["ali"] self.assertEqual(profile["accountId"], "env:ALIBABA_API_KEY") self.assertNotIn( "environment-only-secret", (app.home / "creds.json").read_text() ) with app.api("/auth") as response: accounts = json.load(response)["accounts"] env_account = next( account for account in accounts if account["id"] == "env:ALIBABA_API_KEY" ) self.assertEqual(env_account["endpoint"], provider.url) session = app.session() app.prompt(session, "use the migrated environment key").close() app.idle(session) self.assertEqual( provider.requests[-1]["authorization"], "Bearer environment-only-secret", )
@exclusive def test_unreadable_credentials_are_not_replaced_or_marked_migrated(self): def prepare(app): (app.home / "creds.json").write_text("not json")
with Albedo(prepare=prepare) as app: self.assertEqual((app.home / "creds.json").read_text(), "not json") config = read(app, "config.json") self.assertNotIn("accountsVersion", config) self.assertEqual(config["providers"][app.profile]["apiKey"], "fixture-key")