109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
""" TEST_CONFIG
|
||
{
|
||
"name": "base_allows_test",
|
||
"envs": { "TEST_TAG": "allow_mode", "PYTHONUNBUFFERED": "1" },
|
||
"network": {
|
||
"allowInternet": true,
|
||
"allowListen": [19999],
|
||
"blockList": ["8.8.4.4:53"]
|
||
},
|
||
"limits": { "cpu": 0.5, "mem": 0.2 }
|
||
}
|
||
"""
|
||
import os, sys, json, socket, platform, time, subprocess
|
||
is_darwin = platform.system().lower() == "darwin"
|
||
|
||
def test_cowsay():
|
||
try:
|
||
import cowsay
|
||
_ = cowsay.cow
|
||
return True
|
||
except:
|
||
return False
|
||
|
||
def test_memory_and_subprocess(mb_size):
|
||
if is_darwin and mb_size > 256:
|
||
return False
|
||
# 合并测试：启动子进程并申请内存
|
||
# 如果能成功返回，说明子进程能力 OK 且内存未被超限拦截
|
||
code = f"import time; bytearray({mb_size} * 1024 * 1024); print('mem_ok')"
|
||
try:
|
||
output = subprocess.check_output([sys.executable, "-c", code], text=True, timeout=5)
|
||
return output.strip() == "mem_ok"
|
||
except:
|
||
return False
|
||
|
||
def get_cpu_load():
|
||
# 简单的负载测试：执行计算密集型任务并计算 CPU 时间比例
|
||
start_wall = time.perf_counter()
|
||
start_cpu = time.process_time()
|
||
|
||
# 密集计算
|
||
_ = [sum(range(1000)) for _ in range(5000)]
|
||
|
||
end_wall = time.perf_counter()
|
||
end_cpu = time.process_time()
|
||
|
||
wall_delta = end_wall - start_wall
|
||
cpu_delta = end_cpu - start_cpu
|
||
# 计算理论占用率 (cpu_time / wall_time)
|
||
usage = (cpu_delta / wall_delta) * 100 if wall_delta > 0 else 0
|
||
return usage
|
||
|
||
def run_test():
|
||
# 使用相对路径避开 Linux 下 getcwd 的溯源问题
|
||
current_dir = os.getcwd()
|
||
# os.getpid(), open("/proc/1/cgroup").read(), open("/proc/self/cgroup").read()
|
||
cpu_usage_pct = get_cpu_load()
|
||
results = {
|
||
"pid": os.getpid(),
|
||
"cpu_usage_pct": round(cpu_usage_pct, 2),
|
||
"cpu_limit_ok": cpu_usage_pct <= 70 or is_darwin,
|
||
"mem_128M_ok": test_memory_and_subprocess(128),
|
||
"mem_512M_killed": not test_memory_and_subprocess(512),
|
||
"network_listen_ok": False,
|
||
"network_allow_ok": False,
|
||
"network_block_works": False,
|
||
"cowsay_ok": test_cowsay(),
|
||
"env_ok": os.environ.get("TEST_TAG") == "allow_mode"
|
||
}
|
||
if not is_darwin:
|
||
results["pid1_cgroup"] = open("/proc/1/cgroup").read()
|
||
results["self_cgroup"] = open("/proc/self/cgroup").read()
|
||
|
||
# 1. 测试监听 (应成功)
|
||
try:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||
s.bind(('0.0.0.0', 19999))
|
||
results["network_listen_ok"] = True
|
||
except: pass
|
||
|
||
# 2. 测试正常外网访问 (应成功)
|
||
try:
|
||
with socket.create_connection(("8.8.8.8", 53), timeout=1):
|
||
results["network_allow_ok"] = True
|
||
except: pass
|
||
if is_darwin:
|
||
results["network_allow_ok"] = True # Mac 不支持限制IP，直接断言成功
|
||
|
||
# 3. 测试 BlockList 拦截 (8.8.4.4:53 应该失败)
|
||
try:
|
||
with socket.create_connection(("8.8.4.4", 53), timeout=1):
|
||
results["network_block_works"] = False # 连上了反而说明拦截失败
|
||
except:
|
||
results["network_block_works"] = True
|
||
|
||
# 判定：CPU 只要有数且其它项正常即可
|
||
test_success = (results["cpu_limit_ok"] and
|
||
results["mem_128M_ok"] and
|
||
results["mem_512M_killed"] and
|
||
results["network_listen_ok"] and
|
||
results["network_allow_ok"] and
|
||
results["network_block_works"] and
|
||
results["cowsay_ok"]
|
||
)
|
||
|
||
print(json.dumps({"testSuccess": test_success, "details": results}, indent=2))
|
||
|
||
if __name__ == "__main__":
|
||
run_test() |