ClickHouse/tests/integration/test_settings_constraints/test.py

import pytest
from helpers.cluster import ClickHouseCluster

cluster = ClickHouseCluster(__file__)
instance = cluster.add_instance('instance', user_configs=["configs/users.xml"])


@pytest.fixture(scope="module")
def started_cluster():
    try:
        cluster.start()
        yield cluster

    finally:
        cluster.shutdown()


def test_system_settings(started_cluster):
    assert instance.query(
        "SELECT name, value, min, max, readonly from system.settings WHERE name = 'force_index_by_date'") == \
           "force_index_by_date\t0\t\\N\t\\N\t1\n"

    assert instance.query(
        "SELECT name, value, min, max, readonly from system.settings WHERE name = 'max_memory_usage'") == \
           "max_memory_usage\t10000000000\t5000000000\t20000000000\t0\n"

    assert instance.query("SELECT name, value, min, max, readonly from system.settings WHERE name = 'readonly'") == \
           "readonly\t0\t\\N\t\\N\t0\n"


def test_system_constraints(started_cluster):
    assert_query_settings(instance, "SELECT 1",
                          settings={'readonly': 0},
                          exception="Cannot modify 'readonly'",
                          user="readonly_user")

    assert_query_settings(instance, "SELECT 1",
                          settings={'allow_ddl': 1},
                          exception="Cannot modify 'allow_ddl'",
                          user="no_dll_user")


def test_read_only_constraint(started_cluster):
    # Default value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='force_index_by_date'",
                          settings={},
                          result="0")

    # Invalid value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='force_index_by_date'",
                          settings={'force_index_by_date': 1},
                          result=None,
                          exception="Setting force_index_by_date should not be changed")


def test_min_constraint(started_cluster):
    # Default value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='max_memory_usage'",
                          {},
                          result="10000000000")

    # Valid value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='max_memory_usage'",
                          settings={'max_memory_usage': 5000000000},
                          result="5000000000")

    # Invalid value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='max_memory_usage'",
                          settings={'max_memory_usage': 4999999999},
                          result=None,
                          exception="Setting max_memory_usage shouldn't be less than 5000000000")


def test_max_constraint(started_cluster):
    # Default value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='max_memory_usage'",
                          {},
                          result="10000000000")

    # Valid value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='max_memory_usage'",
                          settings={'max_memory_usage': 20000000000},
                          result="20000000000")

    # Invalid value
    assert_query_settings(instance, "SELECT value FROM system.settings WHERE name='max_memory_usage'",
                          settings={'max_memory_usage': 20000000001},
                          result=None,
                          exception="Setting max_memory_usage shouldn't be greater than 20000000000")


def assert_query_settings(instance, query, settings, result=None, exception=None, user=None):
    """
    Try and send the query with custom settings via all available methods:
    1. TCP Protocol with settings packet
    2. HTTP Protocol with settings params
    3. TCP Protocol with session level settings
    4. TCP Protocol with query level settings
    """

    if not settings:
        settings = {}

    # tcp level settings
    if exception:
        assert exception in instance.query_and_get_error(query, settings=settings, user=user)
    else:
        assert instance.query(query, settings=settings, user=user).strip() == result

    # http level settings
    if exception:
        assert exception in instance.http_query_and_get_error(query, params=settings, user=user)
    else:
        assert instance.http_query(query, params=settings, user=user).strip() == result

    # session level settings
    queries = ""

    for k, v in list(settings.items()):
        queries += "SET {}={};\n".format(k, v)

    queries += query

    if exception:
        assert exception in instance.query_and_get_error(queries, user=user)
    else:
        assert instance.query(queries, user=user).strip() == result

    if settings:
        query += " SETTINGS "
        for ix, (k, v) in enumerate(settings.items()):
            query += "{} = {}".format(k, v)
            if ix != len(settings) - 1:
                query += ", "

    if exception:
        assert exception in instance.query_and_get_error(queries, user=user)
    else:
        assert instance.query(queries, user=user).strip() == result