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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081"""Host resource policy must honor cgroup ancestors and not scale with huge CPU counts.
Daemon E2E cannot change its host's memory or cgroup layout to cover these rules."""
import osfrom pathlib import Pathimport sysimport tempfileimport unittestfrom unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))import resources # noqa: E402
class ResourceTests(unittest.TestCase): def test_big_hosts_are_capped_and_memory_reduces_all_lanes(self): ordinary = resources.plan(24, 4096 * resources.MIB, 10) large = resources.plan(256, 64 * 1024 * resources.MIB, 256) small = resources.plan(256, 1024 * resources.MIB, 256) self.assertEqual( (ordinary["shared"], ordinary["exclusive"], ordinary["go"]), (4, 8, 4) ) self.assertEqual(large, resources.plan(512, 128 * 1024 * resources.MIB, 512)) for field in resources.FIELDS: self.assertLessEqual(small[field], large[field]) self.assertGreaterEqual(small[field], 1) with self.assertRaisesRegex(ValueError, "available memory"): resources.plan(1, 0, 1) self.assertLess(small["exclusive"], ordinary["exclusive"])
def test_concurrent_reservations_fit_one_shared_budget(self): for memory in (1024, 2048, 4096, 8192, 65536): with self.subTest(memory=memory): workers = resources.plan(256, memory * resources.MIB, 256) reserved = ( workers["shared"] * 384 + workers["gleam"] * 4 + workers["harness"] * 32 + workers["build"] * 128 ) self.assertLessEqual(reserved, memory * 3 // 4)
def test_a_frozen_plan_is_not_rediscovered(self): frozen = {name: "1" for name in resources.ENVIRONMENT} with ( mock.patch.dict(os.environ, frozen), mock.patch.object(resources, "available_memory", return_value=0), ): self.assertEqual(set(resources.limits().values()), {1})
def test_cgroup_headroom_includes_parent_usage(self): with tempfile.TemporaryDirectory() as root: parent = Path(root) child = parent / "child" child.mkdir() (parent / "memory.max").write_text(str(1000 * resources.MIB)) (parent / "memory.current").write_text(str(900 * resources.MIB)) (child / "memory.max").write_text("max") with mock.patch.object(resources, "cgroups", return_value=[child, parent]): self.assertEqual(resources.available_memory(), 100 * resources.MIB)
def test_cpu_quota_limits_builds_but_not_waiting_scenarios(self): with tempfile.TemporaryDirectory() as root: directory = Path(root) (directory / "cpu.max").write_text("150000 100000") with ( mock.patch.object(resources, "cgroups", return_value=[directory]), mock.patch.object(os, "cpu_count", return_value=24), mock.patch.object( os, "sched_getaffinity", return_value=set(range(24)), create=True ), ): self.assertEqual(resources.cpu_count(), 2) self.assertEqual(resources.cpu_count(include_quota=False), 24)
if __name__ == "__main__": unittest.main()