jj.wang
31850c4a81
One tick polls in-flight runs, posts results, classifies the first line and moves state/assignee; failures get a structured note with transient retry and backoff, timeouts are killed and handed over, resume failures and long threads rebuild with a handoff summary. Then the lowest Todo items are dispatched within terminal concurrency and serialize groups, and a heartbeat is written. CLI: tick, new, show, comment, state, assign, smoke, check-config.
72 lines
2.7 KiB
Python
72 lines
2.7 KiB
Python
import copy
|
|
import json
|
|
import os
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from agentd import ROOT, config
|
|
from tests.fakes import config_data
|
|
|
|
SCHEMA = json.loads((ROOT / "config" / "schema.json").read_text())
|
|
|
|
|
|
class ConfigTest(unittest.TestCase):
|
|
def test_example_config_is_valid(self):
|
|
cfg = config.load(ROOT / "config" / "agentd.example.yaml")
|
|
self.assertIn("GWENT", cfg.projects)
|
|
self.assertEqual(cfg.token_env("win10"), "PLANE_WINDOW10_API_TOKEN")
|
|
|
|
def test_fixture_is_valid_with_both_validators(self):
|
|
self.assertEqual(config.validate(config_data()), [])
|
|
self.assertEqual(config.mini_validate(config_data(), SCHEMA), [])
|
|
|
|
def check_both(self, data, fragment):
|
|
for errs in (config.validate(data), config.mini_validate(data, SCHEMA) or config.semantic_errors(data)):
|
|
self.assertTrue(errs, "expected an error mentioning %r" % fragment)
|
|
self.assertTrue(any(fragment in e for e in errs), errs)
|
|
|
|
def test_missing_required(self):
|
|
d = config_data()
|
|
del d["terminals"]["win10"]["token_key"]
|
|
self.check_both(d, "token_key")
|
|
|
|
def test_bad_enum(self):
|
|
d = config_data()
|
|
d["terminals"]["oregon"]["transport"] = "telnet"
|
|
self.check_both(d, "telnet")
|
|
|
|
def test_unknown_key(self):
|
|
d = config_data()
|
|
d["projects"]["GWENT"]["routing"]["by_label"]["needs-game"]["serialise"] = "x"
|
|
self.check_both(d, "serialise")
|
|
|
|
def test_wrong_type(self):
|
|
d = config_data()
|
|
d["terminals"]["oregon"]["concurrency"] = "two"
|
|
self.check_both(d, "concurrency")
|
|
|
|
def test_semantic_routing_to_unknown_terminal(self):
|
|
d = config_data()
|
|
d["projects"]["GWENT"]["routing"]["by_assignee"]["mac"] = "mac"
|
|
self.assertTrue(any("unknown terminal 'mac'" in e for e in config.validate(d)))
|
|
|
|
def test_semantic_default_executor_not_on_terminal(self):
|
|
d = config_data()
|
|
d["projects"]["GWENT"]["routing"]["default_executor"]["win10"] = "shell"
|
|
self.assertTrue(any("not in that terminal's executors" in e for e in config.validate(d)))
|
|
|
|
def test_semantic_unknown_executor_on_terminal(self):
|
|
d = copy.deepcopy(config_data())
|
|
d["terminals"]["oregon"]["executors"].append("gemini")
|
|
self.assertTrue(any("unknown executor 'gemini'" in e for e in config.validate(d)))
|
|
|
|
def test_tokens_come_from_env_named_by_key(self):
|
|
cfg = config.Config(config_data())
|
|
with mock.patch.dict(os.environ, {"TOK_W10": "secret"}, clear=False):
|
|
self.assertEqual(cfg.token("win10"), "secret")
|
|
self.assertEqual(cfg.secret_env_args(), ["--secret-env TOK_OR=TOK_OR", "--secret-env TOK_W10=TOK_W10"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|