myxce/src/scheduler/testsuite/test_driver_translate.c

/*
 * 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 <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#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; }

/* dup_val - real reimplementation of snapshot_wire.c's function of the same
 * name (verbatim logic), needed because core_events.c's RESVALS_FULL/
 * RESVALS_DELTA handlers now call it (bug-1585 fix) and this file's
 * white-box #include of ../src/core_events.c does not pull in
 * snapshot_wire.c (same "reimplement the one small function instead of the
 * whole file" rationale this file's own sched_resolve_job_picker copy
 * already documents). This file's own kill_sched_resval is a no-op stub, so
 * the deep copy's contents are never inspected here -- only link-time
 * presence matters for this file's own tests. */
xce_val_t* dup_val(const xce_val_t* src)
{
    xce_val_t* v;

    if (src == NULL) {
        return NULL;
    }

    v = (xce_val_t*)calloc(1, sizeof(*v));
    if (v == NULL) {
        return NULL;
    }

    v->which = src->which;
    if (src->which == XCE_VAL_STRING) {
        v->u.svalue
            = (src->u.svalue != NULL) ? makeString(src->u.svalue) : NULL;
    } else {
        v->u = src->u;
    }

    return v;
}
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 for recording.c/recording_events.c/
 * recording_session.c's real functions -- as of scheduler-core spec task
 * 9.2, driver.c's (real, #included above)
 * xce_sched_driver_translate_pending_events() calls sched_record_write_events()
 * unconditionally (behind a NULL `record_ctx` check, not a call this test ever
 * takes since every call site below passes NULL), and
 * xce_sched_driver_run_normal_mode_round() (out of THIS file's scope, same as
 * the core_session.c/plugin.c/ scheduler.c stubs above) calls the
 * sched_record_session_* trio. This file never links recording.c's real
 * implementations (they need xce_sched_core_save_snapshot(), snapshot.c, an
 * even bigger dependency surface this file's own header comment already avoids
 * pulling in for sched_publish_dispatch()/sched_publish_pend_reason()). Never
 * actually invoked at runtime by any test in this file. */
int sched_record_write_events(xce_scheduler_t* sched, xce_core_event_t* events,
                              int n)
{
    (void)sched;
    (void)events;
    (void)n;
    return 0;
}

/* sched_record_start/sched_record_stop (scheduler-core spec, task 9.4) --
 * driver.c's handle_deferred_admin_record() (new this task) calls both;
 * this file never enqueues an XCE_SCHED_OP_ADMIN_RECORD request, so
 * neither is actually invoked at runtime, same rationale as
 * sched_record_write_events above. */
int sched_record_start(xce_scheduler_t* sched, const char* record_file)
{
    (void)sched;
    (void)record_file;
    return 0;
}

int sched_record_stop(xce_scheduler_t* sched)
{
    (void)sched;
    return 0;
}

sched_record_session_ctx_t* sched_record_session_begin(xce_scheduler_t* sched,
                                                       uint64_t begin_round_ts)
{
    (void)sched;
    (void)begin_round_ts;
    return NULL;
}

void sched_record_session_add_decision(sched_record_session_ctx_t* ctx,
                                       uint64_t ts, xce_decision_t* decision)
{
    (void)ctx;
    (void)ts;
    (void)decision;
}

int sched_record_session_end(xce_scheduler_t*            sched,
                             sched_record_session_ctx_t* ctx, uint64_t end_ts)
{
    (void)sched;
    (void)ctx;
    (void)end_ts;
    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 (bug-1585, FIXED): this
     * comment used to document a PRE-EXISTING production ownership hazard --
     * core_events.c's RESVALS_FULL case used to do `zarray_put(&core->
     * sharedInst, res->idx, rv->val)`, a SHALLOW store of the SAME `val`
     * pointer this function's cleanup then frees via
     * free_ResourceVal->free_Value_ptr, exactly mirroring scheduler.c's
     * sched_pend_update_killer()'s identical (also now-fixed) RESVALS_FULL
     * cleanup -- meaning core->sharedInst's stored pointer for "cpu" used to
     * be a dangling pointer the moment this drain returned, unsafe to
     * dereference (only a non-NULL check was safe). core_events.c's
     * RESVALS_FULL/RESVALS_DELTA handlers now `dup_val()` before
     * `zarray_put()`, so the core owns an independent copy and this
     * function's unconditional free of `full_val`/`delta_val` below is
     * exactly what should happen -- core->sharedInst's stored value is now
     * genuinely safe to read back and assert on (see
     * test_resval_ownership.c for the dedicated regression test proving
     * this end-to-end under a REAL, non-stubbed kill_sched_resval). `which`
     * must be set explicitly here (XCE_VAL_NUMERIC) -- dup_val() now
     * actually inspects it, and a bare calloc() leaves it 0 (==
     * XCE_VAL_STRING), which would make dup_val() misread `u.ivalue32`'s
     * bit pattern as a `char*` and crash in makeString(). */
    full_val              = (xce_val_t*)calloc(1, sizeof(*full_val));
    full_val->which       = XCE_VAL_NUMERIC;
    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 fix as 4 above (per-worker resvals zarray instead of
     * core->sharedInst) -- `which` set explicitly for the same reason. */
    delta_val              = (xce_val_t*)calloc(1, sizeof(*delta_val));
    delta_val->which       = XCE_VAL_NUMERIC;
    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, NULL);

    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, NULL);
    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, NULL);
        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, NULL);
    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, 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, NULL);

    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();
}