From 06cba6ccd165ca8b224797e37fccb9e63f026d77 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sat, 21 Mar 2020 11:28:17 +0100 Subject: Adding upstream version 1.9.1. Signed-off-by: Daniel Baumann --- tests/conftest.py | 137 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 137 insertions(+) create mode 100644 tests/conftest.py (limited to 'tests/conftest.py') diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..4d7d643 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,137 @@ +import os +import tempfile +from textwrap import dedent + +import pexpect +import pytest +import redis + +from iredis.client import Client +from iredis.commands import split_command_args +from iredis.redis_grammar import get_command_grammar +from iredis.exceptions import InvalidArguments +from iredis.config import Config, config as global_config + + +TIMEOUT = 2 +HISTORY_FILE = ".iredis_history" + + +@pytest.fixture +def judge_command(): + def judge_command_func(input_text, expect): + if expect == "invalid": + with pytest.raises(InvalidArguments): + split_command_args(input_text) + return + + command, _ = split_command_args(input_text) + grammar = get_command_grammar(command) + + m = grammar.match(input_text) + + # test on not match + if not expect: + assert m is None + return + + variables = m.variables() + print("Found variables: {}".format(variables)) + for expect_token, expect_value in expect.items(): + all_variables = variables.getall(expect_token) + if len(all_variables) > 1: + assert sorted(all_variables) == sorted(expect_value) + else: + assert variables.get(expect_token) == expect_value + + return judge_command_func + + +@pytest.fixture(scope="function") +def clean_redis(): + """ + Return a empty redis db. (redis-py client) + """ + client = redis.StrictRedis(db=15) + client.flushdb() + return client + + +@pytest.fixture +def iredis_client(): + return Client("127.0.0.1", "6379", db=15) + + +@pytest.fixture +def config(): + newconfig = Config() + global_config.__dict__ = newconfig.__dict__ + config.raw = False + return global_config + + +@pytest.fixture(scope="function") +def cli(): + """Open iredis subprocess to test""" + f = tempfile.TemporaryFile("w") + config_content = dedent( + """ + [main] + log_location = + warning = True + """ + ) + f.write(config_content) + f.close() + env = os.environ + env["PROMPT_TOOLKIT_NO_CPR"] = "1" + + child = pexpect.spawn(f"iredis -n 15 --iredisrc {f.name}", timeout=TIMEOUT, env=env) + child.logfile_read = open("cli_test.log", "ab") + child.expect(["https://iredis.io/issues", "127.0.0.1"]) + yield child + child.close() + + +@pytest.fixture(scope="function") +def raw_cli(): + """Open iredis subprocess to test""" + TEST_IREDISRC = "/tmp/.iredisrc.test" + config_content = dedent( + """ + [main] + log_location = + warning = True + """ + ) + + with open(TEST_IREDISRC, "w+") as test_iredisrc: + test_iredisrc.write(config_content) + + child = pexpect.spawn( + f"iredis --raw -n 15 --iredisrc {TEST_IREDISRC}", timeout=TIMEOUT + ) + child.logfile_read = open("cli_test.log", "ab") + child.expect(["https://iredis.io/issues", "127.0.0.1"]) + yield child + child.close() + + +@pytest.fixture(scope="function") +def cli_without_warning(): + f = tempfile.TemporaryFile("w") + config_content = dedent( + """ + [main] + log_location = /tmp/iredis1.log + warning = False + """ + ) + f.write(config_content) + f.close() + + cli = pexpect.spawn(f"iredis -n 15 --iredisrc {f.name}", timeout=1) + cli.logfile_read = open("cli_test.log", "ab") + yield cli + cli.close() + os.remove("/tmp/iredisrc") -- cgit v1.2.3