############################################################################
# Copyright (C) SchedMD LLC.
############################################################################
import getpass
import json
import logging
import os
import random
import time

import pytest
import requests
import yaml

import atf

pytestmark = pytest.mark.slow

random.seed()

cluster_name = f"test-cluster-{random.randrange(0, 99999999999)}"
cluster2_name = f"{cluster_name}-2"
user_name = f"test-user-{random.randrange(0, 99999999999)}"
user_name2 = f"{user_name}-2"
local_user_name = getpass.getuser()
account_name = f"test-account-{random.randrange(0, 99999999999)}"
account2_name = f"{account_name}-2"
account3_name = f"{account_name}-3"
coord_name = f"{user_name}-commander"
wckey_name = f"test-wckey-{random.randrange(0, 99999999999)}"
wckey2_name = f"{wckey_name}-2"
qos_name = f"test-qos-{random.randrange(0, 99999999999)}"
qos2_name = f"{qos_name}-2"
resv_name = f"test-reservation-{random.randrange(0, 99999999999)}"
req_node_count = 10


@pytest.fixture(scope="module", autouse=True)
def setup():
    global slurmrestd_url, token, slurmrestd
    global local_cluster_name, partition_name

    atf.require_version((26, 11), "sbin/slurmdbd")
    atf.require_version((26, 11), "sbin/slurmctld")
    atf.require_version((26, 11), "sbin/slurmrestd")
    atf.require_slurmrestd("slurmctld,slurmdbd,util", "v0.0.46")

    atf.require_accounting(modify=True)
    atf.require_nodes(req_node_count)
    atf.require_config_parameter("AllowNoDefAcct", "Yes", source="slurmdbd")
    atf.require_config_parameter("TrackWCKey", "Yes", source="slurmdbd")
    atf.require_config_parameter("TrackWCKey", "Yes")
    atf.require_config_parameter_includes("AuthAltTypes", "auth/jwt")
    atf.require_config_parameter_includes("AuthAltTypes", "auth/jwt", source="slurmdbd")
    atf.require_slurm_running()

    # Setup OpenAPI client with OpenAPI-Generator once Slurm(restd) is running
    atf.require_openapi_generator("7.3.0")

    # Uncomment the following two lines to generate a new openapi_spec.json
    # file, then it could be moved to the testsuite_data_dir if we really
    # want to change the OpenAPI specs after .0.
    # with open("openapi_spec.json", "w") as f:
    #     json.dump(atf.properties["openapi_spec"], f, indent=2)

    # Conf reliant variables (put here to avert --auto-config errors)
    local_cluster_name = atf.get_config_parameter("ClusterName")

    partition_name = atf.default_partition()
    if not partition_name:
        partition_name = "debug"


@pytest.fixture(scope="function")
def slurm(setup):
    yield atf.openapi_slurm()


@pytest.fixture(scope="function")
def slurmdb(setup):
    yield atf.openapi_slurmdb()


@pytest.fixture(scope="function")
def admin_level(setup):
    atf.run_command(
        f"sacctmgr -i add user {local_cluster_name} defaultaccount=root AdminLevel=Admin",
        user=atf.properties["slurm-user"],
        fatal=True,
    )
    yield
    atf.cancel_all_jobs()
    atf.run_command(
        f"sacctmgr -i delete user {local_cluster_name}",
        user=atf.properties["slurm-user"],
    )


@pytest.fixture(scope="function")
def non_admin(setup):
    atf.run_command(
        f"sacctmgr -i add user {local_cluster_name} defaultaccount=root",
        user=atf.properties["slurm-user"],
        fatal=True,
    )
    yield
    atf.cancel_all_jobs()
    atf.run_command(
        f"sacctmgr -i delete user {local_cluster_name}",
        user=atf.properties["slurm-user"],
    )


@pytest.fixture(scope="function")
def cleanup_crash(setup):
    yield

    if not atf.is_slurmrestd_running():
        atf.start_slurmrestd()
        # We have a new port
        atf.properties["openapi_config"].host = atf.properties["slurmrestd_url"]


@pytest.fixture(scope="function")
def create_accounts():
    atf.run_command(
        f"sacctmgr -i create account {account_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )
    atf.run_command(
        f"sacctmgr -i create account {account2_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )

    yield

    atf.run_command(
        f"sacctmgr -i delete account {account_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )
    atf.run_command(
        f"sacctmgr -i delete account {account2_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )


@pytest.fixture(scope="function")
def create_users(create_accounts):
    atf.run_command(
        f"sacctmgr -i create user {user_name} cluster={local_cluster_name} account={account_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )

    yield

    atf.cancel_all_jobs()
    atf.run_command(
        f"sacctmgr -i delete user {user_name} cluster={local_cluster_name} account={account_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )


@pytest.fixture(scope="function")
def create_coords(create_users):
    atf.run_command(
        f"sacctmgr -i create user {coord_name} cluster={local_cluster_name} account={account2_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )

    yield

    atf.cancel_all_jobs()
    atf.run_command(
        f"sacctmgr -i delete user {coord_name} cluster={local_cluster_name} account={account2_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )


@pytest.fixture(scope="function")
def create_wckeys():
    atf.run_command(
        f"sacctmgr -i create user {user_name} cluster={local_cluster_name} wckey={wckey_name}",
        user=atf.properties["slurm-user"],
        fatal=True,
    )
    atf.run_command(
        f"sacctmgr -i create user {coord_name} cluster={local_cluster_name} wckey={wckey_name}",
        user=atf.properties["slurm-user"],
        fatal=True,
    )

    yield

    atf.cancel_all_jobs()
    atf.run_command(
        f"sacctmgr -i delete user {user_name} cluster={local_cluster_name} wckey={wckey_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )
    atf.run_command(
        f"sacctmgr -i delete user {coord_name} cluster={local_cluster_name} wckey={wckey_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )


@pytest.fixture(scope="function")
def create_qos(create_coords):
    atf.run_command(
        f"sacctmgr -i create qos {qos_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )
    atf.run_command(
        f"sacctmgr -i create qos {qos2_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )

    yield

    atf.run_command(
        f"sacctmgr -i delete qos {qos_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )
    atf.run_command(
        f"sacctmgr -i delete qos {qos2_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )


@pytest.fixture(scope="function")
def dynamic_node(setup):
    """
    Sets the required MaxNodeCount + cons_tres, and returns a non-existing node.
    """
    nonexistent_node_name = "nonexistent_node"
    config = atf.get_config()

    atf.stop_slurm()
    atf.require_config_parameter("SelectType", "select/cons_tres")
    atf.require_config_parameter("SelectTypeParameters", "CR_CPU")
    atf.require_config_parameter("MaxNodeCount", str(req_node_count + 1))
    atf.start_slurm()

    atf.run_command(
        f"scontrol delete node {nonexistent_node_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
        xfail=True,
    )

    yield nonexistent_node_name

    atf.run_command(
        f"scontrol delete node {nonexistent_node_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )

    atf.stop_slurm()
    atf.require_config_parameter("SelectType", config["SelectType"])
    atf.require_config_parameter("SelectTypeParameters", config["SelectTypeParameters"])
    atf.require_config_parameter("MaxNodeCount", None)
    atf.start_slurm()


@pytest.fixture(scope="function")
def cleanup_created_user2():
    yield
    atf.run_command(
        f"sacctmgr -i delete user {user_name2} "
        f"cluster={local_cluster_name} account={account_name}",
        user=atf.properties["slurm-user"],
    )


@pytest.fixture(scope="function")
def cleanup_created_account3():
    yield
    atf.run_command(
        f"sacctmgr -i delete account {account3_name}",
        user=atf.properties["slurm-user"],
    )


def test_loaded_versions():
    r = atf.request_slurmrestd("openapi/v3")
    assert r.status_code == 200

    spec = r.json()

    # verify older plugins are not loaded
    assert "/slurm/v0.0.45/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.44/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.43/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.42/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.41/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.40/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.39/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.38/jobs" not in spec["paths"].keys()
    assert "/slurmdb/v0.0.38/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.37/jobs" not in spec["paths"].keys()
    assert "/slurmdb/v0.0.37/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.36/jobs" not in spec["paths"].keys()
    assert "/slurmdb/v0.0.36/jobs" not in spec["paths"].keys()
    assert "/slurm/v0.0.35/jobs" not in spec["paths"].keys()

    # verify current plugins are loaded
    assert "/slurm/v0.0.46/jobs/" in spec["paths"].keys()
    assert "/slurmdb/v0.0.46/jobs/" in spec["paths"].keys()


@pytest.mark.skipif(
    atf.get_version() <= (26, 11, 0), reason="Specs may change until .0 is released"
)
@pytest.mark.parametrize("openapi_spec", ["46"], indirect=True)
def test_specification(openapi_spec):
    if atf.get_version("sbin/slurmrestd") >= (28, 5):
        # This is expected to be deprecated in 28.05+
        patch = atf.get_deprecated_openapi_spec_patch(openapi_spec)
        patch.apply(openapi_spec, in_place=True)

    atf.assert_openapi_spec_eq(openapi_spec, atf.properties["openapi_spec"])


def test_db_accounts(slurm, slurmdb, create_wckeys, admin_level):
    from openapi_client import ApiClient as Client
    from openapi_client import Configuration as Config
    from openapi_client.models.v0046_account import V0046Account
    from openapi_client.models.v0046_assoc_short import V0046AssocShort
    from openapi_client.models.v0046_coord import V0046Coord
    from openapi_client.models.v0046_openapi_accounts_resp import (
        V0046OpenapiAccountsResp,
    )

    # make sure account doesn't already exist
    resp = slurmdb.slurmdb_v0046_get_account_with_http_info(account_name)
    assert resp.status_code == 200
    assert len(resp.data.accounts) == 0
    resp = slurmdb.slurmdb_v0046_get_account_with_http_info(account2_name)
    assert resp.status_code == 200
    assert len(resp.data.accounts) == 0

    # create account
    accounts = V0046OpenapiAccountsResp(
        accounts=[
            V0046Account(
                description="test description",
                name=account_name,
                organization="test organization",
            ),
            V0046Account(
                coordinators=[
                    V0046Coord(
                        name=coord_name,
                    )
                ],
                description="test description",
                name=account2_name,
                organization="test organization",
            ),
        ]
    )
    resp = slurmdb.slurmdb_v0046_post_accounts_with_http_info(accounts)
    assert resp.status_code == 200

    accounts = V0046OpenapiAccountsResp(
        accounts=[
            V0046Account(
                name=account2_name,
                description="fail description",
                organization="fail organization",
            )
        ]
    )
    resp = slurmdb.slurmdb_v0046_post_accounts_with_http_info(accounts)
    assert resp.status_code == 200

    # verify account matches modify request
    resp = slurmdb.slurmdb_v0046_get_account(account2_name)
    assert resp.accounts
    for account in resp.accounts:
        assert account.name == account2_name
        assert account.description == accounts.accounts[0].description
        assert account.organization == accounts.accounts[0].organization
        assert not account.flags

    # change account desc and org
    accounts = V0046OpenapiAccountsResp(
        accounts=[
            V0046Account(
                coordinators=[],
                description="test description modified",
                name=account2_name,
                organization="test organization modified",
            )
        ]
    )
    resp = slurmdb.slurmdb_v0046_post_accounts(accounts)
    assert not resp.warnings
    assert len(resp.errors) == 0
    resp = slurmdb.slurmdb_v0046_get_account(account2_name)
    assert resp.accounts
    for account in resp.accounts:
        assert account.name == account2_name
        assert account.description == accounts.accounts[0].description
        assert account.organization == accounts.accounts[0].organization
        assert not account.coordinators

    resp = slurmdb.slurmdb_v0046_get_account(account_name)
    assert resp.accounts
    for account in resp.accounts:
        assert account.name == account_name

    # check full listing works
    resp = slurmdb.slurmdb_v0046_get_accounts(deleted="true")
    assert resp.accounts
    resp = slurmdb.slurmdb_v0046_get_accounts()
    assert resp.accounts

    accounts = V0046OpenapiAccountsResp(
        accounts=[
            V0046Account(
                coordinators=[],
                description="test description modified",
                name=account2_name,
                organization="test organization modified",
            )
        ]
    )

    resp = slurmdb.slurmdb_v0046_delete_account(account_name)
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_delete_account(account2_name)
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_account(account_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.accounts

    resp = slurmdb.slurmdb_v0046_get_account(account2_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.accounts


def test_db_diag(slurmdb, admin_level):
    resp = slurmdb.slurmdb_v0046_get_diag()
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.statistics.time_start > 0


def test_db_wckeys(slurmdb, create_coords, admin_level):
    from openapi_client.models.v0046_openapi_wckey_resp import V0046OpenapiWckeyResp
    from openapi_client.models.v0046_wckey import V0046Wckey

    wckeys = V0046OpenapiWckeyResp(
        wckeys=[
            V0046Wckey(
                cluster=local_cluster_name,
                name=wckey_name,
                user=user_name,
            ),
            V0046Wckey(
                cluster=local_cluster_name,
                name=wckey2_name,
                user=user_name,
            ),
            V0046Wckey(
                cluster=local_cluster_name,
                name=wckey2_name,
                user=coord_name,
            ),
        ]
    )

    resp = slurmdb.slurmdb_v0046_post_wckeys_with_http_info(
        v0046_openapi_wckey_resp=wckeys
    )
    assert resp.status_code == 200

    resp = slurmdb.slurmdb_v0046_get_wckeys()
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert len(resp.wckeys) >= 1

    resp = slurmdb.slurmdb_v0046_get_wckey(wckey_name)
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.wckeys
    for wckey in resp.wckeys:
        assert wckey.name == wckey_name or wckey.name == wckey2_name
        assert wckey.user == user_name or wckey.user == coord_name

    resp = slurmdb.slurmdb_v0046_get_wckey(wckey2_name)
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.wckeys
    for wckey in resp.wckeys:
        assert wckey.name == wckey2_name
        assert wckey.user == user_name or wckey.user == coord_name

    resp = slurmdb.slurmdb_v0046_delete_wckey(wckey_name)
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_delete_wckey(wckey2_name)
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_wckey(wckey_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert len(resp.wckeys) == 0

    resp = slurmdb.slurmdb_v0046_get_wckey(wckey2_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert len(resp.wckeys) == 0


def test_db_clusters(slurmdb, admin_level):
    from openapi_client.models.v0046_cluster_rec import V0046ClusterRec
    from openapi_client.models.v0046_openapi_clusters_resp import (
        V0046OpenapiClustersResp,
    )

    clusters = V0046OpenapiClustersResp(
        clusters=[
            V0046ClusterRec(
                name=cluster_name,
            ),
            V0046ClusterRec(
                name=cluster2_name,
            ),
        ]
    )

    resp = slurmdb.slurmdb_v0046_post_clusters(v0046_openapi_clusters_resp=clusters)
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_clusters()
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.clusters

    resp = slurmdb.slurmdb_v0046_get_cluster(cluster_name)
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.clusters
    for cluster in resp.clusters:
        assert cluster.name == cluster_name
        assert not cluster.nodes

    resp = slurmdb.slurmdb_v0046_get_cluster(cluster2_name)
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.clusters
    for cluster in resp.clusters:
        assert cluster.name == cluster2_name
        assert not cluster.nodes

    resp = slurmdb.slurmdb_v0046_delete_cluster(cluster_name)
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_delete_cluster(cluster2_name)
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_cluster(cluster_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.clusters

    resp = slurmdb.slurmdb_v0046_get_cluster(cluster2_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.clusters


def test_db_users(slurmdb, admin_level):
    from openapi_client.models.v0046_assoc_short import V0046AssocShort
    from openapi_client.models.v0046_coord import V0046Coord
    from openapi_client.models.v0046_openapi_users_resp import V0046OpenapiUsersResp
    from openapi_client.models.v0046_user import V0046User
    from openapi_client.models.v0046_user_default import V0046UserDefault
    from openapi_client.models.v0046_wckey import V0046Wckey

    users = V0046OpenapiUsersResp(
        users=[
            V0046User(
                administrator_level=["None"],
                default=dict(
                    wckey=wckey_name,
                ),
                name=user_name,
            ),
            V0046User(
                administrator_level=["Operator"],
                wckeys=[
                    V0046Wckey(
                        cluster=local_cluster_name, name=wckey_name, user=coord_name
                    ),
                    V0046Wckey(
                        cluster=local_cluster_name,
                        name=wckey2_name,
                        user=coord_name,
                    ),
                ],
                default=V0046UserDefault(
                    wckey=wckey2_name,
                ),
                name=coord_name,
            ),
        ]
    )

    resp = slurmdb.slurmdb_v0046_post_users(v0046_openapi_users_resp=users)
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_users()
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.users

    # Using query parameters (i.e. with_wckeys/with_deleted) results in warnings
    # Slurmrestd expected OpenAPI type=boolean but got OpenAPI type=string

    resp = slurmdb.slurmdb_v0046_get_user(user_name, with_wckeys="true")
    if resp.warnings:
        assert len(resp.warnings) == 1
        assert resp.warnings[0].source == "#/with_wckeys/"
    assert len(resp.errors) == 0
    assert resp.users
    for user in resp.users:
        assert user.name == user_name
        assert user.default.wckey == wckey_name

    resp = slurmdb.slurmdb_v0046_get_user(coord_name, with_wckeys="true")
    if resp.warnings:
        assert len(resp.warnings) == 1
        assert resp.warnings[0].source == "#/with_wckeys/"
    assert len(resp.errors) == 0
    assert resp.users
    for user in resp.users:
        assert user.name == coord_name
        assert user.default.wckey == wckey2_name
        for wckey in user.wckeys:
            assert wckey.name == wckey_name or wckey.name == wckey2_name
            assert wckey.user == coord_name
            assert wckey.cluster == local_cluster_name

    resp = slurmdb.slurmdb_v0046_delete_user(coord_name)
    assert len(resp.errors) == 0

    user_exists = False
    resp = slurmdb.slurmdb_v0046_get_user(coord_name, with_deleted="true")
    assert len(resp.errors) == 0
    for user in resp.users:
        assert user.name == coord_name
        assert user.flags[0] == "DELETED"
        user_exists = True

    if not user_exists:
        users = V0046OpenapiUsersResp(
            users=[
                V0046User(
                    administrator_level=["Administrator"],
                    default=dict(
                        wckey=wckey_name,
                    ),
                    old_name=user_name,
                    name=coord_name,
                )
            ]
        )

        resp = slurmdb.slurmdb_v0046_post_users(v0046_openapi_users_resp=users)
        assert not resp.warnings
        assert len(resp.errors) == 0

        resp = slurmdb.slurmdb_v0046_get_user(coord_name, with_wckeys="true")
        if resp.warnings:
            assert len(resp.warnings) == 1
            assert resp.warnings[0].source == "#/with_wckeys/"
        assert len(resp.errors) == 0
        assert resp.users
        for user in resp.users:
            assert user.name == coord_name
            assert not user.old_name
            assert user.default.wckey == wckey_name
            for wckey in user.wckeys:
                assert wckey.name == wckey_name
                assert wckey.user == coord_name
                assert wckey.cluster == local_cluster_name

        resp = slurmdb.slurmdb_v0046_delete_user(coord_name)
        assert len(resp.errors) == 0

        resp = slurmdb.slurmdb_v0046_get_user(coord_name)
        assert len(resp.warnings) > 0
        assert len(resp.errors) == 0
        assert not resp.users


def test_db_assoc(slurmdb, create_coords, create_qos, admin_level):
    from openapi_client.models.v0046_assoc import V0046Assoc
    from openapi_client.models.v0046_assoc_short import V0046AssocShort
    from openapi_client.models.v0046_coord import V0046Coord
    from openapi_client.models.v0046_openapi_assocs_resp import V0046OpenapiAssocsResp
    from openapi_client.models.v0046_shares_struct import (
        V0046SharesStruct as V0046Shares,
    )
    from openapi_client.models.v0046_uint32_no_val_struct import (
        V0046Uint32NoValStruct as V0046Uint32NoVal,
    )
    from openapi_client.models.v0046_user import V0046User
    from openapi_client.models.v0046_wckey import V0046Wckey

    associations = V0046OpenapiAssocsResp(
        associations=[
            V0046Assoc(
                account=account_name,
                cluster=local_cluster_name,
                default=dict(
                    qos=qos_name,
                ),
                flags=[],
                max=dict(
                    jobs=dict(
                        per=dict(
                            wall_clock=V0046Uint32NoVal(
                                set=True,
                                number=150,
                            )
                        ),
                    ),
                ),
                min=dict(
                    priority_threshold=V0046Uint32NoVal(
                        set=True,
                        number=10,
                    )
                ),
                partition=partition_name,
                priority=V0046Uint32NoVal(number=9, set=True),
                qos=[qos_name, qos2_name],
                shares_raw=V0046Shares(set=True, number=23),
                user=user_name,
            ),
            V0046Assoc(
                account=account_name,
                cluster=local_cluster_name,
                default=dict(
                    qos=qos_name,
                ),
                flags=[],
                max=dict(
                    jobs=dict(
                        per=dict(
                            wall_clock=V0046Uint32NoVal(
                                set=True,
                                number=150,
                            )
                        ),
                    ),
                ),
                min=dict(
                    priority_threshold=V0046Uint32NoVal(
                        set=True,
                        number=10,
                    )
                ),
                priority=V0046Uint32NoVal(number=9, set=True),
                qos=[qos_name, qos2_name],
                shares_raw=V0046Shares(set=True, number=23),
                user=user_name,
            ),
            V0046Assoc(
                account=account2_name,
                cluster=local_cluster_name,
                default=dict(
                    qos=qos2_name,
                ),
                flags=[],
                max=dict(
                    jobs=dict(
                        per=dict(
                            wall_clock=V0046Uint32NoVal(
                                set=True,
                                number=50,
                            )
                        ),
                    ),
                ),
                min=dict(
                    priority_threshold=V0046Uint32NoVal(
                        set=True,
                        number=4,
                    )
                ),
                partition=partition_name,
                priority=V0046Uint32NoVal(number=90, set=True),
                qos=[qos2_name],
                shares_raw=V0046Shares(set=True, number=1012),
                user=user_name,
            ),
        ]
    )

    resp = slurmdb.slurmdb_v0046_post_associations(
        v0046_openapi_assocs_resp=associations
    )
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_associations()
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.associations

    resp = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account_name,
        user=user_name,
        partition=partition_name,
    )
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.associations
    for assoc in resp.associations:
        assert assoc.cluster == local_cluster_name
        assert assoc.account == account_name
        assert assoc.user == user_name
        assert assoc.partition == partition_name
        assert assoc.default.qos == qos_name
        assert not assoc.flags
        assert assoc.max.jobs.per.wall_clock.set
        assert assoc.max.jobs.per.wall_clock.number == 150
        assert assoc.min.priority_threshold.set
        assert assoc.min.priority_threshold.number == 10
        assert assoc.priority.set
        assert assoc.priority.number == 9
        for qos in assoc.qos:
            assert qos == qos_name or qos == qos2_name
        assert assoc.shares_raw.set
        assert not assoc.shares_raw.parent
        assert assoc.shares_raw.number == 23

    associations = V0046OpenapiAssocsResp(
        associations=[
            V0046Assoc(
                account=account_name,
                cluster=local_cluster_name,
                partition=partition_name,
                user=user_name,
                default=dict(
                    qos=qos2_name,
                ),
                qos=[qos2_name],
                max=dict(
                    jobs=dict(
                        per=dict(wall_clock=V0046Uint32NoVal(set=True, number=250)),
                    ),
                ),
                min=dict(
                    priority_threshold=V0046Uint32NoVal(set=True, number=100),
                ),
                priority=V0046Uint32NoVal(number=848, set=True),
                shares_raw=V0046Shares(set=True, number=230),
            )
        ]
    )

    resp = slurmdb.slurmdb_v0046_post_associations(
        v0046_openapi_assocs_resp=associations
    )
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account_name,
        user=user_name,
        partition=partition_name,
    )
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.associations
    for assoc in resp.associations:
        assert assoc.cluster == local_cluster_name
        assert assoc.account == account_name
        assert assoc.user == user_name
        assert assoc.partition == partition_name
        assert assoc.default.qos == qos2_name
        assert not assoc.flags
        assert assoc.max.jobs.per.wall_clock.set
        assert assoc.max.jobs.per.wall_clock.number == 250
        assert assoc.min.priority_threshold.set
        assert assoc.min.priority_threshold.number == 100
        assert assoc.priority.set
        assert assoc.priority.number == 848
        for qos in assoc.qos:
            assert qos == qos2_name
        assert assoc.shares_raw.set
        assert not assoc.shares_raw.parent
        assert assoc.shares_raw.number == 230

    # fairshare=parent (SLURMDB_FS_USE_PARENT) instead of a share count
    associations = V0046OpenapiAssocsResp(
        associations=[
            V0046Assoc(
                account=account_name,
                cluster=local_cluster_name,
                partition=partition_name,
                user=user_name,
                shares_raw=V0046Shares(parent=True),
            )
        ]
    )

    resp = slurmdb.slurmdb_v0046_post_associations(
        v0046_openapi_assocs_resp=associations
    )
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account_name,
        user=user_name,
        partition=partition_name,
    )
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.associations
    for assoc in resp.associations:
        assert assoc.shares_raw.parent
        assert not assoc.shares_raw.set
        # A fairshare-only POST must not clobber the other limits the previous
        # modify set on this association (they survive because the parser leaves
        # omitted fields at NO_VAL and slurmdbd skips NO_VAL).
        assert assoc.cluster == local_cluster_name
        assert assoc.account == account_name
        assert assoc.user == user_name
        assert assoc.partition == partition_name
        assert assoc.default.qos == qos2_name
        assert assoc.priority.set
        assert assoc.priority.number == 848
        assert assoc.max.jobs.per.wall_clock.set
        assert assoc.max.jobs.per.wall_clock.number == 250
        assert assoc.min.priority_threshold.set
        assert assoc.min.priority_threshold.number == 100

    resp = slurmdb.slurmdb_v0046_delete_association(
        cluster=local_cluster_name,
        account=account_name,
        user=user_name,
        partition=partition_name,
    )
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account_name,
        user=user_name,
        partition=partition_name,
    )
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.associations

    resp = slurmdb.slurmdb_v0046_delete_associations(
        cluster=local_cluster_name,
        user=user_name,
    )
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_delete_associations(
        cluster=local_cluster_name,
        account=account_name,
    )
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account_name,
    )
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.associations

    resp = slurmdb.slurmdb_v0046_delete_associations(
        cluster=local_cluster_name,
        account=account2_name,
    )
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account2_name,
    )
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.associations


def _slurmrestd_post(path, body, expect_status=200, expect_error=False):
    """POST raw JSON to slurmrestd and return the decoded response.

    Used for SHARES request shapes the generated OpenAPI client cannot
    express: the bare string "parent" and a bare integer/null shares value (the
    field's generated model is the V0046SharesStruct object).
    """
    r = requests.post(
        f"{atf.properties['slurmrestd_url']}{path}",
        headers=atf.properties["slurmrestd-headers"],
        json=body,
    )
    if expect_error:
        assert r.status_code >= 400, f"POST {path} -> {r.status_code}: {r.text[:300]}"
    else:
        assert (
            r.status_code == expect_status
        ), f"POST {path} -> {r.status_code}: {r.text[:300]}"
    return r.json()


def test_db_assoc_create_user_fairshare_parent(
    slurmdb, create_accounts, admin_level, cleanup_created_user2
):
    """Creating a user association in fairshare=parent mode via POST
    /users_association/ (ASSOC_REC_SET.fairshare) -- the creation path bug 25581
    is about, which the modify-path coverage in test_db_assoc does not exercise.
    """
    from openapi_client.models.v0046_assoc_rec_set import V0046AssocRecSet
    from openapi_client.models.v0046_openapi_users_add_cond_resp import (
        V0046OpenapiUsersAddCondResp,
    )
    from openapi_client.models.v0046_shares_struct import V0046SharesStruct
    from openapi_client.models.v0046_user_short import V0046UserShort
    from openapi_client.models.v0046_users_add_cond import V0046UsersAddCond

    req = V0046OpenapiUsersAddCondResp(
        association_condition=V0046UsersAddCond(
            accounts=[account_name],
            clusters=[local_cluster_name],
            users=[user_name2],
            association=V0046AssocRecSet(fairshare=V0046SharesStruct(parent=True)),
        ),
        user=V0046UserShort(),
    )
    resp = slurmdb.slurmdb_v0046_post_users_association(
        v0046_openapi_users_add_cond_resp=req
    )
    assert not resp.warnings, resp.warnings
    assert len(resp.errors) == 0, resp.errors

    got = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account_name,
        user=user_name2,
    )
    assert len(got.errors) == 0, got.errors
    assert got.associations, "created user association was not returned"
    for assoc in got.associations:
        assert assoc.shares_raw.parent, "created assoc not in parent mode"
        assert not assoc.shares_raw.set, "parent assoc must not set a share count"

    # Cross-check through a second, independent interface: sacctmgr reads the
    # stored value via assoc_mgr/DB, not the REST parse/dump, and prints the
    # literal "parent" for SLURMDB_FS_USE_PARENT. This catches a symmetric
    # parse<->dump mis-mapping the REST round-trip alone would hide (e.g. parent
    # mapped onto INFINITE, which would persist as 1 share).
    share = atf.run_command_output(
        f"sacctmgr -n -P show assoc cluster={local_cluster_name} "
        f"account={account_name} user={user_name2} format=Share",
        fatal=True,
    ).strip()
    assert share == "parent", f"sacctmgr Share should be parent, got {share!r}"


def test_db_assoc_create_user_fairshare_parent_string(
    slurmdb, create_accounts, admin_level, cleanup_created_user2
):
    """Creating a user association with the bare string "parent" via POST
    /users_association/ -- the exact payload reported in bug 25581. The
    generated client cannot express it: the schema types fairshare as the
    V0046SharesStruct object.
    """
    body = {
        "association_condition": {
            "accounts": [account_name],
            "clusters": [local_cluster_name],
            "users": [user_name2],
            "association": {"fairshare": "parent"},
        },
        "user": {},
    }
    resp = _slurmrestd_post("/slurmdb/v0.0.46/users_association/", body)
    assert not resp.get("errors"), resp.get("errors")
    assert not resp.get("warnings"), resp.get("warnings")

    got = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account_name,
        user=user_name2,
    )
    assert len(got.errors) == 0, got.errors
    assert got.associations, "created user association was not returned"
    for assoc in got.associations:
        assert assoc.shares_raw.parent, "created assoc not in parent mode"
        assert not assoc.shares_raw.set, "parent assoc must not set a share count"


def test_db_assoc_create_account_fairshare_parent_string(
    slurmdb, admin_level, cleanup_created_account3
):
    """Creating an account association via POST /accounts_association/ using the
    bare string "parent" (the sacctmgr keyword form; the generated client can
    only send the struct). Covers both the accounts creation endpoint and the
    DATA_TYPE_STRING branch of PARSE_FUNC(SHARES).
    """
    body = {
        "association_condition": {
            "accounts": [account3_name],
            "clusters": [local_cluster_name],
            "association": {"fairshare": "parent"},
        }
    }
    resp = _slurmrestd_post("/slurmdb/v0.0.46/accounts_association/", body)
    assert not resp.get("errors"), resp.get("errors")
    assert not resp.get("warnings"), resp.get("warnings")

    got = slurmdb.slurmdb_v0046_get_association(
        cluster=local_cluster_name,
        account=account3_name,
    )
    assert len(got.errors) == 0, got.errors
    assert got.associations, "created account association was not returned"
    for assoc in got.associations:
        assert assoc.shares_raw.parent, "created account assoc not in parent mode"
        assert not assoc.shares_raw.set, "parent assoc must not set a share count"


def test_db_assoc_shares_infinite_and_unset(slurmdb, create_users, admin_level):
    """The accepted SHARES forms on ASSOC.shares_raw: a bare -1 resets shares to
    1 (like sacctmgr FairShare=-1), null leaves the stored value unchanged (no
    shares= clause emitted), the bare string "parent" selects parent mode, and
    the struct discriminators resolve in the documented order parent > infinite
    > set/number. The null case is the dangerous one -- a regression there
    silently rewrites the share count of any association updated without
    mentioning fairshare.
    """
    ident = {
        "cluster": local_cluster_name,
        "account": account_name,
        "user": user_name,
    }

    def get_shares():
        got = slurmdb.slurmdb_v0046_get_association(**ident)
        assert len(got.errors) == 0, got.errors
        assert (
            len(got.associations) == 1
        ), f"expected exactly one association, got {got.associations}"
        return got.associations[0].shares_raw

    def post_shares(value):
        _slurmrestd_post(
            "/slurmdb/v0.0.46/associations/",
            {"associations": [{**ident, "shares_raw": value}]},
        )

    # Seed a known share count, then -1 must reset it to a single share.
    post_shares(55)
    shares = get_shares()
    assert shares.set and shares.number == 55, "seed of 55 shares did not take"
    post_shares(-1)
    shares = get_shares()
    assert shares.set and shares.number == 1, "fairshare=-1 should reset to 1 share"

    # Re-seed, then null must leave the value untouched.
    post_shares(77)
    assert get_shares().number == 77, "re-seed of 77 shares did not take"
    post_shares(None)
    shares = get_shares()
    assert shares.set and shares.number == 77, "shares=null must not change shares"

    # Same two behaviors via the SHARES struct fields (distinct parser
    # branches from the bare integer/null above): infinite=True resets to 1, and
    # set=False leaves the value unchanged. These are the forms an OpenAPI client
    # sending the generated V0046SharesStruct model actually uses.
    post_shares(55)
    post_shares({"infinite": True})
    shares = get_shares()
    assert shares.set and shares.number == 1, "shares={infinite} should reset to 1"
    post_shares(88)
    post_shares({"set": False})
    assert get_shares().number == 88, "shares={set:false} must not change shares"

    # An update that omits shares_raw entirely must also leave it unchanged (the
    # field stays NO_VAL via slurmdb_init_assoc_rec), explicitly not reset to 0.
    _slurmrestd_post(
        "/slurmdb/v0.0.46/associations/",
        {"associations": [{**ident, "priority": {"set": True, "number": 5}}]},
    )
    got = slurmdb.slurmdb_v0046_get_association(**ident)
    assert len(got.errors) == 0, got.errors
    assert got.associations, "association not found"
    assert got.associations[0].priority.number == 5, "the modify did not land"
    shares = got.associations[0].shares_raw
    assert (
        shares.set and shares.number == 88
    ), "omitting fairshare must not change shares"

    # The bare string form bug 25581 reported against this endpoint.
    post_shares("parent")
    shares = get_shares()
    assert shares.parent, 'shares_raw="parent" should select parent mode'
    assert not shares.set, "parent assoc must not set a share count"

    # Precedence: "parent" outranks every other discriminator, and "infinite"
    # outranks "set"/"number".
    post_shares(99)
    shares = get_shares()
    assert (
        shares.set and shares.number == 99 and not shares.parent
    ), "a numeric POST must clear parent mode"
    post_shares({"parent": True, "infinite": True, "set": True, "number": 42})
    shares = get_shares()
    assert shares.parent, "parent must outrank infinite and set/number"
    assert not shares.set, "parent assoc must not set a share count"
    post_shares(99)
    post_shares({"infinite": True, "set": True, "number": 42})
    shares = get_shares()
    assert shares.set and shares.number == 1, "infinite must outrank set/number"


def test_db_assoc_shares_rejects_invalid(slurmdb, create_users, admin_level):
    """An unrecognized fairshare string is rejected and the stored share count
    survives. sacctmgr.1 documents the value set as {<fairshare_number>|parent},
    so anything outside it must not silently become a share count.
    """
    ident = {
        "cluster": local_cluster_name,
        "account": account_name,
        "user": user_name,
    }

    def get_shares():
        got = slurmdb.slurmdb_v0046_get_association(**ident)
        assert len(got.errors) == 0, got.errors
        assert (
            len(got.associations) == 1
        ), f"expected exactly one association, got {got.associations}"
        return got.associations[0].shares_raw

    _slurmrestd_post(
        "/slurmdb/v0.0.46/associations/",
        {"associations": [{**ident, "shares_raw": 33}]},
    )
    assert get_shares().number == 33, "seed of 33 shares did not take"

    resp = _slurmrestd_post(
        "/slurmdb/v0.0.46/associations/",
        {"associations": [{**ident, "shares_raw": "parnet"}]},
        expect_error=True,
    )
    assert resp["errors"], "an unrecognized fairshare string must be reported"

    shares = get_shares()
    assert shares.set and shares.number == 33, "rejected input must not change shares"


def test_shares_render_parent(slurm, create_users, admin_level):
    """GET /slurm/v0.0.46/shares renders an inherited shares value as "parent".
    This is the third SHARES use-site (ASSOC_SHARES_OBJ_WRAP.shares_raw, a
    parser instance distinct from the ASSOC and ASSOC_REC_SET write parsers),
    so its dump of the parent sentinel is otherwise unobserved. Reads through
    slurmctld's assoc_mgr, which updates asynchronously from slurmdbd.
    """
    atf.run_command(
        f"sacctmgr -i modify account {account_name} where "
        f"cluster={local_cluster_name} set fairshare=7",
        user=atf.properties["slurm-user"],
        fatal=True,
    )
    atf.run_command(
        f"sacctmgr -i modify user {user_name} where cluster={local_cluster_name} "
        f"account={account_name} set fairshare=parent",
        user=atf.properties["slurm-user"],
        fatal=True,
    )

    entry = None
    account_entry = None
    for _ in atf.timer(fatal=True):
        resp = slurm.slurm_v0046_get_shares(accounts=account_name, users=user_name)
        assert len(resp.errors) == 0, resp.errors
        entry = next((s for s in resp.shares.shares if s.name == user_name), None)
        account_entry = next(
            (s for s in resp.shares.shares if s.name == account_name), None
        )
        if entry is not None and entry.shares.parent and account_entry is not None:
            break

    assert entry.shares.parent, "shares of a parent assoc must render parent"
    assert not entry.shares.set, "parent assoc must not report a share count"

    # The numeric account in the same response is the control: a dump stuck on
    # parent would satisfy the user assertions above on its own.
    assert account_entry.shares.set, "numeric assoc must report set"
    assert not account_entry.shares.parent, "numeric assoc must not render parent"
    assert account_entry.shares.number == 7, "numeric assoc must report its shares"


def test_db_qos(slurmdb, create_coords, admin_level):
    from openapi_client.models.v0046_float64_no_val_struct import (
        V0046Float64NoValStruct as V0046Float64NoVal,
    )
    from openapi_client.models.v0046_openapi_slurmdbd_qos_resp import (
        V0046OpenapiSlurmdbdQosResp,
    )
    from openapi_client.models.v0046_qos import V0046Qos
    from openapi_client.models.v0046_tres import V0046Tres
    from openapi_client.models.v0046_uint32_no_val_struct import (
        V0046Uint32NoValStruct as V0046Uint32NoVal,
    )

    qos = V0046OpenapiSlurmdbdQosResp(
        qos=[
            V0046Qos(
                description="test QOS",
                flags=[
                    "PARTITION_MAXIMUM_NODE",
                    "PARTITION_TIME_LIMIT",
                    "ENFORCE_USAGE_THRESHOLD",
                    "NO_RESERVE",
                    "DENY_LIMIT",
                    "OVERRIDE_PARTITION_QOS",
                    "NO_DECAY",
                ],
                limits=dict(
                    min=dict(
                        tres=dict(
                            per=dict(
                                job=[
                                    V0046Tres(
                                        type="cpu",
                                        count=100,
                                    ),
                                    V0046Tres(
                                        type="memory",
                                        count=100000,
                                    ),
                                ],
                            ),
                        ),
                    ),
                ),
                name=qos_name,
                preempt=dict(
                    exempt_time=V0046Uint32NoVal(set=True, number=199),
                ),
                priority=V0046Uint32NoVal(number=180, set=True),
                usage_factor=V0046Float64NoVal(
                    set=True,
                    number=82382.23823,
                ),
                usage_threshold=V0046Float64NoVal(
                    set=True,
                    number=929392.33,
                ),
            ),
            V0046Qos(
                description="test QOS 2",
                name=qos2_name,
            ),
        ]
    )

    resp = slurmdb.slurmdb_v0046_post_qos(v0046_openapi_slurmdbd_qos_resp=qos)
    assert not resp.warnings
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_qos()
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.qos

    resp = slurmdb.slurmdb_v0046_get_single_qos(qos_name)
    assert not resp.warnings
    assert len(resp.errors) == 0
    assert resp.qos
    for qos in resp.qos:
        assert qos.description == "test QOS"
        assert qos.flags
        for flag in qos.flags:
            assert flag in [
                "PARTITION_MAXIMUM_NODE",
                "PARTITION_TIME_LIMIT",
                "ENFORCE_USAGE_THRESHOLD",
                "NO_RESERVE",
                "DENY_LIMIT",
                "OVERRIDE_PARTITION_QOS",
                "NO_DECAY",
            ]
        assert qos.limits.min.tres.per.job
        for tres in qos.limits.min.tres.per.job:
            assert tres.type == "cpu" or tres.type == "memory"
            if tres.type == "cpu":
                assert tres.count == 100
            if tres.type == "memory":
                assert tres.count == 100000
        assert qos.name == qos_name
        assert qos.preempt.exempt_time.set
        assert qos.preempt.exempt_time.number == 199
        assert qos.priority.set
        assert qos.priority.number == 180
        assert qos.usage_factor.set
        assert qos.usage_factor.number == 82382.23823
        assert qos.usage_threshold.set
        assert qos.usage_threshold.number == 929392.33

    resp = slurmdb.slurmdb_v0046_get_single_qos(qos2_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.qos
    for qos in resp.qos:
        assert qos.description == "test QOS 2"
        assert not qos.flags
        assert not qos.limits.min.tres.per.job
        assert qos.name == qos2_name
        assert not qos.preempt.exempt_time.set
        assert qos.priority.set
        assert qos.priority.number == 0
        assert qos.usage_factor.set
        assert qos.usage_factor.number == 1
        assert not qos.usage_threshold.set

    resp = slurmdb.slurmdb_v0046_delete_single_qos(qos_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_single_qos(qos_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.qos

    resp = slurmdb.slurmdb_v0046_delete_single_qos(qos2_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_single_qos(qos2_name)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert not resp.qos


def test_db_tres(slurmdb):
    resp = slurmdb.slurmdb_v0046_get_tres()
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0


def test_db_config(slurmdb, admin_level):
    resp = slurmdb.slurmdb_v0046_get_config()
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0


def test_jobs(slurm, slurmdb, non_admin):
    from openapi_client.models.v0046_job_comment import V0046JobComment
    from openapi_client.models.v0046_job_desc_msg import V0046JobDescMsg
    from openapi_client.models.v0046_job_info import V0046JobInfo
    from openapi_client.models.v0046_job_modify import V0046JobModify
    from openapi_client.models.v0046_job_modify_tres import V0046JobModifyTres
    from openapi_client.models.v0046_job_submit_req import V0046JobSubmitReq
    from openapi_client.models.v0046_openapi_job_modify_req import (
        V0046OpenapiJobModifyReq,
    )
    from openapi_client.models.v0046_process_exit_code_verbose import (
        V0046ProcessExitCodeVerbose,
    )
    from openapi_client.models.v0046_process_exit_code_verbose_signal import (
        V0046ProcessExitCodeVerboseSignal,
    )
    from openapi_client.models.v0046_tres import V0046Tres
    from openapi_client.models.v0046_uint32_no_val_struct import V0046Uint32NoValStruct
    from openapi_client.models.v0046_uint32_no_val_struct import (
        V0046Uint32NoValStruct as V0046Uint32NoVal,
    )

    script = "#!/bin/bash\n/bin/true"
    env = ["PATH=/bin/:/sbin/:/usr/bin/:/usr/sbin/"]

    job = V0046JobSubmitReq(
        script=script,
        job=V0046JobDescMsg(
            partition=partition_name,
            name="runjob",
            environment=env,
            current_working_directory="/tmp/",
        ),
    )

    resp = slurm.slurm_v0046_post_job_submit(job)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.job_id
    assert resp.step_id
    jobid1 = int(resp.job_id)

    resp = slurm.slurm_v0046_get_jobs()
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    for job in resp.jobs:
        assert job.job_id == jobid1
        assert job.name == "runjob"
        assert job.partition == partition_name

    resp = slurm.slurm_v0046_get_job(str(jobid1))
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    for job in resp.jobs:
        assert job.job_id == jobid1
        assert job.name == "runjob"
        assert job.partition == partition_name

    job = V0046JobSubmitReq(
        script=script,
        job=V0046JobDescMsg(
            partition=partition_name,
            name="runjob",
            environment=env,
            current_working_directory="/tmp/",
        ),
    )

    resp = slurm.slurm_v0046_post_job_submit(job)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.job_id
    assert resp.step_id
    jobid2 = int(resp.job_id)

    resp = slurm.slurm_v0046_get_job(str(jobid2))
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert len(resp.jobs) == 1
    for job in resp.jobs:
        assert job.job_id == jobid2
        assert job.name == "runjob"
        assert job.partition == partition_name

    # submit a HELD job to be able to update it
    job = V0046JobSubmitReq(
        script=script,
        job=V0046JobDescMsg(
            partition=partition_name,
            name="test job",
            environment=env,
            priority=V0046Uint32NoVal(number=0, set=True),
            current_working_directory="/tmp/",
        ),
    )

    resp = slurm.slurm_v0046_post_job_submit(job)
    assert len(resp.warnings) > 0
    assert len(resp.errors) == 0
    assert resp.job_id
    assert resp.step_id
    jobid3 = int(resp.job_id)

    job = V0046JobDescMsg(
        environment=env,
        partition=partition_name,
        name="updated test job",
        priority=V0046Uint32NoVal(number=0, set=True),
    )

    resp = slurm.slurm_v0046_post_job(str(jobid3), v0046_job_desc_msg=job)
    assert not len(resp.warnings)
    assert not len(resp.errors)
    if resp.results:
        for result in resp.results:
            assert result.job_id == jobid3
            assert result.error_code == 0

    resp = slurm.slurm_v0046_get_job(str(jobid3))
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    for job in resp.jobs:
        assert job.job_id == jobid3
        assert job.name == "updated test job"
        assert job.partition == partition_name
        assert job.priority.set
        assert job.priority.number == 0
        assert job.user_name == local_user_name

    resp = slurm.slurm_v0046_delete_job(str(jobid3))
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    resp = slurm.slurm_v0046_get_job(str(jobid3))
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    for job in resp.jobs:
        assert job.job_id == jobid3
        assert job.name == "updated test job"
        assert job.partition == partition_name
        assert job.user_name == local_user_name
        assert job.job_state == ["CANCELLED"]

    # Ensure that job is in the DB before querying it
    atf.wait_for_job_accounted(jobid3, fatal=True)

    resp = slurmdb.slurmdb_v0046_get_jobs()
    assert len(resp.errors) == 0

    for t in atf.timer():
        resp = slurmdb.slurmdb_v0046_get_job(str(jobid3))
        assert len(resp.warnings) == 0
        assert len(resp.errors) == 0
        assert resp.jobs
        updated_job_found = False
        for job in resp.jobs:
            if job.name == "updated test job":
                assert job.job_id == jobid3
                assert job.name == "updated test job"
                assert job.partition == partition_name
                updated_job_found = True
                break
        if updated_job_found:
            break
    else:
        assert False, "Updated job should reach the DB"

    resp = slurmdb.slurmdb_v0046_get_jobs(users=local_user_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.jobs
    for job in resp.jobs:
        assert job.user == local_user_name

    # Update job in db -- posting to /job/jobid/
    atf.wait_for_job_accounted(jobid1, "State", "COMPLETED", fatal=True)
    atf.wait_for_job_accounted(jobid2, "State", "COMPLETED", fatal=True)

    atf.run_command(
        f"sacctmgr -i mod user {local_cluster_name} set AdminLevel=Admin",
        user=atf.properties["slurm-user"],
        fatal=True,
    )

    WIFEXIT_CODE = V0046Uint32NoValStruct(
        set=True, infinite=False, number=4  # 1024 >> 8
    )

    job_modify = V0046JobModify(
        comment=V0046JobComment(
            administrator="admin1",
            job="job1",
            system="system1",
        ),
        derived_exit_code=V0046ProcessExitCodeVerbose(
            return_code=V0046Uint32NoValStruct(set=True, infinite=False, number=1024),
        ),
        extra="extra1",
        tres=V0046JobModifyTres(
            allocated=[V0046Tres(type="energy", count=12346, id=3)]
        ),
        wckey="mywckey1",
    )

    resp = slurmdb.slurmdb_v0046_post_job(str(jobid1), v0046_job_modify=job_modify)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_jobs(step=str(jobid1))
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert len(resp.jobs) == 1
    for job in resp.jobs:
        logging.debug(job)
        assert job.comment == job_modify.comment
        assert job.derived_exit_code.return_code == WIFEXIT_CODE
        assert job.extra == job_modify.extra
        found_tres = False
        for tres in job.tres.allocated:
            if tres.type == "energy":
                found_tres = True
                tres == job_modify.tres.allocated[0]
        assert found_tres
        assert job.wckey.wckey == job_modify.wckey

    # Update jobs in db -- posting to /jobs/
    # Update values to be different
    WIFEXIT_CODE = V0046Uint32NoValStruct(
        set=True, infinite=False, number=8  # 2048 >> 8
    )

    job_modify = V0046JobModify(
        comment=V0046JobComment(
            administrator="admin2",
            job="job2",
            system="system2",
        ),
        derived_exit_code=V0046ProcessExitCodeVerbose(
            return_code=V0046Uint32NoValStruct(set=True, infinite=False, number=2048),
        ),
        extra="extra2",
        tres=V0046JobModifyTres(
            allocated=[V0046Tres(type="energy", count=54321, id=3)]
        ),
        wckey="mywckey2",
    )

    job_modify_req = V0046OpenapiJobModifyReq(
        job_id_list=[str(jobid1), str(jobid2)], job_rec=job_modify
    )
    resp = slurmdb.slurmdb_v0046_post_jobs(v0046_openapi_job_modify_req=job_modify_req)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    resp = slurmdb.slurmdb_v0046_get_jobs(job_name="runjob")
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert len(resp.jobs) == 2
    for job in resp.jobs:
        logging.debug(job)
        assert job.comment == job_modify.comment
        assert job.derived_exit_code.return_code == WIFEXIT_CODE
        assert job.extra == job_modify.extra
        found_tres = False
        for tres in job.tres.allocated:
            if tres.type == "energy":
                found_tres = True
                tres == job_modify.tres.allocated[0]
        found_tres = True
        assert job.wckey.wckey == job_modify.wckey


@pytest.fixture(scope="function")
def reservation(setup):
    # Ensure that ALL nodes are idle before reserving them (or scontrol may return an error)
    nodes = atf.get_nodes()
    for node in nodes:
        atf.wait_for_node_state(node, "IDLE", fatal=True)

    atf.run_command(
        f"scontrol create reservation starttime=now duration=120 user=root nodes=ALL ReservationName={resv_name}",
        user=atf.properties["slurm-user"],
        fatal=True,
    )

    yield

    atf.run_command(
        f"scontrol delete ReservationName={resv_name}",
        user=atf.properties["slurm-user"],
        fatal=False,
    )


def test_partitions(slurm):
    resp = slurm.slurm_v0046_get_partition(partition_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.partitions
    for part in resp.partitions:
        assert part.name == partition_name

    resp = slurm.slurm_v0046_get_partitions()
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.partitions


def test_partition_get_post_roundtrip(slurm, admin_level):
    """GET one partition, POST the same partition descriptions as an update.

    Echoing read-only fields (cluster, totals, tres, etc.) may produce warnings
    but must not produce errors.
    """
    from openapi_client.models.v0046_openapi_partitions_mod_req import (
        V0046OpenapiPartitionsModReq,
    )

    get_resp = slurm.slurm_v0046_get_partition(partition_name)
    assert len(get_resp.errors) == 0, "GET partition should not return errors"
    assert len(get_resp.warnings) == 0, "GET partition should not return warnings"
    assert len(get_resp.partitions) == 1, "GET partition should return one partition"

    for part in get_resp.partitions:
        assert part.name == partition_name, "GET should return the requested partition"

    req = V0046OpenapiPartitionsModReq(partitions=get_resp.partitions)
    post_resp = slurm.slurm_v0046_post_partitions(req)
    assert len(post_resp.errors) == 0, "POST echoed partition must not return errors"


def test_post_partitions(slurm, admin_level):
    from openapi_client.models.v0046_openapi_partitions_mod_req import (
        V0046OpenapiPartitionsModReq,
    )
    from openapi_client.models.v0046_partition_info import V0046PartitionInfo
    from openapi_client.models.v0046_partition_info_defaults import (
        V0046PartitionInfoDefaults,
    )
    from openapi_client.models.v0046_partition_info_nodes import V0046PartitionInfoNodes
    from openapi_client.models.v0046_partition_info_partition import (
        V0046PartitionInfoPartition,
    )
    from openapi_client.models.v0046_uint64_no_val_struct import V0046Uint64NoValStruct

    part_1_name = "part_1"
    part_2_name = "part_2"

    nodes = list(atf.get_nodes().keys())
    part_1_nodes = atf.run_command_output(
        f"scontrol show hostlist {','.join(nodes[0:5])}"
    ).strip()
    part_2_nodes = atf.run_command_output(
        f"scontrol show hostlist {','.join(nodes[5:10])}"
    ).strip()

    req = V0046OpenapiPartitionsModReq(
        partitions=[
            V0046PartitionInfo(
                name=part_1_name,
                nodes=V0046PartitionInfoNodes(configured=part_1_nodes),
                flags=[
                    "HIDDEN",
                    "NO_ROOT",
                    "ROOT_ONLY",
                    "REQ_RESV",
                    "LLN",
                    "PDOI",
                ],
                preempt_mode=[
                    "REQUEUE",
                ],
                defaults=V0046PartitionInfoDefaults(
                    partition_memory_per_cpu=V0046Uint64NoValStruct(
                        set=True,
                        number=10,
                    ),
                ),
                partition=V0046PartitionInfoPartition(
                    oversubscribe="FORCE:5", exclusive="USER"
                ),
            ),
            V0046PartitionInfo(
                name=part_2_name,
                nodes=V0046PartitionInfoNodes(configured=part_2_nodes),
                flags=[
                    "LLN",
                ],
                preempt_mode=["DISABLED"],
                defaults=V0046PartitionInfoDefaults(
                    partition_memory_per_node=V0046Uint64NoValStruct(
                        set=True,
                        number=12,
                    )
                ),
                partition=V0046PartitionInfoPartition(
                    oversubscribe="YES:6", exclusive="NO"
                ),
            ),
        ]
    )

    # Create partitions part_1 and part_2
    resp = slurm.slurm_v0046_post_partitions(req)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    # Check that they were created
    resp = slurm.slurm_v0046_get_partitions()
    assert len(resp.errors) == 0
    match_cnt = 0
    inx = 0
    for part in resp.partitions:
        if part.name == part_1_name:
            inx = 0
        elif part.name == part_2_name:
            inx = 1
        else:
            continue

        match_cnt += 1

        assert (
            req.partitions[inx].nodes.configured == part.nodes.configured
        ), "Partition was not created with the correct nodes"
        assert set(req.partitions[inx].flags) == set(
            part.flags
        ), "Partition was not create with the correct flags"
        assert set(req.partitions[inx].preempt_mode) == set(
            part.preempt_mode
        ), "Partition was not create with the correct PreemptMode"
        if req.partitions[inx].defaults.partition_memory_per_cpu:
            assert (
                req.partitions[inx].defaults.partition_memory_per_cpu.number
                == part.defaults.partition_memory_per_cpu.number
            ), "Partition was not create with the correct DefMemPerCpu"
        if req.partitions[inx].defaults.partition_memory_per_node:
            assert (
                req.partitions[inx].defaults.partition_memory_per_node.number
                == part.defaults.partition_memory_per_node.number
            ), "Partition was not create with the correct DefMemPerNode"
        assert req.partitions[inx].partition is not None
        assert part.partition is not None
        assert (
            req.partitions[inx].partition.oversubscribe == part.partition.oversubscribe
        ), "Partition oversubscribe string did not round-trip"

    assert match_cnt == 2

    # Test updating
    req.partitions[0].nodes.configured = part_2_nodes  # swap nodes
    req.partitions[1].nodes.configured = part_1_nodes
    req.partitions[0].flags = [  # clear all flags
        "HIDDEN_CLEAR",
        "NO_ROOT_CLEAR",
        "ROOT_ONLY_CLEAR",
        "REQ_RESV_CLEAR",
        "LLN_CLEAR",
        "PDOI_CLEAR",
    ]
    req.partitions[1].flags = [  # append HIDDEN to existing LLN
        "HIDDEN",
    ]
    req.partitions[0].partition.oversubscribe = "NO"
    req.partitions[0].partition.exclusive = "TOPO"
    req.partitions[1].partition.oversubscribe = "FORCE:1"
    req.partitions[1].partition.exclusive = "USER"

    resp = slurm.slurm_v0046_post_partitions(req)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    # Check that they were updated
    resp = slurm.slurm_v0046_get_partitions()
    assert len(resp.errors) == 0
    match_cnt = 0
    for part in resp.partitions:
        if part.name == part_1_name:
            inx = 0
        elif part.name == part_2_name:
            inx = 1
        else:
            continue

        match_cnt += 1

        assert (
            req.partitions[inx].nodes.configured == part.nodes.configured
        ), "Partition nodes failed to update"
        if inx == 0:
            assert (
                len(part.flags) == 0
            ), f"Partition flags were not all cleared during update: {part.flags}"
        else:
            assert set(part.flags) == {
                "HIDDEN",
                "LLN",
            }, "Partition flags did not update correctly"

        assert (
            req.partitions[inx].partition.oversubscribe == part.partition.oversubscribe
        ), "Partition oversubscribe failed to update"
        assert (
            req.partitions[inx].partition.exclusive == part.partition.exclusive
        ), "Partition exclusive failed to update"

    assert match_cnt == 2

    atf.run_command_output(f"scontrol delete partition {part_1_name}")
    atf.run_command_output(f"scontrol delete partition {part_2_name}")


def test_delete_partition(slurm, admin_level):
    partition_name = "new_partition"
    atf.run_command_output(f"scontrol create partitionname={partition_name}")
    assert partition_name in atf.get_partitions(), "Failed to create partition"
    resp = slurm.slurm_v0046_delete_partition(partition_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert partition_name not in atf.get_partitions(), "Failed to delete partition"


def test_nodes(slurm, admin_level):
    from openapi_client.models.v0046_update_node_msg import V0046UpdateNodeMsg

    node_name = None
    reasonuid = None
    resp = slurm.slurm_v0046_get_nodes()
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.nodes
    for node in resp.nodes:
        if "IDLE" in node.state:
            node_name = node.name
            # comment = node.comment
            extra = node.extra
            feat = node.features
            actfeat = node.active_features
            # state = node.state
            reason = node.reason
            reasonuid = node.reason_set_by_user
            break

    # Skip if no idle nodes are found
    if node_name is None:
        pytest.fail("No idle nodes are found")

    if reasonuid is None or len(reasonuid) <= 0:
        reasonuid = local_user_name

    node = V0046UpdateNodeMsg(
        comment="test node comment",
        extra="test node extra",
        features=[
            "feat1",
            "feat2",
            "feat3",
        ],
        features_act=[
            "feat1",
            "feat3",
        ],
        state=["DRAIN"],
        reason="testing",
        reason_uid=local_user_name,
    )

    resp = slurm.slurm_v0046_post_node(node_name, v0046_update_node_msg=node)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    resp = slurm.slurm_v0046_get_node(node_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.nodes
    for node in resp.nodes:
        assert node.name == node_name
        assert node.comment == "test node comment"
        assert node.extra == "test node extra"
        assert "DRAIN" in node.state
        assert node.reason == "testing"
        assert node.reason_set_by_user == local_user_name

    ncomment = "test comment comment 2"
    node = V0046UpdateNodeMsg(
        comment=ncomment,
        extra=extra,
        features=feat,
        features_act=actfeat,
        state=["RESUME"],
        reason=reason,
        reason_uid=reasonuid,
    )

    resp = slurm.slurm_v0046_post_node(node_name, node)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0

    resp = slurm.slurm_v0046_get_node(node_name)
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.nodes
    for node in resp.nodes:
        assert node.name == node_name
        assert node.comment == ncomment
        assert node.extra == extra


def test_slurmrestd_slurm_nodes_update_time(slurm, admin_level):
    """GET /nodes with update_time from a prior last_update succeeds."""
    resp = slurm.slurm_v0046_get_nodes()
    assert not resp.errors, f"unexpected errors: {resp.errors}"
    assert resp.last_update is not None, "GET /nodes missing last_update"
    node_name = None
    for node in resp.nodes:
        if node.name:
            node_name = node.name
            break
    assert node_name, "expected at least one node from GET /nodes"
    expected_nodes = {node.name for node in resp.nodes if node.name}
    before_update = resp.last_update.number

    atf.run_command(
        f"scontrol update nodename={node_name} " "comment=slurmrestd-nodes-update-time",
        user=atf.properties["slurm-user"],
        fatal=True,
    )

    update_time = str(before_update)
    r = atf.request_slurmrestd(f"slurm/v0.0.46/nodes?update_time={update_time}")
    assert r.status_code == 200, (
        f"GET /nodes with update_time={update_time} "
        f"returned HTTP {r.status_code}: {r.text[:500]}"
    )
    body = r.json()
    assert not body.get("errors"), (
        f"GET /nodes with update_time={update_time} "
        f"returned errors: {body.get('errors')}"
    )
    assert not body.get("warnings"), (
        f"GET /nodes with update_time={update_time} "
        f"returned warnings: {body.get('warnings')}"
    )
    nodes = body.get("nodes")
    assert nodes, "expected the changed node to be returned for a diverged update_time"
    changed = next((n for n in nodes if n.get("name") == node_name), None)
    assert changed, f"changed node {node_name} missing from response"
    assert (
        changed.get("comment") == "slurmrestd-nodes-update-time"
    ), f"changed node {node_name} did not reflect the updated comment"
    returned_nodes = {n.get("name") for n in nodes if n.get("name")}
    assert returned_nodes >= expected_nodes, (
        "expected the entire node collection to be returned for a diverged "
        f"update_time, missing: {expected_nodes - returned_nodes}"
    )
    after_update = body.get("last_update", {}).get("number")
    assert after_update, "response missing last_update"
    assert (
        after_update >= before_update
    ), f"last_update ({after_update}) went backwards past update_time ({before_update})"


def test_slurmrestd_slurm_nodes_update_time_no_change(slurm):
    """GET /nodes with a future update_time returns 304 Not Modified."""
    r = atf.request_slurmrestd("slurm/v0.0.46/nodes?update_time=4102444800")
    assert r.status_code == 304, (
        f"GET /nodes with a future update_time "
        f"returned HTTP {r.status_code}: {r.text[:500]}"
    )
    assert not r.text, f"expected an empty body on 304, got: {r.text[:500]}"


def test_ping(slurm):
    resp = slurm.slurm_v0046_get_ping()
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0


def test_diag(slurm):
    resp = slurm.slurm_v0046_get_diag()
    assert len(resp.warnings) == 0
    assert len(resp.errors) == 0
    assert resp.statistics


def test_licenses(slurm):
    resp = slurm.slurm_v0046_get_licenses()
    assert len(resp.errors) == 0


@pytest.mark.parametrize(
    "flags",
    [[], ["IGNORE_JOBS"], ["IGNORE_JOBS", "MAGNETIC"]],
)
def test_reservations(slurm, flags, admin_level):
    from openapi_client.models.v0046_reservation_desc_msg import V0046ReservationDescMsg
    from openapi_client.models.v0046_reservation_mod_req import V0046ReservationModReq
    from openapi_client.models.v0046_uint32_no_val_struct import V0046Uint32NoValStruct
    from openapi_client.models.v0046_uint64_no_val_struct import V0046Uint64NoValStruct

    resv_name = "test_resv"
    users = ["root", "atf"]
    duration = V0046Uint32NoValStruct(number=300, set=True)
    start_time = V0046Uint64NoValStruct(number=int(time.time()) + 60, set=True)
    end_time = V0046Uint64NoValStruct(
        number=start_time.number + duration.number * 60, set=True
    )
    partition = "primary"
    node_list = ["node1", "node7"]

    # Create a reservation
    logging.debug(f"Creating reservation '{resv_name}'...")
    reservation_info = V0046ReservationDescMsg(
        name=resv_name,
        users=users,
        duration=duration,
        start_time=start_time,
        node_list=node_list,
        partition=partition,
        flags=flags,
    )
    resp = slurm.slurm_v0046_post_reservations(
        V0046ReservationModReq(reservations=[reservation_info])
    )
    assert resp.reservations, f"Reservation {resv_name} should be created"
    assert not resp.warnings and not resp.errors

    # Verify the fields of the response
    retrieved_reservation = resp.reservations[0]
    assert (
        retrieved_reservation.name == resv_name
    ), f"Field 'name' should be {resv_name}"
    assert (
        retrieved_reservation.partition == partition
    ), f"Field 'partition' should be {partition}"
    assert (
        retrieved_reservation.node_list == node_list
    ), f"Field 'node_list' should be {node_list}"
    assert retrieved_reservation.users == users, f"Field 'users' should be {users}"
    assert (
        retrieved_reservation.start_time.number == start_time.number
    ), f"Field 'start_time' should be {start_time}"
    assert (
        retrieved_reservation.duration.number == duration.number
    ), f"Field 'duration' should be {duration}"
    assert (
        # SPEC_NODES is automatically added
        set([flag for flag in retrieved_reservation.flags if flag != "SPEC_NODES"])
        == set(flags)
    ), f"Field 'flags' should be {flags}"

    # Verify fields of reservation created
    resp = slurm.slurm_v0046_get_reservation(resv_name)
    assert len(resp.reservations) == 1, f"Reservation '{resv_name}' should be returned"

    retrieved_reservation = resp.reservations[0]
    assert (
        retrieved_reservation.name == resv_name
    ), f"Field 'name' should be {resv_name}"
    assert (
        retrieved_reservation.partition == partition
    ), f"Field 'partition' should be {partition}"
    assert (
        atf.node_range_to_list(retrieved_reservation.node_list) == node_list
    ), f"Field 'node_list' should be {node_list}"
    assert (
        retrieved_reservation.users.split(",") == users
    ), f"Field 'users' should be {users}"
    assert (
        retrieved_reservation.start_time.number == start_time.number
    ), f"Field 'start_time' should be {start_time}"
    assert (
        retrieved_reservation.end_time.number == end_time.number
    ), f"Field 'end_time' should be {end_time}"
    assert (
        # SPEC_NODES is automatically added
        set([flag for flag in retrieved_reservation.flags if flag != "SPEC_NODES"])
        == set(flags)
    ), f"Field 'flags' should be {flags}"

    # Update reservation
    new_end_time = retrieved_reservation.end_time.number + 300
    new_users = ["root"]
    new_node_list = ["node2"]
    new_start_time = V0046Uint64NoValStruct(number=int(time.time()) + 90, set=True)
    new_flags = ["IGNORE_JOBS", "MAGNETIC", "USER_DELETE"]

    update_info = V0046ReservationDescMsg(
        name=resv_name,
        end_time=V0046Uint64NoValStruct(set=True, number=new_end_time),
        users=new_users,
        node_list=new_node_list,
        start_time=new_start_time,
        flags=new_flags,
    )
    resp = slurm.slurm_v0046_post_reservation(update_info)
    assert resp.reservations, "Reservation should be updated"
    assert not resp.warnings and not resp.errors

    # Validate update
    resp = slurm.slurm_v0046_get_reservation(resv_name)
    assert len(resp.reservations) == 1, f"Reservation '{resv_name}' should be returned"

    updated = resp.reservations[0]

    assert updated.end_time.number == new_end_time
    assert updated.users.split(",") == new_users, f"Field 'users' should be {new_users}"
    assert (
        atf.node_range_to_list(updated.node_list) == new_node_list
    ), f"Field 'node_list' should be {new_node_list}"
    assert (
        updated.start_time.number == new_start_time.number
    ), f"Field 'start_time' should be {new_start_time}"
    assert (
        # SPEC_NODES is automatically added
        set([flag for flag in updated.flags if flag != "SPEC_NODES"])
        == set(new_flags)
    ), f"Field 'flags' should be {flags}"

    # Delete reservation
    resp = slurm.slurm_v0046_delete_reservation(reservation_name=resv_name)
    assert not resp.warnings and not resp.errors
    assert resv_name not in [
        r.name for r in slurm.slurm_v0046_get_reservations().reservations
    ], f"Reservation {resv_name} should be deleted"


@pytest.mark.parametrize("legal_state", ["CLOUD", "FUTURE", "EXTERNAL"])
def test_legal_node_creation(slurm, admin_level, legal_state, dynamic_node):
    from openapi_client.models.v0046_openapi_create_node_req import (
        V0046OpenapiCreateNodeReq,
    )

    request_body = V0046OpenapiCreateNodeReq(
        node_conf=f"nodename={dynamic_node} State={legal_state}"
    )

    response = slurm.slurm_v0046_post_new_node(
        v0046_openapi_create_node_req=request_body
    )

    # Validate there are no errors or warnings, and the meta data exits
    assert not response.errors
    assert not response.warnings
    assert response.meta

    # Validate the node exists and is in the correct state
    node_states = atf.get_node_parameter(dynamic_node, "state")
    assert (
        legal_state in node_states
    ), f"Dynamic node should have {legal_state} in its state only has {node_states}"
    assert (
        "DYNAMIC_NORM" in node_states
    ), "Dynamic node should have DYNAMIC_NORM in its state"


@pytest.mark.parametrize(
    "illegal_state", ["DOWN", "DRAIN", "FAIL", "FAILING", "UNKNOWN"]
)
def test_illegal_node_creation(slurm, admin_level, illegal_state, dynamic_node):
    from openapi_client.api_response import ApiResponse
    from openapi_client.exceptions import ApiException
    from openapi_client.models.v0046_openapi_create_node_req import (
        V0046OpenapiCreateNodeReq,
    )

    request_body = V0046OpenapiCreateNodeReq(
        node_conf=f"nodename={dynamic_node} State={illegal_state}"
    )

    try:
        # This should throw an exception
        slurm.slurm_v0046_post_new_node_with_http_info(
            v0046_openapi_create_node_req=request_body
        )
        assert (
            False
        ), "slurm_v0046_post_new_node_with_http_info should have raised an exception"
    except ApiException as e:
        response = slurm.api_client.deserialize(
            response=ApiResponse(data=e.body), response_type="V0046OpenapiResp"
        )
        # Validate there is an error and no warnings
        assert e.status != 200
        assert len(response.errors) == 1  # error about wrong state
        assert not response.warnings
        assert response.meta

    # Validate the node doesn't exist
    atf.run_command(
        f"scontrol show NodeName={dynamic_node}",
        user="slurm",
        xfail=True,
        fatal=True,
    )


# Test until endpoints
@pytest.fixture
def util_api(setup):
    yield atf.openapi_util()


def test_util_hostnames(util_api):
    from openapi_client.models.v0046_openapi_hostlist_req_resp import (
        V0046OpenapiHostlistReqResp,
    )

    hostnames = ["node01", "node02", "node03"]
    request_body = V0046OpenapiHostlistReqResp(hostlist="node[01-03]")

    response = util_api.util_v0046_post_hostnames(
        v0046_openapi_hostlist_req_resp=request_body
    )
    assert response.hostnames is not None
    assert isinstance(response.hostnames, list)
    assert all(isinstance(host, str) for host in response.hostnames)
    assert hostnames == response.hostnames


def test_util_hostlist(util_api):
    from openapi_client.models.v0046_openapi_hostnames_req_resp import (
        V0046OpenapiHostnamesReqResp,
    )

    request_body = V0046OpenapiHostnamesReqResp(
        hostnames=["node01", "node02", "node03"]
    )
    response = util_api.util_v0046_post_hostlist(
        v0046_openapi_hostnames_req_resp=request_body
    )
    assert response.hostlist is not None
    assert isinstance(response.hostlist, str)
    assert "node[01-03]" == response.hostlist


def test_resv_crash(slurm, admin_level, cleanup_crash):
    """Check for xfree crash (bug 23038)"""
    from openapi_client.models.v0046_reservation_desc_msg import V0046ReservationDescMsg
    from openapi_client.models.v0046_reservation_mod_req import V0046ReservationModReq
    from openapi_client.models.v0046_uint32_no_val_struct import V0046Uint32NoValStruct
    from openapi_client.models.v0046_uint64_no_val_struct import V0046Uint64NoValStruct

    # Don't overlap with other resv in case of restd crash/restart
    resv_name = "crash_test_resv"
    users = ["root", "atf"]
    duration = V0046Uint32NoValStruct(number=1, set=True)
    start_time = V0046Uint64NoValStruct(
        number=int(time.time()) + 10000000,
        set=True,
    )
    partition = "primary"

    # Create a reservation with empty node_list
    reservation_info = V0046ReservationDescMsg(
        name=resv_name,
        users=users,
        duration=duration,
        start_time=start_time,
        partition=partition,
        node_list=[],
    )
    # Exception means restd crashed
    resp = slurm.slurm_v0046_post_reservations(
        V0046ReservationModReq(reservations=[reservation_info])
    )
    assert (
        not resp.warnings and not resp.errors
    ), "We should be able to get the server response from this message"


# Fields that legitimately differ between two independent dumps of the same
# config: runtime counters and the build/boot/update timestamps. The CLI and
# REST calls each sample a separate RPC, so these can disagree without a bug.
conf_volatile_fields = {
    "BOOT_TIME",
    "NEXT_JOB_ID",
    "next_job_id",
    "LastUpdate",
    "controllers",
}


def test_slurmrestd_slurm_conf(slurm):
    """GET /slurm/v0.0.46/conf returns a well-formed slurm.conf payload."""
    resp = slurm.slurm_v0046_get_conf()
    assert not resp.errors, f"unexpected errors: {resp.errors}"
    assert not resp.warnings, f"unexpected warnings: {resp.warnings}"
    assert resp.slurm_conf, "empty slurm_conf payload"
    assert resp.slurm_conf.cluster_name == atf.get_config_parameter(
        "ClusterName"
    ), "slurm_conf ClusterName does not match the running config"


def test_slurmrestd_slurm_conf_update_time_no_change(slurm):
    """A future update_time yields a no-change response with an empty body."""
    resp = slurm.slurm_v0046_get_conf(update_time="4102444800")
    assert not resp.errors, f"unexpected errors: {resp.errors}"
    assert resp.warnings, "expected a no-change warning"
    assert not resp.slurm_conf, "expected empty body on no-change"


def test_slurmrestd_slurm_conf_update_time_past(slurm):
    """A past update_time still returns the full slurm.conf payload."""
    resp = slurm.slurm_v0046_get_conf(update_time="1")
    assert not resp.errors, f"unexpected errors: {resp.errors}"
    assert resp.slurm_conf, "expected a full payload for a past update_time"


def test_slurmrestd_slurm_conf_rejects_non_get():
    """Only GET is registered for /slurm/v0.0.46/conf."""
    r = requests.post(
        f"{atf.properties['slurmrestd_url']}/slurm/v0.0.46/conf",
        headers=atf.properties["slurmrestd-headers"],
        timeout=30,
    )
    # A non-GET on a GET-only path is not served: slurmrestd's router returns
    # 404 (ESLURM_REST_UNKNOWN_URL) today. Accept 405 too so a future change to
    # the more correct status doesn't break this.
    assert r.status_code in (
        404,
        405,
    ), f"expected 404/405, got {r.status_code}: {r.text[:200]}"


def test_scontrol_show_config_json():
    """`scontrol show config --json=v0.0.46` emits a well-formed slurm_conf."""
    blob = json.loads(
        atf.run_command_output("scontrol show config --json=v0.0.46", fatal=True)
    )
    assert blob["errors"] == [], f"unexpected errors: {blob['errors']}"
    assert blob["slurm_conf"], "empty slurm_conf payload"


def test_scontrol_show_config_yaml():
    """`scontrol show config --yaml=v0.0.46` produces the same shape as --json."""
    blob = yaml.safe_load(
        atf.run_command_output("scontrol show config --yaml=v0.0.46", fatal=True)
    )
    assert blob["errors"] == [], f"unexpected errors: {blob['errors']}"
    assert blob["slurm_conf"], "empty slurm_conf payload"


def test_scontrol_show_config_json_list():
    """`scontrol show config --json=list` lists the data_parser plugins."""
    out = atf.run_command_output("scontrol show config --json=list", fatal=True)
    assert "v0.0.46" in out, f"v0.0.46 not listed in: {out}"


def test_slurm_conf_cli_matches_slurmrestd():
    """scontrol --json=v0.0.46 and /slurm/v0.0.46/conf yield equivalent payloads."""
    cli = json.loads(
        atf.run_command_output("scontrol show config --json=v0.0.46", fatal=True)
    )["slurm_conf"]
    rest = atf.request_slurmrestd("slurm/v0.0.46/conf").json()["slurm_conf"]
    assert cli.get("ClusterName"), "slurm_conf payload missing ClusterName"
    cli = {k: v for k, v in cli.items() if k not in conf_volatile_fields}
    rest = {k: v for k, v in rest.items() if k not in conf_volatile_fields}
    assert cli == rest, "scontrol and slurmrestd slurm_conf payloads differ"


def test_slurmrestd_slurmdbd_conf(slurmdb):
    """GET /slurmdb/v0.0.46/conf returns a well-formed slurmdbd.conf payload."""
    resp = slurmdb.slurmdb_v0046_get_conf()
    assert not resp.errors, f"unexpected errors: {resp.errors}"
    assert not resp.warnings, f"unexpected warnings: {resp.warnings}"
    assert resp.slurmdbd_conf, "empty slurmdbd_conf payload"
    assert resp.slurmdbd_conf.dbd_host, "slurmdbd_conf missing DbdHost"


def test_slurmrestd_slurmdbd_conf_rejects_non_get():
    """Only GET is registered for /slurmdb/v0.0.46/conf."""
    r = requests.post(
        f"{atf.properties['slurmrestd_url']}/slurmdb/v0.0.46/conf",
        headers=atf.properties["slurmrestd-headers"],
        timeout=30,
    )
    # See test_slurmrestd_slurm_conf_rejects_non_get: a non-GET on a
    # GET-only path is not served (404 today, 405 acceptable).
    assert r.status_code in (
        404,
        405,
    ), f"expected 404/405, got {r.status_code}: {r.text[:200]}"


def test_sacctmgr_show_config_json():
    """`sacctmgr show config --json=v0.0.46` emits a well-formed slurmdbd_conf."""
    blob = json.loads(
        atf.run_command_output("sacctmgr show config --json=v0.0.46", fatal=True)
    )
    assert blob["errors"] == [], f"unexpected errors: {blob['errors']}"
    assert blob["slurmdbd_conf"], "empty slurmdbd_conf payload"


def test_sacctmgr_show_config_yaml():
    """`sacctmgr show config --yaml=v0.0.46` produces the same shape as --json."""
    blob = yaml.safe_load(
        atf.run_command_output("sacctmgr show config --yaml=v0.0.46", fatal=True)
    )
    assert blob["errors"] == [], f"unexpected errors: {blob['errors']}"
    assert blob["slurmdbd_conf"], "empty slurmdbd_conf payload"


def test_sacctmgr_show_config_json_list():
    """`sacctmgr show config --json=list` lists the data_parser plugins."""
    out = atf.run_command_output("sacctmgr show config --json=list", fatal=True)
    assert "v0.0.46" in out, f"v0.0.46 not listed in: {out}"


def test_slurmdbd_conf_cli_matches_slurmrestd():
    """sacctmgr --json=v0.0.46 and /slurmdb/v0.0.46/conf yield equivalent payloads."""
    cli = json.loads(
        atf.run_command_output("sacctmgr show config --json=v0.0.46", fatal=True)
    )["slurmdbd_conf"]
    rest = atf.request_slurmrestd("slurmdb/v0.0.46/conf").json()["slurmdbd_conf"]
    assert cli.get("DbdHost"), "slurmdbd_conf payload missing DbdHost"
    cli = {k: v for k, v in cli.items() if k not in conf_volatile_fields}
    rest = {k: v for k, v in rest.items() if k not in conf_volatile_fields}
    assert cli == rest, "sacctmgr and slurmrestd slurmdbd_conf payloads differ"