bughz/predixy
1
0
Fork 0
mirror of https://github.com/joyieldInc/predixy.git synced 2026-02-05 01:42:24 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Unify test result formatting

Browse Source
This commit is contained in:
Julien Letessier 2026-01-15 10:55:19 +01:00
parent 9c974494a7
commit 8b8b3ff32a
13 changed files with 123 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
test/basic.py
View File

@ -6,9 +6,9 @@
#
import time
import redis
import sys
import argparse
from test_util import get_host_port, make_client, exit_with_result
c = None
@ -663,9 +663,8 @@ if __name__ == '__main__':
parser.add_argument('case', nargs='*', default=None, help='specify test case')
args = parser.parse_args()
a = set()
host = '127.0.0.1' if not args.h else args.h
port = 7617 if not args.p else args.p
c = redis.StrictRedis(host=host, port=port)
host, port = get_host_port(args, host_attr="h", port_attr="p")
c = make_client(host, port)
if args.case:
a = set(args.case)
fails = []
@ -681,10 +680,12 @@ if __name__ == '__main__':
if not succ:
fails.append('transaction')
print('--------------------------------------------')
if len(fails) > 0:
success = len(fails) == 0
if not success:
print('******* Some case test fail *****')
for cmd in fails:
print(cmd)
else:
print('Good! all Case Pass.')
exit_with_result(success, "basic", "basic")

18
test/eval_cross_shard.py
View File

@ -3,13 +3,12 @@
# Verify EVAL/EVALSHA rejects multi-key cross-shard scripts
#
import argparse
import sys
import redis
from test_util import parse_args, make_client, exit_with_result
def run_test(host, port):
c = redis.StrictRedis(host=host, port=port)
c = make_client(host, port)
script = "return {KEYS[1], KEYS[2]}"
try:
res = c.eval(script, 2, "eval_key1", "eval_key2")
@ -24,13 +23,6 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="EVAL cross-shard test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: eval cross shard")
sys.exit(0)
print("FAIL: eval cross shard")
sys.exit(1)
args = parse_args("EVAL cross-shard test")
success = run_test(args.host, args.port)
exit_with_result(success, "eval cross shard", "eval cross shard")

18
test/mget_wrong_type.py
View File

@ -3,13 +3,12 @@
# Verify MGET returns WRONGTYPE for non-string keys
#
import argparse
import sys
import redis
from test_util import parse_args, make_client, exit_with_result
def run_test(host, port):
c = redis.StrictRedis(host=host, port=port)
c = make_client(host, port)
c.delete("mget_wrong_type")
c.lpush("mget_wrong_type", "v1")
@ -22,13 +21,6 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="MGET wrong type test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: MGET wrong type")
sys.exit(0)
print("FAIL: MGET wrong type")
sys.exit(1)
args = parse_args("MGET wrong type test")
success = run_test(args.host, args.port)
exit_with_result(success, "MGET wrong type", "MGET wrong type")

18
test/msetnx_atomicity.py
View File

@ -3,13 +3,12 @@
# Verify MSETNX does not partially apply changes when it returns 0
#
import argparse
import sys
import redis
from test_util import parse_args, make_client, exit_with_result
def run_test(host, port):
c = redis.StrictRedis(host=host, port=port)
c = make_client(host, port)
c.delete("msetnx_k1", "msetnx_k2")
c.set("msetnx_k1", "existing")
@ -31,13 +30,6 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="MSETNX atomicity test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: MSETNX atomicity")
sys.exit(0)
print("FAIL: MSETNX atomicity")
sys.exit(1)
args = parse_args("MSETNX atomicity test")
success = run_test(args.host, args.port)
exit_with_result(success, "MSETNX atomicity", "MSETNX atomicity")

19
test/null_response_handling.py
View File

@ -3,11 +3,10 @@
# Exercise server write mismatch handling to ensure proxy stays alive
#
import argparse
import socket
import sys
import time
import redis
from test_util import parse_args, make_client, exit_with_result
def run_test(host, port):
@ -21,7 +20,7 @@ def run_test(host, port):
# Ensure the proxy still accepts connections.
try:
c = redis.StrictRedis(host=host, port=port)
c = make_client(host, port)
if c.ping() is not True:
print("FAIL: ping did not return True")
return False
@ -33,13 +32,7 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="Null response handling test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: null response handling")
sys.exit(0)
print("FAIL: null response handling")
sys.exit(1)
args = parse_args("Null response handling test")
success = run_test(args.host, args.port)
exit_with_result(success, "null response handling",
"null response handling")

20
test/pubsub.py
View File

@ -5,9 +5,7 @@
# All rights reserved.
import time
import redis
import sys
import argparse
from test_util import parse_args, make_clients, exit_with_result
c1 = None
c2 = None
@ -111,18 +109,16 @@ def test():
print('Good! PubSub test pass')
else:
print('Oh! PubSub some case fail')
return succ
if __name__ == '__main__':
parser = argparse.ArgumentParser(conflict_handler='resolve')
parser.add_argument('-h', nargs='?', default='127.0.0.1', help='host')
parser.add_argument('-p', nargs='?', default=7617, type=int, help='port')
args = parser.parse_args()
host = '127.0.0.1' if not args.h else args.h
port = 7617 if not args.p else args.p
c1 = redis.StrictRedis(host=host, port=port)
c2 = redis.StrictRedis(host=host, port=port)
test()
args = parse_args("PubSub test")
host = args.host
port = args.port
c1, c2 = make_clients(host, port, count=2)
success = test()
exit_with_result(success, "pubsub", "pubsub")

18
test/pubsub_long_name.py
View File

@ -3,9 +3,8 @@
# Verify subscribe/psubscribe confirmation with long channel/pattern names
#
import argparse
import sys
import redis
from test_util import parse_args, make_client, exit_with_result
def normalize_bytes(value):
@ -18,7 +17,7 @@ def run_test(host, port):
long_name = "ch_" + ("x" * 200)
pattern = long_name + "*"
c1 = redis.StrictRedis(host=host, port=port)
c1 = make_client(host, port)
ps = c1.pubsub()
ps.subscribe(long_name)
@ -53,13 +52,6 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="Long pubsub name test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: long pubsub name")
sys.exit(0)
print("FAIL: long pubsub name")
sys.exit(1)
args = parse_args("Long pubsub name test")
success = run_test(args.host, args.port)
exit_with_result(success, "long pubsub name", "long pubsub name")

20
test/pubsub_message_response.py
View File

@ -3,9 +3,8 @@
# Verify pubsub message responses include message data
#
import argparse
import sys
import redis
from test_util import parse_args, make_clients, exit_with_result
def normalize_bytes(value):
@ -15,8 +14,7 @@ def normalize_bytes(value):
def run_test(host, port):
c1 = redis.StrictRedis(host=host, port=port)
c2 = redis.StrictRedis(host=host, port=port)
c1, c2 = make_clients(host, port, count=2)
ps = c1.pubsub()
ps.subscribe("ch_resp")
@ -44,13 +42,7 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="Pubsub message response test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: pubsub message response")
sys.exit(0)
print("FAIL: pubsub message response")
sys.exit(1)
args = parse_args("Pubsub message response test")
success = run_test(args.host, args.port)
exit_with_result(success, "pubsub message response",
"pubsub message response")

21
test/pubsub_minimal.py
View File

@ -3,16 +3,14 @@
# Minimal test to reproduce pubsub message queueing issue in Predixy
# Tests pass against Redis (6379) but fail against Predixy (7617)
import redis
import sys
import argparse
from test_util import parse_args, make_clients, exit_with_result
def test_redis(host, port):
"""Test pubsub against Redis or Predixy"""
print(f"\n=== Testing against {host}:{port} ===\n")
c1 = redis.StrictRedis(host=host, port=port)
c2 = redis.StrictRedis(host=host, port=port)
c1, c2 = make_clients(host, port, count=2)
ps = c1.pubsub()
@ -76,24 +74,15 @@ def test_redis(host, port):
return False
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Minimal pubsub test')
parser.add_argument('--host', default='127.0.0.1', help='Host (default: 127.0.0.1)')
parser.add_argument('-p', '--port', type=int, required=True, help='Port (6379 for Redis, 7617 for Predixy)')
args = parser.parse_args()
args = parse_args("Minimal pubsub test", require_port=True)
host = args.host
port = args.port
try:
success = test_redis(host, port)
if success:
print("\n✓ Test PASSED")
sys.exit(0)
else:
print("\n✗ Test FAILED")
sys.exit(1)
exit_with_result(success, "pubsub minimal", "pubsub minimal")
except Exception as e:
print(f"\n✗ Test ERROR: {e}")
print(f"\n🔴 pubsub minimal error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)

20
test/pubsub_parser_reset.py
View File

@ -3,14 +3,12 @@
# Verify pubsub parser state does not reuse old messages
#
import argparse
import sys
import redis
from test_util import parse_args, make_clients, exit_with_result
def run_test(host, port):
c1 = redis.StrictRedis(host=host, port=port)
c2 = redis.StrictRedis(host=host, port=port)
c1, c2 = make_clients(host, port, count=2)
ps = c1.pubsub()
ps.subscribe("ch_reset")
@ -35,13 +33,7 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="Pubsub parser reset test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: pubsub parser reset")
sys.exit(0)
print("FAIL: pubsub parser reset")
sys.exit(1)
args = parse_args("Pubsub parser reset test")
success = run_test(args.host, args.port)
exit_with_result(success, "pubsub parser reset",
"pubsub parser reset")

20
test/pubsub_subscription_order.py
View File

@ -3,15 +3,13 @@
# Verify subscription confirmations arrive before messages
#
import argparse
import sys
import redis
from test_util import parse_args, make_clients, exit_with_result
import time
def run_test(host, port):
c1 = redis.StrictRedis(host=host, port=port)
c2 = redis.StrictRedis(host=host, port=port)
c1, c2 = make_clients(host, port, count=2)
ps = c1.pubsub()
ps.subscribe("ch_order")
@ -43,13 +41,7 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="Pubsub subscription order test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: pubsub subscription order")
sys.exit(0)
print("FAIL: pubsub subscription order")
sys.exit(1)
args = parse_args("Pubsub subscription order test")
success = run_test(args.host, args.port)
exit_with_result(success, "pubsub subscription order",
"pubsub subscription order")

55
test/test_util.py Normal file
View File

@ -0,0 +1,55 @@
#!/usr/bin/env python3
#
# Shared test utilities for argument parsing and Redis connections.
#
import argparse
import sys
import redis
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 7617
def add_host_port_args(parser, default_port=DEFAULT_PORT, require_port=False):
parser.add_argument("-h", "--host", default=DEFAULT_HOST)
parser.add_argument("-p", "--port", type=int, default=default_port,
required=require_port)
return parser
def parse_args(description, default_port=DEFAULT_PORT, require_port=False):
parser = argparse.ArgumentParser(conflict_handler="resolve",
description=description)
add_host_port_args(parser, default_port=default_port,
require_port=require_port)
return parser.parse_args()
def get_host_port(args, host_attr="host", port_attr="port",
default_host=DEFAULT_HOST, default_port=DEFAULT_PORT):
host = getattr(args, host_attr, None) or default_host
port = getattr(args, port_attr, None) or default_port
return host, port
def make_client(host, port):
return redis.StrictRedis(host=host, port=port)
def make_clients(host, port, count=2):
return [make_client(host, port) for _ in range(count)]
def report_result(success, pass_msg, fail_msg):
if success:
print(f"🟢 {pass_msg}")
return True
print(f"🔴 {fail_msg}")
return False
def exit_with_result(success, pass_msg, fail_msg):
report_result(success, pass_msg, fail_msg)
sys.exit(0 if success else 1)

18
test/transaction_forbid.py
View File

@ -3,13 +3,12 @@
# Verify forbidden command in transaction returns error without closing connection
#
import argparse
import sys
import redis
from test_util import parse_args, make_client, exit_with_result
def run_test(host, port):
c = redis.StrictRedis(host=host, port=port)
c = make_client(host, port)
try:
r = c.execute_command("MULTI")
if r not in (b"OK", "OK"):
@ -46,13 +45,6 @@ def run_test(host, port):
if __name__ == "__main__":
parser = argparse.ArgumentParser(conflict_handler='resolve', description="Transaction forbid test")
parser.add_argument("-h", "--host", default="127.0.0.1")
parser.add_argument("-p", "--port", type=int, default=7617)
args = parser.parse_args()
if run_test(args.host, args.port):
print("PASS: transaction forbid")
sys.exit(0)
print("FAIL: transaction forbid")
sys.exit(1)
args = parse_args("Transaction forbid test")
success = run_test(args.host, args.port)
exit_with_result(success, "transaction forbid", "transaction forbid")