输出日志:<\/div>
+
{}<\/div>
+ <\/div>
+""".format(status_class, test_case["name"], status_style, status_text, test_case["output"])
+
+ html_content += """
+
+
+
+
+
+"""
+
+ with open(output_file, 'w', encoding='utf-8') as f:
+ f.write(html_content)
+
+def run_install_test(test_case):
+
+ """运行单个安装测试"""
+ name = test_case.get('name', 'Unknown Test')
+ chooses = test_case.get('chooses', [])
+
+ print("开始测试: {}".format(name))
+
+ # 创建临时配置文件路径
+ temp_config = "/tmp/fish_install_test_temp.yaml"
+
+ # 确保 /tmp 目录存在
+ os.makedirs("/tmp", exist_ok=True)
+
+ # 创建临时配置文件
+ config_data = {'chooses': chooses}
+ try:
+ with open(temp_config, 'w', encoding='utf-8') as f:
+ yaml.dump(config_data, f, allow_unicode=True)
+ print("已创建临时配置文件: {}".format(temp_config))
+ except Exception as e:
+ print("创建临时配置文件失败: {}".format(e))
+ return False, ""
+
+ # 备份原始的 fish_install.yaml (如果存在)
+ original_config = "../fish_install.yaml"
+ backup_config = "../fish_install.yaml.backup"
+ if os.path.exists(original_config):
+ try:
+ os.rename(original_config, backup_config)
+ print("已备份原始配置文件至: {}".format(backup_config))
+ except Exception as e:
+ print("备份原始配置文件失败: {}".format(e))
+ # 即使备份失败也继续执行，因为我们会在最后恢复
+
+ # 将临时配置文件复制为当前配置文件和/tmp/fishinstall/tools/fish_install.yaml
+ try:
+ import shutil
+ shutil.copy(temp_config, original_config)
+ print("已将临时配置文件复制为: {}".format(original_config))
+
+ # 同时将配置文件复制到/tmp/fishinstall/tools/目录下
+ fishinstall_config = "/tmp/fishinstall/tools/fish_install.yaml"
+ # 确保目录存在
+ os.makedirs(os.path.dirname(fishinstall_config), exist_ok=True)
+ shutil.copy(temp_config, fishinstall_config)
+ print("已将临时配置文件复制为: {}".format(fishinstall_config))
+ except Exception as e:
+ print("复制配置文件失败: {}".format(e))
+ # 恢复备份的配置文件
+ if os.path.exists(backup_config):
+ try:
+ os.rename(backup_config, original_config)
+ print(f"已恢复备份的配置文件: {original_config}")
+ except:
+ pass
+ # 清理临时文件
+ if os.path.exists(temp_config):
+ os.remove(temp_config)
+ return False, ""
+
+ # 初始化输出和错误信息
+ output = ""
+ error = ""
+
+ # 运行安装脚本
+ try:
+ # 使用 -u 参数确保输出不被缓冲，以便实时查看日志
+ # 直接运行 install.py，它会自动检测并使用 ../fish_install.yaml
+ # 增加超时时间为 2 小时 (7200 秒)
+ process = subprocess.Popen(
+ [sys.executable, "../install.py"],
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ universal_newlines=True,
+ bufsize=1,
+ env={**os.environ, 'FISH_INSTALL_CONFIG': '../fish_install.yaml'}
+ )
+
+ # 实时打印输出
+ print("=== 脚本输出开始 ===")
+ while True:
+ output_line = process.stdout.readline()
+ if output_line == '' and process.poll() is not None:
+ break
+ if output_line:
+ print(output_line.strip())
+ output += output_line
+ # 确保实时刷新输出
+ sys.stdout.flush()
+ print("=== 脚本输出结束 ===")
+
+ # 等待进程结束，超时时间为 2 小时
+ # stdout, _ = process.communicate(timeout=7200)
+ # output = stdout
+
+ # # 打印输出
+ # print("=== 脚本输出开始 ===")
+ # print(stdout)
+ # print("=== 脚本输出结束 ===")
+
+ # 检查退出码和输出中的错误信息
+ if process.returncode == 0 and not check_output_for_errors(output):
+ print(f"测试通过: {name}")
+ return True, output
+ else:
+ if process.returncode != 0:
+ print(f"测试失败: {name} (退出码: {process.returncode})")
+ else:
+ print(f"测试失败: {name} (脚本中检测到错误)")
+ return False, output
+ except subprocess.TimeoutExpired:
+ print(f"测试超时: {name} (超过7200秒)")
+ # 终止进程
+ process.kill()
+ stdout, _ = process.communicate()
+ output = stdout
+ return False, output
+ except Exception as e:
+ print(f"运行测试时发生异常: {e}")
+ error = str(e)
+ return False, output + "\n" + error
+ finally:
+ # 恢复备份的配置文件 (如果存在)
+ if os.path.exists(backup_config):
+ try:
+ os.rename(backup_config, original_config)
+ print(f"已恢复备份的配置文件: {original_config}")
+ except Exception as e:
+ print(f"恢复备份的配置文件失败: {e}")
+ # 如果没有备份文件，但创建了原始配置文件，则删除它
+ elif os.path.exists(original_config):
+ try:
+ os.remove(original_config)
+ print(f"已删除临时创建的配置文件: {original_config}")
+ except Exception as e:
+ print(f"删除临时配置文件失败: {e}")
+ # 清理临时配置文件
+ if os.path.exists(temp_config):
+ try:
+ os.remove(temp_config)
+ print(f"已清理临时配置文件: {temp_config}")
+ except Exception as e:
+ print(f"清理临时配置文件失败: {e}")
+
+def main():
+ """主函数"""
+ # 解析命令行参数
+ parser = argparse.ArgumentParser(description='运行一键安装工具测试')
+ parser.add_argument('--target-os-version', type=str, help='目标Ubuntu版本代号 (例如: bionic, focal, jammy, noble)')
+ args = parser.parse_args()
+
+ target_os_version = args.target_os_version
+ if target_os_version:
+ print(f"目标系统版本: {target_os_version}")
+ else:
+ # 自动检测系统版本代号
+ target_os_version = get_ubuntu_codename()
+ if target_os_version:
+ print(f"自动检测到系统版本: {target_os_version}")
+ else:
+ print("未指定目标系统版本，也未能自动检测到系统版本")
+
+ config_file = "fish_install_test.yaml"
+
+ # 检查配置文件是否存在
+ if not os.path.exists(config_file):
+ print(f"错误: 找不到测试配置文件 {config_file}")
+ sys.exit(1)
+
+ # 加载测试用例
+ all_test_cases = load_test_cases(config_file)
+ if not all_test_cases:
+ print("错误: 没有找到有效的测试用例")
+ sys.exit(1)
+
+ # 根据目标系统版本过滤测试用例
+ if target_os_version:
+ test_cases = [tc for tc in all_test_cases if tc.get('target_os_version') == target_os_version]
+ if not test_cases:
+ # 如果没有找到特定于该系统的测试用例，则运行所有没有指定target_os_version的测试用例
+ test_cases = [tc for tc in all_test_cases if 'target_os_version' not in tc]
+ if not test_cases:
+ print(f"错误: 没有找到适用于系统版本 {target_os_version} 的测试用例")
+ sys.exit(1)
+ else:
+ # 如果没有指定目标系统版本，则运行所有没有指定target_os_version的测试用例
+ test_cases = [tc for tc in all_test_cases if 'target_os_version' not in tc]
+ if not test_cases:
+ print("错误: 没有找到适用于所有系统的通用测试用例")
+ sys.exit(1)
+
+ print(f"共找到 {len(test_cases)} 个适用于当前系统版本的测试用例")
+
+ # 运行所有测试用例并收集结果
+ results = []
+ passed = 0
+ failed = 0
+
+ for i, test_case in enumerate(test_cases):
+ print("\n--- 测试用例 {}/{} ---".format(i+1, len(test_cases)))
+ success, output = run_install_test(test_case)
+ case_name = test_case.get('name', 'Test Case {}'.format(i+1))
+
+ result = {
+ "name": case_name,
+ "success": success,
+ "output": output
+ }
+ results.append(result)
+
+ if success:
+ passed += 1
+ else:
+ failed += 1
+ # 在测试用例之间添加延迟，避免系统资源冲突
+ time.sleep(2)
+
+ # 生成详细的测试报告
+ report = {
+ "summary": {
+ "total": len(test_cases),
+ "passed": passed,
+ "failed": failed
+ },
+ "details": results
+ }
+
+ # 将报告保存为 JSON 文件
+ report_file = "test_report.json"
+ try:
+ with open(report_file, 'w', encoding='utf-8') as f:
+ json.dump(report, f, ensure_ascii=False, indent=2)
+ print("\n详细测试报告已保存至: {}".format(report_file))
+ except Exception as e:
+ print("保存测试报告失败: {}".format(e))
+
+ # 生成HTML格式的测试报告
+ html_report_file = "test_report.html"
+ try:
+ generate_html_report(report, html_report_file)
+ print("HTML测试报告已保存至: {}".format(html_report_file))
+ except Exception as e:
+ print("生成HTML测试报告失败: {}".format(e))
+
+ # 输出测试结果摘要
+ print("\n=== 测试结果摘要 ===")
+ print(f"通过: {passed}")
+ print(f"失败: {failed}")
+ print(f"总计: {len(test_cases)}")
+
+ if failed > 0:
+ print("部分测试失败，请检查日志和测试报告。")
+ sys.exit(1)
+ else:
+ print("所有测试通过！")
+ sys.exit(0)
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
diff --git a/tools/base.py b/tools/base.py
index 22ae6b9..b432863 100644
--- a/tools/base.py
+++ b/tools/base.py
@@ -33,7 +33,9 @@ def __init__(self,record_file=None):
self.record_input_queue = Queue()
self.record_file = record_file
- if self.record_file==None: self.record_file = "./fish_install.yaml"
+ if self.record_file==None:
+ # 首先检查环境变量
+ self.record_file = os.environ.get('FISH_INSTALL_CONFIG', "./fish_install.yaml")
self.default_input_queue = self.get_default_queue(self.record_file)
def record_input(self,item):
@@ -62,12 +64,16 @@ def gen_config_file(self):
# 检查目标文件是否存在
if os.path.exists(target_path):
- print("检测到已存在的配置文件: {}".format(target_path))
- user_input = input("是否替换该文件？[y/N]: ")
- if user_input.lower() not in ['y', 'yes']:
- print("取消替换，保留原配置文件")
- os.remove(temp_path) # 删除临时文件
- return
+ # 在自动化测试环境中，直接覆盖原配置文件
+ if os.environ.get('GITHUB_ACTIONS') == 'true':
+ print("检测到GitHub Actions环境，直接覆盖已存在的配置文件: {}".format(target_path))
+ else:
+ print("检测到已存在的配置文件: {}".format(target_path))
+ user_input = input("是否替换该文件？[y/N]: ")
+ if user_input.lower() not in ['y', 'yes']:
+ print("取消替换，保留原配置文件")
+ os.remove(temp_path) # 删除临时文件
+ return
# 先尝试删除目标文件（如果存在），避免mv命令的交互提示
if os.path.exists(target_path):
@@ -130,8 +136,6 @@ def get_default_queue(self,param_file_path):
else:
config_yaml = yaml.load(config_data)
- for choose in config_yaml['chooses']:
- choose_queue.put(choose)
for choose in config_yaml['chooses']:
choose_queue.put(choose)
@@ -1112,8 +1116,17 @@ def __choose(data,tips,array):
choose = str(choose_item['choose'])
PrintUtils.print_text(tr.tr("为您从配置文件找到默认选项：")+str(choose_item))
else:
- choose = input(tr.tr("请输入[]内的数字以选择:"))
- choose_item = None
+ try:
+ choose = input(tr.tr("请输入[]内的数字以选择:"))
+ choose_item = None
+ except EOFError:
+ # 在自动化测试环境中，input()可能会引发EOFError
+ # 如果是从配置文件读取的选项，则使用它，否则返回默认值0（退出）
+ if choose_item:
+ choose = str(choose_item['choose'])
+ else:
+ choose = "0" # 默认选择退出
+ PrintUtils.print_text(tr.tr("检测到自动化环境，使用默认选项: {}").format(choose))
# Input From Queue
if choose.isdecimal() :
if (int(choose) in dic.keys() ) or (int(choose)==0):
@@ -1163,8 +1176,17 @@ def __choose(data,tips,array,categories):
choose_id = str(choose_item['choose'])
print(tr.tr("为您从配置文件找到默认选项：")+str(choose_item))
else:
- choose_id = input(tr.tr("请输入[]内的数字以选择:"))
- choose_item = None
+ try:
+ choose_id = input(tr.tr("请输入[]内的数字以选择:"))
+ choose_item = None
+ except EOFError:
+ # 在自动化测试环境中，input()可能会引发EOFError
+ # 如果是从配置文件读取的选项，则使用它，否则返回默认值0（退出）
+ if choose_item:
+ choose_id = str(choose_item['choose'])
+ else:
+ choose_id = "0" # 默认选择退出
+ PrintUtils.print_text(tr.tr("检测到自动化环境，使用默认选项: {}").format(choose_id))
# Input From Queue
if choose_id.isdecimal() :
if int(choose_id) in tool_ids :
@@ -1260,10 +1282,17 @@ def new(path,name=None,data=''):
CmdTask("sudo mkdir -p {}".format(path),3).run()
if name!=None:
# 使用临时文件和sudo权限来创建受保护的文件
- temp_file = "/tmp/{}".format(name)
- with open(temp_file, "w") as f:
- f.write(data)
- CmdTask("sudo mv {} {}".format(temp_file, path+name), 3).run()
+ # 修复：使用 uuid 生成唯一临时文件名，避免权限冲突
+ import uuid
+ temp_file = "/tmp/{}_{}".format(uuid.uuid4(), name)
+ try:
+ with open(temp_file, "w") as f:
+ f.write(data)
+ CmdTask("sudo mv {} {}".format(temp_file, path+name), 3).run()
+ finally:
+ # 确保临时文件被清理
+ if os.path.exists(temp_file):
+ os.remove(temp_file)
return True
@staticmethod
diff --git a/tools/tool_config_system_source.py b/tools/tool_config_system_source.py
index b9ba0a4..82ed7e1 100644
--- a/tools/tool_config_system_source.py
+++ b/tools/tool_config_system_source.py
@@ -88,8 +88,6 @@ def clean_old_source(self):
dic_source_method = {1:"自动测速选择最快的源", 2:"根据测速结果手动选择源"}
self.source_method_code, _ = ChooseTask(dic_source_method, "请选择源的选择方式").run()
-
-
def get_source_by_system(self,system,codename,arch,failed_sources=[], return_all=False):
# 实际测试发现，阿里云虽然延时很低，但是带宽也低的离谱，一点都不用心，删掉了
ubuntu_amd64_sources = [
@@ -170,9 +168,6 @@ def get_source_by_system(self,system,codename,arch,failed_sources=[], return_all
else:
return fast_source[0],template
return None,None
-
-
-
for source in sources:
if "tsinghua" in source:
@@ -235,8 +230,6 @@ def get_source_by_system(self,system,codename,arch,failed_sources=[], return_all
return source, template
-
-
# 去除末尾的斜杠
if source.endswith("/"):
source = source[:-1]
@@ -291,6 +284,7 @@ def replace_source(self,failed_sources=[]):
else:
PrintUtils.print_success('为您选择最快镜像源:{}'.format(source))
+ # 使用已修复的 FileUtils.new 方法
FileUtils.new('/etc/apt/','sources.list',template.replace("
",codename).replace('',source))
return source
diff --git a/tools/translation/translator.py b/tools/translation/translator.py
index 50eeb57..5c6ce98 100644
--- a/tools/translation/translator.py
+++ b/tools/translation/translator.py
@@ -8,6 +8,8 @@
import os
import tools.base
from tools.base import CmdTask
+import subprocess
+import time
_suported_languages = ['zh_CN', 'en_US']
url_prefix = os.environ.get('FISHROS_URL','http://mirror.fishros.com/install')
@@ -33,7 +35,19 @@ def __init__(self):
# Create directory for downloads
CmdTask("mkdir -p /tmp/fishinstall/tools/translation/assets").run()
for lang in _suported_languages:
- CmdTask("wget {} -O /tmp/fishinstall/{} --no-check-certificate".format(lang_url.format(lang), lang_url.format(lang).replace(url_prefix, ''))).run()
+ # Add timeout and retry mechanism for downloading language files
+ # Use /tmp/ directory directly to avoid permission issues
+ temp_file = "/tmp/fishros_lang_{}.py".format(lang)
+ final_path = "/tmp/fishinstall/{}".format(lang_url.format(lang).replace(url_prefix, ''))
+ download_cmd = "wget {} -O {} --no-check-certificate --timeout=10 --tries=3".format(lang_url.format(lang), temp_file)
+ result = CmdTask(download_cmd).run()
+ # Move file to final destination if download was successful
+ if result[0] == 0:
+ CmdTask("mkdir -p $(dirname {})".format(final_path)).run()
+ CmdTask("mv {} {}".format(temp_file, final_path)).run()
+ else:
+ # Clean up temp file if download failed
+ CmdTask("rm -f {}".format(temp_file)).run()
self.loadTranslationFile()
tools.base.tr = self
@@ -62,20 +76,30 @@ def isCN(self) -> bool:
def getLocalFromIP(self) -> str:
local_str = ""
+ temp_file = "/tmp/fishros_check_country.json"
try:
- os.system("""wget --header="Accept: application/json" --no-check-certificate "https://ip.renfei.net/" -O /tmp/fishros_check_country.json -qq""")
- with open('/tmp/fishros_check_country.json', 'r') as json_file:
- data = json.loads(json_file.read())
- self.ip_info = data
- self.country = data['location']['countryCode']
- if data['location']['countryCode'] in COUNTRY_CODE_MAPPING:
- local_str = COUNTRY_CODE_MAPPING[data['location']['countryCode']]
- else:
- local_str = "en_US"
+ # Add timeout for IP detection
+ result = subprocess.run(["wget", "--header=Accept: application/json", "--no-check-certificate",
+ "https://ip.renfei.net/", "-O", temp_file, "-qq", "--timeout=10"],
+ capture_output=True, text=True, timeout=15)
+ if result.returncode == 0:
+ with open(temp_file, 'r') as json_file:
+ data = json.loads(json_file.read())
+ self.ip_info = data
+ self.country = data['location']['countryCode']
+ if data['location']['countryCode'] in COUNTRY_CODE_MAPPING:
+ local_str = COUNTRY_CODE_MAPPING[data['location']['countryCode']]
+ else:
+ local_str = "en_US"
+ else:
+ local_str = "en_US"
except Exception:
local_str = "en_US"
finally:
- os.system("rm -f /tmp/fishros_check_country.json")
+ try:
+ os.remove(temp_file)
+ except:
+ pass
return local_str