"""Unit tests for ``xcecommon.mapping`` (design.md, ``xcecommon.mapping``; Requirements 1.1, 1.7, 1.8, 1.9, 1.10, 1.11, 1.12, 5.3, 9.1, 9.2, 9.3). Pure unit tests against in-memory dicts (no file I/O) per task 2.3/5.1 -- except where a test deliberately loads the real ``src/configs/*.yaml`` fixtures to prove the mapper handles this repo's actual example configs, not just synthetic minimal dicts (task 2.3's TDD approach explicitly asks for both). These tests build real capnp ``ClusterConfig`` messages via ``xcecommon.schema.load_schema()`` -- the same loaded module ``mapping.py`` targets -- so any drift between this module's hardcoded field names and the real ``controller.capnp`` schema surfaces as an ``AttributeError``/ ``KjException`` here, not silently. """ from __future__ import annotations import inspect from pathlib import Path import pytest import yaml from xcecommon import mapping from xcecommon.schema import load_schema REPO_CONFIGS_DIR = Path(__file__).resolve().parents[2] / "configs" @pytest.fixture def cluster_config(): """A fresh ``ClusterConfig`` builder -- never reused across tests, so every test starts from an all-default message. """ return mapping.new_cluster_config_builder() def _load_real_yaml(name: str) -> dict: path = REPO_CONFIGS_DIR / name with open(path, encoding="utf-8") as fh: return yaml.safe_load(fh) # --------------------------------------------------------------------------- # SECTION_MAP shape # --------------------------------------------------------------------------- def test_section_map_has_eight_entries_one_per_yaml_capnp_field_pair(): assert len(mapping.SECTION_MAP) == 8 def test_section_map_hosts_yaml_feeds_two_distinct_capnp_fields(): hosts_entries = [s for s in mapping.SECTION_MAP if s["yaml_file"] == "hosts.yaml"] assert len(hosts_entries) == 2 capnp_fields = {s["capnp_field"] for s in hosts_entries} assert capnp_fields == {"hgroups", "hostConfigs"} yaml_keys = {s["yaml_key"] for s in hosts_entries} assert yaml_keys == {"hostgroups", "hosts"} def test_section_map_covers_all_14_clusterconfig_fields_or_documents_exclusion(): # 7 scalar (xce.yaml, including worker-rejoin-resync's workerResyncWindowMs/ # workerResyncTimeoutMs, task 1.5) + 7 list capnp_fields covers 14 of the # 15 ClusterConfig fields (@0-@14); configGeneration (@12) is the one # deliberately excluded field (Requirement 1.9), documented in the # module docstring, not present in SECTION_MAP at all. list_fields = {s["capnp_field"] for s in mapping.SECTION_MAP if s["kind"] == "list"} assert list_fields == { "resources", "queues", "ugroups", "hgroups", "hostConfigs", "hproviders", "limits", } scalar_entries = [s for s in mapping.SECTION_MAP if s["kind"] == "scalar"] assert len(scalar_entries) == 1 assert {f[1] for f in mapping._XCE_SCALAR_FIELDS} == { "clusterName", "heartbeatInterval", "inactiveHeartbeatCount", "fetchJobInterval", "netioTimeout", "workerResyncWindowMs", "workerResyncTimeoutMs", } # --------------------------------------------------------------------------- # map_scalar_section (xce.yaml) # --------------------------------------------------------------------------- def test_map_scalar_section_sets_all_7_fields(cluster_config): spec = mapping.SECTION_MAP[0] values = { "cluster_name": "eda1", "heartbeat_interval": 30, "inactive_heartbeat_count": 3, "fetch_job_interval": 10, "netio_timeout": 30, "worker_resync_window_ms": 900000, "worker_resync_timeout_ms": 15000, } mapping.map_scalar_section(spec, values, cluster_config) assert cluster_config.clusterName == "eda1" assert cluster_config.heartbeatInterval == 30 assert cluster_config.inactiveHeartbeatCount == 3 assert cluster_config.fetchJobInterval == 10 assert cluster_config.netioTimeout == 30 assert cluster_config.workerResyncWindowMs == 900000 assert cluster_config.workerResyncTimeoutMs == 15000 def test_map_scalar_section_leaves_missing_optional_keys_at_default(cluster_config): spec = mapping.SECTION_MAP[0] values = {"cluster_name": "eda1", "heartbeat_interval": 30, "inactive_heartbeat_count": 3} mapping.map_scalar_section(spec, values, cluster_config) assert cluster_config.clusterName == "eda1" assert cluster_config.fetchJobInterval == 0 assert cluster_config.netioTimeout == 0 assert cluster_config.workerResyncWindowMs == 0 assert cluster_config.workerResyncTimeoutMs == 0 @pytest.mark.parametrize("empty", [None, {}]) def test_map_scalar_section_noop_on_empty_values(cluster_config, empty): spec = mapping.SECTION_MAP[0] mapping.map_scalar_section(spec, empty, cluster_config) assert cluster_config.clusterName == "" assert cluster_config.heartbeatInterval == 0 # --------------------------------------------------------------------------- # map_list_section: resources.yaml -> resources # --------------------------------------------------------------------------- def _spec_for(capnp_field: str) -> mapping.SectionSpec: for spec in mapping.SECTION_MAP: if spec["capnp_field"] == capnp_field: return spec raise AssertionError(f"no SECTION_MAP entry for {capnp_field!r}") def test_map_list_section_resources_populates_mapped_fields_only(cluster_config): items = [ {"name": "slots", "type": "numeric", "order": "decrease", "consumable": True, "description": "job slot count", "interval": 0}, {"name": "mem", "type": "numeric", "interval": 60, "description": "memory"}, ] mapping.map_list_section(_spec_for("resources"), items, cluster_config) assert len(cluster_config.resources) == 2 r0, r1 = cluster_config.resources assert r0.name == "slots" assert r0.desc == "job slot count" assert r0.interval == 0 assert r0.valtype == 1 # XCE_VAL_NUMERIC assert r0.ordtype == 1 # XCE_ORD_DECR assert r0.flags == 0x80 # XCE_RESF_CONSUMABLE ("release" absent -> not OR'd in) assert r1.name == "mem" assert r1.interval == 60 assert r1.valtype == 1 # XCE_VAL_NUMERIC ("type" present) assert r1.ordtype == 0 # "order" absent on this item -> untouched default assert r1.flags == 0 # "consumable"/"release" both absent -> no bits set # Deliberately-unmapped fields (see module docstring) stay at default. assert r0.idx == 0 assert r0.lastReport == 0 def test_map_list_section_resources_builtin_flag(cluster_config): """``builtin: true`` (resources.yaml) -> XCE_RESF_BUILTIN (0x01) OR'd on top of the other bits: the load indices (mem/io/r1m...) and host attributes (type/model/maxmem...) are produced by hardcoded worker/ controller reporters, and the flag tells them apart from user-declared resources (the CLI's BUILTIN column renders it). Absent/false must stay 0 -- the same convention as consumable/release. """ items = [ {"name": "mem", "type": "numeric", "interval": 60, "consumable": True, "builtin": True, "description": "memory"}, {"name": "mylic", "type": "numeric", "consumable": True, "builtin": False, "description": "user pool"}, {"name": "plain", "type": "numeric"}, # builtin absent ] mapping.map_list_section(_spec_for("resources"), items, cluster_config) r0, r1, r2 = cluster_config.resources assert r0.flags == 0x80 | 0x01 # CONSUMABLE | BUILTIN assert r1.flags == 0x80 # explicit false -> no BUILTIN bit assert r2.flags == 0 # absent -> no BUILTIN bit def test_map_list_section_noop_on_empty_or_missing_items(cluster_config): mapping.map_list_section(_spec_for("resources"), [], cluster_config) mapping.map_list_section(_spec_for("resources"), None, cluster_config) assert len(cluster_config.resources) == 0 def test_map_list_section_resourcemap_maps_distributions_and_all_flag(cluster_config): """resourcemap -> distributions (admin-plane-gaps 1.10): one ResourceDistribution per resourcemap item in YAML order (location -> hosts, value -> the Value union member picked by YAML type). Only the location==["all"] + value shape sets XCE_RESF_SHARED|XCE_RESF_GLOBAL (the cluster-wide static shared pool / license-pool shape); other shapes map definition-only. """ items = [ { "name": "licenses", "type": "numeric", "consumable": True, "resourcemap": [{"location": ["all"], "value": 100}], }, { "name": "hostpool", "type": "numeric", "resourcemap": [ {"value": 100, "location": ["host1", "host2"]}, {"location": ["host5", "host6"]}, # dynamic, worker-reported ], }, { "name": "bigmem", "type": "boolean", "resourcemap": [{"value": True, "location": ["host1"]}], }, { "name": "strres2", "type": "string", "resourcemap": [{"value": "v1", "location": ["host1"]}], }, {"name": "plain", "type": "numeric"}, # no resourcemap at all ] mapping.map_list_section(_spec_for("resources"), items, cluster_config) r0, r1, r2, r3, r4 = cluster_config.resources # [all]+value: SHARED|GLOBAL OR'd on top of CONSUMABLE assert r0.flags == 0x80 | 0x08 | 0x04 # CONSUMABLE | SHARED | GLOBAL assert len(r0.distributions) == 1 assert [str(h) for h in r0.distributions[0].hosts] == ["all"] assert r0.distributions[0].value.u.which == "fvalue" assert r0.distributions[0].value.u.fvalue == 100.0 # host-subset static pool + dynamic shape: distributions mapped # faithfully, flags untouched (no SHARED|GLOBAL without location [all]) assert r1.flags == 0 assert len(r1.distributions) == 2 assert [str(h) for h in r1.distributions[0].hosts] == ["host1", "host2"] assert r1.distributions[0].value.u.fvalue == 100.0 assert [str(h) for h in r1.distributions[1].hosts] == ["host5", "host6"] assert r1.distributions[1].value.u.which == "ivalue32" # capnp union default # boolean/string values pick their union member assert r2.distributions[0].value.u.which == "bvalue" assert bool(r2.distributions[0].value.u.bvalue) is True assert r3.distributions[0].value.u.which == "svalue" assert str(r3.distributions[0].value.u.svalue) == "v1" # no resourcemap -> no distributions, no extra flags assert len(r4.distributions) == 0 assert r4.flags == 0 def test_map_list_section_resourcemap_external_location_flag(cluster_config): """resourcemap location==["external"] (external-license-resources): the resource carries XCE_RESF_SHARED | XCE_RESF_GLOBAL | XCE_RESF_EXTERNAL (0x08|0x04|0x10) -- cluster-wide pool, but its values come only from the extres daemon's free-count pushes, never from a config distribution or worker reports. A bare ["external"] (no value) also maps definition-only -- the daemon owns the values entirely. """ items = [ { "name": "verilog_xl", "type": "numeric", "consumable": True, "resourcemap": [{"location": ["external"]}], }, { "name": "extlic_valued", "type": "numeric", "resourcemap": [{"location": ["external"], "value": 7}], }, ] mapping.map_list_section(_spec_for("resources"), items, cluster_config) r0, r1 = cluster_config.resources # CONSUMABLE | SHARED | GLOBAL | EXTERNAL assert r0.flags == 0x80 | 0x08 | 0x04 | 0x10 assert len(r0.distributions) == 1 assert [str(h) for h in r0.distributions[0].hosts] == ["external"] # valued variant: same flags, the distribution value still maps (the # controller refuses to seed it -- EXTERNAL skips seed_sharedinst) assert r1.flags == 0x08 | 0x04 | 0x10 assert r1.distributions[0].value.u.fvalue == 7.0 def test_map_list_section_resourcemap_real_repo_configs(cluster_config): """The mapper handles the real src/configs/resources.yaml: every resourcemap entry lands as a distribution, and the two [all]+value resources (staticres4, res1) carry SHARED|GLOBAL. """ data = _load_real_yaml("resources.yaml") mapping.map_list_section(_spec_for("resources"), data["resources"], cluster_config) by_name = {r.name: r for r in cluster_config.resources} for r in by_name.values(): expected_maps = [item for item in data["resources"] if item["name"] == r.name][0].get("resourcemap") or [] assert len(r.distributions) == len(expected_maps) for name in ("staticres4", "res1"): assert by_name[name].flags & 0x08 | 0x04 == 0x08 | 0x04 assert len(by_name[name].distributions) == 1 assert [str(h) for h in by_name[name].distributions[0].hosts] == ["all"] assert by_name[name].distributions[0].value.u.fvalue == 100.0 # per-host/per-default shapes stay definition-only for name in ("staticres1", "staticres2", "staticres3", "host_quota"): assert by_name[name].flags & (0x08 | 0x04) == 0 # builtin:true resources (the worker load indices + host attributes the # repo config declares) carry XCE_RESF_BUILTIN; user-declared shapes # (staticres*, extlic) do not. for name in ("slots", "mem", "r1m", "r15m", "r15s", "ut", "pg", "ls", "it", "swp", "io", "tmp", "type", "model", "cpuf", "maxcpus", "maxmem", "maxswap", "maxtmp", "ndisks"): assert by_name[name].flags & 0x01 == 0x01, f"{name} missing BUILTIN" for name in ("dynres1", "dynres2", "staticres1", "staticres2", "staticres3", "staticres4", "res1", "sres1", "bigmem", "strres1", "strres2", "host_quota", "gstring", "extlic"): assert by_name[name].flags & 0x01 == 0, f"{name} unexpectedly BUILTIN" # --------------------------------------------------------------------------- # map_list_section: users.yaml/hosts.yaml -> ugroups/hgroups (named groups) # --------------------------------------------------------------------------- def test_map_list_section_ugroups_maps_groupname_members_exclude(cluster_config): items = [{"groupname": "ug1", "members": ["user1", "user2"], "exclude_members": ["user3"]}] mapping.map_list_section(_spec_for("ugroups"), items, cluster_config) g = cluster_config.ugroups[0] assert g.name == "ug1" assert list(g.members) == ["user1", "user2"] assert list(g.exmembers) == ["user3"] def test_map_list_section_hgroups_omits_exclude_members_when_absent(cluster_config): items = [{"groupname": "hg1", "members": ["all"]}] mapping.map_list_section(_spec_for("hgroups"), items, cluster_config) g = cluster_config.hgroups[0] assert g.name == "hg1" assert list(g.members) == ["all"] assert list(g.exmembers) == [] # --------------------------------------------------------------------------- # map_list_section: hosts.yaml -> hostConfigs (the "feeds two fields" file) # --------------------------------------------------------------------------- def test_hosts_yaml_feeds_hgroups_and_hostconfigs_independently(cluster_config): """The one file that feeds two separate output fields (design.md / task 5.1) -- build both from the same parsed hosts.yaml dict and assert each populated correctly and independently. """ hosts_data = { "hosts": [{"hostname": "h1", "type": "linux", "max_slot": 4}], "hostgroups": [{"groupname": "hg1", "members": ["all"]}], } mapping.map_list_section(_spec_for("hgroups"), hosts_data.get("hostgroups"), cluster_config) mapping.map_list_section(_spec_for("hostConfigs"), hosts_data.get("hosts"), cluster_config) assert cluster_config.hgroups[0].name == "hg1" assert cluster_config.hostConfigs[0].name == "h1" assert cluster_config.hostConfigs[0].mxj == 4 # Independence: touching hostConfigs must not disturb hgroups. assert len(cluster_config.hgroups) == 1 def test_map_list_section_host_config_full_field_mapping(cluster_config): items = [ { "hostname": "default", "type": "x86_64-linux-gnu", "model": "x86_64-linux-gnu", "max_slot": -1, "description": "default host", "resource": [{"name": "mg"}, {"name": "maxswap", "value": 10000}], "sched_stop": [ {"resname": "r1m", "sched": 3.5, "stop": 4.5}, {"resname": "r15s", "sched": 4.0, "stop": 10.0}, ], "dispatch_window": [ {"start_time": "9:00", "end_time": "18:00"}, {"start_time": "21:00", "end_time": "7:00"}, ], }, ] mapping.map_list_section(_spec_for("hostConfigs"), items, cluster_config) h = cluster_config.hostConfigs[0] assert h.name == "default" assert h.type == "x86_64-linux-gnu" assert h.model == "x86_64-linux-gnu" # max_slot == -1 (auto-detect sentinel) cannot be represented directly # by UInt32 mxj -- mapped to the 0xFFFFFFFF bit-pattern instead, which # the controller's `int mxj` field reinterprets back to -1 (see module # docstring and HOSTCONFIG_MXJ_AUTO_DETECT). assert h.mxj == mapping.HOSTCONFIG_MXJ_AUTO_DETECT assert [r.name for r in h.resources] == ["mg", "maxswap"] assert h.resources[0].value == "" assert h.resources[1].value == "10000" assert [t.name for t in h.thresholds] == ["r1m", "r15s"] assert h.thresholds[0].sched == 3.5 assert h.thresholds[0].stop == 4.5 segs = h.dispatchwindows.segs assert len(segs) == 2 assert segs[0].open.hour == 9 assert segs[0].open.min == 0 assert segs[0].close.hour == 18 assert segs[1].open.hour == 21 assert segs[1].close.hour == 7 def test_map_list_section_host_config_sets_mxj_when_max_slot_nonnegative(cluster_config): items = [{"hostname": "worker2", "max_slot": 256}] mapping.map_list_section(_spec_for("hostConfigs"), items, cluster_config) assert cluster_config.hostConfigs[0].mxj == 256 def test_map_list_section_host_config_sets_mxj_auto_detect_sentinel_when_max_slot_absent( cluster_config, ): """The production case: a host entry with no `max_slot` key at all (the normal way to request auto-detect, per hosts.schema.json) must NOT silently ship mxj=0 -- that reads to the controller as "explicitly configure 0 slots" and permanently blocks scheduling on that host (worker.c apply_join_state: `if (nslots < 0) nslots = maxcpus;` never fires for 0). It must ship the 0xFFFFFFFF sentinel instead.""" items = [{"hostname": "worker3"}] mapping.map_list_section(_spec_for("hostConfigs"), items, cluster_config) assert cluster_config.hostConfigs[0].mxj == mapping.HOSTCONFIG_MXJ_AUTO_DETECT assert mapping.HOSTCONFIG_MXJ_AUTO_DETECT != 0 @pytest.mark.parametrize( ("value", "expected"), [ (3.5, 3.5), (2000, 2000.0), (0, 0.0), ("500MB", 500 * 1024**2), ("2GB", 2 * 1024**3), ("1.5GB", 1.5 * 1024**3), ("500 MB", 500 * 1024**2), # space between number and unit ("500mb", 500 * 1024**2), # lowercase unit ("1024B", 1024.0), ("-1KB", -1024.0), ], ) def test_parse_size_or_number_accepts_plain_numbers_and_size_strings(value, expected): assert mapping.parse_size_or_number(value) == expected @pytest.mark.parametrize( "value", [ "not a number", "500XB", # unrecognized unit "GB", # missing number "500", # missing unit -- ambiguous, must be a plain number instead None, [1, 2], ], ) def test_parse_size_or_number_rejects_unparseable_values(value): with pytest.raises((TypeError, ValueError)): mapping.parse_size_or_number(value) def test_map_list_section_host_config_sched_stop_accepts_size_strings(cluster_config): """The production case for parse_size_or_number: an operator writes a human-friendly size string for a byte-valued resource's threshold (e.g. maxmem/mem/tmp/maxswap/maxtmp -- see test.c's is_bytes_valued_resource) instead of computing the raw byte count by hand.""" items = [ { "hostname": "worker4", "sched_stop": [{"resname": "tmp", "sched": "500MB", "stop": "100MB"}], } ] mapping.map_list_section(_spec_for("hostConfigs"), items, cluster_config) threshold = cluster_config.hostConfigs[0].thresholds[0] assert threshold.name == "tmp" assert threshold.sched == 500 * 1024**2 assert threshold.stop == 100 * 1024**2 def test_time_window_day_prefixed_spec_parses_day_hour_minute(cluster_config): items = [ { "name": "night", "priority": 30, "run_window": [ {"start_time": "5:19:00", "end_time": "1:8:30"}, ], } ] mapping.map_list_section(_spec_for("queues"), items, cluster_config) seg = cluster_config.queues[0].runWindow.segs[0] assert seg.open.day == 5 assert seg.open.hour == 19 assert seg.open.min == 0 assert seg.close.day == 1 assert seg.close.hour == 8 assert seg.close.min == 30 # --------------------------------------------------------------------------- # map_list_section: queues.yaml -> queues # --------------------------------------------------------------------------- def test_map_list_section_queue_full_field_mapping(cluster_config): items = [ { "name": "night", "description": "night queue", "priority": 30, "res_req": "mem>1000 rusage[mem=1000]", "gpu_req": "num=1:mode=shared", "sched_stop": [{"resname": "tmp", "sched": 2000, "stop": 1000}], "dispatch_window": [{"start_time": "19:00", "end_time": "8:30"}], "run_window": [{"start_time": "20:00", "end_time": "8:30"}], "pre_exec": "/path/to/preexec.sh", "post_exec": "/path/to/postexec.sh", "users": {"members": ["ug1", "ug2"], "exclude_members": ["user3"]}, "hosts": {"members": ["hgroup1", "hgroup2"], "exclude_members": ["host3"]}, }, ] mapping.map_list_section(_spec_for("queues"), items, cluster_config) q = cluster_config.queues[0] assert q.name == "night" assert q.priority == 30 assert q.resreq == "mem>1000 rusage[mem=1000]" assert q.preExec == "/path/to/preexec.sh" assert q.postExec == "/path/to/postexec.sh" assert list(q.users.members) == ["ug1", "ug2"] assert list(q.users.exmembers) == ["user3"] assert q.users.name == "" # embedded Membership never carries a name assert list(q.hosts.members) == ["hgroup1", "hgroup2"] assert q.thresholds[0].name == "tmp" assert q.thresholds[0].sched == 2000.0 assert q.dispatchWindow.segs[0].open.hour == 19 assert q.runWindow.segs[0].open.hour == 20 # Deliberately-unmapped fields: description, gpu_req, attrs, status, # acceptIntervals, schedDelay have no destination or no data source. assert q.attrs == 0 assert q.status == 0 assert q.acceptIntervals == 0 assert q.schedDelay == 0 def test_map_list_section_queue_minimal_required_fields_only(cluster_config): items = [{"name": "xce", "priority": 30}] mapping.map_list_section(_spec_for("queues"), items, cluster_config) q = cluster_config.queues[0] assert q.name == "xce" assert q.priority == 30 assert q.resreq == "" assert len(q.thresholds) == 0 assert len(q.dispatchWindow.segs) == 0 # --------------------------------------------------------------------------- # map_list_section: scaler.yaml -> hproviders # --------------------------------------------------------------------------- def test_map_list_section_scaler_config_full_field_mapping(cluster_config): items = [ { "name": "aws-asg", "type": "script", "conf_path": "/etc/xce/scalers/aws-asg.conf", "script_path": "/opt/xce/scalers/aws_asg.sh", "script_option": "--region us-east-1", }, {"name": "k8s-pool", "type": "script", "script_path": "/opt/xce/scalers/k8s_pool.sh"}, ] mapping.map_list_section(_spec_for("hproviders"), items, cluster_config) p0, p1 = cluster_config.hproviders assert p0.name == "aws-asg" assert p0.type == "script" assert p0.confPath == "/etc/xce/scalers/aws-asg.conf" assert p0.scriptPath == "/opt/xce/scalers/aws_asg.sh" assert p0.scriptOption == "--region us-east-1" assert p1.name == "k8s-pool" assert p1.confPath == "" assert p1.scriptOption == "" # --------------------------------------------------------------------------- # map_list_section: limits.yaml -> limits # --------------------------------------------------------------------------- def test_map_list_section_limit_config_full_field_mapping(cluster_config): items = [ { "name": "ug2_slots_per_user", "users": {"members": ["ug2"], "exclude_members": ["user3"]}, "resname": "slots", "value": 4, "per_user": True, }, { "name": "night_mem_pct", "queues": {"members": ["night"]}, "resname": "mem", "value": 50, "percent": True, }, { "name": "ug1_normal_slots", "users": {"members": ["ug1"]}, "queues": {"members": ["normal"]}, "resname": "slots", "value": 8, }, ] mapping.map_list_section(_spec_for("limits"), items, cluster_config) l0, l1, l2 = cluster_config.limits assert l0.name == "ug2_slots_per_user" assert list(l0.users.members) == ["ug2"] assert list(l0.users.exmembers) == ["user3"] assert l0.val.resname == "slots" assert l0.val.value == 4.0 assert l0.val.percentage is False assert l0.perUser is True assert l0.perQueue is False assert l1.val.resname == "mem" assert l1.val.percentage is True assert list(l2.users.members) == ["ug1"] assert list(l2.queues.members) == ["normal"] # --------------------------------------------------------------------------- # Order preservation (Requirement 9.1) # --------------------------------------------------------------------------- def test_list_order_preserved_not_sorted(cluster_config): items = [{"name": n} for n in ["zebra", "apple", "mango"]] mapping.map_list_section(_spec_for("resources"), items, cluster_config) assert [r.name for r in cluster_config.resources] == ["zebra", "apple", "mango"] def test_group_membership_order_preserved(cluster_config): items = [{"groupname": "hg1", "members": ["h3", "h1", "h2"]}] mapping.map_list_section(_spec_for("hgroups"), items, cluster_config) assert list(cluster_config.hgroups[0].members) == ["h3", "h1", "h2"] # --------------------------------------------------------------------------- # Determinism (Requirements 9.1, 9.2, 9.3) # --------------------------------------------------------------------------- def _build_full_sections() -> dict[str, dict]: return { "xce": { "cluster_name": "eda1", "heartbeat_interval": 30, "inactive_heartbeat_count": 3, "fetch_job_interval": 10, "netio_timeout": 30, }, "resources": {"resources": [{"name": "slots", "type": "numeric"}, {"name": "mem", "type": "numeric"}]}, "queues": {"queues": [{"name": "xce", "priority": 30, "users": {"members": ["all"]}}]}, "users": {"usergroups": [{"groupname": "ug1", "members": ["user1", "user2"]}]}, "hosts": { "hosts": [{"hostname": "h1", "max_slot": 4}], "hostgroups": [{"groupname": "hg1", "members": ["all"]}], }, "scaler": {"scalers": [{"name": "aws-asg", "type": "script", "script_path": "/x.sh"}]}, "limits": {"limits": [{"name": "l1", "resname": "slots", "value": 10}]}, } def test_map_all_sections_building_same_input_twice_produces_identical_bytes(): sections = _build_full_sections() target1 = mapping.new_cluster_config_builder() mapping.map_all_sections(sections, target1) bytes1 = target1.to_bytes() target2 = mapping.new_cluster_config_builder() mapping.map_all_sections(sections, target2) bytes2 = target2.to_bytes() assert bytes1 == bytes2 def test_map_all_sections_missing_file_matches_present_empty_section(): """A schema_name absent from `sections` (missing YAML file) must produce byte-identical output to that section being present but empty -- Requirement 1.5's rule, enforced structurally. """ sections_missing = { "xce": {"cluster_name": "c", "heartbeat_interval": 1, "inactive_heartbeat_count": 1} } sections_present_empty = dict(sections_missing, resources={"resources": []}) target_missing = mapping.new_cluster_config_builder() mapping.map_all_sections(sections_missing, target_missing) target_empty = mapping.new_cluster_config_builder() mapping.map_all_sections(sections_present_empty, target_empty) assert target_missing.to_bytes() == target_empty.to_bytes() def test_map_all_sections_populates_all_capnp_fields_and_skips_configgeneration(): sections = _build_full_sections() target = mapping.new_cluster_config_builder() mapping.map_all_sections(sections, target) assert target.clusterName == "eda1" assert len(target.resources) == 2 assert len(target.queues) == 1 assert len(target.ugroups) == 1 assert len(target.hgroups) == 1 assert len(target.hostConfigs) == 1 assert len(target.hproviders) == 1 assert len(target.limits) == 1 # configGeneration (@12): controller-assigned, never touched (1.9). assert target.configGeneration == 0 def test_no_timestamp_hostname_or_username_source_anywhere_in_module(): """Static regression guard for Requirement 9.2: no wall-clock, hostname, or username source is ever imported or called by this module. """ source = inspect.getsource(mapping) forbidden_tokens = [ "datetime", "time.time", "gethostname", "getlogin", "getuser", "uuid.uuid", "os.environ", ] for token in forbidden_tokens: assert token not in source, f"forbidden non-deterministic source {token!r} found in mapping.py" # --------------------------------------------------------------------------- # Real src/configs/*.yaml fixtures # --------------------------------------------------------------------------- @pytest.mark.skipif(not REPO_CONFIGS_DIR.is_dir(), reason="src/configs/ fixtures not present") def test_maps_real_repo_config_fixtures_without_error(): sections = { "xce": _load_real_yaml("xce.yaml"), "resources": _load_real_yaml("resources.yaml"), "queues": _load_real_yaml("queues.yaml"), "users": _load_real_yaml("users.yaml"), "hosts": _load_real_yaml("hosts.yaml"), "scaler": _load_real_yaml("scaler.yaml"), "limits": _load_real_yaml("limits.yaml"), } target = mapping.new_cluster_config_builder() mapping.map_all_sections(sections, target) assert target.clusterName == "eda1" assert len(target.resources) == len(sections["resources"]["resources"]) assert len(target.queues) == len(sections["queues"]["queues"]) assert len(target.hostConfigs) == len(sections["hosts"]["hosts"]) assert len(target.hgroups) == len(sections["hosts"]["hostgroups"]) assert len(target.limits) == len(sections["limits"]["limits"]) # Round-trip through to_bytes()/from_bytes() using the real loaded # schema module -- catches drift between mapping.py's hardcoded field # names and the real controller.capnp schema. mod = load_schema() target_bytes = target.to_bytes() with mod.ClusterConfig.from_bytes(target_bytes) as reader: assert reader.clusterName == "eda1" assert reader.queues[3].name == "short" # Determinism against the real fixtures too. target2 = mapping.new_cluster_config_builder() mapping.map_all_sections(sections, target2) assert target_bytes == target2.to_bytes()