Function run_successful_activation

tests/test_daemon_smoke.py:534–642

Run one native activation and bind its durable audit trail to its PID.

(
    binary,
    env,
    tmpdir,
    activation_log,
    label,
    action,
    arguments,
    expected_source_build,
    minimum_daemon_clients=1,
)

def run_successful_activation(
    binary,
    env,
    tmpdir,
    activation_log,
    label,
    action,
    arguments,
    expected_source_build,
    minimum_daemon_clients=1,
):
    """Run one native activation and bind its durable audit trail to its PID."""
    records_before = len(json_records(activation_log))
    process = subprocess.Popen(
        [str(binary)] + arguments,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        env=env,
    )
    try:
        stdout, stderr = process.communicate(timeout=SHUTDOWN_TIMEOUT)
    except subprocess.TimeoutExpired:
        process.terminate()
        try:
            process.communicate(timeout=3)
        except subprocess.TimeoutExpired:
            process.kill()
            process.communicate(timeout=3)
        raise
    (tmpdir / (label + ".out")).write_text(stdout, encoding="utf-8")
    (tmpdir / (label + ".err")).write_text(stderr, encoding="utf-8")
    check(
        process.returncode == 0,
        label + " failed: stdout={!r}, stderr={!r}".format(stdout, stderr),
    )
    check(
        "Stopping active CBM sessions and operations for {}...".format(action)
        in stdout,
        label + " did not show coordination progress: " + repr(stdout),
    )
    check(
        RESTART_GUIDANCE in stdout,
        label + " did not print the exact restart guidance: " + repr(stdout),
    )

    records = json_records(activation_log)[records_before:]
    records = [
        record
        for record in records
        if record.get("event") == "cbm.activation"
        and record.get("requester_pid") == process.pid
    ]
    phases = [record.get("phase") for record in records]
    check(
        phases == ["requested", "daemon_stopped", "completed"],
        label + " activation audit sequence mismatch: " + repr(records),