from types import SimpleNamespace
import unittest
from yr.cli.component.base import ComponentConfig
from yr.cli.component.registry import get_depends_on_overrides
from yr.cli.const import StartMode
from yr.cli.system_launcher import SystemLauncher
class TestableSystemLauncher(SystemLauncher):
def apply_component_overrides_for_test(self, comp_name, launcher):
self._apply_component_overrides(comp_name, launcher)
def get_start_order_for_test(self):
return self._get_start_order()
class TestCliSystemLauncher(unittest.TestCase):
def make_launcher(self, name: str):
return SimpleNamespace(component_config=ComponentConfig(name=name))
def test_disabled_ds_worker_dependency_is_not_silently_removed(self):
system_launcher = TestableSystemLauncher.__new__(TestableSystemLauncher)
system_launcher.mode = StartMode.MASTER
system_launcher.prepend_char_overrides = {}
system_launcher.depends_on_overrides = {
"function_proxy": ["ds_worker", "runtime_launcher"],
}
system_launcher.resolver = SimpleNamespace(
rendered_config={
"mode": {
StartMode.MASTER.value: {
"function_proxy": True,
"runtime_launcher": True,
"ds_worker": False,
}
}
}
)
function_proxy = self.make_launcher("function_proxy")
runtime_launcher = self.make_launcher("runtime_launcher")
system_launcher.apply_component_overrides_for_test("function_proxy", function_proxy)
system_launcher.components = {
"function_proxy": function_proxy,
"runtime_launcher": runtime_launcher,
}
with self.assertRaisesRegex(ValueError, "depends on unknown component 'ds_worker'"):
system_launcher.get_start_order_for_test()
def test_disabled_etcd_dependency_can_be_provided_externally(self):
system_launcher = TestableSystemLauncher.__new__(TestableSystemLauncher)
system_launcher.mode = StartMode.MASTER
system_launcher.prepend_char_overrides = {}
system_launcher.depends_on_overrides = {
"function_master": ["etcd"],
}
system_launcher.resolver = SimpleNamespace(
rendered_config={
"mode": {
StartMode.MASTER.value: {
"function_master": True,
"etcd": False,
}
}
}
)
function_master = self.make_launcher("function_master")
system_launcher.apply_component_overrides_for_test("function_master", function_master)
system_launcher.components = {
"function_master": function_master,
}
self.assertEqual(system_launcher.get_start_order_for_test(), ["function_master"])
def test_function_proxy_does_not_require_runtime_launcher(self):
self.assertNotIn(
"runtime_launcher",
get_depends_on_overrides(StartMode.MASTER)["function_proxy"],
)
self.assertNotIn(
"runtime_launcher",
get_depends_on_overrides(StartMode.AGENT)["function_proxy"],
)
if __name__ == "__main__":
unittest.main()