/* * test_driver_translate.c - contract test for the normal-mode driver's * controller event translation path (scheduler-core spec, task 5.1, Req 5 * AC 8 / AC 9's translation half). * * Observable under test: a sequence of synthetic xce_sched_update_request_t * records enqueued into a real dataq (the same pend_updates mechanism * handler.c already populates from capnp push/notify, unchanged by this * task), once drained by xce_sched_driver_translate_pending_events(), * produces an xce_core_event_t batch whose contents match the input * records one-to-one (same count, same types, same order, same field * values) -- and xce_core_push_events() genuinely gets called with that * batch, proven via its REAL, observable side effects (core state * mutation), not a mocked stand-in. * * White-box #include of ../src/core.c, ../src/core_clock.c (needed for * xce_core_now(), which driver.c calls to stamp each translated batch), * ../src/core_events.c, and ../src/driver.c (same convention as * test_core_events.c) so this test * exercises the REAL xce_sched_core_init()/xce_core_push_events()/ * xce_sched_driver_translate_pending_events(), and can additionally call * driver.c's static translate_one_pending_update() directly for tight, * per-op-type field assertions on the translation step in isolation from * the batch/push machinery around it. * * No picker plugin is registered in this file's fixture (unlike * test_core_events.c's g_full_plugin): core->job_pick_plugin stays NULL, * so sched_resolve_job_picker() (this file's own real reimplementation, * copied from test_core_events.c's identical rationale -- schedobj.c is not * pulled in) returns -EINVAL, xce_core_push_events() WARNs and proceeds * with picker=NULL. Every case in core_events.c already NULL-guards its * picker hook call, so this is safe -- and this file's Observable is core * STATE mutation (proving the real push happened), not hook-firing, which * task 4.3's own test suite already covers exhaustively. * * Stubs below (kill_sched_job/kill_sched_queue/kill_sched_worker/ * kill_sched_user/kill_sched_project/kill_sched_resval/plugin_unregister) * are pure no-ops, identical to test_core_events.c's own -- fixtures in * this file are never torn down (this directory's established convention), * so these killers are never actually invoked; present only to satisfy the * linker for xce_sched_core_init()'s kvmgr_params_t.datakiller fields. */ #include #include #include #include #include #include "controller.capnp.h" #include "kvmgr.h" #include "scheduler.h" #include "xstring.h" /* Pull in the real xce_sched_core_init()/xce_core_push_events()/ * xce_sched_driver_translate_pending_events() under test into this TU. */ #include "../src/core.c" #include "../src/core_clock.c" #include "../src/core_events.c" #include "../src/driver.c" void kill_sched_job(void* arg) { (void)arg; } void kill_sched_queue(void* arg) { (void)arg; } void kill_sched_worker(void* arg) { (void)arg; } void kill_sched_user(void* arg) { (void)arg; } void kill_sched_project(void* arg) { (void)arg; } void kill_sched_resval(xce_val_t* val) { (void)val; } void plugin_unregister(void* arg) { (void)arg; } /* Link-satisfying stubs for scheduler.c's sched_publish_dispatch()/ * sched_publish_pend_reason() -- as of scheduler-core spec task 5.2, * driver.c's (real, #included above) xce_sched_driver_on_decision() calls * both, but this file never exercises that function (it is out of THIS * file's scope -- see test_driver_decision_cb.c) and never links * scheduler.c (same rationale as the sched_resolve_job_picker * reimplementation below: this file's own fixtures never need a real * xcecli_ctx_t/apictx). Never actually invoked at runtime by any test in * this file. */ int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* d) { (void)sched; (void)d; return 0; } int sched_publish_pend_reason(xce_scheduler_t* sched, xce_preason_set_t* r) { (void)sched; (void)r; return 0; } /* Link-satisfying stubs: as of scheduler-core spec task 5.2, * xce_sched_driver_run_normal_mode_round() (driver.c, real, #included * above, alongside xce_sched_driver_translate_pending_events() this file * DOES exercise) references xce_core_begin_session()/xce_core_signal_go()/ * xce_core_end_session() (core_session.c) and plugin_begin_session()/ * plugin_end_session() (plugin.c) / sched_account_jobs() (scheduler.c). * This file never exercises xce_sched_driver_run_normal_mode_round() itself * (out of THIS file's scope -- see test_driver_session_ordering.c) and * never links core_session.c/plugin.c/scheduler.c. Never actually invoked * at runtime by any test in this file. */ int xce_core_begin_session(xce_sched_core_t* core) { (void)core; return 0; } int xce_core_signal_go(xce_sched_core_t* core) { (void)core; return 0; } int xce_core_end_session(xce_sched_core_t* core) { (void)core; return 0; } int plugin_begin_session(void* arg) { (void)arg; return 0; } int plugin_end_session(void* arg) { (void)arg; return 0; } int sched_account_jobs(xce_scheduler_t* sched) { (void)sched; return 0; } /* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see * file header comment (same rationale as test_core_events.c's own copy). */ int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out) { xce_sched_plugin_t* plugin = NULL; int rc = 0; if ((core == NULL) || (out == NULL)) { return -EINVAL; } *out = NULL; if (core->job_pick_plugin == NULL) { return -EINVAL; } lock(core->plugins, RTTI_LOCK_READ); rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin); unlock(core->plugins); if (rc <= 0) { return (rc < 0) ? rc : -ENOENT; } if ((plugin == NULL) || (plugin->picker_iface == NULL) || (plugin->picker_iface->next_job == NULL) || (plugin->picker_iface->begin_session == NULL) || (plugin->picker_iface->end_session == NULL) || (plugin->picker_iface->on_dispatch == NULL) || (plugin->picker_iface->on_pend == NULL)) { return -EINVAL; } *out = plugin; return 0; } void setUp(void) { return; } void tearDown(void) { return; } /* ---------------------------------------------------------------------- */ /* Fixture: real xce_sched_core_init() + manually dataq_init'd */ /* pend_updates (core.c's own header comment: dataq_init for pend_updates */ /* stays driver/scheduler.c-owned, not xce_sched_core_init()'s job) -- */ /* same pattern test_handlers.c's fixture_init() uses. Never torn down. */ /* ---------------------------------------------------------------------- */ static void fixture_init(xce_sched_core_t* core) { int rc; rc = xce_sched_core_init(core); TEST_ASSERT_EQUAL_INT(0, rc); core->pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t)); dataq_init(core->pend_updates[0], NULL); core->pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t)); dataq_init(core->pend_updates[1], NULL); } static void enqueue(xce_sched_core_t* core, xce_sched_update_request_t* req) { dataq_enq(core->pend_updates[0], &(req->link)); } static xce_sched_update_request_t* make_req(xce_sched_op_t op) { xce_sched_update_request_t* req = (xce_sched_update_request_t*)calloc(1, sizeof(*req)); req->op = op; return req; } /* Every string field here is strdup()'d, and the wrapper structs below are * all individually calloc()'d (never a stack local or a shared array) -- * matching the real shape release_translated_payload()'s free_Job_ptr() etc. * (capnp-generated, controller.capnp.c) expect to free: every free_X_ptr() * in this codebase does `free(field); ...; free(*d);` unconditionally on * every non-NULL string/wrapper it owns. A string literal or stack address * here would crash the moment the real cleanup path (this task's own new * code, exercised for real by xce_sched_driver_translate_pending_events()) * tries to free it -- this is what a genuinely capnp-decoded xce_job_t * looks like in production, not a test shortcut. */ static xce_job_t* make_wire_job(uint64_t jobid, uint32_t state, const char* queue, const char* user, const char* project) { xce_job_t* j = (xce_job_t*)calloc(1, sizeof(*j)); j->jobid = jobid; j->state = state; j->sub_time = 1000; j->queue_time = 1000; j->submit = (job_submit_data_t*)calloc(1, sizeof(*(j->submit))); j->submit->queue = strdup(queue); j->submit->user = strdup(user); j->submit->project_name = strdup(project); j->submit->resreq = strdup("type == any"); j->submit->mintasks = 1; j->submit->maxtasks = 1; return j; } /* ---------------------------------------------------------------------- */ /* Per-op unit tests of translate_one_pending_update() (white-box static */ /* call) -- one per op type, asserting event type + every field matches. */ /* ---------------------------------------------------------------------- */ static void test_translate_config_push(void) { xce_sched_core_t core; xce_conf4sched_t config; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_CONFIG); xce_core_event_t ev; int rc; fixture_init(&core); memset(&config, 0, sizeof(config)); req->u.config = &config; rc = translate_one_pending_update(&core, req, 12345, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_CONFIG_PUSH, ev.type); TEST_ASSERT_EQUAL_UINT64(12345, ev.ts); TEST_ASSERT_EQUAL_PTR(&config, ev.u.config_push.config); } static void test_translate_jobs_push(void) { xce_sched_core_t core; xce_job_t* job1 = make_wire_job(1, JOB_STAT_PEND, "normal", "a", "p"); xce_job_t* jobs[1]; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB); xce_core_event_t ev; int rc; fixture_init(&core); jobs[0] = job1; req->u.jobupdate.njobs = 1; req->u.jobupdate.jobs = jobs; rc = translate_one_pending_update(&core, req, 99, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOBS_PUSH, ev.type); TEST_ASSERT_EQUAL_UINT64(99, ev.ts); TEST_ASSERT_EQUAL_INT(1, ev.u.jobs_push.njobs); TEST_ASSERT_EQUAL_PTR(jobs, ev.u.jobs_push.jobs); TEST_ASSERT_EQUAL_PTR(job1, ev.u.jobs_push.jobs[0]); } static void test_translate_job_fields_update(void) { xce_sched_core_t core; xce_job_t* job = make_wire_job(7, JOB_STAT_RUN, "normal", "a", "p"); xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_FIELDS); xce_core_event_t ev; int rc; fixture_init(&core); req->u.jobfieldsupdate.job = job; rc = translate_one_pending_update(&core, req, 5, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_UPDATE, ev.type); TEST_ASSERT_EQUAL_UINT64(7, ev.u.job_update.jobid); TEST_ASSERT_EQUAL_PTR(job, ev.u.job_update.job); TEST_ASSERT_EQUAL_UINT32(0, ev.u.job_update.changed_fields); } static void test_translate_job_fields_update_null_job_skips(void) { xce_sched_core_t core; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_JOB_FIELDS); xce_core_event_t ev; int rc; fixture_init(&core); req->u.jobfieldsupdate.job = NULL; rc = translate_one_pending_update(&core, req, 5, &ev); TEST_ASSERT_EQUAL_INT(0, rc); } static void test_translate_workers_push(void) { xce_sched_core_t core; xce_worker_t wire_worker; xce_worker_t* workers[1]; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_WORKER); xce_core_event_t ev; int rc; fixture_init(&core); memset(&wire_worker, 0, sizeof(wire_worker)); wire_worker.name = "hostA"; workers[0] = &wire_worker; req->u.workerupdate.nworkers = 1; req->u.workerupdate.workers = workers; rc = translate_one_pending_update(&core, req, 3, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_WORKERS_PUSH, ev.type); TEST_ASSERT_EQUAL_INT(1, ev.u.workers_push.nworkers); TEST_ASSERT_EQUAL_PTR(workers, ev.u.workers_push.workers); } static void test_translate_jstate_change_looks_up_old_state(void) { xce_sched_core_t core; xce_sched_job_t existing_job; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JSTATE_CHANGE); xce_core_event_t ev; int rc; fixture_init(&core); memset(&existing_job, 0, sizeof(existing_job)); existing_job.jobid = 42; existing_job.state = JOB_STAT_PEND; sched_add_job(&core, &existing_job); req->u.job_state_change.jobid = 42; req->u.job_state_change.state = JOB_STAT_RUN; rc = translate_one_pending_update(&core, req, 8, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_STATE_CHANGE, ev.type); TEST_ASSERT_EQUAL_UINT64(42, ev.u.job_state_change.jobid); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, ev.u.job_state_change.old_state); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, ev.u.job_state_change.new_state); TEST_ASSERT_EQUAL_INT32(0, ev.u.job_state_change.exit_code); } static void test_translate_jstate_change_unknown_job_old_state_zero(void) { xce_sched_core_t core; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JSTATE_CHANGE); xce_core_event_t ev; int rc; fixture_init(&core); req->u.job_state_change.jobid = 999; req->u.job_state_change.state = JOB_STAT_EXIT; rc = translate_one_pending_update(&core, req, 8, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_UINT32(0, ev.u.job_state_change.old_state); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, ev.u.job_state_change.new_state); } static void test_translate_queue_update(void) { xce_sched_core_t core; xce_queue_t wire_queue; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_QUEUE); xce_core_event_t ev; int rc; fixture_init(&core); memset(&wire_queue, 0, sizeof(wire_queue)); wire_queue.queue = "batch"; req->u.queueupdate.queue = &wire_queue; rc = translate_one_pending_update(&core, req, 1, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_QUEUE_UPDATE, ev.type); TEST_ASSERT_EQUAL_PTR(&wire_queue, ev.u.queue_update.queue); } static void test_translate_resources_push(void) { xce_sched_core_t core; xce_resource_t wire_resource; xce_resource_t* resources[1]; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_RESOURCES); xce_core_event_t ev; int rc; fixture_init(&core); memset(&wire_resource, 0, sizeof(wire_resource)); wire_resource.name = "cpu"; resources[0] = &wire_resource; req->u.resourceupdate.nresources = 1; req->u.resourceupdate.resources = resources; rc = translate_one_pending_update(&core, req, 2, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_RESOURCES_PUSH, ev.type); TEST_ASSERT_EQUAL_INT(1, ev.u.resources_push.n); TEST_ASSERT_EQUAL_PTR(resources, ev.u.resources_push.resources); } static void test_translate_resvals_full(void) { xce_sched_core_t core; xce_val_t val; xce_resval_t wire_resval; xce_resval_t* vals[1]; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_RESVALS_FULL); xce_core_event_t ev; int rc; fixture_init(&core); memset(&val, 0, sizeof(val)); memset(&wire_resval, 0, sizeof(wire_resval)); wire_resval.name = "cpu"; wire_resval.val = &val; vals[0] = &wire_resval; req->u.resvalfullupdate.nvals = 1; req->u.resvalfullupdate.vals = vals; rc = translate_one_pending_update(&core, req, 6, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_RESVALS_FULL, ev.type); TEST_ASSERT_EQUAL_INT(1, ev.u.resvals_full.n); TEST_ASSERT_EQUAL_PTR(vals, ev.u.resvals_full.resvals); } static void test_translate_resvals_delta(void) { xce_sched_core_t core; xce_resval_list_t deltas; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_RESVALS); xce_core_event_t ev; int rc; fixture_init(&core); memset(&deltas, 0, sizeof(deltas)); req->u.resvalupdate.resvals = &deltas; rc = translate_one_pending_update(&core, req, 4, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_RESVALS_DELTA, ev.type); TEST_ASSERT_EQUAL_PTR(&deltas, ev.u.resvals_delta.deltas); } static void test_translate_job_rusage(void) { xce_sched_core_t core; xce_job_rusages_t rusages; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JOB_RUSAGE); xce_core_event_t ev; int rc; fixture_init(&core); memset(&rusages, 0, sizeof(rusages)); rusages.jobid = 55; req->u.job_rusage.rusages = &rusages; rc = translate_one_pending_update(&core, req, 9, &ev); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_EQUAL_INT(XCE_CORE_EVENT_JOB_RUSAGE, ev.type); TEST_ASSERT_EQUAL_UINT64(55, ev.u.job_rusage.jobid); TEST_ASSERT_EQUAL_PTR(&rusages, ev.u.job_rusage.rusages); } static void test_translate_job_rusage_null_rusages_skips(void) { xce_sched_core_t core; xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_JOB_RUSAGE); xce_core_event_t ev; int rc; fixture_init(&core); req->u.job_rusage.rusages = NULL; rc = translate_one_pending_update(&core, req, 9, &ev); TEST_ASSERT_EQUAL_INT(0, rc); } static void test_translate_unknown_op_skips(void) { xce_sched_core_t core; xce_sched_update_request_t* req = make_req((xce_sched_op_t)999); xce_core_event_t ev; int rc; fixture_init(&core); rc = translate_one_pending_update(&core, req, 9, &ev); TEST_ASSERT_EQUAL_INT(0, rc); } /* ---------------------------------------------------------------------- */ /* End-to-end: enqueue all 10 op types into a real dataq, drain via the */ /* public entry point, assert real core-state mutation for every one -- */ /* proves xce_core_push_events() genuinely ran with the translated batch. */ /* ---------------------------------------------------------------------- */ static void test_drain_translates_and_pushes_all_ten_op_types(void) { xce_sched_core_t core; xce_conf4sched_t* config; xce_job_t** jobs_batch; xce_job_t* job1; xce_job_t* job_fields_payload; xce_worker_t* wire_worker; xce_worker_t** workers; xce_queue_t* wire_queue; xce_resource_t* wire_resource; xce_resource_t** resources; xce_val_t* full_val; xce_resval_t* full_resval; xce_resval_t** full_vals; xce_val_t* delta_val; xce_resval_t* delta_resval; xce_resval_t** delta_vals; xce_resval_list_t* deltas; xce_job_rusages_t rusages; int rc; fixture_init(&core); /* Every payload below is individually calloc()'d (wrapper structs, * arrays, and strings via strdup()) -- see make_wire_job()'s own * comment for why: release_translated_payload() (driver.c) genuinely * calls the real capnp-generated free_X_ptr() on each of these after * translation, exactly mirroring scheduler.c's sched_pend_update_killer() * in production. A stack local or string literal here would crash the * moment that real cleanup path runs -- this is what genuinely * capnp-decoded wire payloads look like, not a test shortcut. */ /* 1. UPDATE_JOB -> JOBS_PUSH: creates job 42 (needed by later ops). */ job1 = make_wire_job(42, JOB_STAT_PEND, "normal", "alice", "proj1"); jobs_batch = (xce_job_t**)calloc(1, sizeof(*jobs_batch)); jobs_batch[0] = job1; xce_sched_update_request_t* r1 = make_req(XCE_SCHED_OP_UPDATE_JOB); r1->u.jobupdate.njobs = 1; r1->u.jobupdate.jobs = jobs_batch; enqueue(&core, r1); /* 2. UPDATE_WORKER -> WORKERS_PUSH */ wire_worker = (xce_worker_t*)calloc(1, sizeof(*wire_worker)); wire_worker->name = strdup("hostA"); workers = (xce_worker_t**)calloc(1, sizeof(*workers)); workers[0] = wire_worker; xce_sched_update_request_t* r2 = make_req(XCE_SCHED_OP_UPDATE_WORKER); r2->u.workerupdate.nworkers = 1; r2->u.workerupdate.workers = workers; enqueue(&core, r2); /* 3. UPDATE_RESOURCES -> RESOURCES_PUSH: registers "cpu" (needed by 4). * makeResource() (resobj.c) deep-copies name/desc via makeString() -- * verified by reading resobj_init()'s source -- so freeing * wire_resource afterward never dangles the resmgr's own copy. */ wire_resource = (xce_resource_t*)calloc(1, sizeof(*wire_resource)); wire_resource->name = strdup("cpu"); resources = (xce_resource_t**)calloc(1, sizeof(*resources)); resources[0] = wire_resource; xce_sched_update_request_t* r3 = make_req(XCE_SCHED_OP_UPDATE_RESOURCES); r3->u.resourceupdate.nresources = 1; r3->u.resourceupdate.resources = resources; enqueue(&core, r3); /* 4. UPDATE_RESVALS_FULL -> RESVALS_FULL. NOTE: core_events.c's * RESVALS_FULL case does `zarray_put(&core->sharedInst, res->idx, * rv->val)` -- a SHALLOW store of the SAME `val` pointer (verified by * reading zarray_put_ex()'s source, src/lib/zarray.c) -- and this * function's cleanup then frees `full_resval` (and, via * free_ResourceVal->free_Value_ptr, `full_val` too), exactly mirroring * scheduler.c's existing sched_pend_update_killer()'s identical * RESVALS_FULL cleanup. This is a PRE-EXISTING production ownership * characteristic (the same hazard already exists in sched_update()'s * own live path today, untouched by this task) -- so after the drain * below, core->sharedInst's stored pointer for "cpu" is only safe to * check for non-NULL (the slot was populated), never to dereference. */ full_val = (xce_val_t*)calloc(1, sizeof(*full_val)); full_val->u.ivalue32 = 16; full_resval = (xce_resval_t*)calloc(1, sizeof(*full_resval)); full_resval->name = strdup("cpu"); full_resval->val = full_val; full_vals = (xce_resval_t**)calloc(1, sizeof(*full_vals)); full_vals[0] = full_resval; xce_sched_update_request_t* r4 = make_req(XCE_SCHED_OP_UPDATE_RESVALS_FULL); r4->u.resvalfullupdate.nvals = 1; r4->u.resvalfullupdate.vals = full_vals; enqueue(&core, r4); /* 5. UPDATE_RESVALS -> RESVALS_DELTA: attributes to worker "hostA". * Same shallow-store-then-free hazard as 4 above (per-worker resvals * zarray instead of core->sharedInst) -- same "non-NULL only" caveat. */ delta_val = (xce_val_t*)calloc(1, sizeof(*delta_val)); delta_val->u.ivalue32 = 4; delta_resval = (xce_resval_t*)calloc(1, sizeof(*delta_resval)); delta_resval->name = strdup("cpu"); delta_resval->host = strdup("hostA"); delta_resval->val = delta_val; delta_vals = (xce_resval_t**)calloc(1, sizeof(*delta_vals)); delta_vals[0] = delta_resval; deltas = (xce_resval_list_t*)calloc(1, sizeof(*deltas)); deltas->count = 1; deltas->vals = delta_vals; xce_sched_update_request_t* r5 = make_req(XCE_SCHED_OP_UPDATE_RESVALS); r5->u.resvalupdate.resvals = deltas; enqueue(&core, r5); /* 6. UPDATE_QUEUE -> QUEUE_UPDATE. core_build_queue() deep-copies the * queue name via makeString() and priority is a plain int copy, so * freeing wire_queue afterward never dangles the core's own queue * object -- safe to assert on post-drain. */ wire_queue = (xce_queue_t*)calloc(1, sizeof(*wire_queue)); wire_queue->queue = strdup("batch"); wire_queue->priority = 9; wire_queue->status = QUEUE_STAT_OPEN; xce_sched_update_request_t* r6 = make_req(XCE_SCHED_OP_UPDATE_QUEUE); r6->u.queueupdate.queue = wire_queue; enqueue(&core, r6); /* 7. UPDATE_JOB_FIELDS -> JOB_UPDATE: mutates job 42's resreq. */ job_fields_payload = make_wire_job(42, JOB_STAT_PEND, "normal", "alice", "proj1"); xce_sched_update_request_t* r7 = make_req(XCE_SCHED_OP_UPDATE_JOB_FIELDS); r7->u.jobfieldsupdate.job = job_fields_payload; enqueue(&core, r7); /* 8. JSTATE_CHANGE -> JOB_STATE_CHANGE: job 42 PEND -> RUN. */ xce_sched_update_request_t* r8 = make_req(XCE_SCHED_OP_JSTATE_CHANGE); r8->u.job_state_change.jobid = 42; r8->u.job_state_change.state = JOB_STAT_RUN; enqueue(&core, r8); /* 9. JOB_RUSAGE -> JOB_RUSAGE: job 42 (already pushed by #1, in the same * batch, in FIFO order) adopts `rusages` by pointer * (core_events.c: `job->rusages = rusages`) -- release_translated_payload() * detects the adoption via job->rusages == rusages and does NOT free it, * so a stack local is safe here (unlike every payload above). */ memset(&rusages, 0, sizeof(rusages)); rusages.jobid = 42; xce_sched_update_request_t* r9 = make_req(XCE_SCHED_OP_JOB_RUSAGE); r9->u.job_rusage.rusages = &rusages; enqueue(&core, r9); /* 10. UPDATE_CONFIG -> CONFIG_PUSH (dead in the live enqueue path today * -- handler.c never creates this op -- but the translation must still * be real and correct per this task's scope). core->config adopts this * pointer permanently on a successful push -- never freed -- so it must * outlive this function's return; heap-allocated for that reason. */ config = (xce_conf4sched_t*)calloc(1, sizeof(*config)); xce_sched_update_request_t* r10 = make_req(XCE_SCHED_OP_UPDATE_CONFIG); r10->u.config = config; enqueue(&core, r10); rc = xce_sched_driver_translate_pending_events(&core); TEST_ASSERT_EQUAL_INT(10, rc); /* Real, observable core-state mutation proves xce_core_push_events() * genuinely ran with the translated batch -- not a mocked call. */ TEST_ASSERT_NOT_NULL(sched_get_job(&core, 42)); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, sched_get_job(&core, 42)->state); TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 42)->rusages); TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostA")); { xce_resource_t* found = NULL; int frc = key_find(core.resmgr, (void*)"cpu", (void**)&found); TEST_ASSERT_GREATER_THAN_INT(0, frc); TEST_ASSERT_NOT_NULL(found); /* Slot-populated check only -- see comment on payload #4/#5 above * for why the stored pointers are not safe to dereference here. */ TEST_ASSERT_NOT_NULL(zarray_get(&(core.sharedInst), found->idx)); { xce_sched_worker_t* w = sched_get_worker(&core, "hostA"); TEST_ASSERT_NOT_NULL(zarray_get(&(w->resvals), found->idx)); } } { xce_sched_queue_t* q = sched_get_queue(&core, "batch"); TEST_ASSERT_NOT_NULL(q); TEST_ASSERT_EQUAL_INT(9, q->priority); } TEST_ASSERT_EQUAL_PTR(config, core.config); } /* ---------------------------------------------------------------------- */ /* Buffering discipline (Req 5 AC 8): events accumulate between drains; */ /* an empty drain is a valid no-op returning 0, not an error. */ /* ---------------------------------------------------------------------- */ static void test_accumulates_between_drains_and_empty_drain_is_noop(void) { xce_sched_core_t core; int rc; fixture_init(&core); /* Nothing pending yet. */ rc = xce_sched_driver_translate_pending_events(&core); TEST_ASSERT_EQUAL_INT(0, rc); /* One record enqueued -- drained on the next call. Heap-allocated (see * make_wire_job()'s comment above): the real drain call below genuinely * frees this payload via free_Worker_ptr(). */ { xce_worker_t* wire_worker = (xce_worker_t*)calloc(1, sizeof(*wire_worker)); xce_worker_t** workers = (xce_worker_t**)calloc(1, sizeof(*workers)); xce_sched_update_request_t* req = make_req(XCE_SCHED_OP_UPDATE_WORKER); wire_worker->name = strdup("hostB"); workers[0] = wire_worker; req->u.workerupdate.nworkers = 1; req->u.workerupdate.workers = workers; enqueue(&core, req); rc = xce_sched_driver_translate_pending_events(&core); TEST_ASSERT_EQUAL_INT(1, rc); TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostB")); } /* Nothing new pending -- must not re-process anything. */ rc = xce_sched_driver_translate_pending_events(&core); TEST_ASSERT_EQUAL_INT(0, rc); } /* ---------------------------------------------------------------------- */ /* Bad-argument guard clauses. */ /* ---------------------------------------------------------------------- */ static void test_null_core_returns_einval(void) { int rc = xce_sched_driver_translate_pending_events(NULL); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); } static void test_uninitialized_pend_updates_returns_einval(void) { xce_sched_core_t core; int rc; memset(&core, 0, sizeof(core)); rc = xce_sched_driver_translate_pending_events(&core); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); } int main(void) { UNITY_BEGIN(); RUN_TEST(test_translate_config_push); RUN_TEST(test_translate_jobs_push); RUN_TEST(test_translate_job_fields_update); RUN_TEST(test_translate_job_fields_update_null_job_skips); RUN_TEST(test_translate_workers_push); RUN_TEST(test_translate_jstate_change_looks_up_old_state); RUN_TEST(test_translate_jstate_change_unknown_job_old_state_zero); RUN_TEST(test_translate_queue_update); RUN_TEST(test_translate_resources_push); RUN_TEST(test_translate_resvals_full); RUN_TEST(test_translate_resvals_delta); RUN_TEST(test_translate_job_rusage); RUN_TEST(test_translate_job_rusage_null_rusages_skips); RUN_TEST(test_translate_unknown_op_skips); RUN_TEST(test_drain_translates_and_pushes_all_ten_op_types); RUN_TEST(test_accumulates_between_drains_and_empty_drain_is_noop); RUN_TEST(test_null_core_returns_einval); RUN_TEST(test_uninitialized_pend_updates_returns_einval); return UNITY_END(); }