Skip to content

自动测试脚本 #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

自动测试脚本 #1

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8092073
自动测试脚本
Catheriany Jun 9, 2025
3626ffd
打印执行过程Log
Catheriany Jun 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
168 changes: 150 additions & 18 deletions testbot/testbot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,9 @@
from io import BytesIO
import platform
import abc
import json
import schedule

from datetime import datetime
import sys

CONFIG_JSON = "config.json"

Expand All @@ -26,7 +26,6 @@ class Notifier(abc.ABC):
def notify_results(self, meta):
raise NotImplementedError()


class FeishuNotifier(Notifier):
def __init__(self, config):
self.webhook_url = config["url"]
Expand Down Expand Up @@ -205,30 +204,163 @@ def __init__(self, config):

def install(self, config_flags=""):
name = "安装InfiniCore"
try:
os.chdir(self.project_dir)
self.test_cmd(f"python scripts/install.py {config_flags}", name=name)

except:
raise
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
log_file = os.path.join(self.project_dir, f"install_{timestamp}.log")
cmd = f"python scripts/install.py {config_flags}".strip()

with open(log_file, "w", encoding="utf-8", errors="replace") as log:
proc = subprocess.Popen(
cmd,
shell=True,
cwd=self.project_dir,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
bufsize=1,
)
for line in proc.stdout:
log.write(line)
sys.stdout.write(line)
sys.stdout.flush()

returncode = proc.wait()
if returncode != 0:
err = f"\n[ERROR] {name} failed with return code {returncode}\n"
log.write(err)
sys.stdout.write(err)
raise subprocess.CalledProcessError(returncode, cmd)

def run_python_tests(self, flags=""):
name = "运行算子python测试"
try:
os.chdir(self.project_dir)
self.test_cmd(
f"python scripts/python_test.py {flags}",
name=name,
break_on_error=False,
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
log_file = os.path.join(self.project_dir, f"python_test_{timestamp}.log")

cmd = f"python scripts/python_test.py {flags}".strip()
with open(log_file, "w", encoding="utf-8", errors="replace") as log:
process = subprocess.Popen(
cmd,
shell=True,
cwd=self.project_dir,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
)

except:
raise
for line in process.stdout:
log.write(line)
sys.stdout.write(line)
sys.stdout.flush()

returncode = process.wait()

if returncode != 0:
err_msg = f"\n[ERROR] Python tests failed with return code {returncode}\n"
log.write(err_msg)
sys.stdout.write(err_msg)
raise subprocess.CalledProcessError(returncode, cmd)


def run_gguf_tests(self, flags=""):
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
log_file = os.path.join(self.project_dir, f"gguf_test_{timestamp}.log")

cmd = f"python scripts/gguf_test.py {flags}".strip()
with open(log_file, "w", encoding="utf-8", errors="replace") as log:
proc = subprocess.Popen(
cmd,
shell=True,
cwd=self.project_dir,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
bufsize=1,
)

# 实时输出并写日志
for line in proc.stdout:
log.write(line)
sys.stdout.write(line)
sys.stdout.flush()

returncode = proc.wait()
if returncode != 0:
err_msg = f"\n[ERROR] GGUF test failed with return code {returncode}\n"
log.write(err_msg)
sys.stdout.write(err_msg)
raise subprocess.CalledProcessError(returncode, cmd)


def run_infiniccl_test(self, flags=""):
os.chdir(self.project_dir)
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
log_file = os.path.join(self.project_dir, f"infiniccl_test_{timestamp}.log")

with open(log_file, "w", encoding="utf-8", errors="replace") as log:

def run_and_log(cmd, header):
log.write(f"\n=== {header} ===\n")
result = subprocess.run(
cmd,
shell=True,
cwd=self.project_dir,
stdout=log,
stderr=subprocess.STDOUT,
text=True,
encoding="utf-8",
errors="replace",
)
if result.returncode != 0:
log.write(f"\n[ERROR] '{cmd}' exited with code {result.returncode}\n")
raise subprocess.CalledProcessError(result.returncode, cmd)

run_and_log("xmake build infiniccl-test", "Building infiniccl-test")
run_and_log("xmake install", "Installing infiniccl-test")
exe_cmd = f"build/linux/x86_64/release/infiniccl-test {flags}"
run_and_log(exe_cmd, "Running infiniccl-test")


def run_infinirt_test(self, flags=""):
os.chdir(self.project_dir)
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
log_file = os.path.join(self.project_dir, f"infinirt_test_{timestamp}.log")

with open(log_file, "w", encoding="utf-8", errors="replace") as log:

def run_and_log(cmd, header):
log.write(f"\n=== {header} ===\n")
result = subprocess.run(
cmd,
shell=True,
stdout=log,
stderr=subprocess.STDOUT,
cwd=self.project_dir,
text=True,
encoding="utf-8",
errors="replace",
)
if result.returncode != 0:
log.write(f"\n[ERROR] Command failed with exit code {result.returncode}\n")
raise subprocess.CalledProcessError(result.returncode, cmd)

run_and_log("xmake build infinirt-test", "Building infinirt-test")
run_and_log("xmake install", "Installing infinirt-test")

exe_cmd = f"build/linux/x86_64/release/infinirt-test {flags}"
run_and_log(exe_cmd, "Running infinirt-test")


def run_tests(self):
def _run_test():
self.install(self.get_xmake_config_flags())
self.run_python_tests(self.get_python_test_flags())
self.run_gguf_tests(self.get_python_test_flags())
self.run_infiniccl_test(self.get_python_test_flags())
self.run_infinirt_test(self.get_python_test_flags())

try:
self.clone_or_update()
Expand Down