diff --git a/README.md b/README.md index 8d8e85c..f98abd3 100644 --- a/README.md +++ b/README.md @@ -4,12 +4,13 @@ ## **API 说明** -| **端点** | **方法** | **功能** | **参数** | **成功返回** | **失败返回** | -| ------------- | ---------- | ---------------- | ------------------------------------------ | -------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- | -| **`/screen`** | `GET` | 获取屏幕截图 | - `r`(高斯模糊半径)
- `k`(API 密钥) | - `200 OK`,返回 `image/jpeg` 截图 | - `401 Unauthorized`:配置了 `api_key` 且低模糊度密钥错误
- `403 Forbidden`:私密模式
- `500 Internal Server Error`:截图失败 | -| **`/record`** | `GET` | 获取最近录音 | 无 | - `200 OK`,返回 `audio/wav` 录音文件 | - `403 Forbidden`:私密模式
- `500 Internal Server Error`:录音失败 | -| **`/idle`** | `GET` | 获取用户空闲时间 | 无 | - `200 OK`,返回 JSON:`{"idle_seconds": 123.456, "last_input_time": "2026-02-09T23:00:00+08:00"}` | - `403 Forbidden`:私密模式 | -| **`/check`** | `GET/POST` | 检查是否运行 | 无 | - `200 OK` | 无 | +| **端点** | **方法** | **功能** | **参数** | **成功返回** | **失败返回** | +| ------------- | ---------- | ---------------- | ------------------------------------------ | ----------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- | +| **`/screen`** | `GET` | 获取屏幕截图 | - `r`(高斯模糊半径)
- `k`(API 密钥) | - `200 OK`,返回 `image/jpeg` 截图 | - `401 Unauthorized`:配置了 `api_key` 且低模糊度密钥错误
- `403 Forbidden`:私密模式
- `500 Internal Server Error`:截图失败 | +| **`/record`** | `GET` | 获取最近录音 | 无 | - `200 OK`,返回 `audio/wav` 录音文件 | - `403 Forbidden`:私密模式
- `500 Internal Server Error`:录音失败 | +| **`/idle`** | `GET` | 获取用户空闲时间 | 无 | - `200 OK`,返回 JSON:`{"idle_seconds": 123.456, "last_input_time": "..."}` | - `403 Forbidden`:私密模式 | +| **`/info`** | `GET` | 获取设备信息 | 无 | - `200 OK`,返回 JSON:`{"hostname": "PC", "cpu": "Intel...", "gpus": [...]}` | - `403 Forbidden`:私密模式 | +| **`/check`** | `GET/POST` | 检查是否运行 | 无 | - `200 OK` | 无 | ## **使用** @@ -50,6 +51,7 @@ uv run pyinstaller peekapi.spec ```toml [basic] is_public = true # 程序启动时默认是否为公开模式 +device_name = "" # 设备名称,留空则使用系统主机名 api_key = "" # 低模糊度下获取截图的key,留空则不需要key host = "0.0.0.0" # 监听IP port = 1920 # 监听端口 @@ -68,6 +70,7 @@ gain = 20 # 音量增益倍数 | **参数** | **说明** | **默认值** | | ---------------------- | -------------------------------------------------- | ----------- | | **`is_public`** | 程序启动时默认是否为公开模式 | `true` | +| **`device_name`** | 设备名称,留空则使用系统主机名 | `""` | | **`api_key`** | 低模糊度下获取截图的密钥,留空则不需要key | `""` | | **`host`** | 监听 IP | `"0.0.0.0"` | | **`port`** | 监听端口 | `1920` | diff --git a/justfile b/justfile index 30e7f93..5632eca 100644 --- a/justfile +++ b/justfile @@ -56,10 +56,13 @@ check: # ===== 版本管理 ===== -# 版本发布(更新版本号、更新 lock 文件) +# 版本发布(更新版本号、更新 lock 文件、自动修正标签) bump: uv run cz bump uv lock + git add uv.lock + git commit --amend --no-edit --no-verify + $version = (uv run cz version --project).Trim(); git tag -f "v$version" # 生成 changelog changelog: @@ -74,3 +77,15 @@ hooks: # 更新 prek hooks update: uv run prek auto-update + +# 从 dev 向 main 创建 PR +pr: + gh pr create --base main --head dev --fill + gh pr view --web + +# PR 合并后同步 dev 到 main +sync: + git fetch origin + git checkout dev + git reset --hard origin/main + git push origin dev --force-with-lease diff --git a/memory-bank/activeContext.md b/memory-bank/activeContext.md index 00fdba7..59ec580 100644 --- a/memory-bank/activeContext.md +++ b/memory-bank/activeContext.md @@ -4,11 +4,16 @@ 项目处于稳定运行状态,核心功能已完成,CI 检查全部通过。当前关注点: - CI/CD 配置优化(已完成) +- 添加设备信息端点 (`/info`)(已完成) - Linux 平台支持(计划中) - GitHub Release 自动发布(计划中) ## 最近变更 +- 2026-02-10: 完成设备信息端点 (`/info`) + - 使用 PowerShell WMI 查询硬件信息 + - 返回主机名、电脑型号、主板、CPU、显卡 + - 支持 device_name 配置覆盖主机名 - 2026-02-09: 添加用户空闲时间端点 (`/idle`) - 使用 Windows GetLastInputInfo API 获取最后操作时间 - 返回空闲秒数和最后操作时间 (ISO 格式) diff --git a/memory-bank/tasks/TASK010-github-release-exe.md b/memory-bank/tasks/TASK010-github-release-exe.md index 6e73592..7589ced 100644 --- a/memory-bank/tasks/TASK010-github-release-exe.md +++ b/memory-bank/tasks/TASK010-github-release-exe.md @@ -1,8 +1,8 @@ # [TASK010] - GitHub Release 自动发布 EXE -**Status:** Pending +**Status:** Completed **Added:** 2026-02-02 -**Updated:** 2026-02-02 +**Updated:** 2026-02-10 ## Original Request @@ -220,14 +220,14 @@ jobs: ## Implementation Plan - [x] 1.1 创建基础 release.yml 工作流文件(含版本校验逻辑) -- [ ] 1.2 集成 TheDoctor0/zip-release 打包 -- [ ] 1.3 配置 Release 产出物名称 -- [ ] 1.4 可选:添加 SHA256 校验和 -- [ ] 1.5 测试完整发布流程 +- [x] 1.2 集成 TheDoctor0/zip-release 打包 +- [x] 1.3 配置 Release 产出物名称 +- [-] 1.4 可选:添加 SHA256 校验和 (已跳过) +- [x] 1.5 测试完整发布流程 ## Progress Tracking -**Overall Status:** Pending - 15% +**Overall Status:** Completed - 100% ### Subtasks | ID | Description | Status | Updated | Notes | @@ -250,3 +250,7 @@ jobs: - 参数: filename, directory, path, type, exclusions 等 - 设计完整的发布方案,采用 TheDoctor0/zip-release 作为压缩方案 +### 2026-02-10 +- 完成 release.yml 配置,修复版本检查逻辑 +- 标记任务完成 + diff --git a/memory-bank/tasks/TASK014-fix-gettickcount-overflow.md b/memory-bank/tasks/TASK014-fix-gettickcount-overflow.md new file mode 100644 index 0000000..d162a3a --- /dev/null +++ b/memory-bank/tasks/TASK014-fix-gettickcount-overflow.md @@ -0,0 +1,89 @@ +# [TASK014] - 修复 GetTickCount 溢出问题 + +**Status:** Completed +**Added:** 2026-02-10 +**Updated:** 2026-02-10 +**Priority:** High + +## Original Request + +`GetTickCount` 默认 `restype` 是有符号 `c_int`,在系统运行超过约 24.9 天后会变成负数,或在 49.7 天左右发生回绕,导致 `idle_seconds` 计算错误甚至为负数。 + +## 问题分析 + +### 当前代码(修复前) + +```python +# idle.py 第 38-40 行 +current_tick = ctypes.windll.kernel32.GetTickCount() +idle_ms = current_tick - lii.dwTime +idle_seconds = idle_ms / 1000.0 +``` + +### 问题 + +1. `ctypes.windll.kernel32.GetTickCount()` 默认返回类型是 `c_int`(有符号 32 位) +2. `GetTickCount` 实际返回 `DWORD`(无符号 32 位),范围 0 ~ 4,294,967,295 +3. 当值超过 2,147,483,647(约 24.9 天)时,Python 会将其解释为负数 +4. 49.7 天后发生回绕(wrap around),导致计算错误 + +### 影响 + +- 系统运行超过 ~25 天后,`idle_seconds` 可能变为负数 +- 用户操作时间显示错误 +- 多主机选择功能(nonebot-plugin-peek TASK003)依赖此端点,会导致选择错误主机 + +## Implementation Plan + +### 采用方案 A:使用 GetTickCount64 + +```python +def get_idle_info() -> tuple[float, datetime]: + lii = LASTINPUTINFO() + lii.cbSize = sizeof(LASTINPUTINFO) + ctypes.windll.user32.GetLastInputInfo(ctypes.byref(lii)) + + # 使用 GetTickCount64 避免溢出(返回 64 位无符号整数) + GetTickCount64 = ctypes.windll.kernel32.GetTickCount64 + GetTickCount64.restype = ctypes.c_uint64 + current_tick = GetTickCount64() + + # dwTime 仍是 32 位,需处理回绕 + current_tick_32 = current_tick & 0xFFFFFFFF + if current_tick_32 >= lii.dwTime: + idle_ms = current_tick_32 - lii.dwTime + else: + # 回绕:dwTime 在 current_tick 回绕之前设置 + idle_ms = (0xFFFFFFFF - lii.dwTime) + current_tick_32 + 1 + + idle_seconds = idle_ms / 1000.0 + last_input_time = datetime.now(_BEIJING_TZ) - timedelta(seconds=idle_seconds) + + return idle_seconds, last_input_time +``` + +## Progress Tracking + +**Overall Status:** Completed - 100% + +### Subtasks + +| ID | Description | Status | Updated | Notes | +| --- | -------------------------------- | -------- | ---------- | --------------------- | +| 1 | 修改 idle.py 使用 GetTickCount64 | Complete | 2026-02-10 | 使用 c_uint64 | +| 2 | 添加 32 位回绕处理 | Complete | 2026-02-10 | dwTime 仍是 32 位 | +| 3 | 添加单元测试覆盖边界情况 | Complete | 2026-02-10 | 增加 2 个回绕测试用例 | +| 4 | 更新文档 | Complete | 2026-02-10 | 任务文档已更新 | + +## Progress Log + +### 2026-02-10 +- 创建任务 +- 分析问题根因 +- 设计修复方案 +- 实现 GetTickCount64 修复(方案 A) +- 添加 32 位回绕边界测试用例 + - `test_get_idle_info_handles_32bit_wraparound` - 测试回绕情况 + - `test_get_idle_info_large_tick_count` - 测试接近上限值 +- 所有 7 个 idle 测试通过 +- 任务完成 diff --git a/memory-bank/tasks/TASK015-add-system-info.md b/memory-bank/tasks/TASK015-add-system-info.md new file mode 100644 index 0000000..41113a4 --- /dev/null +++ b/memory-bank/tasks/TASK015-add-system-info.md @@ -0,0 +1,43 @@ +# TASK015 - 添加设备信息端点与主机名配置 + +**Status:** Completed +**Added:** 2026-02-09 +**Updated:** 2026-02-10 + +## Original Request +用户希望通过 API 获取设备硬件信息(主机名、型号、主板、CPU、显卡),并希望在配置文件中添加一个选项,如果设置了该选项,API 返回的主机名将使用配置值替代系统真实主机名。 + +## Thought Process +用户需要了解运行服务的宿主机硬件情况,用于多设备管理或展示。 +考虑到跨平台兼容性和轻量化,决定使用 PowerShell 命令获取 Windows WMI 信息,无需额外依赖。 +配置项 `device_name` 添加到 `[basic]` 节中,留空则使用系统主机名。 + +## Implementation Plan +- [x] Update `README.md` to document the new `/info` endpoint and `device_name` config. +- [x] Create `src/peekapi/system_info.py` to handle PowerShell commands and parsing. +- [x] Update `src/peekapi/config.py` to include `device_name` field. +- [x] Add `/info` route in `src/peekapi/server.py`. +- [x] Add unit tests for system info retrieval and config override. + +## Progress Tracking + +**Overall Status:** Completed - 100% + +### Subtasks +| ID | Description | Status | Updated | Notes | +| --- | ---------------------- | -------- | ---------- | ------------------------------ | +| 1 | Update Documentation | Complete | 2026-02-10 | README, TASK 文件 | +| 2 | Implement System Info | Complete | 2026-02-10 | system_info.py 使用 PowerShell | +| 3 | Implement Config Logic | Complete | 2026-02-10 | device_name 字段 | +| 4 | Add API Endpoint | Complete | 2026-02-10 | /info 路由 | +| 5 | Add Unit Tests | Complete | 2026-02-10 | 13 测试通过 | + +## Progress Log +### 2026-02-10 +- Implemented `system_info.py` module with PowerShell WMI queries +- Added `device_name` config option to `BasicConfig` +- Added `/info` endpoint to `server.py` +- Created comprehensive unit tests (13 tests) +- Fixed existing mock test for `test_idle.py` (GetTickCount64 compatibility) +- All 106 tests pass +- ruff and basedpyright checks pass diff --git a/memory-bank/tasks/_index.md b/memory-bank/tasks/_index.md index fe483be..c50649e 100644 --- a/memory-bank/tasks/_index.md +++ b/memory-bank/tasks/_index.md @@ -18,12 +18,14 @@ ## Pending - [TASK001] 支持 Linux 平台 - 替换 Windows 专用依赖,实现跨平台支持 -- [TASK010] GitHub Release 自动发布 EXE - 配置 CI 自动打包并发布到 GitHub Releases --- ## Completed +- [TASK014] 修复 GetTickCount 溢出问题 - ✅ 2026-02-10 完成(使用 GetTickCount64) +- [TASK010] GitHub Release 自动发布 EXE - ✅ 2026-02-10 完成(配置 CI 自动打包并发布到 GitHub Releases) +- [TASK015] 添加设备信息端点 - ✅ 2026-02-10 完成(/info 端点获取硬件信息) - [TASK013] 添加用户空闲时间端点 - ✅ 2026-02-09 完成(/idle 端点获取用户空闲时间) - [TASK012] 系统托盘图标不显示 - ✅ 2026-02-03 完成(修复 stderr 为 None 导致崩溃) - [TASK011] CI 代码检查修复 - ✅ 2026-02-02 完成(ruff、basedpyright、pytest 全部通过) @@ -49,11 +51,11 @@ | 状态 | 数量 | | ----------- | ------ | | In Progress | 0 | -| Pending | 2 | -| Completed | 10 | +| Pending | 1 | +| Completed | 13 | | Abandoned | 0 | -| **总计** | **12** | +| **总计** | **14** | --- -*最后更新: 2026-02-09* +*最后更新: 2026-02-10* diff --git a/src/peekapi/config.py b/src/peekapi/config.py index d1c3890..9de8258 100644 --- a/src/peekapi/config.py +++ b/src/peekapi/config.py @@ -9,6 +9,7 @@ class BasicConfig(Struct): """基础配置""" is_public: bool = True + device_name: str = "" # 设备名称,留空则使用系统主机名 api_key: str = "" host: str = "0.0.0.0" port: int = 1920 diff --git a/src/peekapi/idle.py b/src/peekapi/idle.py index 09cf2a0..6c7fb7b 100644 --- a/src/peekapi/idle.py +++ b/src/peekapi/idle.py @@ -34,9 +34,20 @@ def get_idle_info() -> tuple[float, datetime]: lii.cbSize = sizeof(LASTINPUTINFO) ctypes.windll.user32.GetLastInputInfo(ctypes.byref(lii)) - # GetTickCount 返回系统启动后的毫秒数 - current_tick = ctypes.windll.kernel32.GetTickCount() - idle_ms = current_tick - lii.dwTime + # 使用 GetTickCount64 避免 49.7 天溢出问题 + GetTickCount64 = ctypes.windll.kernel32.GetTickCount64 + GetTickCount64.restype = ctypes.c_uint64 + current_tick = GetTickCount64() + + # dwTime 是 32 位,需要处理回绕 + # 取 current_tick 的低 32 位与 dwTime 比较 + current_tick_32 = current_tick & 0xFFFFFFFF + if current_tick_32 >= lii.dwTime: + idle_ms = current_tick_32 - lii.dwTime + else: + # 回绕情况:dwTime 在 GetTickCount 回绕之前设置 + idle_ms = (0xFFFFFFFF - lii.dwTime) + current_tick_32 + 1 + idle_seconds = idle_ms / 1000.0 # 计算最后操作时间 diff --git a/src/peekapi/screenshot.py b/src/peekapi/screenshot.py index 6facbec..53f0bf2 100644 --- a/src/peekapi/screenshot.py +++ b/src/peekapi/screenshot.py @@ -1,4 +1,5 @@ import io +import math import mss from PIL import Image, ImageFilter @@ -15,7 +16,7 @@ def screenshot(radius: float, main_screen_only: bool) -> bytes: img_pil = Image.frombytes("RGB", img.size, img.rgb) - if radius > 0: + if math.isfinite(radius) and radius > 0: img_pil = img_pil.filter(ImageFilter.GaussianBlur(radius=radius)) img_byte = io.BytesIO() diff --git a/src/peekapi/server.py b/src/peekapi/server.py index 0dcf105..0108f9b 100644 --- a/src/peekapi/server.py +++ b/src/peekapi/server.py @@ -1,3 +1,4 @@ +import math from contextlib import asynccontextmanager from threading import Thread @@ -10,6 +11,7 @@ from .logging import logger, setup_logging from .record import recorder from .screenshot import screenshot +from .system_info import get_system_info from .system_tray import start_system_tray @@ -53,6 +55,11 @@ def screen_route( """获取屏幕截图""" client_ip = request.client.host if request.client else "unknown" + # 拒绝 NaN / Inf 等非有限浮点数,防止绕过鉴权和模糊 + if not math.isfinite(r): + logger.info(f"[{client_ip}] 截图请求被拒绝: 非法半径值 (r={r})") + raise HTTPException(status_code=401, detail="模糊半径必须为有限数值") + if not config.basic.is_public: logger.info(f"[{client_ip}] 截图请求被拒绝: 私密模式") raise HTTPException(status_code=403, detail="瑟瑟中") @@ -112,6 +119,20 @@ def idle_route(request: Request): } +@app.get("/info") +def info_route(request: Request): + """获取设备信息""" + client_ip = request.client.host if request.client else "unknown" + + if not config.basic.is_public: + logger.info(f"[{client_ip}] 设备信息请求被拒绝: 私密模式") + raise HTTPException(status_code=403, detail="瑟瑟中") + + info = get_system_info(config.basic.device_name) + logger.info(f"[{client_ip}] 设备信息请求成功") + return info + + @app.get("/check") @app.post("/check") def check_route(): diff --git a/src/peekapi/system_info.py b/src/peekapi/system_info.py new file mode 100644 index 0000000..3458dc6 --- /dev/null +++ b/src/peekapi/system_info.py @@ -0,0 +1,109 @@ +"""系统信息获取模块 + +使用 PowerShell 查询 Windows WMI 获取硬件信息。 +""" + +import json +import socket +import subprocess +from typing import TypedDict + +from .logging import logger + + +class SystemInfo(TypedDict): + """系统信息类型""" + + hostname: str + computer_model: str + motherboard: str + cpu: str + gpus: list[str] + + +def _run_powershell(command: str) -> dict | list | None: + """执行 PowerShell 命令并返回 JSON 解析结果""" + try: + result = subprocess.run( + ["powershell", "-Command", command], + capture_output=True, + text=True, + timeout=10, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + if result.returncode == 0 and result.stdout.strip(): + return json.loads(result.stdout) + except (subprocess.TimeoutExpired, json.JSONDecodeError, Exception) as e: + logger.warning(f"PowerShell 命令执行失败: {e}") + return None + + +def _get_computer_model() -> str: + """获取电脑型号""" + data = _run_powershell( + "Get-CimInstance Win32_ComputerSystem | Select-Object Model | ConvertTo-Json" + ) + if isinstance(data, dict): + return data.get("Model", "Unknown") + return "Unknown" + + +def _get_motherboard() -> str: + """获取主板信息""" + data = _run_powershell( + "Get-CimInstance Win32_BaseBoard | Select-Object Manufacturer, Product | ConvertTo-Json" + ) + if isinstance(data, dict): + manufacturer = data.get("Manufacturer", "") + product = data.get("Product", "") + if manufacturer and product: + return f"{manufacturer} {product}" + return manufacturer or product or "Unknown" + return "Unknown" + + +def _get_cpu() -> str: + """获取 CPU 型号""" + data = _run_powershell( + "Get-CimInstance Win32_Processor | Select-Object Name | ConvertTo-Json" + ) + if isinstance(data, dict): + return data.get("Name", "Unknown") + # 多个 CPU 的情况 + if isinstance(data, list) and data: + return data[0].get("Name", "Unknown") + return "Unknown" + + +def _get_gpus() -> list[str]: + """获取显卡型号列表""" + data = _run_powershell( + "Get-CimInstance Win32_VideoController | Select-Object Name | ConvertTo-Json" + ) + if isinstance(data, dict): + name = data.get("Name", "") + return [name] if name else [] + if isinstance(data, list): + return [item.get("Name", "") for item in data if item.get("Name")] + return [] + + +def get_system_info(device_name_override: str = "") -> SystemInfo: + """ + 获取系统硬件信息 + + Args: + device_name_override: 可选的设备名称覆盖,如果提供则替代系统主机名 + + Returns: + SystemInfo: 包含主机名、电脑型号、主板、CPU、显卡信息的字典 + """ + hostname = device_name_override if device_name_override else socket.gethostname() + + return SystemInfo( + hostname=hostname, + computer_model=_get_computer_model(), + motherboard=_get_motherboard(), + cpu=_get_cpu(), + gpus=_get_gpus(), + ) diff --git a/tests/unit/test_idle.py b/tests/unit/test_idle.py index 93c74d5..87896c6 100644 --- a/tests/unit/test_idle.py +++ b/tests/unit/test_idle.py @@ -66,8 +66,9 @@ def test_get_idle_info_with_mocked_windows_api(self): """测试使用 mock 的 Windows API""" mock_windll = MagicMock() - # 模拟 GetTickCount 返回 10000ms (10秒) - mock_windll.kernel32.GetTickCount.return_value = 10000 + # 模拟 GetTickCount64 返回 10000ms (10秒) + mock_get_tick_count_64 = MagicMock(return_value=10000) + mock_windll.kernel32.GetTickCount64 = mock_get_tick_count_64 # LASTINPUTINFO.dwTime 模拟返回 5000ms (5秒前的最后输入) # 这意味着空闲时间为 10000 - 5000 = 5000ms = 5秒 @@ -87,3 +88,65 @@ def mock_get_last_input_info(lii_ref): # 验证最后输入时间 assert isinstance(last_input_time, datetime) + + def test_get_idle_info_handles_32bit_wraparound(self): + """测试 32 位回绕情况(dwTime 在回绕前设置,current_tick 在回绕后)""" + mock_windll = MagicMock() + + # 模拟 GetTickCount64 返回值:假设已超过 49.7 天 + # current_tick 的低 32 位为 1000(刚回绕后) + # 即 current_tick = 0x100000000 + 1000 = 4294968296 + current_tick_64 = 0x100000000 + 1000 # 回绕后 1 秒 + mock_get_tick_count_64 = MagicMock(return_value=current_tick_64) + mock_windll.kernel32.GetTickCount64 = mock_get_tick_count_64 + + # dwTime 在回绕前设置,假设为 0xFFFFFFFF - 4000 = 4294963295 + # 即在回绕前 4 秒 + dwTime_before_wrap = 0xFFFFFFFF - 4000 + + def mock_get_last_input_info(lii_ref): + lii_ref._obj.dwTime = dwTime_before_wrap + return True + + mock_windll.user32.GetLastInputInfo.side_effect = mock_get_last_input_info + + with patch("peekapi.idle.ctypes.windll", mock_windll): + from peekapi.idle import get_idle_info + + idle_seconds, last_input_time = get_idle_info() + + # 预期空闲时间:从 dwTime 到回绕点 (4001ms) + 回绕后到 current_tick (1000ms) + # = (0xFFFFFFFF - dwTime) + current_tick_32 + 1 + # = 4001 + 1000 = 5001ms = 5.001 秒 + expected_idle_ms = (0xFFFFFFFF - dwTime_before_wrap) + 1000 + 1 + expected_idle_seconds = expected_idle_ms / 1000.0 + + assert abs(idle_seconds - expected_idle_seconds) < 0.1 + assert isinstance(last_input_time, datetime) + + def test_get_idle_info_large_tick_count(self): + """测试大 tick count 值(接近 32 位上限但未回绕)""" + mock_windll = MagicMock() + + # 模拟系统运行接近 49.7 天,current_tick 接近 32 位上限 + current_tick_64 = 0xFFFFF000 # 接近 32 位最大值 + mock_get_tick_count_64 = MagicMock(return_value=current_tick_64) + mock_windll.kernel32.GetTickCount64 = mock_get_tick_count_64 + + # dwTime 为 current_tick - 3000(3 秒前) + dwTime = current_tick_64 - 3000 + + def mock_get_last_input_info(lii_ref): + lii_ref._obj.dwTime = dwTime + return True + + mock_windll.user32.GetLastInputInfo.side_effect = mock_get_last_input_info + + with patch("peekapi.idle.ctypes.windll", mock_windll): + from peekapi.idle import get_idle_info + + idle_seconds, last_input_time = get_idle_info() + + # 预期空闲时间:3 秒 + assert abs(idle_seconds - 3.0) < 0.1 + assert isinstance(last_input_time, datetime) diff --git a/tests/unit/test_screenshot.py b/tests/unit/test_screenshot.py index f81b7d3..42dd61f 100644 --- a/tests/unit/test_screenshot.py +++ b/tests/unit/test_screenshot.py @@ -120,3 +120,21 @@ def test_screenshot_blur_applies_gaussian(self, mock_mss): # 获取传入 filter 的参数 call_args = mock_filter.call_args[0][0] assert isinstance(call_args, ImageFilter.GaussianBlur) + + def test_screenshot_nan_radius_no_blur(self, mock_mss): + """验证 radius=NaN 时不调用 filter(防御性检查)""" + from peekapi.screenshot import screenshot + + with patch("PIL.Image.Image.filter") as mock_filter: + screenshot(radius=float("nan"), main_screen_only=True) + + mock_filter.assert_not_called() + + def test_screenshot_inf_radius_no_blur(self, mock_mss): + """验证 radius=Inf 时不调用 filter(防御性检查)""" + from peekapi.screenshot import screenshot + + with patch("PIL.Image.Image.filter") as mock_filter: + screenshot(radius=float("inf"), main_screen_only=True) + + mock_filter.assert_not_called() diff --git a/tests/unit/test_server.py b/tests/unit/test_server.py index 2f909f8..ce7f136 100644 --- a/tests/unit/test_server.py +++ b/tests/unit/test_server.py @@ -211,6 +211,37 @@ def test_screen_wrong_api_key(self, app_client): assert response.status_code == 401 + def test_screen_nan_radius_rejected(self, app_client): + """NaN radius 返回 422,防止绕过鉴权和模糊""" + app_client["config"].basic.api_key = "secret123" + app_client["config"].screenshot.radius_threshold = 10 + + with patch("peekapi.server.screenshot", return_value=b"test") as mock_ss: + response = app_client["client"].get("/screen?r=nan") + + assert response.status_code == 401 + mock_ss.assert_not_called() + + def test_screen_inf_radius_rejected(self, app_client): + """Inf radius 返回 422""" + app_client["config"].basic.api_key = "secret123" + + with patch("peekapi.server.screenshot", return_value=b"test") as mock_ss: + response = app_client["client"].get("/screen?r=inf") + + assert response.status_code == 401 + mock_ss.assert_not_called() + + def test_screen_negative_inf_radius_rejected(self, app_client): + """-Inf radius 返回 422""" + app_client["config"].basic.api_key = "secret123" + + with patch("peekapi.server.screenshot", return_value=b"test") as mock_ss: + response = app_client["client"].get("/screen?r=-inf") + + assert response.status_code == 401 + mock_ss.assert_not_called() + # ============ /record 端点测试 ============ def test_record_public_mode_returns_audio(self, app_client): diff --git a/tests/unit/test_system_info.py b/tests/unit/test_system_info.py new file mode 100644 index 0000000..9049c72 --- /dev/null +++ b/tests/unit/test_system_info.py @@ -0,0 +1,249 @@ +"""系统信息模块测试""" + +import socket +import sys +from unittest.mock import MagicMock, patch + +import pytest + + +class TestGetSystemInfo: + """get_system_info 函数测试""" + + @pytest.mark.skipif(sys.platform != "win32", reason="仅 Windows 平台支持") + def test_get_system_info_returns_dict(self): + """验证返回类型为字典""" + from peekapi.system_info import get_system_info + + result = get_system_info() + + assert isinstance(result, dict) + assert "hostname" in result + assert "computer_model" in result + assert "motherboard" in result + assert "cpu" in result + assert "gpus" in result + + @pytest.mark.skipif(sys.platform != "win32", reason="仅 Windows 平台支持") + def test_get_system_info_hostname_default(self): + """验证默认使用系统主机名""" + from peekapi.system_info import get_system_info + + result = get_system_info() + + # 不传参数时应该使用系统主机名 + assert result["hostname"] == socket.gethostname() + + @pytest.mark.skipif(sys.platform != "win32", reason="仅 Windows 平台支持") + def test_get_system_info_hostname_override(self): + """验证 device_name 覆盖主机名""" + from peekapi.system_info import get_system_info + + custom_name = "MyCustomDevice" + result = get_system_info(device_name_override=custom_name) + + assert result["hostname"] == custom_name + + @pytest.mark.skipif(sys.platform != "win32", reason="仅 Windows 平台支持") + def test_get_system_info_hostname_empty_string_uses_system(self): + """验证空字符串使用系统主机名""" + from peekapi.system_info import get_system_info + + result = get_system_info(device_name_override="") + + assert result["hostname"] == socket.gethostname() + + @pytest.mark.skipif(sys.platform != "win32", reason="仅 Windows 平台支持") + def test_get_system_info_gpus_is_list(self): + """验证 gpus 是列表类型""" + from peekapi.system_info import get_system_info + + result = get_system_info() + + assert isinstance(result["gpus"], list) + + @pytest.mark.skipif(sys.platform != "win32", reason="仅 Windows 平台支持") + def test_get_system_info_all_fields_are_strings(self): + """验证除 gpus 外所有字段都是字符串""" + from peekapi.system_info import get_system_info + + result = get_system_info() + + assert isinstance(result["hostname"], str) + assert isinstance(result["computer_model"], str) + assert isinstance(result["motherboard"], str) + assert isinstance(result["cpu"], str) + + +class TestSystemInfoMocked: + """使用 mock 的系统信息测试(跨平台)""" + + def test_get_system_info_with_mocked_powershell(self): + """测试使用 mock 的 PowerShell 调用""" + mock_run = MagicMock() + + # 模拟 PowerShell 返回的 JSON + def side_effect(args, **kwargs): + cmd = args[2] if len(args) > 2 else "" + result = MagicMock() + result.returncode = 0 + + if "Win32_ComputerSystem" in cmd: + result.stdout = '{"Model": "TestModel"}' + elif "Win32_BaseBoard" in cmd: + result.stdout = '{"Manufacturer": "TestMfg", "Product": "TestBoard"}' + elif "Win32_Processor" in cmd: + result.stdout = '{"Name": "TestCPU"}' + elif "Win32_VideoController" in cmd: + result.stdout = '{"Name": "TestGPU"}' + else: + result.stdout = "{}" + + return result + + mock_run.side_effect = side_effect + + with patch("peekapi.system_info.subprocess.run", mock_run): + from peekapi.system_info import get_system_info + + result = get_system_info("MockedHost") + + assert result["hostname"] == "MockedHost" + assert result["computer_model"] == "TestModel" + assert result["motherboard"] == "TestMfg TestBoard" + assert result["cpu"] == "TestCPU" + assert result["gpus"] == ["TestGPU"] + + def test_get_system_info_handles_powershell_failure(self): + """测试 PowerShell 失败时返回 Unknown""" + mock_run = MagicMock() + mock_run.return_value.returncode = 1 + mock_run.return_value.stdout = "" + + with patch("peekapi.system_info.subprocess.run", mock_run): + from peekapi.system_info import get_system_info + + result = get_system_info("TestHost") + + assert result["hostname"] == "TestHost" + assert result["computer_model"] == "Unknown" + assert result["motherboard"] == "Unknown" + assert result["cpu"] == "Unknown" + assert result["gpus"] == [] + + def test_get_system_info_handles_multiple_gpus(self): + """测试多显卡情况""" + mock_run = MagicMock() + + def side_effect(args, **kwargs): + cmd = args[2] if len(args) > 2 else "" + result = MagicMock() + result.returncode = 0 + + if "Win32_VideoController" in cmd: + result.stdout = '[{"Name": "GPU1"}, {"Name": "GPU2"}]' + else: + result.stdout = '{"Model": "Test", "Manufacturer": "Test", "Product": "Test", "Name": "Test"}' + + return result + + mock_run.side_effect = side_effect + + with patch("peekapi.system_info.subprocess.run", mock_run): + from peekapi.system_info import get_system_info + + result = get_system_info() + + assert result["gpus"] == ["GPU1", "GPU2"] + + def test_get_system_info_handles_timeout(self): + """测试 PowerShell 超时情况""" + import subprocess + + mock_run = MagicMock() + mock_run.side_effect = subprocess.TimeoutExpired(cmd="test", timeout=10) + + with patch("peekapi.system_info.subprocess.run", mock_run): + from peekapi.system_info import get_system_info + + result = get_system_info("TestHost") + + # 超时时应返回 Unknown + assert result["hostname"] == "TestHost" + assert result["computer_model"] == "Unknown" + + +class TestInfoRouteIntegration: + """Server /info 路由集成测试""" + + @pytest.fixture + def app_client(self): + """创建 FastAPI 测试客户端""" + from io import BytesIO + + from fastapi.testclient import TestClient + + with patch("peekapi.server.recorder") as mock_recorder: + with patch("peekapi.server.config") as mock_config: + # 设置默认配置 + mock_config.basic.is_public = True + mock_config.basic.device_name = "" + mock_config.basic.api_key = "" + mock_config.basic.host = "127.0.0.1" + mock_config.basic.port = 8000 + mock_config.screenshot.radius_threshold = 10 + mock_config.screenshot.main_screen_only = True + + # Mock recorder + mock_audio = BytesIO(b"RIFF" + b"\x00" * 40) + mock_audio.seek(0) + mock_recorder.get_audio.return_value = mock_audio + + from peekapi.server import app + + client = TestClient(app, raise_server_exceptions=False) + + yield { + "client": client, + "config": mock_config, + "recorder": mock_recorder, + } + + def test_info_route_public_mode(self, app_client): + """测试公开模式下 /info 返回成功""" + with patch("peekapi.server.get_system_info") as mock_info: + mock_info.return_value = { + "hostname": "TestPC", + "computer_model": "TestModel", + "motherboard": "TestBoard", + "cpu": "TestCPU", + "gpus": ["TestGPU"], + } + + response = app_client["client"].get("/info") + + assert response.status_code == 200 + data = response.json() + assert data["hostname"] == "TestPC" + assert data["cpu"] == "TestCPU" + + def test_info_route_private_mode(self, app_client): + """测试私密模式下 /info 返回 403""" + app_client["config"].basic.is_public = False + + response = app_client["client"].get("/info") + + assert response.status_code == 403 + assert "瑟瑟中" in response.content.decode("utf-8") + + def test_info_route_uses_device_name_config(self, app_client): + """测试 /info 使用配置的 device_name""" + app_client["config"].basic.device_name = "ConfiguredName" + + with patch("peekapi.server.get_system_info") as mock_info: + mock_info.return_value = {"hostname": "ConfiguredName"} + + app_client["client"].get("/info") + + # 验证调用时传入了配置的 device_name + mock_info.assert_called_once_with("ConfiguredName")