Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_timer():
with patch("iredis.utils.logger") as mock_logger:
timer("foo")
time.sleep(0.1)
timer("bar")
mock_logger.debug.assert_called()
args, kwargs = mock_logger.debug.call_args
matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])
assert matched.group(1) == str(3)
assert 0.1 <= float(matched.group(2)) <= 0.2
# --- test again ---
timer("foo")
time.sleep(0.2)
timer("bar")
mock_logger.debug.assert_called()
args, kwargs = mock_logger.debug.call_args
matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])
assert matched.group(1) == str(5)
assert 0.2 <= float(matched.group(2)) <= 0.3
def test_timer():
with patch("iredis.utils.logger") as mock_logger:
timer("foo")
time.sleep(0.1)
timer("bar")
mock_logger.debug.assert_called()
args, kwargs = mock_logger.debug.call_args
matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])
assert matched.group(1) == str(3)
assert 0.1 <= float(matched.group(2)) <= 0.2
# --- test again ---
timer("foo")
time.sleep(0.2)
timer("bar")
mock_logger.debug.assert_called()
args, kwargs = mock_logger.debug.call_args
matched = re.match(r"\[timer (\d)\] (0\.\d+) -> bar", args[0])
assert matched.group(1) == str(5)
group2commands = copy.deepcopy(group)
# add lowercase commands
group2command_res = {}
for syntax in group.keys():
commands = group[syntax]
lower_commands = [command.lower() for command in commands]
commands += lower_commands
# Space in command cloud be mutiple spaces
re_commands = [command.replace(" ", r"\s+") for command in commands]
group2command_res[syntax] = "|".join(re_commands)
return group2commands, group2command_res, command2callback
timer("[Loader] Start loading commands file...")
group2commands, group2command_res, command2callback = load_command()
# all redis command strings, in UPPER case
# NOTE: Must sort by length, to match longest command first
all_commands = sorted(
list(command2callback.keys()) + ["HELP"] + ["HELP"],
key=lambda x: len(x),
reverse=True,
)
commands_summary = load_command_summary()
timer("[Loader] Finished loading commands.")
def repl(client, session, start_time):
command_holder = UserInputCommand()
timer(f"First REPL command enter, time cost: {time.time() - start_time}")
while True:
logger.info("↓↓↓↓" * 10)
logger.info("REPL waiting for command...")
if config.compiling != COMPILING_DONE:
# auto refresh to display animation...
_interval = 0.1
else:
_interval = None
try:
command = session.prompt(
"{hostname}> ".format(hostname=str(client)),
bottom_toolbar=BottomToolbar(command_holder).render,
refresh_interval=_interval,
input_processors=[GetCommandProcessor(command_holder)],
group2command_res[syntax] = "|".join(re_commands)
return group2commands, group2command_res, command2callback
timer("[Loader] Start loading commands file...")
group2commands, group2command_res, command2callback = load_command()
# all redis command strings, in UPPER case
# NOTE: Must sort by length, to match longest command first
all_commands = sorted(
list(command2callback.keys()) + ["HELP"] + ["HELP"],
key=lambda x: len(x),
reverse=True,
)
commands_summary = load_command_summary()
timer("[Loader] Finished loading commands.")