🐛 修复自检在ARM上的问题

This commit is contained in:
AkashiCoin 2024-09-07 15:00:11 +08:00
parent 62fd93c1ff
commit 3c3881af02
2 changed files with 145 additions and 14 deletions

View File

@ -0,0 +1,92 @@
from typing import cast
from pathlib import Path
from collections.abc import Callable
import nonebot
from nonebug import App
from respx import MockRouter
from pytest_mock import MockerFixture
from nonebot.adapters.onebot.v11 import Bot
from nonebot.adapters.onebot.v11.event import GroupMessageEvent
from tests.utils import _v11_group_message_event
from tests.config import BotId, UserId, GroupId, MessageId
async def test_check(
app: App,
mocker: MockerFixture,
mocked_api: MockRouter,
create_bot: Callable,
tmp_path: Path,
) -> None:
"""
测试自检
"""
from zhenxun.configs.config import BotConfig
from zhenxun.builtin_plugins.check import _matcher
from zhenxun.builtin_plugins.check.data_source import __get_version
mocker.patch("zhenxun.builtin_plugins.check.data_source.psutil")
mock_cpuinfo = mocker.patch("zhenxun.builtin_plugins.check.data_source.cpuinfo")
mock_platform = mocker.patch("zhenxun.builtin_plugins.check.data_source.platform")
mock_template_to_pic = mocker.patch("zhenxun.builtin_plugins.check.template_to_pic")
mock_template_to_pic_return = mocker.AsyncMock()
mock_template_to_pic.return_value = mock_template_to_pic_return
mock_build_message = mocker.patch(
"zhenxun.builtin_plugins.check.MessageUtils.build_message"
)
mock_build_message_return = mocker.AsyncMock()
mock_build_message.return_value = mock_build_message_return
mock_template_path_new = tmp_path / "resources" / "template"
mocker.patch(
"zhenxun.builtin_plugins.check.TEMPLATE_PATH", new=mock_template_path_new
)
async with app.test_matcher(_matcher) as ctx:
bot = create_bot(ctx)
bot: Bot = cast(Bot, bot)
raw_message = "自检"
event: GroupMessageEvent = _v11_group_message_event(
message=raw_message,
self_id=BotId.QQ_BOT,
user_id=UserId.SUPERUSER,
group_id=GroupId.GROUP_ID_LEVEL_5,
message_id=MessageId.MESSAGE_ID_3,
to_me=True,
)
ctx.receive_event(bot=bot, event=event)
mock_template_to_pic.assert_awaited_once_with(
template_path=str((mock_template_path_new / "check").absolute()),
template_name="main.html",
templates={
"data": {
"cpu_info": "1.0% - 1.0Ghz [1 core]",
"cpu_process": 1.0,
"ram_info": "1.0 / 1.0 GB",
"ram_process": 100.0,
"swap_info": "1.0 / 1.0 GB",
"swap_process": 100.0,
"disk_info": "1.0 / 1.0 GB",
"disk_process": 100.0,
"brand_raw": mock_cpuinfo.get_cpu_info().get(),
"baidu": "red",
"google": "red",
"system": f"{mock_platform.uname().system} "
f"{mock_platform.uname().release}",
"version": __get_version(),
"plugin_count": len(nonebot.get_loaded_plugins()),
"nickname": BotConfig.self_nickname,
}
},
pages={
"viewport": {"width": 195, "height": 750},
"base_url": f"file://{mock_template_path_new.absolute()}",
},
wait=2,
)
mock_build_message.assert_called_once_with(mock_template_to_pic_return)
mock_build_message_return.send.assert_awaited_once()

View File

@ -1,22 +1,24 @@
import os
import platform import platform
from dataclasses import dataclass import subprocess
from pathlib import Path from pathlib import Path
from dataclasses import dataclass
import psutil
import cpuinfo import cpuinfo
import nonebot import nonebot
import psutil
from httpx import ConnectTimeout, NetworkError
from nonebot.utils import run_sync
from pydantic import BaseModel from pydantic import BaseModel
from nonebot.utils import run_sync
from zhenxun.configs.config import BotConfig
from zhenxun.services.log import logger from zhenxun.services.log import logger
from zhenxun.configs.config import BotConfig
from zhenxun.utils.http_utils import AsyncHttpx from zhenxun.utils.http_utils import AsyncHttpx
BAIDU_URL = "https://www.baidu.com/" BAIDU_URL = "https://www.baidu.com/"
GOOGLE_URL = "https://www.google.com/" GOOGLE_URL = "https://www.google.com/"
VERSION_FILE = Path() / "__version__" VERSION_FILE = Path() / "__version__"
ARM_KEY = "aarch64"
@dataclass @dataclass
@ -31,9 +33,11 @@ class CPUInfo:
@classmethod @classmethod
def get_cpu_info(cls): def get_cpu_info(cls):
cpu_core = psutil.cpu_count(logical=False) cpu_core = psutil.cpu_count(logical=False)
cpu_usage = psutil.cpu_percent(interval=1) cpu_usage = psutil.cpu_percent(interval=0.1)
cpu_freq = round(psutil.cpu_freq().current / 1000, 2) if _cpu_freq := psutil.cpu_freq():
cpu_freq = round(_cpu_freq.current / 1000, 2)
else:
cpu_freq = 0
return CPUInfo(core=cpu_core, usage=cpu_usage, freq=cpu_freq) return CPUInfo(core=cpu_core, usage=cpu_usage, freq=cpu_freq)
@ -102,8 +106,9 @@ class SystemInfo(BaseModel):
def get_system_info(self): def get_system_info(self):
return { return {
"cpu_info": f"{self.cpu.usage}% - {self.cpu.freq}Ghz [{self.cpu.core} core]", "cpu_info": f"{self.cpu.usage}% - {self.cpu.freq}Ghz "
"cpu_process": psutil.cpu_percent(), f"[{self.cpu.core} core]",
"cpu_process": self.cpu.usage,
"ram_info": f"{self.ram.usage} / {self.ram.total} GB", "ram_info": f"{self.ram.usage} / {self.ram.total} GB",
"ram_process": ( "ram_process": (
0 if self.ram.total == 0 else (self.ram.usage / self.ram.total * 100) 0 if self.ram.total == 0 else (self.ram.usage / self.ram.total * 100)
@ -120,14 +125,14 @@ class SystemInfo(BaseModel):
@run_sync @run_sync
def __build_status() -> dict: def __build_status() -> SystemInfo:
"""获取 `CPU` `RAM` `SWAP` `DISK` 信息""" """获取 `CPU` `RAM` `SWAP` `DISK` 信息"""
cpu = CPUInfo.get_cpu_info() cpu = CPUInfo.get_cpu_info()
ram = RAMInfo.get_ram_info() ram = RAMInfo.get_ram_info()
swap = SwapMemory.get_swap_info() swap = SwapMemory.get_swap_info()
disk = DiskInfo.get_disk_info() disk = DiskInfo.get_disk_info()
return SystemInfo(cpu=cpu, ram=ram, swap=swap, disk=disk).get_system_info() return SystemInfo(cpu=cpu, ram=ram, swap=swap, disk=disk)
async def __get_network_info(): async def __get_network_info():
@ -155,16 +160,50 @@ def __get_version() -> str | None:
return None return None
def __get_arm_cpu():
env = os.environ.copy()
env["LC_ALL"] = "en_US.UTF-8"
cpu_info = subprocess.check_output(["lscpu"], env=env).decode()
model_name = ""
cpu_freq = 0
for line in cpu_info.splitlines():
if "Model name" in line:
model_name = line.split(":")[1].strip()
if "CPU MHz" in line:
cpu_freq = float(line.split(":")[1].strip())
return model_name, cpu_freq
def __get_arm_oracle_cpu_freq():
cpu_freq = subprocess.check_output(
["dmidecode", "-s", "processor-frequency"]
).decode()
return round(float(cpu_freq.split()[0]) / 1000, 2)
async def get_status_info() -> dict: async def get_status_info() -> dict:
"""获取信息""" """获取信息"""
data = await __build_status() data = await __build_status()
system = platform.uname()
if system.machine == ARM_KEY and not (
cpuinfo.get_cpu_info().get("brand_raw") and data.cpu.freq
):
model_name, cpu_freq = __get_arm_cpu()
if not data.cpu.freq:
data.cpu.freq = cpu_freq or __get_arm_oracle_cpu_freq()
data = data.get_system_info()
data["brand_raw"] = model_name
else:
data = data.get_system_info()
data["brand_raw"] = cpuinfo.get_cpu_info().get("brand_raw", "Unknown")
baidu, google = await __get_network_info() baidu, google = await __get_network_info()
data["baidu"] = "#8CC265" if baidu else "red" data["baidu"] = "#8CC265" if baidu else "red"
data["google"] = "#8CC265" if google else "red" data["google"] = "#8CC265" if google else "red"
system = platform.uname()
data["system"] = f"{system.system} {system.release}" data["system"] = f"{system.system} {system.release}"
data["version"] = __get_version() data["version"] = __get_version()
data["brand_raw"] = cpuinfo.get_cpu_info()["brand_raw"]
data["plugin_count"] = len(nonebot.get_loaded_plugins()) data["plugin_count"] = len(nonebot.get_loaded_plugins())
data["nickname"] = BotConfig.self_nickname data["nickname"] = BotConfig.self_nickname
return data return data