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()
|