Google Git Google Git
Sign in
third-party-mirror/SchedMD/slurm/04ea0cdbdba2c80ca9dd0f6cb72a005d66d780bf/./testsuite/python/tests/test_112_45.py
blob: 9a37c714c53474e24a9d9fdaf4337e7c4d6b2b54 [file]
############################################################################
# Copyright (C) SchedMD LLC.
############################################################################
import getpass
import json
import logging
import os
import pwd
import random
import time
import jsonpatch
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_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_slurmrestd("slurmctld,slurmdbd,util", "v0.0.45")
atf.require_version((26, 5), "sbin/slurmdbd")
atf.require_version((26, 5), "sbin/slurmctld")
atf.require_version((26, 5), "sbin/slurmrestd")
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()
def _delete_user(delete_args):
"""Delete a user once slurmdbd no longer accounts active jobs for it.
Cancelled jobs are done for slurmctld before slurmdbd receives their end
records, and sacctmgr refuses to delete a user with active jobs.
"""
for t in atf.timer(poll_interval=0.2, fatal=True):
result = atf.run_command(
f"sacctmgr -i delete user {delete_args}",
user=atf.properties["slurm-user"],
)
if "Job(s) active" not in result["stderr"]:
break
@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()
_delete_user(local_cluster_name)
@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,
)
_wait_for_assoc_mgr_account(local_cluster_name, "root")
yield
atf.cancel_all_jobs()
_delete_user(local_cluster_name)
@pytest.fixture(scope="function")
def resv_crash(setup):
"""Ensure that slurmrestd is running and reservation is deleted"""
resv_name = "crash_resv_name"
yield resv_name
if not atf.is_slurmrestd_running():
atf.start_slurmrestd()
# We have a new port
atf.properties["openapi_config"].host = atf.properties["slurmrestd_url"]
atf.run_command(f"scontrol delete reservationname={resv_name}", user="root")
@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()
_delete_user(f"{user_name} cluster={local_cluster_name} account={account_name}")
@pytest.fixture(scope="function")
def association_enforcement():
"""Require AccountingStorageEnforce=associations.
In local-config mode only the tests using this fixture are skipped.
In auto-config mode enforcement is enabled and, like any configuration
change, stays in effect until the module is torn down.
"""
atf.require_config_parameter_includes("AccountingStorageEnforce", "associations")
def _wait_for_assoc_mgr_account(user, account):
"""Wait until slurmctld's association cache reflects a sacctmgr change.
sacctmgr modifies slurmdbd directly; slurmctld's in-memory assoc_mgr
cache is updated asynchronously, so job submissions right after a
sacctmgr add/delete can otherwise race a stale cache.
"""
for t in atf.timer(fatal=True):
output = atf.run_command_output(
f"scontrol show assoc_mgr users={user} flags=assoc", fatal=True
)
if f"Account={account}" in output:
break
def _wait_for_no_assoc_mgr_user(user):
"""Wait until slurmctld's association cache no longer has any
association for user (see _wait_for_assoc_mgr_account)."""
for t in atf.timer(fatal=True):
output = atf.run_command_output(
f"scontrol show assoc_mgr users={user} flags=assoc", fatal=True
)
if f"UserName={user}(" not in output:
break
@pytest.fixture(scope="function")
def create_local_user_assoc(create_accounts):
"""Create association for local user when AccountingStorageEnforce=associations."""
atf.run_command(
f"sacctmgr -i add user {local_user_name} cluster={local_cluster_name} account={account_name}",
user=atf.properties["slurm-user"],
fatal=True,
)
_wait_for_assoc_mgr_account(local_user_name, account_name)
yield
atf.run_command(
f"sacctmgr -i delete user {local_user_name} cluster={local_cluster_name} account={account_name}",
user=atf.properties["slurm-user"],
)
@pytest.fixture(scope="function")
def no_local_user_assoc():
"""Ensure local_user_name has no association for the duration of the test.
Removes any association before the test runs (so tests relying on this
do not depend on another test's teardown having already removed one)
and again afterwards, waiting for slurmctld's association cache to
reflect the removal so job submissions aren't raced against a stale
cache.
"""
atf.require_auto_config("deletes all associations of the local user")
def _remove_and_wait():
atf.run_command(
f"sacctmgr -i delete user {local_user_name} cluster={local_cluster_name}",
user=atf.properties["slurm-user"],
)
_wait_for_no_assoc_mgr_user(local_user_name)
_remove_and_wait()
yield
_remove_and_wait()
@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()
_delete_user(f"{coord_name} cluster={local_cluster_name} account={account2_name}")
@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()
_delete_user(f"{user_name} cluster={local_cluster_name} wckey={wckey_name}")
_delete_user(f"{coord_name} cluster={local_cluster_name} wckey={wckey_name}")
@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()
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.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.45/jobs/" in spec["paths"].keys()
assert "/slurmdb/v0.0.45/jobs/" in spec["paths"].keys()
@pytest.mark.skipif(
atf.get_version() <= (26, 5, 0), reason="Specs may change until .0 is released"
)
@pytest.mark.parametrize("openapi_spec", ["45"], indirect=True)
def test_specification(openapi_spec):
base_path = "/components/schemas/v0.0.45_"
if atf.get_version("sbin/slurmrestd") >= (26, 11):
# Ticket 25180: In 26.11 we removed gres_drained
path = base_path + "node/properties/gres_drained/deprecated"
patch = jsonpatch.JsonPatch([{"op": "add", "path": path, "value": True}])
patch.apply(openapi_spec, in_place=True)
if atf.get_version("sbin/slurmrestd") >= (27, 11):
# This is expected to be deprecated in 27.11+
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.v0045_account import V0045Account
from openapi_client.models.v0045_assoc_short import V0045AssocShort
from openapi_client.models.v0045_coord import V0045Coord
from openapi_client.models.v0045_openapi_accounts_resp import (
V0045OpenapiAccountsResp,
)
# make sure account doesn't already exist
resp = slurmdb.slurmdb_v0045_get_account_with_http_info(account_name)
assert resp.status_code == 200
assert len(resp.data.accounts) == 0
resp = slurmdb.slurmdb_v0045_get_account_with_http_info(account2_name)
assert resp.status_code == 200
assert len(resp.data.accounts) == 0
# create account
accounts = V0045OpenapiAccountsResp(
accounts=[
V0045Account(
description="test description",
name=account_name,
organization="test organization",
),
V0045Account(
coordinators=[
V0045Coord(
name=coord_name,
)
],
description="test description",
name=account2_name,
organization="test organization",
),
]
)
resp = slurmdb.slurmdb_v0045_post_accounts_with_http_info(accounts)
assert resp.status_code == 200
accounts = V0045OpenapiAccountsResp(
accounts=[
V0045Account(
name=account2_name,
description="fail description",
organization="fail organization",
)
]
)
resp = slurmdb.slurmdb_v0045_post_accounts_with_http_info(accounts)
assert resp.status_code == 200
# verify account matches modify request
resp = slurmdb.slurmdb_v0045_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 = V0045OpenapiAccountsResp(
accounts=[
V0045Account(
coordinators=[],
description="test description modified",
name=account2_name,
organization="test organization modified",
)
]
)
resp = slurmdb.slurmdb_v0045_post_accounts(accounts)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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_v0045_get_account(account_name)
assert resp.accounts
for account in resp.accounts:
assert account.name == account_name
# check full listing works
resp = slurmdb.slurmdb_v0045_get_accounts(deleted="true")
assert resp.accounts
resp = slurmdb.slurmdb_v0045_get_accounts()
assert resp.accounts
accounts = V0045OpenapiAccountsResp(
accounts=[
V0045Account(
coordinators=[],
description="test description modified",
name=account2_name,
organization="test organization modified",
)
]
)
resp = slurmdb.slurmdb_v0045_delete_account(account_name)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_delete_account(account2_name)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_account(account_name)
assert len(resp.warnings) > 0
assert len(resp.errors) == 0
assert not resp.accounts
resp = slurmdb.slurmdb_v0045_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_v0045_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.v0045_openapi_wckey_resp import V0045OpenapiWckeyResp
from openapi_client.models.v0045_wckey import V0045Wckey
wckeys = V0045OpenapiWckeyResp(
wckeys=[
V0045Wckey(
cluster=local_cluster_name,
name=wckey_name,
user=user_name,
),
V0045Wckey(
cluster=local_cluster_name,
name=wckey2_name,
user=user_name,
),
V0045Wckey(
cluster=local_cluster_name,
name=wckey2_name,
user=coord_name,
),
]
)
resp = slurmdb.slurmdb_v0045_post_wckeys_with_http_info(
v0045_openapi_wckey_resp=wckeys
)
assert resp.status_code == 200
resp = slurmdb.slurmdb_v0045_get_wckeys()
assert not resp.warnings
assert len(resp.errors) == 0
assert len(resp.wckeys) >= 1
resp = slurmdb.slurmdb_v0045_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_v0045_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_v0045_delete_wckey(wckey_name)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_delete_wckey(wckey2_name)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_wckey(wckey_name)
assert len(resp.warnings) > 0
assert len(resp.errors) == 0
assert len(resp.wckeys) == 0
resp = slurmdb.slurmdb_v0045_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.v0045_cluster_rec import V0045ClusterRec
from openapi_client.models.v0045_openapi_clusters_resp import (
V0045OpenapiClustersResp,
)
clusters = V0045OpenapiClustersResp(
clusters=[
V0045ClusterRec(
name=cluster_name,
),
V0045ClusterRec(
name=cluster2_name,
),
]
)
resp = slurmdb.slurmdb_v0045_post_clusters(v0045_openapi_clusters_resp=clusters)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_clusters()
assert not resp.warnings
assert len(resp.errors) == 0
assert resp.clusters
resp = slurmdb.slurmdb_v0045_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_v0045_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_v0045_delete_cluster(cluster_name)
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_delete_cluster(cluster2_name)
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_cluster(cluster_name)
assert len(resp.warnings) > 0
assert len(resp.errors) == 0
assert not resp.clusters
resp = slurmdb.slurmdb_v0045_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.v0045_assoc_short import V0045AssocShort
from openapi_client.models.v0045_coord import V0045Coord
from openapi_client.models.v0045_openapi_users_resp import V0045OpenapiUsersResp
from openapi_client.models.v0045_user import V0045User
from openapi_client.models.v0045_user_default import V0045UserDefault
from openapi_client.models.v0045_wckey import V0045Wckey
users = V0045OpenapiUsersResp(
users=[
V0045User(
administrator_level=["None"],
default=dict(
wckey=wckey_name,
),
name=user_name,
),
V0045User(
administrator_level=["Operator"],
wckeys=[
V0045Wckey(
cluster=local_cluster_name, name=wckey_name, user=coord_name
),
V0045Wckey(
cluster=local_cluster_name,
name=wckey2_name,
user=coord_name,
),
],
default=V0045UserDefault(
wckey=wckey2_name,
),
name=coord_name,
),
]
)
resp = slurmdb.slurmdb_v0045_post_users(v0045_openapi_users_resp=users)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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_v0045_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_v0045_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_v0045_delete_user(coord_name)
assert len(resp.errors) == 0
user_exists = False
resp = slurmdb.slurmdb_v0045_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 = V0045OpenapiUsersResp(
users=[
V0045User(
administrator_level=["Administrator"],
default=dict(
wckey=wckey_name,
),
old_name=user_name,
name=coord_name,
)
]
)
resp = slurmdb.slurmdb_v0045_post_users(v0045_openapi_users_resp=users)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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_v0045_delete_user(coord_name)
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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.v0045_assoc import V0045Assoc
from openapi_client.models.v0045_assoc_short import V0045AssocShort
from openapi_client.models.v0045_coord import V0045Coord
from openapi_client.models.v0045_openapi_assocs_resp import V0045OpenapiAssocsResp
from openapi_client.models.v0045_uint32_no_val_struct import (
V0045Uint32NoValStruct as V0045Uint32NoVal,
)
from openapi_client.models.v0045_user import V0045User
from openapi_client.models.v0045_wckey import V0045Wckey
associations = V0045OpenapiAssocsResp(
associations=[
V0045Assoc(
account=account_name,
cluster=local_cluster_name,
default=dict(
qos=qos_name,
),
flags=[],
max=dict(
jobs=dict(
per=dict(
wall_clock=V0045Uint32NoVal(
set=True,
number=150,
)
),
),
),
min=dict(
priority_threshold=V0045Uint32NoVal(
set=True,
number=10,
)
),
partition=partition_name,
priority=V0045Uint32NoVal(number=9, set=True),
qos=[qos_name, qos2_name],
shares_raw=23,
user=user_name,
),
V0045Assoc(
account=account_name,
cluster=local_cluster_name,
default=dict(
qos=qos_name,
),
flags=[],
max=dict(
jobs=dict(
per=dict(
wall_clock=V0045Uint32NoVal(
set=True,
number=150,
)
),
),
),
min=dict(
priority_threshold=V0045Uint32NoVal(
set=True,
number=10,
)
),
priority=V0045Uint32NoVal(number=9, set=True),
qos=[qos_name, qos2_name],
shares_raw=23,
user=user_name,
),
V0045Assoc(
account=account2_name,
cluster=local_cluster_name,
default=dict(
qos=qos2_name,
),
flags=[],
max=dict(
jobs=dict(
per=dict(
wall_clock=V0045Uint32NoVal(
set=True,
number=50,
)
),
),
),
min=dict(
priority_threshold=V0045Uint32NoVal(
set=True,
number=4,
)
),
partition=partition_name,
priority=V0045Uint32NoVal(number=90, set=True),
qos=[qos2_name],
shares_raw=1012,
user=user_name,
),
]
)
resp = slurmdb.slurmdb_v0045_post_associations(
v0045_openapi_assocs_resp=associations
)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_associations()
assert not resp.warnings
assert len(resp.errors) == 0
assert resp.associations
resp = slurmdb.slurmdb_v0045_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 == 23
associations = V0045OpenapiAssocsResp(
associations=[
V0045Assoc(
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=V0045Uint32NoVal(set=True, number=250)),
),
),
min=dict(
priority_threshold=V0045Uint32NoVal(set=True, number=100),
),
priority=V0045Uint32NoVal(number=848, set=True),
shares_raw=230,
)
]
)
resp = slurmdb.slurmdb_v0045_post_associations(
v0045_openapi_assocs_resp=associations
)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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 == 230
resp = slurmdb.slurmdb_v0045_delete_association(
cluster=local_cluster_name,
account=account_name,
user=user_name,
partition=partition_name,
)
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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_v0045_delete_associations(
cluster=local_cluster_name,
user=user_name,
)
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_delete_associations(
cluster=local_cluster_name,
account=account_name,
)
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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_v0045_delete_associations(
cluster=local_cluster_name,
account=account2_name,
)
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_association(
cluster=local_cluster_name,
account=account2_name,
)
assert len(resp.warnings) > 0
assert len(resp.errors) == 0
assert not resp.associations
def test_db_qos(slurmdb, create_coords, admin_level):
from openapi_client.models.v0045_float64_no_val_struct import (
V0045Float64NoValStruct as V0045Float64NoVal,
)
from openapi_client.models.v0045_openapi_slurmdbd_qos_resp import (
V0045OpenapiSlurmdbdQosResp,
)
from openapi_client.models.v0045_qos import V0045Qos
from openapi_client.models.v0045_tres import V0045Tres
from openapi_client.models.v0045_uint32_no_val_struct import (
V0045Uint32NoValStruct as V0045Uint32NoVal,
)
qos = V0045OpenapiSlurmdbdQosResp(
qos=[
V0045Qos(
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=[
V0045Tres(
type="cpu",
count=100,
),
V0045Tres(
type="memory",
count=100000,
),
],
),
),
),
),
name=qos_name,
preempt=dict(
exempt_time=V0045Uint32NoVal(set=True, number=199),
),
priority=V0045Uint32NoVal(number=180, set=True),
usage_factor=V0045Float64NoVal(
set=True,
number=82382.23823,
),
usage_threshold=V0045Float64NoVal(
set=True,
number=929392.33,
),
),
V0045Qos(
description="test QOS 2",
name=qos2_name,
),
]
)
resp = slurmdb.slurmdb_v0045_post_qos(v0045_openapi_slurmdbd_qos_resp=qos)
assert not resp.warnings
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_qos()
assert not resp.warnings
assert len(resp.errors) == 0
assert resp.qos
resp = slurmdb.slurmdb_v0045_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_v0045_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_v0045_delete_single_qos(qos_name)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_get_single_qos(qos_name)
assert len(resp.warnings) > 0
assert len(resp.errors) == 0
assert not resp.qos
resp = slurmdb.slurmdb_v0045_delete_single_qos(qos2_name)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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_v0045_get_tres()
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
def test_db_config(slurmdb, admin_level):
resp = slurmdb.slurmdb_v0045_get_config()
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
def test_jobs(slurm, slurmdb, non_admin):
from openapi_client.models.v0045_job_comment import V0045JobComment
from openapi_client.models.v0045_job_desc_msg import V0045JobDescMsg
from openapi_client.models.v0045_job_info import V0045JobInfo
from openapi_client.models.v0045_job_modify import V0045JobModify
from openapi_client.models.v0045_job_modify_tres import V0045JobModifyTres
from openapi_client.models.v0045_job_submit_req import V0045JobSubmitReq
from openapi_client.models.v0045_openapi_job_modify_req import (
V0045OpenapiJobModifyReq,
)
from openapi_client.models.v0045_process_exit_code_verbose import (
V0045ProcessExitCodeVerbose,
)
from openapi_client.models.v0045_process_exit_code_verbose_signal import (
V0045ProcessExitCodeVerboseSignal,
)
from openapi_client.models.v0045_tres import V0045Tres
from openapi_client.models.v0045_uint32_no_val_struct import V0045Uint32NoValStruct
from openapi_client.models.v0045_uint32_no_val_struct import (
V0045Uint32NoValStruct as V0045Uint32NoVal,
)
script = "#!/bin/bash\n/bin/true"
env = ["PATH=/bin/:/sbin/:/usr/bin/:/usr/sbin/"]
job = V0045JobSubmitReq(
script=script,
job=V0045JobDescMsg(
partition=partition_name,
name="runjob",
environment=env,
current_working_directory="/tmp/",
),
)
resp = slurm.slurm_v0045_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_v0045_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_v0045_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 = V0045JobSubmitReq(
script=script,
job=V0045JobDescMsg(
partition=partition_name,
name="runjob",
environment=env,
current_working_directory="/tmp/",
),
)
resp = slurm.slurm_v0045_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_v0045_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 = V0045JobSubmitReq(
script=script,
job=V0045JobDescMsg(
partition=partition_name,
name="test job",
environment=env,
priority=V0045Uint32NoVal(number=0, set=True),
current_working_directory="/tmp/",
),
)
resp = slurm.slurm_v0045_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 = V0045JobDescMsg(
environment=env,
partition=partition_name,
name="updated test job",
priority=V0045Uint32NoVal(number=0, set=True),
)
resp = slurm.slurm_v0045_post_job(str(jobid3), v0045_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_v0045_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_v0045_delete_job(str(jobid3))
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
resp = slurm.slurm_v0045_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_v0045_get_jobs()
assert len(resp.errors) == 0
for t in atf.timer():
resp = slurmdb.slurmdb_v0045_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_v0045_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 = V0045Uint32NoValStruct(
set=True, infinite=False, number=4 # 1024 >> 8
)
job_modify = V0045JobModify(
comment=V0045JobComment(
administrator="admin1",
job="job1",
system="system1",
),
derived_exit_code=V0045ProcessExitCodeVerbose(
return_code=V0045Uint32NoValStruct(set=True, infinite=False, number=1024),
),
extra="extra1",
tres=V0045JobModifyTres(
allocated=[V0045Tres(type="energy", count=12345, id=3)]
),
wckey="mywckey1",
)
resp = slurmdb.slurmdb_v0045_post_job(str(jobid1), v0045_job_modify=job_modify)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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 = V0045Uint32NoValStruct(
set=True, infinite=False, number=8 # 2048 >> 8
)
job_modify = V0045JobModify(
comment=V0045JobComment(
administrator="admin2",
job="job2",
system="system2",
),
derived_exit_code=V0045ProcessExitCodeVerbose(
return_code=V0045Uint32NoValStruct(set=True, infinite=False, number=2048),
),
extra="extra2",
tres=V0045JobModifyTres(
allocated=[V0045Tres(type="energy", count=54321, id=3)]
),
wckey="mywckey2",
)
job_modify_req = V0045OpenapiJobModifyReq(
job_id_list=[str(jobid1), str(jobid2)], job_rec=job_modify
)
resp = slurmdb.slurmdb_v0045_post_jobs(v0045_openapi_job_modify_req=job_modify_req)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
resp = slurmdb.slurmdb_v0045_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
def _assert_invalid_account_rejection(exc):
"""Check a rejected job submission against the documented response."""
resp = json.loads(exc.body)
assert resp["errors"], f"Expected errors in the response, got: {resp}"
assert any(
"invalid account" in err["error"].lower() for err in resp["errors"]
), f"Expected an Invalid account error, got: {resp['errors']}"
assert resp.get("job_id") is None, f"Rejected job got a job_id: {resp}"
def test_jobs_associated_account(
slurm, association_enforcement, create_local_user_assoc
):
"""Test that job submission with an associated account is accepted.
Verifies that when AccountingStorageEnforce=associations is set,
submitting a job with an account the user is associated with succeeds.
"""
from openapi_client.models.v0045_job_desc_msg import V0045JobDescMsg
from openapi_client.models.v0045_job_submit_req import V0045JobSubmitReq
script = "#!/bin/bash\n/bin/true"
env = ["PATH=/bin/:/sbin/:/usr/bin/:/usr/sbin/"]
job = V0045JobSubmitReq(
script=script,
job=V0045JobDescMsg(
account=account_name,
partition=partition_name,
name="test-associated-account",
environment=env,
current_working_directory="/tmp/",
),
)
resp = slurm.slurm_v0045_post_job_submit(job)
assert len(resp.errors) == 0, f"Expected no errors, got: {resp.errors}"
assert resp.job_id, f"Expected a job_id for an associated account: {resp}"
# The account and association can't be removed while the job is active.
atf.wait_for_job_accounted(resp.job_id, "End", r"^\d", fatal=True)
@pytest.mark.parametrize(
"invalid_account",
["totally_invalid_account_name", account2_name],
ids=["nonexistent", "not_associated"],
)
def test_jobs_invalid_account(
slurm, association_enforcement, create_local_user_assoc, invalid_account
):
"""Test that job submission with invalid account is rejected.
Verifies that when AccountingStorageEnforce=associations is set,
submitting a job with an account that does not exist, or that exists
but the user is not associated with, is properly rejected.
"""
from openapi_client.exceptions import ApiException
from openapi_client.models.v0045_job_desc_msg import V0045JobDescMsg
from openapi_client.models.v0045_job_submit_req import V0045JobSubmitReq
script = "#!/bin/bash\n/bin/true"
env = ["PATH=/bin/:/sbin/:/usr/bin/:/usr/sbin/"]
# Test that job submission with invalid account is rejected
invalid_job = V0045JobSubmitReq(
script=script,
job=V0045JobDescMsg(
account=invalid_account,
partition=partition_name,
name="test-invalid-account",
environment=env,
current_working_directory="/tmp/",
),
)
with pytest.raises(ApiException) as exc_info:
slurm.slurm_v0045_post_job_submit(invalid_job)
_assert_invalid_account_rejection(exc_info.value)
def test_jobs_missing_user_association(
slurm, association_enforcement, no_local_user_assoc
):
"""Test that job submission by user with no associations is rejected.
Verifies that when AccountingStorageEnforce=associations is set,
a user without any associations cannot submit jobs.
"""
from openapi_client.exceptions import ApiException
from openapi_client.models.v0045_job_desc_msg import V0045JobDescMsg
from openapi_client.models.v0045_job_submit_req import V0045JobSubmitReq
script = "#!/bin/bash\n/bin/true"
env = ["PATH=/bin/:/sbin/:/usr/bin/:/usr/sbin/"]
# Test that job submission without user associations is rejected
# Don't specify account or user - job will run as local_user_name with no associations
job = V0045JobSubmitReq(
script=script,
job=V0045JobDescMsg(
partition=partition_name,
name="test-no-association",
environment=env,
current_working_directory="/tmp/",
),
)
# A user with no default account/association is rejected the same
# way as an invalid account (ESLURM_INVALID_ACCOUNT).
with pytest.raises(ApiException) as exc_info:
slurm.slurm_v0045_post_job_submit(job)
_assert_invalid_account_rejection(exc_info.value)
@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_v0045_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_v0045_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.v0045_openapi_partitions_mod_req import (
V0045OpenapiPartitionsModReq,
)
get_resp = slurm.slurm_v0045_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 = V0045OpenapiPartitionsModReq(partitions=get_resp.partitions)
post_resp = slurm.slurm_v0045_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.v0045_openapi_partitions_mod_req import (
V0045OpenapiPartitionsModReq,
)
from openapi_client.models.v0045_partition_info import V0045PartitionInfo
from openapi_client.models.v0045_partition_info_defaults import (
V0045PartitionInfoDefaults,
)
from openapi_client.models.v0045_partition_info_nodes import V0045PartitionInfoNodes
from openapi_client.models.v0045_partition_info_partition import (
V0045PartitionInfoPartition,
)
from openapi_client.models.v0045_uint64_no_val_struct import V0045Uint64NoValStruct
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 = V0045OpenapiPartitionsModReq(
partitions=[
V0045PartitionInfo(
name=part_1_name,
nodes=V0045PartitionInfoNodes(configured=part_1_nodes),
flags=[
"HIDDEN",
"NO_ROOT",
"ROOT_ONLY",
"REQ_RESV",
"LLN",
"PDOI",
],
preempt_mode=[
"REQUEUE",
],
defaults=V0045PartitionInfoDefaults(
partition_memory_per_cpu=V0045Uint64NoValStruct(
set=True,
number=10,
),
),
partition=V0045PartitionInfoPartition(
oversubscribe="FORCE:5", exclusive="USER"
),
),
V0045PartitionInfo(
name=part_2_name,
nodes=V0045PartitionInfoNodes(configured=part_2_nodes),
flags=[
"LLN",
],
preempt_mode=["DISABLED"],
defaults=V0045PartitionInfoDefaults(
partition_memory_per_node=V0045Uint64NoValStruct(
set=True,
number=12,
)
),
partition=V0045PartitionInfoPartition(
oversubscribe="YES:6", exclusive="NO"
),
),
]
)
# Create partitions part_1 and part_2
resp = slurm.slurm_v0045_post_partitions(req)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
# Check that they were created
resp = slurm.slurm_v0045_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_v0045_post_partitions(req)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
# Check that they were updated
resp = slurm.slurm_v0045_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_v0045_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.v0045_update_node_msg import V0045UpdateNodeMsg
node_name = None
reasonuid = None
resp = slurm.slurm_v0045_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 = V0045UpdateNodeMsg(
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_v0045_post_node(node_name, v0045_update_node_msg=node)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
resp = slurm.slurm_v0045_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 = V0045UpdateNodeMsg(
comment=ncomment,
extra=extra,
features=feat,
features_act=actfeat,
state=["RESUME"],
reason=reason,
reason_uid=reasonuid,
)
resp = slurm.slurm_v0045_post_node(node_name, node)
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
resp = slurm.slurm_v0045_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_v0045_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.45/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.45/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_v0045_get_ping()
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
def test_diag(slurm):
resp = slurm.slurm_v0045_get_diag()
assert len(resp.warnings) == 0
assert len(resp.errors) == 0
assert resp.statistics
def test_licenses(slurm):
resp = slurm.slurm_v0045_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.v0045_reservation_desc_msg import V0045ReservationDescMsg
from openapi_client.models.v0045_reservation_mod_req import V0045ReservationModReq
from openapi_client.models.v0045_uint32_no_val_struct import V0045Uint32NoValStruct
from openapi_client.models.v0045_uint64_no_val_struct import V0045Uint64NoValStruct
resv_name = "test_resv"
users = ["root", "atf"]
duration = V0045Uint32NoValStruct(number=300, set=True)
start_time = V0045Uint64NoValStruct(number=int(time.time()) + 60, set=True)
end_time = V0045Uint64NoValStruct(
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 = V0045ReservationDescMsg(
name=resv_name,
users=users,
duration=duration,
start_time=start_time,
node_list=node_list,
partition=partition,
flags=flags,
)
resp = slurm.slurm_v0045_post_reservations(
V0045ReservationModReq(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_v0045_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 = V0045Uint64NoValStruct(number=int(time.time()) + 90, set=True)
new_flags = ["IGNORE_JOBS", "MAGNETIC", "USER_DELETE"]
update_info = V0045ReservationDescMsg(
name=resv_name,
end_time=V0045Uint64NoValStruct(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_v0045_post_reservation(update_info)
assert resp.reservations, "Reservation should be updated"
assert not resp.warnings and not resp.errors
# Validate update
resp = slurm.slurm_v0045_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_v0045_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_v0045_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.v0045_openapi_create_node_req import (
V0045OpenapiCreateNodeReq,
)
request_body = V0045OpenapiCreateNodeReq(
node_conf=f"nodename={dynamic_node} State={legal_state}"
)
response = slurm.slurm_v0045_post_new_node(
v0045_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.v0045_openapi_create_node_req import (
V0045OpenapiCreateNodeReq,
)
request_body = V0045OpenapiCreateNodeReq(
node_conf=f"nodename={dynamic_node} State={illegal_state}"
)
try:
# This should throw an exception
slurm.slurm_v0045_post_new_node_with_http_info(
v0045_openapi_create_node_req=request_body
)
assert (
False
), "slurm_v0045_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="V0045OpenapiResp"
)
# 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.v0045_openapi_hostlist_req_resp import (
V0045OpenapiHostlistReqResp,
)
hostnames = ["node01", "node02", "node03"]
request_body = V0045OpenapiHostlistReqResp(hostlist="node[01-03]")
response = util_api.util_v0045_post_hostnames(
v0045_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.v0045_openapi_hostnames_req_resp import (
V0045OpenapiHostnamesReqResp,
)
request_body = V0045OpenapiHostnamesReqResp(
hostnames=["node01", "node02", "node03"]
)
response = util_api.util_v0045_post_hostlist(
v0045_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, resv_crash):
"""Check for xfree crash (bug 23038)"""
from openapi_client.models.v0045_reservation_desc_msg import V0045ReservationDescMsg
from openapi_client.models.v0045_reservation_mod_req import V0045ReservationModReq
from openapi_client.models.v0045_uint32_no_val_struct import V0045Uint32NoValStruct
from openapi_client.models.v0045_uint64_no_val_struct import V0045Uint64NoValStruct
# Don't overlap with other resv in case of restd crash/restart
resv_name = resv_crash
users = ["root", "atf"]
duration = V0045Uint32NoValStruct(number=1, set=True)
start_time = V0045Uint64NoValStruct(
number=int(time.time()) + 10000000,
set=True,
)
partition = "primary"
# Create a reservation with empty node_list
reservation_info = V0045ReservationDescMsg(
name=resv_name,
users=users,
duration=duration,
start_time=start_time,
partition=partition,
node_list=[],
)
# Exception means restd crashed
resp = slurm.slurm_v0045_post_reservations(
V0045ReservationModReq(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.45/conf returns a well-formed slurm.conf payload."""
resp = slurm.slurm_v0045_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_v0045_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_v0045_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.45/conf."""
r = requests.post(
f"{atf.properties['slurmrestd_url']}/slurm/v0.0.45/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.45` emits a well-formed slurm_conf."""
blob = json.loads(
atf.run_command_output("scontrol show config --json=v0.0.45", 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.45` produces the same shape as --json."""
blob = yaml.safe_load(
atf.run_command_output("scontrol show config --yaml=v0.0.45", 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.45" in out, f"v0.0.45 not listed in: {out}"
def test_slurm_conf_cli_matches_slurmrestd():
"""scontrol --json=v0.0.45 and /slurm/v0.0.45/conf yield equivalent payloads."""
cli = json.loads(
atf.run_command_output("scontrol show config --json=v0.0.45", fatal=True)
)["slurm_conf"]
rest = atf.request_slurmrestd("slurm/v0.0.45/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.45/conf returns a well-formed slurmdbd.conf payload."""
resp = slurmdb.slurmdb_v0045_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.45/conf."""
r = requests.post(
f"{atf.properties['slurmrestd_url']}/slurmdb/v0.0.45/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.45` emits a well-formed slurmdbd_conf."""
blob = json.loads(
atf.run_command_output("sacctmgr show config --json=v0.0.45", 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.45` produces the same shape as --json."""
blob = yaml.safe_load(
atf.run_command_output("sacctmgr show config --yaml=v0.0.45", 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.45" in out, f"v0.0.45 not listed in: {out}"
def test_slurmdbd_conf_cli_matches_slurmrestd():
"""sacctmgr --json=v0.0.45 and /slurmdb/v0.0.45/conf yield equivalent payloads."""
cli = json.loads(
atf.run_command_output("sacctmgr show config --json=v0.0.45", fatal=True)
)["slurmdbd_conf"]
rest = atf.request_slurmrestd("slurmdb/v0.0.45/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"
def _assert_job_requeue(jobid, expected_state, expected_reason, held):
for t in atf.timer(fatal=True):
if int(atf.get_job_parameter(jobid, "Restarts", quiet=True)) > 0:
break
atf.wait_for_job_state(
jobid,
expected_state,
desired_reason=expected_reason,
fatal=True,
)
if held:
assert (
atf.get_job_parameter(jobid, "Priority", quiet=True) == 0
), "Held job should have priority 0"
def _get_running_array_tasks(jobid, task_count):
atf.wait_for_job_state(jobid, "RUNNING", fatal=True)
task_ids = list(atf.get_jobs(job_id=jobid, quiet=True))
assert len(task_ids) == task_count, f"Array job should have {task_count} tasks"
for task_id in task_ids:
atf.wait_for_job_state(task_id, "RUNNING", fatal=True)
return task_ids
@pytest.mark.parametrize(
"query_params,expected_state,expected_reason",
[
pytest.param({}, "PENDING", "BeginTime", id="no_flags"),
pytest.param({"incomplete": "True"}, "PENDING", "BeginTime", id="incomplete"),
pytest.param(
{"hold": "True"}, "PENDING", "job_requeued_in_held_state", id="hold"
),
pytest.param(
{"hold": "True", "special_exit": "True"},
"SPECIAL_EXIT",
"job_requeued_in_held_state",
id="special_exit",
),
],
)
def test_job_requeue(slurm, query_params, expected_state, expected_reason, non_admin):
"""Test single job requeue endpoint"""
jobid = atf.submit_job_sbatch("--requeue --wrap='sleep infinity' -a1-2", fatal=True)
task_ids = _get_running_array_tasks(jobid, 2)
# Requeue it
resp = slurm.slurm_v0045_get_job_requeue(str(jobid), **query_params)
assert not resp.warnings, "Requeue response should have no warnings"
assert not resp.errors, "Requeue response should have no errors"
# Verify that the response has a status since it is an array job
assert len(resp.status) == 1, "Array job requeue should populate status"
assert resp.status[0].job_id == jobid, "Status job_id should match requeued job"
assert resp.status[0].error_code == 0, "Requeue status should have no error"
# Make sure every task of the array requeued
for task_id in task_ids:
_assert_job_requeue(
task_id, expected_state, expected_reason, "hold" in query_params
)
def test_job_requeue_incomplete(slurm, non_admin):
"""The Incomplete flag skips jobs that have already completed"""
jobid = atf.submit_job_sbatch("--requeue --wrap='exit 0'", fatal=True)
atf.wait_for_job_state(jobid, "COMPLETED", fatal=True)
# With the Incomplete flag the completed job is left untouched
resp = slurm.slurm_v0045_get_job_requeue(str(jobid), incomplete="True")
assert not resp.errors, "Requeue response should have no errors"
assert (
atf.get_job_parameter(jobid, "JobState", quiet=True) == "COMPLETED"
), "Incomplete flag should not requeue a completed job"
assert (
int(atf.get_job_parameter(jobid, "Restarts", quiet=True)) == 0
), "Incomplete flag should not requeue a completed job"
# Without the flag the same completed job is requeued
resp = slurm.slurm_v0045_get_job_requeue(str(jobid))
assert not resp.errors, "Requeue response should have no errors"
_assert_job_requeue(jobid, "PENDING", "BeginTime", False)
@pytest.mark.parametrize(
"requeue_flags,expected_state,expected_reason,held",
[
pytest.param(None, "PENDING", "BeginTime", False, id="no_flags"),
pytest.param(
["Hold"], "PENDING", "job_requeued_in_held_state", True, id="hold"
),
pytest.param(
["Hold", "SpecialExit"],
"SPECIAL_EXIT",
"job_requeued_in_held_state",
True,
id="special_exit",
),
],
)
def test_jobs_requeue(
slurm, requeue_flags, expected_state, expected_reason, held, non_admin
):
"""Test batch jobs requeue endpoint"""
from openapi_client.models.v0045_openapi_jobs_requeue_query import (
V0045OpenapiJobsRequeueQuery,
)
jobids = []
jobids.append(
atf.submit_job_sbatch("--requeue --wrap='sleep infinity' -a1-2", fatal=True)
)
jobids.append(
atf.submit_job_sbatch("--requeue --wrap='sleep infinity' -a1-2", fatal=True)
)
# Wait for every task of both arrays to go from PENDING to RUNNING
task_ids = []
for jobid in jobids:
task_ids.extend(_get_running_array_tasks(jobid, 2))
# Batch requeue both jobs
requeue_params = {
"jobs": [str(jid) for jid in jobids],
}
if requeue_flags:
requeue_params["flags"] = requeue_flags
requeue_req = V0045OpenapiJobsRequeueQuery(**requeue_params)
resp = slurm.slurm_v0045_post_jobs_requeue(
v0045_openapi_jobs_requeue_query=requeue_req,
)
assert not resp.warnings, "Requeue response should have no warnings"
assert not resp.errors, "Requeue response should have no errors"
# Verify that the response has a status since these are array jobs
assert len(resp.status) == 2, "Each requeued array job should have a status"
for i in range(len(jobids)):
assert len(resp.status[i]) == 1, "Array job status should have one entry"
assert resp.status[i][0].job_id == jobids[i], "Status job_id should match"
assert resp.status[i][0].error_code == 0, "Requeue status should have no error"
# Make sure every task of both arrays requeued
for task_id in task_ids:
_assert_job_requeue(task_id, expected_state, expected_reason, held)
def test_jobs_requeue_incomplete(slurm, non_admin):
"""The batch Incomplete flag skips only the jobs that have completed"""
from openapi_client.models.v0045_openapi_jobs_requeue_query import (
V0045OpenapiJobsRequeueQuery,
)
done_id = atf.submit_job_sbatch("--requeue --wrap='exit 0'", fatal=True)
atf.wait_for_job_state(done_id, "COMPLETED", fatal=True)
running_id = atf.submit_job_sbatch("--requeue --wrap='sleep infinity'", fatal=True)
atf.wait_for_job_state(running_id, "RUNNING", fatal=True)
requeue_req = V0045OpenapiJobsRequeueQuery(
jobs=[str(done_id), str(running_id)], flags=["Incomplete"]
)
resp = slurm.slurm_v0045_post_jobs_requeue(
v0045_openapi_jobs_requeue_query=requeue_req,
)
assert not resp.warnings, "Requeue response should have no warnings"
assert not resp.errors, "Requeue response should have no errors"
assert (
atf.get_job_parameter(done_id, "JobState", quiet=True) == "COMPLETED"
), "Incomplete flag should not requeue a completed job"
assert (
int(atf.get_job_parameter(done_id, "Restarts", quiet=True)) == 0
), "Incomplete flag should not requeue a completed job"
_assert_job_requeue(running_id, "PENDING", "BeginTime", False)
def test_job_requeue_special_exit_requires_hold(slurm, non_admin):
"""SpecialExit without Hold is rejected by both requeue endpoints"""
from openapi_client.api_response import ApiResponse
from openapi_client.exceptions import ApiException
from openapi_client.models.v0045_openapi_jobs_requeue_query import (
V0045OpenapiJobsRequeueQuery,
)
jobid = atf.submit_job_sbatch("--requeue --wrap='sleep infinity'", fatal=True)
atf.wait_for_job_state(jobid, "RUNNING", fatal=True)
# GET single job endpoint rejects SpecialExit without Hold
try:
slurm.slurm_v0045_get_job_requeue_with_http_info(
str(jobid), special_exit="True"
)
assert False, "SpecialExit without Hold should raise an exception"
except ApiException as e:
response = slurm.api_client.deserialize(
response=ApiResponse(data=e.body), response_type="V0045OpenapiResp"
)
assert e.status != 200, "Rejected request should not return HTTP 200"
assert len(response.errors) == 1, "Rejected request should report one error"
assert not response.warnings, "Rejected request should have no warnings"
# POST batch endpoint rejects SpecialExit without Hold
requeue_req = V0045OpenapiJobsRequeueQuery(jobs=[str(jobid)], flags=["SpecialExit"])
try:
slurm.slurm_v0045_post_jobs_requeue_with_http_info(
v0045_openapi_jobs_requeue_query=requeue_req
)
assert False, "SpecialExit without Hold should raise an exception"
except ApiException as e:
response = slurm.api_client.deserialize(
response=ApiResponse(data=e.body), response_type="V0045OpenapiResp"
)
assert e.status != 200, "Rejected request should not return HTTP 200"
assert len(response.errors) == 1, "Rejected request should report one error"
assert not response.warnings, "Rejected request should have no warnings"
# The rejected requests did not requeue or hold the job
assert (
atf.get_job_parameter(jobid, "JobState", quiet=True) == "RUNNING"
), "Rejected requeue should leave the job running"
def test_job_requeue_invalid_job(slurm):
"""Requeuing a nonexistent job reports an error"""
from openapi_client.api_response import ApiResponse
from openapi_client.exceptions import ApiException
from openapi_client.models.v0045_openapi_jobs_requeue_query import (
V0045OpenapiJobsRequeueQuery,
)
# A job id above MaxJobId is never assigned, so it cannot exist
bad_jobid = int(atf.get_config_parameter("MaxJobId")) + 1
try:
slurm.slurm_v0045_get_job_requeue_with_http_info(str(bad_jobid))
assert False, "Requeuing a nonexistent job should raise an exception"
except ApiException as e:
response = slurm.api_client.deserialize(
response=ApiResponse(data=e.body), response_type="V0045OpenapiResp"
)
assert e.status != 200, "Nonexistent job requeue should not return HTTP 200"
assert response.errors, "Nonexistent job requeue should report an error"
requeue_req = V0045OpenapiJobsRequeueQuery(jobs=[str(bad_jobid)])
try:
slurm.slurm_v0045_post_jobs_requeue_with_http_info(
v0045_openapi_jobs_requeue_query=requeue_req
)
assert False, "Requeuing a nonexistent job should raise an exception"
except ApiException as e:
response = slurm.api_client.deserialize(
response=ApiResponse(data=e.body), response_type="V0045OpenapiResp"
)
assert e.status != 200, "Nonexistent job requeue should not return HTTP 200"
assert response.errors, "Nonexistent job requeue should report an error"
@pytest.mark.skipif(
"daemon" not in {p.pw_name for p in pwd.getpwall()},
reason="Issue 50672: nobody is reserved, so a local daemon user is needed",
)
def test_job_requeue_permission_denied(non_admin):
"""A denied requeue reports a per-job error in the status array"""
import openapi_client
from openapi_client.models.v0045_openapi_jobs_requeue_query import (
V0045OpenapiJobsRequeueQuery,
)
# Slurm reserves "nobody" and rejects its tokens outright
non_owner = "daemon"
jobid = atf.submit_job_sbatch("--requeue --wrap='sleep infinity' -a1-2", fatal=True)
atf.wait_for_job_state(jobid, "RUNNING", fatal=True)
# Mint a token for an unprivileged user that does not own the job
token = (
atf.run_command_output(
f"scontrol token username={non_owner}",
user=atf.properties["slurm-user"],
fatal=True,
quiet=True,
)
.strip()
.split("=", 1)[1]
.strip()
)
config = openapi_client.Configuration()
config.host = atf.properties["slurmrestd_url"]
config.access_token = token
other = openapi_client.SlurmApi(openapi_client.ApiClient(config))
# A user that does not own the job cannot requeue it, so the denial comes
# back as a per-job error in the status array, not a top-level error
resp = other.slurm_v0045_get_job_requeue(str(jobid))
assert not resp.errors, "Per-job denial should not report a top-level error"
assert len(resp.status) == 1, "Denied array requeue should report one status"
assert (
resp.status[0].error_code == 2002
), "Denied requeue should report ESLURM_ACCESS_DENIED"
requeue_req = V0045OpenapiJobsRequeueQuery(jobs=[str(jobid)])
resp = other.slurm_v0045_post_jobs_requeue(
v0045_openapi_jobs_requeue_query=requeue_req,
)
assert not resp.errors, "Per-job denial should not report a top-level error"
assert len(resp.status) == 1, "Batch requeue should report one status per job"
assert len(resp.status[0]) == 1, "Denied array requeue should report one status"
assert (
resp.status[0][0].error_code == 2002
), "Denied requeue should report ESLURM_ACCESS_DENIED"
# The job was left running, not requeued
assert (
int(atf.get_job_parameter(jobid, "Restarts", quiet=True)) == 0
), "Denied requeue should not requeue the job"
def test_job_requeue_single_task(slurm, non_admin):
"""Both requeue endpoints requeue a single array task by its #_# id"""
from openapi_client.models.v0045_openapi_jobs_requeue_query import (
V0045OpenapiJobsRequeueQuery,
)
jobid = atf.submit_job_sbatch("--requeue --wrap='sleep infinity' -a1-3", fatal=True)
_get_running_array_tasks(jobid, 3)
# Resolve the raw job ids of every task (stable across the requeue)
task_1, task_2, task_3 = [
next(iter(atf.get_jobs(job_id=f"{jobid}_{i}", quiet=True))) for i in (1, 2, 3)
]
# Hold so each requeued task stays PENDING for a deterministic check
resp = slurm.slurm_v0045_get_job_requeue(f"{jobid}_1", hold="True")
assert not resp.errors, "Requeue response should have no errors"
_assert_job_requeue(task_1, "PENDING", "job_requeued_in_held_state", True)
for task_id in (task_2, task_3):
assert (
atf.get_job_parameter(task_id, "JobState", quiet=True) == "RUNNING"
), "The other array tasks should keep running"
requeue_req = V0045OpenapiJobsRequeueQuery(jobs=[f"{jobid}_2"], flags=["Hold"])
resp = slurm.slurm_v0045_post_jobs_requeue(
v0045_openapi_jobs_requeue_query=requeue_req,
)
assert not resp.errors, "Requeue response should have no errors"
_assert_job_requeue(task_2, "PENDING", "job_requeued_in_held_state", True)
assert (
atf.get_job_parameter(task_3, "JobState", quiet=True) == "RUNNING"
), "The other array task should keep running"
def test_job_requeue_non_array(slurm, non_admin):
"""A non-array job requeues with an empty status list"""
jobid = atf.submit_job_sbatch("--requeue --wrap='sleep infinity'", fatal=True)
atf.wait_for_job_state(jobid, "RUNNING", fatal=True)
resp = slurm.slurm_v0045_get_job_requeue(str(jobid))
assert not resp.errors, "Requeue response should have no errors"
# A non-array job does not populate the status list
assert len(resp.status) == 0, "Non-array requeue should have an empty status"
_assert_job_requeue(jobid, "PENDING", "BeginTime", False)
def test_job_requeue_not_requeueable(slurm, non_admin):
"""Neither requeue endpoint requeues a job submitted with --no-requeue"""
from openapi_client.models.v0045_openapi_jobs_requeue_query import (
V0045OpenapiJobsRequeueQuery,
)
jobid = atf.submit_job_sbatch(
"--no-requeue --wrap='sleep infinity' -a1-2", fatal=True
)
task_ids = _get_running_array_tasks(jobid, 2)
resp = slurm.slurm_v0045_get_job_requeue(str(jobid))
assert not resp.errors, "Per-job refusal should not report a top-level error"
assert len(resp.status) == 1, "Refused array requeue should report one status"
assert (
resp.status[0].error_code == 2037
), "Refused requeue should report ESLURM_DISABLED"
requeue_req = V0045OpenapiJobsRequeueQuery(jobs=[str(jobid)])
resp = slurm.slurm_v0045_post_jobs_requeue(
v0045_openapi_jobs_requeue_query=requeue_req,
)
assert not resp.errors, "Per-job refusal should not report a top-level error"
assert len(resp.status) == 1, "Batch requeue should report one status per job"
assert len(resp.status[0]) == 1, "Refused array requeue should report one status"
assert (
resp.status[0][0].error_code == 2037
), "Refused requeue should report ESLURM_DISABLED"
for task_id in task_ids:
assert (
int(atf.get_job_parameter(task_id, "Restarts", quiet=True)) == 0
), "A --no-requeue job should never be requeued"
assert (
atf.get_job_parameter(task_id, "JobState", quiet=True) == "RUNNING"
), "A --no-requeue job should keep running"
def test_job_stdio_expanded(slurmdb, create_local_user_assoc):
"""Verify the slurmdb job record exposes raw and expanded stdio paths."""
# --expand-patterns is a text-only flag for sacct/squeue; the data parser
# emits the raw and the *_expanded fields unconditionally.
directory = os.getcwd()
name = "expandpat"
stdin_pattern = f"{directory}/restd_%%.in"
atf.run_command(f"touch '{directory}/restd_%.in'", fatal=True)
job_id = atf.submit_job_sbatch(
f"-J {name} -A {account_name} --output={directory}/restd_%j.out"
f" --error={directory}/restd_%x.err --input={stdin_pattern}"
f" --wrap 'true'",
fatal=True,
)
# The stdio fields are set at submit time, so the job need not run: waiting
# for the record to reach the DB is enough, and keeps this test independent
# of whatever reservations the rest of the module has created.
atf.wait_for_job_accounted(job_id, fatal=True)
resp = slurmdb.slurmdb_v0045_get_job(str(job_id))
assert not resp.errors, "Getting the job record should not return errors"
assert len(resp.jobs) == 1, f"Expected exactly one record for job {job_id}"
job = resp.jobs[0]
assert job.stdout == f"{directory}/restd_%j.out", "stdout should be unexpanded"
assert job.stderr == f"{directory}/restd_%x.err", "stderr should be unexpanded"
assert job.stdin == stdin_pattern, "stdin should be unexpanded"
assert (
job.stdout_expanded == f"{directory}/restd_{job_id}.out"
), "stdout_expanded should replace %j with the job id"
assert (
job.stderr_expanded == f"{directory}/restd_{name}.err"
), "stderr_expanded should replace %x with the job name"
assert (
job.stdin_expanded == f"{directory}/restd_%.in"
), "stdin_expanded should replace %% with a single percent"
# The account and association can't be removed while the job is active.
atf.run_command(f"scancel {job_id}", fatal=True)
atf.wait_for_job_accounted(job_id, "End", r"^\d", fatal=True)