myxce/src/scheduler/testsuite/test_core_events.c

/*
 * test_core_events.c - contract test for the core event interface and
 * batch event push (scheduler-core spec, task 4.3, Req 5 AC 4 / AC 8).
 *
 * Observable under test: pushing a batch of events with the picker idle
 * (core->session_active == 0) fires the corresponding state-change hook
 * exactly once per event, synchronously, before xce_core_push_events()
 * returns -- and the session-active rejection path (-EBUSY, no event
 * applied) when a session IS active.
 *
 * White-box #include of both ../src/core.c and ../src/core_events.c (same
 * convention as test_core_isolation.c / test_core_jbuf_capacity_pin.c) so
 * this test exercises the REAL xce_sched_core_init()/xce_core_push_events(),
 * not a re-implementation -- xce_core_push_events() lives in core_events.c
 * (split out of core.c purely to respect the 800-line-per-file cap; see
 * that file's own header comment), which itself calls straight back into
 * core.c's real sched_get_job()/sched_add_job()/etc. accessors, all within
 * this one translation unit. Stubs below are only for scheduler-internal
 * externs these two files reference (as kvmgr datakiller callback
 * addresses, or -- for sched_resolve_job_picker -- because the real
 * implementation lives in schedobj.c, a separate translation unit this
 * file does not pull in):
 *
 *   - kill_sched_job/kill_sched_queue/kill_sched_worker/kill_sched_user/
 *     kill_sched_project/kill_sched_resval/plugin_unregister: pure no-op
 *     stubs, identical to test_core_jbuf_capacity_pin.c's own -- this
 *     file's fixtures are never torn down (never call
 *     xce_sched_core_fini()), same "fixtures aren't torn down" convention
 *     established across this directory, so these killers are never
 *     actually invoked; present only to satisfy the linker for
 *     xce_sched_core_init()'s kvmgr_params_t.datakiller fields.
 *   - sched_resolve_job_picker: a real, working reimplementation of
 *     schedobj.c's function of the same name (identical algorithm --
 *     key_find core->job_pick_plugin in core->plugins, validate the
 *     picker MUST hooks) -- needed because the true implementation lives
 *     in schedobj.c, and pulling that whole file in would drag in its own
 *     large stub surface (plugin_queue_update, plugin_job_update, etc.)
 *     for functions this file's tests never exercise. This IS real logic,
 *     not a null stub, because the hook-firing contract under test
 *     depends on picker resolution actually working.
 *
 * make_job_account/free_job_account/free_JobAlloc_ptr/free_JobRusages_ptr/
 * makeResource/parse_resreq/makeString are all real, linked from the
 * shared libxce library (src/lib), not stubbed -- same as every other
 * scheduler white-box test in this directory.
 */
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "kvmgr.h"
#include "scheduler.h"
#include "xstring.h"

/* Pull in the real xce_sched_core_init()/xce_core_push_events() under test
 * into this TU (white-box #include, same convention as
 * test_core_jbuf_capacity_pin.c's `#include "../src/core.c"`). */
#include "../src/core.c"
#include "../src/core_events.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; }

/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- see
 * file header comment for why this file provides its own copy instead of
 * #include-ing schedobj.c. */
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; }

/* ---------------------------------------------------------------------- */
/* Test picker: records call counts + last-seen arguments per hook, so     */
/* tests can assert "exactly once, with the right argument" per event.     */
/* ---------------------------------------------------------------------- */

typedef struct {
    int                config_calls;
    int                job_push_calls;
    int                job_state_change_calls;
    int                job_update_calls;
    int                queue_update_calls;
    int                job_rusage_calls;
    xce_conf4sched_t*  last_config;
    xce_sched_job_t*   last_job_push;
    xce_sched_job_t*   last_job_state_change_job;
    uint32_t           last_job_state_change_state;
    xce_sched_job_t*   last_job_update;
    xce_sched_queue_t* last_queue_update;
    xce_sched_job_t*   last_job_rusage_job;
    xce_job_rusages_t* last_job_rusage_rusages;
} test_picker_counts_t;

static test_picker_counts_t g_counts;

static int test_begin_session(void* picker)
{
    (void)picker;
    return 0;
}
static xce_sched_job_t* test_next_job(void* picker)
{
    (void)picker;
    return NULL;
}
static int test_end_session(void* picker)
{
    (void)picker;
    return 0;
}
static int test_on_dispatch(void* picker, uint64_t jobid)
{
    (void)picker;
    (void)jobid;
    return 0;
}
static int test_on_pend(void* picker, uint64_t jobid)
{
    (void)picker;
    (void)jobid;
    return 0;
}

static int test_on_config(void* picker, xce_conf4sched_t* config)
{
    (void)picker;
    g_counts.config_calls++;
    g_counts.last_config = config;
    return 0;
}

static int test_on_job_push(void* picker, xce_sched_job_t* job)
{
    (void)picker;
    g_counts.job_push_calls++;
    g_counts.last_job_push = job;
    return 0;
}

static int test_on_job_state_change(void* picker, xce_sched_job_t* job,
                                    uint32_t state)
{
    (void)picker;
    g_counts.job_state_change_calls++;
    g_counts.last_job_state_change_job   = job;
    g_counts.last_job_state_change_state = state;
    return 0;
}

static int test_on_job_update(void* picker, xce_sched_job_t* job)
{
    (void)picker;
    g_counts.job_update_calls++;
    g_counts.last_job_update = job;
    return 0;
}

static int test_on_queue_update(void* picker, xce_sched_queue_t* queue)
{
    (void)picker;
    g_counts.queue_update_calls++;
    g_counts.last_queue_update = queue;
    return 0;
}

static int test_on_job_rusage(void* picker, xce_sched_job_t* job,
                              xce_job_rusages_t* rusages)
{
    (void)picker;
    g_counts.job_rusage_calls++;
    g_counts.last_job_rusage_job     = job;
    g_counts.last_job_rusage_rusages = rusages;
    return 0;
}

/* Full picker: every hook wired, including the two optional ones
 * (on_config/on_job_rusage). */
static xce_job_picker_iface_t g_full_picker_iface = {
    .init                = NULL,
    .fini                = NULL,
    .begin_session       = test_begin_session,
    .next_job            = test_next_job,
    .end_session         = test_end_session,
    .on_dispatch         = test_on_dispatch,
    .on_pend             = test_on_pend,
    .on_queue_update     = test_on_queue_update,
    .on_job_push         = test_on_job_push,
    .on_job_state_change = test_on_job_state_change,
    .on_job_update       = test_on_job_update,
    .on_config           = test_on_config,
    .on_job_rusage       = test_on_job_rusage,
};

/* Picker with on_job_rusage left NULL (Req 5 AC 4: "optional -- NULL if
 * not registered") -- proves JOB_RUSAGE applies state without erroring
 * when the hook is absent. */
static xce_job_picker_iface_t g_no_rusage_picker_iface = {
    .init                = NULL,
    .fini                = NULL,
    .begin_session       = test_begin_session,
    .next_job            = test_next_job,
    .end_session         = test_end_session,
    .on_dispatch         = test_on_dispatch,
    .on_pend             = test_on_pend,
    .on_queue_update     = test_on_queue_update,
    .on_job_push         = test_on_job_push,
    .on_job_state_change = test_on_job_state_change,
    .on_job_update       = test_on_job_update,
    .on_config           = test_on_config,
    .on_job_rusage       = NULL,
};

static xce_sched_plugin_t g_full_plugin
    = { .name = "test_full_picker", .picker_iface = &g_full_picker_iface };
static xce_sched_plugin_t g_no_rusage_plugin
    = { .name         = "test_no_rusage_picker",
        .picker_iface = &g_no_rusage_picker_iface };

/* ---------------------------------------------------------------------- */
/* Fixture: real xce_sched_core_init(), a test picker registered directly  */
/* into core.plugins (key_add -- the same mechanism plugin_register()      */
/* itself uses, schedobj.c not included so its wrapper is not available),  */
/* and core.job_pick_plugin pointed at it. Never torn down (this           */
/* directory's established convention -- see test_core_jbuf_capacity_pin.c */
/* header comment). */
/* ---------------------------------------------------------------------- */

static void fixture_init(xce_sched_core_t* core, xce_sched_plugin_t* plugin)
{
    int rc;

    memset(&g_counts, 0, sizeof(g_counts));

    rc = xce_sched_core_init(core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    lock(core->plugins, RTTI_LOCK_WRITE);
    rc = key_add(core->plugins, (void*)plugin->name, (void*)plugin);
    unlock(core->plugins);
    TEST_ASSERT_GREATER_OR_EQUAL(0, rc);

    core->job_pick_plugin = plugin->name;
}

static xce_job_t* make_job(uint64_t jobid, uint32_t state, char* queue,
                           char* user, 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->start_time     = 0;
    j->submit         = (job_submit_data_t*)calloc(1, sizeof(*(j->submit)));
    j->submit->queue        = queue;
    j->submit->user         = user;
    j->submit->project_name = project;
    j->submit->resreq       = "type == any";
    j->submit->mintasks     = 1;
    j->submit->maxtasks     = 1;

    return j;
}

/* ---------------------------------------------------------------------- */
/* Tests                                                                  */
/* ---------------------------------------------------------------------- */

/* One event of each of the 6 hook-having kinds in a single batch --
 * asserts each corresponding hook fired EXACTLY once, synchronously
 * (already true by the time push_events returns, in this single-threaded
 * test), with the right argument. */
static void
test_push_events_fires_each_hook_once_per_event_with_picker_idle(void)
{
    xce_sched_core_t core;
    xce_core_event_t events[6];
    xce_conf4sched_t config;
    xce_job_t*       job1;
    xce_queue_t      wire_queue;
    xce_job_t        job_update_payload;
    xce_job_rusages_t rusages;
    int              rc;

    fixture_init(&core, &g_full_plugin);

    memset(&config, 0, sizeof(config));
    job1 = make_job(42, JOB_STAT_PEND, "normal", "alice", "proj1");

    memset(&wire_queue, 0, sizeof(wire_queue));
    wire_queue.queue    = "normal";
    wire_queue.priority = 5;
    wire_queue.status   = QUEUE_STAT_OPEN;

    memset(events, 0, sizeof(events));

    events[0].type                    = XCE_CORE_EVENT_CONFIG_PUSH;
    events[0].u.config_push.config    = &config;

    events[1].type                 = XCE_CORE_EVENT_QUEUE_UPDATE;
    events[1].u.queue_update.queue = &wire_queue;

    events[2].type               = XCE_CORE_EVENT_JOBS_PUSH;
    events[2].u.jobs_push.njobs  = 1;
    events[2].u.jobs_push.jobs   = &job1;

    events[3].type                            = XCE_CORE_EVENT_JOB_STATE_CHANGE;
    events[3].u.job_state_change.jobid         = 42;
    events[3].u.job_state_change.old_state     = JOB_STAT_PEND;
    events[3].u.job_state_change.new_state     = JOB_STAT_RUN;
    events[3].u.job_state_change.exit_code     = 0;

    memset(&job_update_payload, 0, sizeof(job_update_payload));
    job_update_payload.jobid = 42;
    job_update_payload.state = JOB_STAT_RUN;

    events[4].type                       = XCE_CORE_EVENT_JOB_UPDATE;
    events[4].u.job_update.jobid         = 42;
    events[4].u.job_update.changed_fields = 0;
    events[4].u.job_update.job           = &job_update_payload;

    memset(&rusages, 0, sizeof(rusages));
    rusages.jobid = 42;

    events[5].type                  = XCE_CORE_EVENT_JOB_RUSAGE;
    events[5].u.job_rusage.jobid    = 42;
    events[5].u.job_rusage.rusages  = &rusages;

    rc = xce_core_push_events(&core, events, 6);

    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls);
    TEST_ASSERT_EQUAL_PTR(&config, g_counts.last_config);

    TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls);
    TEST_ASSERT_NOT_NULL(g_counts.last_queue_update);
    TEST_ASSERT_EQUAL_STRING("normal", g_counts.last_queue_update->name);

    TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls);
    TEST_ASSERT_NOT_NULL(g_counts.last_job_push);
    TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_push->jobid);

    TEST_ASSERT_EQUAL_INT(1, g_counts.job_state_change_calls);
    TEST_ASSERT_NOT_NULL(g_counts.last_job_state_change_job);
    TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_state_change_job->jobid);
    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, g_counts.last_job_state_change_state);
    /* state mutation actually applied on the core-owned job object */
    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, sched_get_job(&core, 42)->state);

    TEST_ASSERT_EQUAL_INT(1, g_counts.job_update_calls);
    TEST_ASSERT_NOT_NULL(g_counts.last_job_update);
    TEST_ASSERT_EQUAL_UINT64(42, g_counts.last_job_update->jobid);

    TEST_ASSERT_EQUAL_INT(1, g_counts.job_rusage_calls);
    TEST_ASSERT_NOT_NULL(g_counts.last_job_rusage_job);
    TEST_ASSERT_EQUAL_PTR(&rusages, g_counts.last_job_rusage_rusages);
    TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 42)->rusages);
}

/* JOBS_PUSH with njobs > 1: on_job_push must fire once PER JOB, not once
 * per event -- Req 5 AC 4's "on_job_push(picker, job) for each job in the
 * batch". */
static void test_jobs_push_fires_on_job_push_once_per_job(void)
{
    xce_sched_core_t core;
    xce_core_event_t event;
    xce_job_t*       jobs[3];
    int              rc;

    fixture_init(&core, &g_full_plugin);

    jobs[0] = make_job(1, JOB_STAT_PEND, "normal", "alice", "proj1");
    jobs[1] = make_job(2, JOB_STAT_PEND, "normal", "bob", "proj1");
    jobs[2] = make_job(3, JOB_STAT_PEND, "normal", "carol", "proj1");

    memset(&event, 0, sizeof(event));
    event.type               = XCE_CORE_EVENT_JOBS_PUSH;
    event.u.jobs_push.njobs  = 3;
    event.u.jobs_push.jobs   = jobs;

    rc = xce_core_push_events(&core, &event, 1);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(3, g_counts.job_push_calls);
    TEST_ASSERT_NOT_NULL(sched_get_job(&core, 1));
    TEST_ASSERT_NOT_NULL(sched_get_job(&core, 2));
    TEST_ASSERT_NOT_NULL(sched_get_job(&core, 3));
}

/* WORKERS_PUSH/RESOURCES_PUSH/RESVALS_FULL/RESVALS_DELTA are state-update
 * only -- Req 1 AC 2: "Hooks not defined on the picker interface (worker
 * updates, resource updates, resval updates) are not forwarded to the
 * picker." Every hook counter must stay at 0, while the underlying state
 * IS genuinely mutated (not a no-op placeholder). */
static void test_non_picker_events_mutate_state_but_never_call_hooks(void)
{
    xce_sched_core_t core;
    xce_core_event_t events[4];
    xce_worker_t     wire_worker;
    xce_worker_t*    worker_ptrs[1];
    xce_resource_t   wire_resource;
    xce_resource_t*  resource_ptrs[1];
    xce_resval_t     full_resval;
    xce_resval_t*    full_vals[1];
    xce_resval_t     delta_resval;
    xce_resval_t*    delta_vals[1];
    xce_resval_list_t deltas;
    xce_val_t        full_val;
    xce_val_t        delta_val;
    int              rc;

    fixture_init(&core, &g_full_plugin);

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name              = "hostA";
    wire_worker.status            = 0;
    wire_worker.nresvals_to_sched = 0;
    worker_ptrs[0]                = &wire_worker;

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name = "cpu";
    wire_resource.idx  = 0;
    resource_ptrs[0]   = &wire_resource;

    memset(&full_val, 0, sizeof(full_val));
    memset(&full_resval, 0, sizeof(full_resval));
    full_resval.name = "cpu";
    full_resval.val  = &full_val;
    full_vals[0]     = &full_resval;

    memset(&delta_val, 0, sizeof(delta_val));
    memset(&delta_resval, 0, sizeof(delta_resval));
    delta_resval.name = "cpu";
    delta_resval.host = "hostA";
    delta_resval.val  = &delta_val;
    delta_vals[0]     = &delta_resval;
    deltas.count      = 1;
    deltas.vals       = delta_vals;

    memset(events, 0, sizeof(events));

    events[0].type                     = XCE_CORE_EVENT_WORKERS_PUSH;
    events[0].u.workers_push.nworkers  = 1;
    events[0].u.workers_push.workers   = worker_ptrs;

    events[1].type                    = XCE_CORE_EVENT_RESOURCES_PUSH;
    events[1].u.resources_push.n         = 1;
    events[1].u.resources_push.resources = resource_ptrs;

    events[2].type                = XCE_CORE_EVENT_RESVALS_FULL;
    events[2].u.resvals_full.n    = 1;
    events[2].u.resvals_full.resvals = full_vals;

    events[3].type                  = XCE_CORE_EVENT_RESVALS_DELTA;
    events[3].u.resvals_delta.deltas = &deltas;

    /* worker must exist before RESVALS_DELTA can attribute a value to it */
    rc = xce_core_push_events(&core, events, 1); /* WORKERS_PUSH only, first */
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = xce_core_push_events(&core, &events[1], 3); /* RESOURCES/RESVALS_FULL/RESVALS_DELTA */
    TEST_ASSERT_EQUAL_INT(0, rc);

    TEST_ASSERT_EQUAL_INT(0, g_counts.config_calls);
    TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls);
    TEST_ASSERT_EQUAL_INT(0, g_counts.job_state_change_calls);
    TEST_ASSERT_EQUAL_INT(0, g_counts.job_update_calls);
    TEST_ASSERT_EQUAL_INT(0, g_counts.queue_update_calls);
    TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls);

    TEST_ASSERT_NOT_NULL(sched_get_worker(&core, "hostA"));
}

/*
 * bug-1568 regression: a worker connecting triggers multiple WORKERS_PUSH
 * events in quick succession (live-confirmed against a real deploy -- an
 * early push before the controller finishes writing a worker's real
 * capacity values, carrying stale/zero data, followed shortly after by a
 * push carrying the real values). Before this fix, WORKERS_PUSH always
 * built a brand-new xce_sched_worker_t and called sched_add_worker() (a
 * plain key_add(), add-only) unconditionally -- so the SECOND push for an
 * already-registered worker name silently failed -EEXIST and its resource
 * values were discarded, leaving core->workers[name] stuck at whatever the
 * FIRST push happened to carry, forever. Two WORKERS_PUSH events for the
 * SAME worker name, with DIFFERENT resource values, must both be applied
 * -- the worker's resvals end up at the SECOND push's values, not the
 * first, and no duplicate/orphaned worker object is created.
 */
static void test_workers_push_twice_updates_existing_worker_in_place(void)
{
    xce_sched_core_t core;
    xce_core_event_t event;
    xce_resource_t   wire_resource;
    xce_worker_t     wire_worker;
    xce_worker_t*    worker_ptrs[1];
    xce_val_t        val1;
    xce_name_val_t   nv1;
    xce_name_val_t*  nv1_ptrs[1];
    xce_val_t        val2;
    xce_name_val_t   nv2;
    xce_name_val_t*  nv2_ptrs[1];
    xce_sched_worker_t* w;
    xce_resource_t*     r = NULL;
    xce_val_t*           applied;
    int                  rc;

    fixture_init(&core, &g_full_plugin);

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name = "mem";
    TEST_ASSERT_NOT_NULL(makeResource(core.resmgr, &wire_resource));

    TEST_ASSERT_TRUE(key_find(core.resmgr, (void*)"mem", (void**)&r) >= 0);
    TEST_ASSERT_NOT_NULL(r);

    /* First push: worker "hostA" joins with mem=100.0. */
    memset(&val1, 0, sizeof(val1));
    val1.which   = Value_u_fvalue;
    val1.u.fvalue = 100.0;
    memset(&nv1, 0, sizeof(nv1));
    nv1.name     = "mem";
    nv1.val      = &val1;
    nv1_ptrs[0]  = &nv1;

    memset(&wire_worker, 0, sizeof(wire_worker));
    wire_worker.name              = "hostA";
    wire_worker.status            = 0;
    wire_worker.nresvals_to_sched = 1;
    wire_worker.resvals_to_sched  = nv1_ptrs;
    worker_ptrs[0]                = &wire_worker;

    memset(&event, 0, sizeof(event));
    event.type                    = XCE_CORE_EVENT_WORKERS_PUSH;
    event.u.workers_push.nworkers = 1;
    event.u.workers_push.workers  = worker_ptrs;

    rc = xce_core_push_events(&core, &event, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    w = sched_get_worker(&core, "hostA");
    TEST_ASSERT_NOT_NULL(w);
    applied = (xce_val_t*)zarray_get(&(w->resvals), r->idx);
    TEST_ASSERT_NOT_NULL(applied);
    TEST_ASSERT_EQUAL_FLOAT(100.0f, (float)applied->u.fvalue);

    /* Second push: SAME worker name, mem now really 200.0 (the "arrived
     * late with the real value" push bug-1568 traced live). */
    memset(&val2, 0, sizeof(val2));
    val2.which    = Value_u_fvalue;
    val2.u.fvalue = 200.0;
    memset(&nv2, 0, sizeof(nv2));
    nv2.name    = "mem";
    nv2.val     = &val2;
    nv2_ptrs[0] = &nv2;

    wire_worker.resvals_to_sched = nv2_ptrs;

    rc = xce_core_push_events(&core, &event, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Still exactly one "hostA" worker -- updated in place, not a second,
     * silently-discarded object leaving the first one stale. */
    w = sched_get_worker(&core, "hostA");
    TEST_ASSERT_NOT_NULL(w);
    applied = (xce_val_t*)zarray_get(&(w->resvals), r->idx);
    TEST_ASSERT_NOT_NULL(applied);
    TEST_ASSERT_EQUAL_FLOAT(200.0f, (float)applied->u.fvalue);
}

/* JOB_RUSAGE's on_job_rusage hook is OPTIONAL (Req 5 AC 4) -- a picker
 * without it must still apply the state mutation and must not error or
 * crash. */
static void test_job_rusage_applies_state_when_hook_is_null(void)
{
    xce_sched_core_t core;
    xce_core_event_t events[2];
    xce_job_t*       job1;
    xce_job_rusages_t rusages;
    int              rc;

    fixture_init(&core, &g_no_rusage_plugin);

    job1 = make_job(7, JOB_STAT_RUN, "normal", "alice", "proj1");

    memset(&rusages, 0, sizeof(rusages));
    rusages.jobid = 7;

    memset(events, 0, sizeof(events));
    events[0].type              = XCE_CORE_EVENT_JOBS_PUSH;
    events[0].u.jobs_push.njobs = 1;
    events[0].u.jobs_push.jobs  = &job1;

    events[1].type                 = XCE_CORE_EVENT_JOB_RUSAGE;
    events[1].u.job_rusage.jobid   = 7;
    events[1].u.job_rusage.rusages = &rusages;

    rc = xce_core_push_events(&core, events, 2);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(0, g_counts.job_rusage_calls);
    TEST_ASSERT_EQUAL_PTR(&rusages, sched_get_job(&core, 7)->rusages);
}

/* CONFIG_PUSH must decompose the pushed xce_conf4sched_t's embedded
 * queues[]/resources[]/resvals[] arrays into core state -- not just store
 * the config pointer and bump configver (Req 5 AC 4's CONFIG_PUSH row:
 * "All queues, resource thresholds, ... " is part of the payload, and
 * decomposing it is the core's job, mirroring scheduler.c's
 * sched_init_resources()/sched_init_sharedInst()/sched_init_queues(), all
 * three of which sched_init_config() calls today for this exact same
 * xce_conf4sched_t payload). Also asserts on_queue_update fires once per
 * embedded queue (design intent: a config push's queues behave the same as
 * individual queue updates, just batched). */
static void test_config_push_decomposes_queues_resources_and_resvals(void)
{
    xce_sched_core_t   core;
    xce_core_event_t   event;
    xce_conf4sched_t   config;
    xce_resource_t     wire_resource;
    xce_resource_t*    resources[1];
    xce_val_t          resval_value;
    xce_name_val_t     wire_resval;
    xce_name_val_t*    resvals[1];
    xce_queue_t        wire_queue;
    xce_queue_t*       queues[1];
    xce_resource_t*    found_resource = NULL;
    xce_sched_queue_t* found_queue;
    xce_val_t*         shared_val;
    int                rc;

    fixture_init(&core, &g_full_plugin);

    memset(&wire_resource, 0, sizeof(wire_resource));
    wire_resource.name = "cpu";
    resources[0]        = &wire_resource;

    memset(&resval_value, 0, sizeof(resval_value));
    resval_value.u.ivalue32 = 8;

    memset(&wire_resval, 0, sizeof(wire_resval));
    wire_resval.name = "cpu";
    wire_resval.val  = &resval_value;
    resvals[0]        = &wire_resval;

    memset(&wire_queue, 0, sizeof(wire_queue));
    wire_queue.queue    = "batch";
    wire_queue.priority = 7;
    wire_queue.status   = QUEUE_STAT_OPEN;
    queues[0]           = &wire_queue;

    memset(&config, 0, sizeof(config));
    config.n_resources = 1;
    config.resources   = resources;
    config.nresvals    = 1;
    config.resvals     = resvals;
    config.n_queues    = 1;
    config.queues      = queues;

    memset(&event, 0, sizeof(event));
    event.type                 = XCE_CORE_EVENT_CONFIG_PUSH;
    event.u.config_push.config = &config;

    rc = xce_core_push_events(&core, &event, 1);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_INT(1, g_counts.config_calls);

    /* resource decomposed into core->resmgr */
    rc = key_find(core.resmgr, (void*)"cpu", (void**)&found_resource);
    TEST_ASSERT_GREATER_THAN_INT(0, rc);
    TEST_ASSERT_NOT_NULL(found_resource);

    /* resval decomposed into core->sharedInst at the resource's idx */
    shared_val = (xce_val_t*)zarray_get(&(core.sharedInst), found_resource->idx);
    TEST_ASSERT_NOT_NULL(shared_val);
    TEST_ASSERT_EQUAL_INT(8, shared_val->u.ivalue32);

    /* queue decomposed into core->queues, and on_queue_update fired once */
    found_queue = sched_get_queue(&core, "batch");
    TEST_ASSERT_NOT_NULL(found_queue);
    TEST_ASSERT_EQUAL_INT(7, found_queue->priority);
    TEST_ASSERT_EQUAL_INT(1, g_counts.queue_update_calls);
    TEST_ASSERT_EQUAL_PTR(found_queue, g_counts.last_queue_update);
}

/* Session-active rejection (Req 5 AC 8 / design.md precondition):
 * core_push_events() is only valid between sessions. */
static void
test_push_events_rejects_with_ebusy_when_session_active_no_mutation(void)
{
    xce_sched_core_t core;
    xce_core_event_t event;
    xce_job_t*       job1;
    int              rc;

    fixture_init(&core, &g_full_plugin);

    atomic_store(&(core.session_active), 1);

    job1 = make_job(99, JOB_STAT_PEND, "normal", "alice", "proj1");

    memset(&event, 0, sizeof(event));
    event.type               = XCE_CORE_EVENT_JOBS_PUSH;
    event.u.jobs_push.njobs  = 1;
    event.u.jobs_push.jobs   = &job1;

    rc = xce_core_push_events(&core, &event, 1);

    TEST_ASSERT_EQUAL_INT(-EBUSY, rc);
    TEST_ASSERT_EQUAL_INT(0, g_counts.job_push_calls);
    TEST_ASSERT_NULL(sched_get_job(&core, 99));
}

/* Bad-argument guard clauses. */
static void test_push_events_null_core_returns_einval(void)
{
    xce_core_event_t event;
    int              rc;

    memset(&event, 0, sizeof(event));
    event.type = XCE_CORE_EVENT_CONFIG_PUSH;

    rc = xce_core_push_events(NULL, &event, 1);

    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}

static void test_push_events_null_events_with_positive_nevents_returns_einval(void)
{
    xce_sched_core_t core;
    int              rc;

    fixture_init(&core, &g_full_plugin);

    rc = xce_core_push_events(&core, NULL, 1);

    TEST_ASSERT_EQUAL_INT(-EINVAL, rc);
}

/* Zero-length batch is a valid no-op (not an error). */
static void test_push_events_zero_nevents_is_a_valid_noop(void)
{
    xce_sched_core_t core;
    int              rc;

    fixture_init(&core, &g_full_plugin);

    rc = xce_core_push_events(&core, NULL, 0);

    TEST_ASSERT_EQUAL_INT(0, rc);
}

int main(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_push_events_fires_each_hook_once_per_event_with_picker_idle);
    RUN_TEST(test_jobs_push_fires_on_job_push_once_per_job);
    RUN_TEST(test_non_picker_events_mutate_state_but_never_call_hooks);
    RUN_TEST(test_workers_push_twice_updates_existing_worker_in_place);
    RUN_TEST(test_job_rusage_applies_state_when_hook_is_null);
    RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals);
    RUN_TEST(test_push_events_rejects_with_ebusy_when_session_active_no_mutation);
    RUN_TEST(test_push_events_null_core_returns_einval);
    RUN_TEST(test_push_events_null_events_with_positive_nevents_returns_einval);
    RUN_TEST(test_push_events_zero_nevents_is_a_valid_noop);
    return UNITY_END();
}