testsuite/python/tests/test_112_1.py

############################################################################
# Copyright (C) SchedMD LLC.
############################################################################
import pytest

import atf

# import re


@pytest.fixture(scope="module", autouse=True)
def setup():
    atf.require_accounting()
    atf.require_slurm_running()


def test_help():
    """Verify -h is a valid option"""

    assert atf.run_command_exit("slurmrestd -h") == 0


def test_invalid_option():
    """Verify failure for invalid option"""

    assert atf.run_command_exit("slurmrestd --invalid") != 0


def test_a():
    """Check -a option"""

    assert atf.run_command_exit("slurmrestd -a list") == 0
    assert atf.run_command_exit("slurmrestd -a invalid_plugin") != 0


def test_s():
    """Check -s option"""

    assert atf.run_command_exit("slurmrestd -s list") == 0
    assert atf.run_command_exit("slurmrestd -s invalid_plugin") != 0


def test_d():
    """Check -d option"""

    assert atf.run_command_exit("slurmrestd -d list") == 0
    assert atf.run_command_exit("slurmrestd -d invalid_plugin") != 0


@pytest.mark.xfail(
    atf.get_version("sbin/slurmrestd") < (26, 5, 5),
    reason="Issue 50192: repeated -d data_parser entries abort slurmrestd. Fixed in 26.05.5",
)
def test_d_repeated_plugin():
    """Verify a -d list naming one plugin more than once is accepted

    LATEST_PLUGIN_NAME resolves to an explicit version, so "latest,latest"
    asks for one data_parser twice. Only the version reaches a {data_parser}
    URL, so the repeat cannot be served and must be dropped rather than built
    a second time.
    """

    assert (
        atf.run_command_exit(
            "slurmrestd -a rest_auth/local -d latest,latest",
            input="GET /openapi HTTP/1.1\r\nConnection: Close\r\n\r\n",
        )
        == 0
    )


def test_get_a_rest_auth():
    """Check response to GET with -a rest_auth"""

    assert (
        atf.run_command_exit(
            "slurmrestd -a rest_auth/local",
            input="GET /openapi HTTP/1.1\r\nConnection: Close\r\n\r\n",
        )
        == 0
    )
    assert (
        atf.run_command_exit(
            "slurmrestd -a rest_auth/invalid",
            input="GET /openapi HTTP/1.1\r\nConnection: Close\r\n\r\n",
        )
        != 0
    )


# Subtests related to bug 10388


def test_invalid_input():
    """Check response to invalid input (bug 10388)"""

    assert atf.run_command_exit("slurmrestd", input="INVALID") != 0
    assert atf.run_command_exit("slurmrestd", input="INVALID\r\n\r\n") != 0
    assert atf.run_command_exit("slurmrestd", input="INVALID\r\nINVALID\r\n\r\n") != 0
    assert (
        atf.run_command_exit(
            "slurmrestd", input="GET /openapi HTTP/1.1\r\nINVALID\r\n\r\n"
        )
        != 0
    )