pulsareon/ShowenV2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: M1.1 完成 + M1.2 启动 — 全量更新

Browse Source 
M1.1 收尾:
- 24项 P0/P1/P2 bug 修复 (Rust 107 tests + Flutter 15 tests)
- Flutter App v0.3: cupertino_icons 修复, 单元测试, 调试面板, APK 52.6MB
- 示例插件完善: manifest.json + 请求/响应示范 + 7个测试
- API 文档重写 (以 routes.rs 为唯一权威)
- MILESTONES.md 更新至 100%

M1.2 启动:
- P0: 插件管理 API 闭环 (handle_manager_message Custom 分支 + broadcast_plugin_states)
- ServiceManager 集成测试 8/8 (tests/m1_2_service_manager.rs)
- M1.2 测试计划 (docs/M1.2_TEST_PLAN.md, 18个E2E场景)
- 动态插件系统: auto_rollback + version_manager GC + 路径穿越防护

总计: Rust 115/115 测试, Flutter 15/15 测试, 零 warning

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
showen
2026-03-14 18:12:42 +08:00
parent 8ed9cb2d9d
commit d30c111c71
68 changed files with 8115 additions and 1201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/core/config.rs
View File

@@ -17,9 +17,9 @@ pub struct AppConfig {
pub remote_control: RemoteControlConfig,
#[serde(default)]
pub ble: BleConfig,
#[serde(skip)]
#[serde(default)]
pub source_path: PathBuf,
#[serde(skip)]
#[serde(default)]
pub source_dir: PathBuf,
}

162
src/core/dynamic_plugin.rs
View File

@@ -11,7 +11,30 @@ use crate::core::plugin_abi::{
use anyhow::{anyhow, Context, Result};
use libloading::Library;
use std::ffi::CString;
use std::sync::mpsc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{mpsc, Arc};
struct CallbackState {
active: AtomicBool,
tx: mpsc::Sender<Envelope>,
}
impl CallbackState {
fn new(tx: mpsc::Sender<Envelope>) -> Self {
Self {
active: AtomicBool::new(true),
tx,
}
}
fn deactivate(&self) {
self.active.store(false, Ordering::Release);
}
fn is_active(&self) -> bool {
self.active.load(Ordering::Acquire)
}
}
/// 动态加载的插件
pub struct DynamicPlugin {
@@ -29,8 +52,22 @@ pub struct DynamicPlugin {
dependencies: Vec<String>,
/// .so 文件路径（用于调试/日志）
so_path: String,
/// Sender 上下文指针（堆分配的 mpsc::Sender）
/// Sender 上下文指针（持有一份 Arc 强引用，供插件跨线程回调期间保活）
sender_ctx: *mut std::ffi::c_void,
/// 宿主持有的回调状态，用于卸载前熔断回调并延长 Sender 生命周期
sender_state: Option<Arc<CallbackState>>,
/// stop 是否已经执行成功，避免卸载阶段重复 stop
stopped: bool,
}
impl std::fmt::Debug for DynamicPlugin {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DynamicPlugin")
.field("id", &self.id)
.field("so_path", &self.so_path)
.field("stopped", &self.stopped)
.finish()
}
}
// PluginHandle 是 *mut c_void，需要手动声明 Send
@@ -80,6 +117,8 @@ impl DynamicPlugin {
dependencies,
so_path: so_path.to_string(),
sender_ctx: std::ptr::null_mut(),
sender_state: None,
stopped: false,
})
}
@@ -115,6 +154,25 @@ impl DynamicPlugin {
unsafe { (vtable.free_string)(ffi_str) };
string
}
fn deactivate_callback(&self) {
if let Some(state) = &self.sender_state {
state.deactivate();
}
}
fn release_sender_ctx(&mut self) {
self.deactivate_callback();
if !self.sender_ctx.is_null() {
unsafe {
drop(Arc::from_raw(self.sender_ctx as *const CallbackState));
}
self.sender_ctx = std::ptr::null_mut();
}
self.sender_state = None;
}
}
impl Plugin for DynamicPlugin {
@@ -153,9 +211,13 @@ impl Plugin for DynamicPlugin {
.context("failed to serialize config for dynamic plugin")?;
let config_cstr = CString::new(config_json).context("config JSON contains null byte")?;
// 将 Sender 分配到堆上，生命周期由 DynamicPlugin 管理
let sender_box = Box::new(ctx.tx);
self.sender_ctx = Box::into_raw(sender_box) as *mut std::ffi::c_void;
// 通过 Arc 为回调上下文保活：
// - 宿主持有一份 Arc，控制 active flag
// - FFI ctx 持有另一份 Arc，保证插件后台线程回调期间 sender_ctx 不悬空
let sender_state = Arc::new(CallbackState::new(ctx.tx));
self.sender_ctx = Arc::into_raw(Arc::clone(&sender_state)) as *mut std::ffi::c_void;
self.sender_state = Some(sender_state);
self.stopped = false;
let result = unsafe {
(self.vtable.init)(
@@ -165,11 +227,18 @@ impl Plugin for DynamicPlugin {
ffi_send_callback,
)
};
unsafe { self.check_result(result, "init") }
match unsafe { self.check_result(result, "init") } {
Ok(()) => Ok(()),
Err(err) => {
self.release_sender_ctx();
Err(err)
}
}
}
fn start(&mut self) -> Result<()> {
let result = unsafe { (self.vtable.start)(self.handle) };
self.stopped = false;
unsafe { self.check_result(result, "start") }
}
@@ -183,30 +252,49 @@ impl Plugin for DynamicPlugin {
}
fn stop(&mut self) -> Result<()> {
self.deactivate_callback();
let result = unsafe { (self.vtable.stop)(self.handle) };
unsafe { self.check_result(result, "stop") }
match unsafe { self.check_result(result, "stop") } {
Ok(()) => {
self.stopped = true;
Ok(())
}
Err(err) => Err(err),
}
}
}
impl Drop for DynamicPlugin {
fn drop(&mut self) {
self.deactivate_callback();
if !self.handle.is_null() && !self.stopped {
let result = unsafe { (self.vtable.stop)(self.handle) };
if let Err(err) = unsafe { self.check_result(result, "stop during unload") } {
eprintln!(
"[DynamicPlugin] plugin '{}' stop during unload failed: {err}",
self.id
);
}
}
if !self.handle.is_null() {
unsafe { (self.vtable.destroy)(self.handle) };
self.handle = std::ptr::null_mut();
}
if !self.sender_ctx.is_null() {
unsafe {
drop(Box::from_raw(
self.sender_ctx as *mut mpsc::Sender<Envelope>,
));
}
self.sender_ctx = std::ptr::null_mut();
}
self.release_sender_ctx();
}
}
// ── SendCallback 实现 ──
/// C FFI 回调：插件调用此函数向主程序发消息
///
/// # Safety
/// - `ctx` 必须来自 `DynamicPlugin::init` 传入的 `sender_ctx`。
/// - 动态插件必须在 `stop()` 返回前停止所有可能继续调用该回调的后台线程。
/// - 宿主在卸载前会先把回调熔断为 no-op，再执行 `stop()`/`destroy()`，避免卸载期间 UAF。
unsafe extern "C" fn ffi_send_callback(
ctx: *mut std::ffi::c_void,
envelope_json: crate::core::plugin_abi::FfiStr,
@@ -232,8 +320,48 @@ unsafe extern "C" fn ffi_send_callback(
}
};
let tx = unsafe { &*(ctx as *const mpsc::Sender<Envelope>) };
if let Err(e) = tx.send(envelope) {
let state = unsafe { &*(ctx as *const CallbackState) };
if !state.is_active() {
return;
}
if let Err(e) = state.tx.send(envelope) {
eprintln!("[DynamicPlugin] failed to send envelope: {e}");
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::message::{Destination, Message};
use std::time::Duration;
#[test]
fn ffi_send_callback_becomes_noop_after_deactivate() {
let (tx, rx) = mpsc::channel();
let state = Arc::new(CallbackState::new(tx));
let ctx = Arc::into_raw(Arc::clone(&state)) as *mut std::ffi::c_void;
let envelope = Envelope {
from: "dynamic-test".to_string(),
to: Destination::Manager,
message: Message::Shutdown,
};
let json = serde_json::to_string(&envelope).expect("envelope should serialize");
let json = CString::new(json).expect("json should not contain null");
unsafe { ffi_send_callback(ctx, json.as_ptr()) };
let received = rx
.recv_timeout(Duration::from_millis(100))
.expect("active callback should deliver envelope");
assert_eq!(received.from, "dynamic-test");
state.deactivate();
unsafe { ffi_send_callback(ctx, json.as_ptr()) };
assert!(rx.recv_timeout(Duration::from_millis(100)).is_err());
unsafe {
drop(Arc::from_raw(ctx as *const CallbackState));
}
}
}

39
src/core/plugin_abi.rs
View File

@@ -9,8 +9,16 @@ use std::ptr;
/// 插件实例的不透明句柄
pub type PluginHandle = *mut c_void;
/// FFI 安全的字符串：指向 C 字符串 + 长度
/// 调用方读取内容后，必须通过分配方提供的 free 函数释放
/// FFI 安全的字符串：指向 C 字符串 + 长度。
///
/// 该类型本身不携带 allocator 元数据，以保持现有 `repr(C)` ABI 稳定；
/// 调用方必须通过 API 约定追踪所有权，并使用分配该字符串的一侧提供的释放函数。
///
/// # Safety
/// - 宿主返回给宿主的 `FfiString` 只能由宿主分配器释放。
/// - 动态插件返回给宿主的 `FfiString` 只能通过对应 `PluginVTable::free_string`
/// 释放，不能调用宿主侧释放逻辑。
/// - 跨 allocator 释放会导致未定义行为（可能崩溃或内存损坏）。
#[repr(C)]
pub struct FfiString {
pub ptr: *mut c_char,
@@ -18,7 +26,9 @@ pub struct FfiString {
}
impl FfiString {
/// 从 Rust String 创建 FfiString（转移所有权到 C 侧）
/// 从 Rust String 创建 `FfiString`（转移所有权到调用方一侧）
///
/// 生成的指针必须回到同一 allocator 释放。
pub fn from_string(s: String) -> Self {
match CString::new(s) {
Ok(cstr) => {
@@ -43,7 +53,8 @@ impl FfiString {
/// 复制为 Rust String（不释放底层内存）
///
/// # Safety
/// ptr 必须是由 CString::into_raw 产生的有效指针
/// `ptr` 必须是由当前 allocator 的 `CString::into_raw` 产生的有效指针。
/// 调用此函数不会改变所有权，原始分配方仍负责释放该内存。
pub unsafe fn to_string(&self) -> Option<String> {
if self.ptr.is_null() {
return None;
@@ -99,6 +110,10 @@ impl FfiResult {
/// 插件向主程序发消息的回调函数类型
/// envelope_json: JSON 序列化的 Envelope
///
/// # Safety
/// 插件若把该回调保存到后台线程，必须保证在线程完全退出后再让 `stop()` 返回；
/// 一旦宿主开始卸载插件，回调会先被熔断为 no-op，随后执行 `stop()`/`destroy()`。
pub type SendCallback = unsafe extern "C" fn(ctx: *mut c_void, envelope_json: FfiStr);
/// 插件虚函数表 — 每个动态插件导出一个此结构体
@@ -131,7 +146,11 @@ pub struct PluginVTable {
/// 停止插件
pub stop: unsafe extern "C" fn(handle: PluginHandle) -> FfiResult,
/// 释放插件分配的 FfiString
/// 释放插件分配的 `FfiString`
///
/// # Safety
/// 只能用于释放该插件自己返回的字符串。宿主分配的 `FfiString` 绝不能传给这里，
/// 否则会发生跨 allocator 释放。
pub free_string: unsafe extern "C" fn(s: FfiString),
/// 销毁插件实例，释放资源
@@ -157,13 +176,3 @@ pub unsafe fn ffi_str_to_str<'a>(ptr: FfiStr) -> Option<&'a str> {
}
unsafe { CStr::from_ptr(ptr) }.to_str().ok()
}
/// 释放 FfiString 占用的内存
///
/// # Safety
/// ptr 必须是由 FfiString::from_string 创建的
pub unsafe fn ffi_string_free(s: FfiString) {
if !s.ptr.is_null() {
drop(unsafe { CString::from_raw(s.ptr) });
}
}

112
src/core/plugin_loader.rs
View File

@@ -123,8 +123,8 @@ impl PluginLoader {
/// 保存全局注册表
pub fn save_registry(&self, registry: &PluginRegistry) -> Result<()> {
let registry_path = self.store_path.join("registry.json");
let content = serde_json::to_string_pretty(registry)
.context("failed to serialize registry")?;
let content =
serde_json::to_string_pretty(registry).context("failed to serialize registry")?;
std::fs::write(&registry_path, content)
.with_context(|| format!("failed to write {}", registry_path.display()))
}
@@ -201,9 +201,7 @@ impl PluginLoader {
.plugins
.get(plugin_id)
.map(|e| e.active_version.clone())
.ok_or_else(|| {
anyhow!("plugin '{plugin_id}' not found in registry")
})?
.ok_or_else(|| anyhow!("plugin '{plugin_id}' not found in registry"))?
}
};
@@ -211,18 +209,28 @@ impl PluginLoader {
let manifest_path = version_dir.join("manifest.json");
let manifest = self.read_manifest(&manifest_path)?;
let so_path = version_dir.join(&manifest.so_filename);
if !so_path.exists() {
if manifest.id != plugin_id {
return Err(anyhow!(
"plugin .so not found: {}",
so_path.display()
"plugin manifest id mismatch: requested '{plugin_id}', found '{}'",
manifest.id
));
}
if manifest.version != version {
return Err(anyhow!(
"plugin manifest version mismatch: requested '{version}', found '{}'",
manifest.version
));
}
let so_path = version_dir.join(&manifest.so_filename);
if !so_path.exists() {
return Err(anyhow!("plugin .so not found: {}", so_path.display()));
}
let so_path_str = so_path.to_string_lossy().to_string();
let mut plugin = unsafe {
DynamicPlugin::load(&so_path_str, manifest.dependencies.clone())?
};
let mut plugin =
unsafe { DynamicPlugin::load(&so_path_str, manifest.dependencies.clone())? };
plugin.set_id(manifest.id.clone());
Ok((plugin, manifest))
@@ -255,6 +263,14 @@ mod tests {
use super::*;
use std::fs;
fn unique_test_dir(name: &str) -> std::path::PathBuf {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system time should be after unix epoch")
.as_nanos();
std::env::temp_dir().join(format!("showen_plugin_loader_{name}_{nanos}"))
}
fn setup_test_store(base: &Path) {
let plugin_dir = base.join("test-plugin").join("1.0.0");
fs::create_dir_all(&plugin_dir).unwrap();
@@ -387,4 +403,76 @@ mod tests {
assert_eq!(manifest.test_timeout_ms, 5000);
assert!(manifest.auto_test);
}
#[test]
fn manifest_test_timeout_ms_is_configurable_from_manifest() {
let json = r#"{
"id": "timed-plugin",
"version": "1.0.0",
"sdk_version": "0.2.0",
"so_filename": "libtimed_plugin.so",
"test_timeout_ms": 12000
}"#;
let manifest: PluginManifest = serde_json::from_str(json).unwrap();
assert_eq!(manifest.test_timeout_ms, 12000);
}
#[test]
fn load_plugin_rejects_manifest_id_mismatch() {
let tmp = unique_test_dir("id_mismatch");
let plugin_dir = tmp.join("expected-plugin").join("1.0.0");
fs::create_dir_all(&plugin_dir).unwrap();
fs::write(
plugin_dir.join("manifest.json"),
r#"{
"id": "other-plugin",
"version": "1.0.0",
"sdk_version": "0.2.0",
"so_filename": "libexpected_plugin.so"
}"#,
)
.unwrap();
let loader = PluginLoader::new(&tmp);
let error = match loader.load_plugin("expected-plugin", Some("1.0.0")) {
Ok(_) => panic!("id mismatch should be rejected"),
Err(error) => error,
};
assert!(error.to_string().contains(
"plugin manifest id mismatch: requested 'expected-plugin', found 'other-plugin'"
));
let _ = fs::remove_dir_all(&tmp);
}
#[test]
fn load_plugin_rejects_manifest_version_mismatch() {
let tmp = unique_test_dir("version_mismatch");
let plugin_dir = tmp.join("expected-plugin").join("1.0.0");
fs::create_dir_all(&plugin_dir).unwrap();
fs::write(
plugin_dir.join("manifest.json"),
r#"{
"id": "expected-plugin",
"version": "2.0.0",
"sdk_version": "0.2.0",
"so_filename": "libexpected_plugin.so"
}"#,
)
.unwrap();
let loader = PluginLoader::new(&tmp);
let error = match loader.load_plugin("expected-plugin", Some("1.0.0")) {
Ok(_) => panic!("version mismatch should be rejected"),
Err(error) => error,
};
assert!(error
.to_string()
.contains("plugin manifest version mismatch: requested '1.0.0', found '2.0.0'"));
let _ = fs::remove_dir_all(&tmp);
}
}

380
src/core/plugin_repo.rs
View File

@@ -16,6 +16,31 @@ use anyhow::{anyhow, Context, Result};
use flate2::read::GzDecoder;
use serde::{Deserialize, Serialize};
use std::io::Read;
use std::path::{Component, Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
struct TempDirGuard {
path: PathBuf,
keep: bool,
}
impl TempDirGuard {
fn new(path: PathBuf) -> Self {
Self { path, keep: false }
}
fn disarm(&mut self) {
self.keep = true;
}
}
impl Drop for TempDirGuard {
fn drop(&mut self) {
if !self.keep && self.path.exists() {
let _ = std::fs::remove_dir_all(&self.path);
}
}
}
/// 远程仓库索引
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -93,15 +118,9 @@ impl PluginRepository {
}
/// 下载并安装插件到 plugin_store/
pub fn download_and_install(
&self,
plugin_id: &str,
version: &str,
) -> Result<()> {
pub fn download_and_install(&self, plugin_id: &str, version: &str) -> Result<()> {
let url = format!("{}/{}/{}.tar.gz", self.base_url, plugin_id, version);
println!(
"[PluginRepo] 下载插件 '{plugin_id}' v{version} 从 {url}"
);
println!("[PluginRepo] 下载插件 '{plugin_id}' v{version} 从 {url}");
let response = ureq::get(&url)
.call()
@@ -114,12 +133,16 @@ impl PluginRepository {
.read_to_end(&mut body)
.context("failed to read download body")?;
// 解压 tar.gz 到临时目录
let target_dir = self
.loader
.store_path()
.join(plugin_id)
.join(version);
self.install_archive_bytes(plugin_id, version, &body)
}
fn install_archive_bytes(&self, plugin_id: &str, version: &str, body: &[u8]) -> Result<()> {
let plugin_dir = self.loader.store_path().join(plugin_id);
std::fs::create_dir_all(&plugin_dir)
.with_context(|| format!("failed to create {}", plugin_dir.display()))?;
let target_dir = plugin_dir.join(version);
if target_dir.exists() {
return Err(anyhow!(
@@ -127,27 +150,30 @@ impl PluginRepository {
));
}
std::fs::create_dir_all(&target_dir).with_context(|| {
format!("failed to create {}", target_dir.display())
})?;
let staging_dir = self.staging_dir_path(&plugin_dir, version);
std::fs::create_dir_all(&staging_dir)
.with_context(|| format!("failed to create {}", staging_dir.display()))?;
let mut cleanup = TempDirGuard::new(staging_dir.clone());
// 解压 tar.gz
let gz = GzDecoder::new(body.as_slice());
let mut archive = tar::Archive::new(gz);
archive
.unpack(&target_dir)
.with_context(|| format!("failed to unpack archive to {}", target_dir.display()))?;
self.extract_archive_securely(body, &staging_dir)?;
// 验证 manifest.json 存在
let manifest_path = target_dir.join("manifest.json");
if !manifest_path.exists() {
// 清理
let _ = std::fs::remove_dir_all(&target_dir);
let manifest_path = staging_dir.join("manifest.json");
if !manifest_path.is_file() {
return Err(anyhow!(
"downloaded archive for '{plugin_id}' v{version} missing manifest.json"
));
}
std::fs::rename(&staging_dir, &target_dir).with_context(|| {
format!(
"failed to atomically move extracted plugin from {} to {}",
staging_dir.display(),
target_dir.display()
)
})?;
cleanup.disarm();
println!(
"[PluginRepo] 插件 '{plugin_id}' v{version} 安装成功到 {}",
target_dir.display()
@@ -155,10 +181,119 @@ impl PluginRepository {
Ok(())
}
fn staging_dir_path(&self, plugin_dir: &Path, version: &str) -> PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
plugin_dir.join(format!(".{version}.tmp-{}-{nanos}", std::process::id()))
}
fn extract_archive_securely(&self, body: &[u8], target_dir: &Path) -> Result<()> {
let gz = GzDecoder::new(body);
let mut archive = tar::Archive::new(gz);
let canonical_target = std::fs::canonicalize(target_dir)
.with_context(|| format!("failed to canonicalize {}", target_dir.display()))?;
for entry in archive
.entries()
.context("failed to read archive entries")?
{
let mut entry = entry.context("failed to read archive entry")?;
let entry_path = entry.path().context("failed to read archive entry path")?;
let entry_path = entry_path.into_owned();
Self::validate_archive_path(&entry_path)?;
let entry_type = entry.header().entry_type();
if entry_type.is_symlink() || entry_type.is_hard_link() {
return Err(anyhow!(
"archive entry '{}' uses unsupported link type",
entry_path.display()
));
}
let destination = target_dir.join(&entry_path);
if entry_type.is_dir() {
std::fs::create_dir_all(&destination).with_context(|| {
format!("failed to create directory {}", destination.display())
})?;
Self::ensure_path_within_root(&canonical_target, &destination)?;
continue;
}
if !entry_type.is_file() {
return Err(anyhow!(
"archive entry '{}' has unsupported type {:?}",
entry_path.display(),
entry_type
));
}
let parent = destination.parent().ok_or_else(|| {
anyhow!(
"archive entry '{}' has no valid parent directory",
entry_path.display()
)
})?;
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
Self::ensure_path_within_root(&canonical_target, parent)?;
entry.unpack(&destination).with_context(|| {
format!(
"failed to unpack archive entry to {}",
destination.display()
)
})?;
}
Ok(())
}
fn validate_archive_path(path: &Path) -> Result<()> {
if path.as_os_str().is_empty() {
return Err(anyhow!("archive contains empty path"));
}
for component in path.components() {
match component {
Component::Normal(_) | Component::CurDir => {}
Component::ParentDir => {
return Err(anyhow!(
"archive entry '{}' attempts path traversal",
path.display()
));
}
Component::RootDir | Component::Prefix(_) => {
return Err(anyhow!(
"archive entry '{}' uses absolute path",
path.display()
));
}
}
}
Ok(())
}
fn ensure_path_within_root(root: &Path, path: &Path) -> Result<()> {
let canonical_path = std::fs::canonicalize(path)
.with_context(|| format!("failed to canonicalize {}", path.display()))?;
if canonical_path.starts_with(root) {
Ok(())
} else {
Err(anyhow!(
"archive entry resolves outside extraction root: {}",
canonical_path.display()
))
}
}
/// 批量检查所有已安装插件的更新
pub fn check_all_updates(
&self,
) -> Result<Vec<(String, String, String)>> {
pub fn check_all_updates(&self) -> Result<Vec<(String, String, String)>> {
// (plugin_id, current_version, new_version)
let registry = self.loader.load_registry()?;
let mut updates = Vec::new();
@@ -166,17 +301,11 @@ impl PluginRepository {
for (plugin_id, entry) in &registry.plugins {
match self.check_update(plugin_id, &entry.active_version) {
Ok(Some(new_version)) => {
updates.push((
plugin_id.clone(),
entry.active_version.clone(),
new_version,
));
updates.push((plugin_id.clone(), entry.active_version.clone(), new_version));
}
Ok(None) => {}
Err(e) => {
eprintln!(
"[PluginRepo] 检查 '{plugin_id}' 更新失败: {e}"
);
eprintln!("[PluginRepo] 检查 '{plugin_id}' 更新失败: {e}");
}
}
}
@@ -184,3 +313,178 @@ impl PluginRepository {
Ok(updates)
}
}
#[cfg(test)]
mod tests {
use super::*;
use flate2::write::GzEncoder;
use flate2::Compression;
use std::fs;
use tar::{Builder, EntryType, Header};
fn unique_test_dir(name: &str) -> PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("system time should be after unix epoch")
.as_nanos();
std::env::temp_dir().join(format!(
"showen_plugin_repo_{name}_{}_{}",
std::process::id(),
nanos
))
}
fn gzip_bytes(bytes: &[u8]) -> Vec<u8> {
let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
std::io::Write::write_all(&mut encoder, bytes).expect("gzip encoder should accept bytes");
encoder.finish().expect("gzip encoder should finish")
}
fn build_plain_tar(entries: Vec<(&str, &[u8])>) -> Vec<u8> {
let mut builder = Builder::new(Vec::new());
for (path, contents) in entries {
let mut header = Header::new_gnu();
header.set_entry_type(EntryType::Regular);
header.set_mode(0o644);
header.set_size(contents.len() as u64);
header.set_cksum();
builder
.append_data(&mut header, path, contents)
.expect("archive entry should be appended");
}
builder.finish().expect("archive builder should finish");
builder.into_inner().expect("tar bytes should be returned")
}
fn build_archive(entries: Vec<(&str, &[u8])>) -> Vec<u8> {
let tar_bytes = build_plain_tar(entries);
gzip_bytes(&tar_bytes)
}
fn build_symlink_archive(path: &str, target: &str) -> Vec<u8> {
let mut builder = Builder::new(Vec::new());
let mut header = Header::new_gnu();
header.set_entry_type(EntryType::Symlink);
header.set_mode(0o777);
header.set_size(0);
header.set_cksum();
builder
.append_link(&mut header, path, target)
.expect("symlink entry should be appended");
builder.finish().expect("archive builder should finish");
let tar_bytes = builder.into_inner().expect("tar bytes should be returned");
gzip_bytes(&tar_bytes)
}
fn build_traversal_archive() -> Vec<u8> {
fn write_octal(field: &mut [u8], value: u64) {
field.fill(0);
let octal = format!("{:0width$o}\0", value, width = field.len() - 1);
field[..octal.len()].copy_from_slice(octal.as_bytes());
}
let contents = br#"{}"#;
let mut tar_bytes = vec![0u8; 2048];
let (header, rest) = tar_bytes.split_at_mut(512);
let path = b"../manifest.json";
header[..path.len()].copy_from_slice(path);
write_octal(&mut header[100..108], 0o644);
write_octal(&mut header[108..116], 0);
write_octal(&mut header[116..124], 0);
write_octal(&mut header[124..136], contents.len() as u64);
write_octal(&mut header[136..148], 0);
header[148..156].fill(b' ');
header[156] = b'0';
header[257..263].copy_from_slice(b"ustar\0");
header[263..265].copy_from_slice(b"00");
{
let data = &mut rest[..512];
data[..contents.len()].copy_from_slice(contents);
}
let checksum = header.iter().map(|byte| *byte as u32).sum::<u32>();
let checksum_field = format!("{:06o}\0 ", checksum);
header[148..156].copy_from_slice(checksum_field.as_bytes());
gzip_bytes(&tar_bytes)
}
#[test]
fn install_archive_rejects_path_traversal_and_cleans_staging_dir() {
let store = unique_test_dir("traversal");
fs::create_dir_all(&store).expect("store dir should be created");
let loader = PluginLoader::new(&store);
let repo = PluginRepository::new("https://plugins.example.com", loader);
let archive = build_traversal_archive();
let err = repo
.install_archive_bytes("sensor", "1.0.0", &archive)
.expect_err("path traversal archive should be rejected");
assert!(err.to_string().contains("path traversal"));
let plugin_dir = store.join("sensor");
let entries = fs::read_dir(&plugin_dir)
.expect("plugin dir should exist")
.collect::<std::result::Result<Vec<_>, _>>()
.expect("plugin dir entries should be readable");
assert!(entries.is_empty());
assert!(!plugin_dir.join("1.0.0").exists());
let _ = fs::remove_dir_all(&store);
}
#[test]
fn install_archive_rejects_symlinks() {
let store = unique_test_dir("symlink");
fs::create_dir_all(&store).expect("store dir should be created");
let loader = PluginLoader::new(&store);
let repo = PluginRepository::new("https://plugins.example.com", loader);
let archive = build_symlink_archive("manifest.json", "/etc/passwd");
let err = repo
.install_archive_bytes("sensor", "1.0.0", &archive)
.expect_err("symlink archive should be rejected");
assert!(err.to_string().contains("unsupported link type"));
assert!(!store.join("sensor").join("1.0.0").exists());
let _ = fs::remove_dir_all(&store);
}
#[test]
fn install_archive_extracts_to_staging_then_moves_into_place() {
let store = unique_test_dir("valid");
fs::create_dir_all(&store).expect("store dir should be created");
let loader = PluginLoader::new(&store);
let repo = PluginRepository::new("https://plugins.example.com", loader);
let archive = build_archive(vec![
(
"manifest.json",
br#"{"id":"sensor","version":"1.0.0","sdk_version":"0.2.0","so_filename":"libsensor.so"}"#,
),
("libsensor.so", b"fake so bytes"),
]);
repo.install_archive_bytes("sensor", "1.0.0", &archive)
.expect("valid archive should install");
let target_dir = store.join("sensor").join("1.0.0");
assert!(target_dir.join("manifest.json").is_file());
assert!(target_dir.join("libsensor.so").is_file());
let leftover_tmp = fs::read_dir(store.join("sensor"))
.expect("plugin dir should be readable")
.collect::<std::result::Result<Vec<_>, _>>()
.expect("plugin dir entries should be readable")
.into_iter()
.any(|entry| entry.file_name().to_string_lossy().contains(".tmp-"));
assert!(!leftover_tmp);
let _ = fs::remove_dir_all(&store);
}
}

549
src/core/service_manager.rs
View File

@@ -1,12 +1,29 @@
use crate::core::config::AppConfig;
use crate::core::message::{Destination, Envelope, Message};
use crate::core::plugin::{CapabilityTestResult, Plugin, PluginContext};
use crate::core::plugin_loader::ErrorPolicy;
use crate::core::plugin_loader::{ErrorPolicy, PluginLoader, PluginRegistryEntry};
use crate::core::plugin_repo::PluginRepository;
use crate::core::version_manager::VersionManager;
use anyhow::{anyhow, Result};
use serde::Deserialize;
use std::collections::{HashMap, HashSet};
use std::sync::{mpsc, Arc};
const DEFAULT_PLUGIN_REPO_URL: &str = "https://plugins.example.com";
#[derive(Deserialize)]
struct PluginSwitchCommand {
id: String,
version: String,
}
#[derive(Deserialize)]
struct PluginInstallCommand {
id: String,
#[serde(default)]
version: Option<String>,
}
/// 插件运行时状态包装
struct PluginState {
plugin: Box<dyn Plugin>,
@@ -92,6 +109,46 @@ pub struct ServiceManager {
}
impl ServiceManager {
fn plugin_context(&self) -> PluginContext {
PluginContext {
tx: self.tx.clone(),
config: Arc::clone(&self.config),
}
}
fn init_and_start_plugin_with_context(
state: &mut PluginState,
ctx: PluginContext,
) -> Result<()> {
if let Err(init_error) = state.plugin.init(ctx) {
let cleanup_error = state.plugin.stop().err();
return match cleanup_error {
Some(stop_error) => Err(anyhow!(
"plugin '{}' init failed: {}; cleanup stop failed: {}",
state.id(),
init_error,
stop_error
)),
None => Err(init_error),
};
}
if let Err(start_error) = state.plugin.start() {
let cleanup_error = state.plugin.stop().err();
return match cleanup_error {
Some(stop_error) => Err(anyhow!(
"plugin '{}' start failed: {}; cleanup stop failed: {}",
state.id(),
start_error,
stop_error
)),
None => Err(start_error),
};
}
Ok(())
}
pub fn new(config: AppConfig) -> Self {
let (tx, rx) = mpsc::channel();
Self {
@@ -240,15 +297,17 @@ impl ServiceManager {
match &state.error_policy {
ErrorPolicy::AutoRollback => {
eprintln!(
"[ServiceManager] 动态插件 '{}' 必须能力自测失败，禁用 (待回退)",
"[ServiceManager] 动态插件 '{}' 必须能力自测失败，尝试自动回退到稳定版本",
state.id()
);
state.needs_rollback = true;
}
ErrorPolicy::DisableAndLog => {
eprintln!(
"[ServiceManager] 动态插件 '{}' 必须能力自测失败，禁用",
state.id()
);
state.needs_rollback = false;
}
}
state.enabled = false;
@@ -258,6 +317,15 @@ impl ServiceManager {
}
}
for idx in 0..self.plugins.len() {
if !self.plugins[idx].needs_rollback {
continue;
}
let plugin_id = self.plugins[idx].id().to_string();
self.rollback_dynamic_plugin(idx, &plugin_id);
}
// Phase 3: start
for state in &mut self.plugins {
if !state.enabled {
@@ -279,6 +347,8 @@ impl ServiceManager {
}
}
self.broadcast_plugin_states();
Ok(())
}
@@ -329,18 +399,29 @@ impl ServiceManager {
/// 启用/禁用指定插件
pub fn set_plugin_enabled(&mut self, plugin_id: &str, enabled: bool) -> Result<()> {
let state = self
let idx = self
.plugins
.iter_mut()
.find(|s| s.id() == plugin_id)
.iter()
.position(|s| s.id() == plugin_id)
.ok_or_else(|| anyhow!("plugin '{plugin_id}' not found"))?;
if enabled && !state.enabled {
// 重新启用：reset 错误计数
if enabled && !self.plugins[idx].enabled {
let ctx = self.plugin_context();
let state = &mut self.plugins[idx];
state.error_count = 0;
state.enabled = true;
println!("[ServiceManager] 插件 '{plugin_id}' 已启用");
} else if !enabled && state.enabled {
match Self::init_and_start_plugin_with_context(state, ctx) {
Ok(()) => {
state.enabled = true;
println!("[ServiceManager] 插件 '{plugin_id}' 已启用");
}
Err(error) => {
state.enabled = false;
return Err(anyhow!("failed to enable plugin '{plugin_id}': {error}"));
}
}
} else if !enabled && self.plugins[idx].enabled {
let state = &mut self.plugins[idx];
state.plugin.stop()?;
state.enabled = false;
println!("[ServiceManager] 插件 '{plugin_id}' 已禁用");
}
@@ -348,6 +429,23 @@ impl ServiceManager {
Ok(())
}
pub fn rollback_plugin(&mut self, plugin_id: &str) -> Result<()> {
let idx = self
.plugins
.iter()
.position(|state| state.id() == plugin_id)
.ok_or_else(|| anyhow!("plugin '{plugin_id}' not found"))?;
if !self.plugins[idx].is_dynamic {
return Err(anyhow!(
"plugin '{plugin_id}' is not dynamic and cannot be rolled back"
));
}
self.rollback_dynamic_plugin(idx, plugin_id);
Ok(())
}
/// 查询插件状态信息（供 HTTP API 使用）
pub fn plugin_states(&self) -> Vec<PluginStateInfo> {
self.plugins
@@ -389,18 +487,50 @@ impl ServiceManager {
new_state.capabilities = capabilities;
new_state.auto_test = auto_test;
let ctx = PluginContext {
tx: self.tx.clone(),
config: Arc::clone(&self.config),
};
new_state.plugin.init(ctx)?;
new_state.plugin.start()?;
let ctx = self.plugin_context();
let mut old_state = self.plugins.remove(idx);
let old_was_enabled = old_state.enabled;
if self.plugins[idx].enabled {
let _ = self.plugins[idx].plugin.stop();
if old_was_enabled {
// 先停旧插件，避免热替换窗口内新旧实例同时持有端口、文件句柄等独占资源。
old_state.plugin.stop()?;
}
let replace_result = Self::init_and_start_plugin_with_context(&mut new_state, ctx);
match replace_result {
Ok(()) => {
new_state.enabled = true;
self.plugins.insert(idx, new_state);
}
Err(new_error) => {
if old_was_enabled {
let restore_ctx = self.plugin_context();
match Self::init_and_start_plugin_with_context(&mut old_state, restore_ctx) {
Ok(()) => {
old_state.enabled = true;
self.plugins.insert(idx, old_state);
return Err(anyhow!(
"failed to replace plugin '{plugin_id}': {new_error}; restored previous plugin"
));
}
Err(restore_error) => {
old_state.enabled = false;
self.plugins.insert(idx, old_state);
return Err(anyhow!(
"failed to replace plugin '{plugin_id}': {new_error}; failed to restore previous plugin: {restore_error}"
));
}
}
}
old_state.enabled = false;
self.plugins.insert(idx, old_state);
return Err(anyhow!(
"failed to replace plugin '{plugin_id}': {new_error}"
));
}
}
self.plugins[idx] = new_state;
println!("[ServiceManager] 插件 '{plugin_id}' 热替换成功");
Ok(())
}
@@ -441,6 +571,192 @@ impl ServiceManager {
)
}
fn plugin_loader(&self) -> Result<PluginLoader> {
let version_manager = self
.version_manager
.as_ref()
.ok_or_else(|| anyhow!("plugin version manager is not configured"))?;
Ok(PluginLoader::new(version_manager.loader().store_path()))
}
fn plugin_repository(&self) -> Result<PluginRepository> {
Ok(PluginRepository::new(
&std::env::var("SHOWEN_PLUGIN_REPO_URL")
.ok()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| DEFAULT_PLUGIN_REPO_URL.to_string()),
self.plugin_loader()?,
))
}
fn register_dynamic_plugin_runtime(
&mut self,
plugin_id: &str,
plugin: Box<dyn Plugin>,
error_policy: ErrorPolicy,
max_errors: u32,
required_capabilities: Vec<String>,
capabilities: Vec<String>,
auto_test: bool,
) -> Result<()> {
let mut state = PluginState::new_dynamic(plugin, error_policy, max_errors);
state.required_capabilities = required_capabilities;
state.capabilities = capabilities;
state.auto_test = auto_test;
let ctx = self.plugin_context();
Self::init_and_start_plugin_with_context(&mut state, ctx)
.map_err(|error| anyhow!("failed to start installed plugin '{plugin_id}': {error}"))?;
state.enabled = true;
self.plugins.push(state);
println!("[ServiceManager] 插件 '{plugin_id}' 已安装并启动");
Ok(())
}
fn switch_plugin_version(&mut self, plugin_id: &str, version: &str) -> Result<()> {
let version_manager = self
.version_manager
.as_ref()
.ok_or_else(|| anyhow!("plugin version manager is not configured"))?;
version_manager.switch_version(plugin_id, version)?;
let (plugin, manifest) = version_manager
.loader()
.load_plugin(plugin_id, Some(version))?;
let loader = PluginLoader::new(version_manager.loader().store_path());
let registry = loader.load_registry()?;
let entry = registry
.plugins
.get(plugin_id)
.ok_or_else(|| anyhow!("plugin '{plugin_id}' not in registry"))?;
if let Some(idx) = self
.plugins
.iter()
.position(|state| state.id() == plugin_id)
{
self.replace_dynamic_plugin_at_index(
idx,
plugin_id,
Box::new(plugin),
manifest.error_policy,
entry.max_errors,
manifest.required_capabilities,
manifest.capabilities,
manifest.auto_test,
)?;
} else {
self.register_dynamic_plugin_runtime(
plugin_id,
Box::new(plugin),
manifest.error_policy,
entry.max_errors,
manifest.required_capabilities,
manifest.capabilities,
manifest.auto_test,
)?;
}
println!("[ServiceManager] 插件 '{plugin_id}' 已切换到版本 {version}");
Ok(())
}
fn install_plugin(&mut self, request: PluginInstallCommand) -> Result<()> {
let plugin_id = request.id.clone();
let repo = self.plugin_repository()?;
let version = match request.version.clone() {
Some(version) => version,
None => repo.check_update(&plugin_id, "0.0.0")?.ok_or_else(|| {
anyhow!(
"repo did not report an installable version for '{}'",
plugin_id
)
})?,
};
repo.download_and_install(&plugin_id, &version)?;
let loader = self.plugin_loader()?;
let (plugin, manifest) = loader.load_plugin(&plugin_id, Some(&version))?;
let mut registry = loader.load_registry()?;
let existing = registry.plugins.get(&plugin_id).cloned();
let entry = PluginRegistryEntry {
active_version: version.clone(),
last_stable_version: existing
.as_ref()
.and_then(|entry| entry.last_stable_version.clone()),
enabled: true,
error_policy: existing
.as_ref()
.map(|entry| entry.error_policy.clone())
.unwrap_or_else(|| manifest.error_policy.clone()),
max_errors: existing.as_ref().map(|entry| entry.max_errors).unwrap_or(5),
};
registry.plugins.insert(plugin_id.clone(), entry.clone());
loader.save_registry(&registry)?;
if let Some(idx) = self
.plugins
.iter()
.position(|state| state.id() == plugin_id)
{
self.replace_dynamic_plugin_at_index(
idx,
&plugin_id,
Box::new(plugin),
manifest.error_policy,
entry.max_errors,
manifest.required_capabilities,
manifest.capabilities,
manifest.auto_test,
)?;
} else {
self.register_dynamic_plugin_runtime(
&plugin_id,
Box::new(plugin),
manifest.error_policy,
entry.max_errors,
manifest.required_capabilities,
manifest.capabilities,
manifest.auto_test,
)?;
}
Ok(())
}
fn check_plugin_updates(&self) -> Result<()> {
let repo = self.plugin_repository()?;
let registry = self.plugin_loader()?.load_registry()?;
for (plugin_id, entry) in &registry.plugins {
match repo.check_update(plugin_id, &entry.active_version)? {
Some(version) => println!(
"[ServiceManager] 插件 '{plugin_id}' 发现可用更新: {} -> {version}",
entry.active_version
),
None => println!(
"[ServiceManager] 插件 '{plugin_id}' 已是最新版本 {}",
entry.active_version
),
}
}
Ok(())
}
fn broadcast_plugin_states(&mut self) {
match serde_json::to_string(&self.plugin_states()) {
Ok(payload) => self.broadcast_message(Message::Custom {
kind: "plugin_states".to_string(),
payload,
}),
Err(error) => eprintln!("[ServiceManager] 序列化 plugin_states 失败: {error}"),
}
}
/// 处理发给管理层自身的消息
fn handle_manager_message(&mut self, msg: Message) -> Result<()> {
match msg {
@@ -484,6 +800,79 @@ impl ServiceManager {
println!("[ServiceManager] 插件 '{}' 就绪", id);
self.broadcast_message(Message::PluginReady(id));
}
Message::Custom { kind, payload } => {
let should_broadcast = match kind.as_str() {
"plugin_enable" => {
if let Err(error) = self.set_plugin_enabled(&payload, true) {
eprintln!(
"[ServiceManager] plugin_enable('{}') 失败: {error}",
payload
);
}
true
}
"plugin_disable" => {
if let Err(error) = self.set_plugin_enabled(&payload, false) {
eprintln!(
"[ServiceManager] plugin_disable('{}') 失败: {error}",
payload
);
}
true
}
"plugin_rollback" => {
if let Err(error) = self.rollback_plugin(&payload) {
eprintln!(
"[ServiceManager] plugin_rollback('{}') 失败: {error}",
payload
);
}
true
}
"plugin_switch" => {
match serde_json::from_str::<PluginSwitchCommand>(&payload) {
Ok(command) => {
if let Err(error) =
self.switch_plugin_version(&command.id, &command.version)
{
eprintln!(
"[ServiceManager] plugin_switch('{}', '{}') 失败: {error}",
command.id, command.version
);
}
}
Err(error) => {
eprintln!("[ServiceManager] plugin_switch payload 非法: {error}");
}
}
true
}
"plugin_install" => {
match serde_json::from_str::<PluginInstallCommand>(&payload) {
Ok(command) => {
if let Err(error) = self.install_plugin(command) {
eprintln!("[ServiceManager] plugin_install 失败: {error}");
}
}
Err(error) => {
eprintln!("[ServiceManager] plugin_install payload 非法: {error}");
}
}
true
}
"plugin_check_updates" => {
if let Err(error) = self.check_plugin_updates() {
eprintln!("[ServiceManager] plugin_check_updates 失败: {error}");
}
true
}
_ => false,
};
if should_broadcast {
self.broadcast_plugin_states();
}
}
_ => {}
}
Ok(())
@@ -665,75 +1054,79 @@ impl ServiceManager {
"[ServiceManager] 插件 '{}' 错误次数达到阈值，尝试自动回退到稳定版本",
plugin_id
);
self.rollback_dynamic_plugin(idx, plugin_id);
}
}
}
let rollback_result = {
let Some(version_manager) = self.version_manager.as_ref() else {
eprintln!(
"[ServiceManager] 插件 '{}' 未配置 VersionManager，标记为待回退",
plugin_id
fn rollback_dynamic_plugin(&mut self, idx: usize, plugin_id: &str) {
let rollback_result = {
let Some(version_manager) = self.version_manager.as_ref() else {
eprintln!(
"[ServiceManager] 插件 '{}' 未配置 VersionManager，标记为待回退",
plugin_id
);
self.plugins[idx].needs_rollback = true;
return;
};
match version_manager.rollback(plugin_id) {
Ok(version) => match version_manager
.loader()
.load_plugin(plugin_id, Some(&version))
{
Ok((plugin, manifest)) => {
Ok((version, Box::new(plugin) as Box<dyn Plugin>, manifest))
}
Err(e) => Err((Some(version), e)),
},
Err(e) => Err((None, e)),
}
};
match rollback_result {
Ok((version, plugin, manifest)) => {
let max_errors = self.plugins[idx].max_errors;
match self.replace_dynamic_plugin_at_index(
idx,
plugin_id,
plugin,
manifest.error_policy,
max_errors,
manifest.required_capabilities,
manifest.capabilities,
manifest.auto_test,
) {
Ok(()) => {
self.plugins[idx].needs_rollback = false;
println!(
"[ServiceManager] 插件 '{}' 已回退并重新加载稳定版本 {}",
plugin_id, version
);
self.plugins[idx].needs_rollback = true;
return;
};
match version_manager.rollback(plugin_id) {
Ok(version) => match version_manager
.loader()
.load_plugin(plugin_id, Some(&version))
{
Ok((plugin, manifest)) => {
Ok((version, Box::new(plugin) as Box<dyn Plugin>, manifest))
}
Err(e) => Err((Some(version), e)),
},
Err(e) => Err((None, e)),
}
};
match rollback_result {
Ok((version, plugin, manifest)) => {
let max_errors = self.plugins[idx].max_errors;
match self.replace_dynamic_plugin_at_index(
idx,
plugin_id,
plugin,
manifest.error_policy,
max_errors,
manifest.required_capabilities,
manifest.capabilities,
manifest.auto_test,
) {
Ok(()) => {
println!(
"[ServiceManager] 插件 '{}' 已回退并重新加载稳定版本 {}",
plugin_id, version
);
}
Err(e) => {
eprintln!(
"[ServiceManager] 插件 '{}' 已切换到稳定版本 {}，但热替换失败: {}",
plugin_id, version, e
);
self.plugins[idx].needs_rollback = true;
}
}
}
Err((Some(version), e)) => {
Err(e) => {
eprintln!(
"[ServiceManager] 插件 '{}' 已切换到稳定版本 {}，但加载回退版本失败: {}",
"[ServiceManager] 插件 '{}' 已切换到稳定版本 {}，但热替换失败: {}",
plugin_id, version, e
);
self.plugins[idx].needs_rollback = true;
}
Err((None, e)) => {
eprintln!(
"[ServiceManager] 插件 '{}' 自动回退失败，标记为待回退: {}",
plugin_id, e
);
self.plugins[idx].needs_rollback = true;
}
}
}
Err((Some(version), e)) => {
eprintln!(
"[ServiceManager] 插件 '{}' 已切换到稳定版本 {}，但加载回退版本失败: {}",
plugin_id, version, e
);
self.plugins[idx].needs_rollback = true;
}
Err((None, e)) => {
eprintln!(
"[ServiceManager] 插件 '{}' 自动回退失败，标记为待回退: {}",
plugin_id, e
);
self.plugins[idx].needs_rollback = true;
}
}
}

576
src/core/tests.rs
View File

@@ -2,7 +2,7 @@ use super::config::{parse_str, AppConfig};
use super::message::{Destination, Envelope, Message};
use super::plugin::{CapabilityTestResult, Platform, Plugin, PluginContext, PluginInfo};
use super::plugin_loader::{ErrorPolicy, PluginLoader, PluginRegistry, PluginRegistryEntry};
use super::service_manager::ServiceManager;
use super::service_manager::{PluginStateInfo, ServiceManager};
use super::version_manager::VersionManager;
use anyhow::Result;
use std::fs;
@@ -58,6 +58,29 @@ fn has_event(events: &Arc<Mutex<Vec<String>>>, expected: &str) -> bool {
lock_events(events).iter().any(|event| event == expected)
}
fn clear_events(events: &Arc<Mutex<Vec<String>>>) {
lock_events(events).clear();
}
fn latest_plugin_states(events: &Arc<Mutex<Vec<String>>>, plugin_id: &str) -> Vec<PluginStateInfo> {
let prefix = format!("msg:{plugin_id}:custom:plugin_states:");
let payload = lock_events(events)
.iter()
.rev()
.find_map(|event| event.strip_prefix(&prefix).map(str::to_owned))
.unwrap_or_else(|| panic!("missing plugin_states event for {}", plugin_id));
serde_json::from_str(&payload).expect("plugin_states payload should deserialize")
}
fn non_plugin_state_events(events: &Arc<Mutex<Vec<String>>>) -> Vec<String> {
lock_events(events)
.iter()
.filter(|event| !event.contains("custom:plugin_states:"))
.cloned()
.collect()
}
fn message_label(message: &Message) -> String {
match message {
Message::Custom { kind, payload } => format!("custom:{kind}:{payload}"),
@@ -89,6 +112,88 @@ struct TestPlugin {
events: Arc<Mutex<Vec<String>>>,
}
#[derive(Clone, Default)]
struct PluginFailurePlan {
fail_init: bool,
fail_start: bool,
fail_stop: bool,
}
struct LifecyclePlugin {
id: String,
events: Arc<Mutex<Vec<String>>>,
plan: Arc<Mutex<PluginFailurePlan>>,
label: String,
}
impl LifecyclePlugin {
fn new(
id: &str,
label: &str,
events: Arc<Mutex<Vec<String>>>,
plan: PluginFailurePlan,
) -> (Self, Arc<Mutex<PluginFailurePlan>>) {
let plan = Arc::new(Mutex::new(plan));
(
Self {
id: id.to_string(),
events,
plan: plan.clone(),
label: label.to_string(),
},
plan,
)
}
fn record(&self, phase: &str) {
lock_events(&self.events).push(format!("{phase}:{}:{}", self.id, self.label));
}
}
impl Plugin for LifecyclePlugin {
fn id(&self) -> &str {
&self.id
}
fn info(&self) -> PluginInfo {
PluginInfo {
name: format!("{}-{}", self.id, self.label),
version: "test".to_string(),
description: "lifecycle test plugin".to_string(),
platform: Platform::Any,
}
}
fn init(&mut self, _ctx: PluginContext) -> Result<()> {
self.record("init");
if self.plan.lock().expect("plan mutex poisoned").fail_init {
anyhow::bail!("init failure:{}:{}", self.id, self.label);
}
Ok(())
}
fn start(&mut self) -> Result<()> {
self.record("start");
if self.plan.lock().expect("plan mutex poisoned").fail_start {
anyhow::bail!("start failure:{}:{}", self.id, self.label);
}
Ok(())
}
fn handle_message(&mut self, msg: Message) -> Result<()> {
self.record(&format!("msg:{}", message_label(&msg)));
Ok(())
}
fn stop(&mut self) -> Result<()> {
self.record("stop");
if self.plan.lock().expect("plan mutex poisoned").fail_stop {
anyhow::bail!("stop failure:{}:{}", self.id, self.label);
}
Ok(())
}
}
impl TestPlugin {
fn new(id: &str, deps: Vec<&str>, events: Arc<Mutex<Vec<String>>>) -> Self {
Self {
@@ -153,8 +258,9 @@ fn service_manager_register_start_and_stop_flow() {
manager.start_all().expect("start_all should succeed");
manager.stop_all().expect("stop_all should succeed");
let events = non_plugin_state_events(&events);
assert_eq!(
lock_events(&events).clone(),
events,
vec![
"init:alpha",
"init:beta",
@@ -283,8 +389,9 @@ fn start_all_sorts_plugins_topologically() {
.expect("start_all should sort dependencies");
manager.stop_all().expect("stop_all should succeed");
let events = non_plugin_state_events(&events);
assert_eq!(
lock_events(&events).clone(),
events,
vec![
"init:alpha",
"init:beta",
@@ -759,6 +866,59 @@ fn auto_rollback_updates_registry_and_marks_pending_when_reload_fails() {
let _ = fs::remove_dir_all(&tmp);
}
#[test]
fn self_test_auto_rollback_updates_registry_and_marks_pending_when_reload_fails() {
let tmp = std::env::temp_dir().join("showen_test_self_test_autorollback");
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
manager.set_version_manager(setup_rollback_store(&tmp, "sensor"));
let plugin = TestPluginWithSelfTest::new(
"sensor",
events.clone(),
vec!["temperature".into()],
vec![CapabilityTestResult {
capability: "temperature".into(),
passed: false,
message: "sensor not connected".into(),
}],
);
manager.register_dynamic_with_manifest(
Box::new(plugin),
ErrorPolicy::AutoRollback,
5,
vec!["temperature".into()],
vec!["temperature".into()],
true,
);
manager.start_all().expect("start_all should succeed");
let registry = PluginLoader::new(&tmp).load_registry().unwrap();
assert_eq!(registry.plugins["sensor"].active_version, "1.0.0");
let states = manager.plugin_states();
assert!(
!states[0].enabled,
"plugin should stay disabled after failed self-test"
);
assert!(
states[0].needs_rollback,
"plugin should be marked for restart-time reload when rollback reload fails"
);
let log = lock_events(&events);
assert!(log.contains(&"init:sensor".to_string()));
assert!(log.contains(&"self_test:sensor".to_string()));
assert!(
!log.contains(&"start:sensor".to_string()),
"plugin should not start after failed required capability"
);
let _ = fs::remove_dir_all(&tmp);
}
#[test]
fn message_config_reload_request_round_trips_through_json() {
let json = serde_json::to_string(&Message::ConfigReloadRequest)
@@ -781,6 +941,8 @@ fn message_config_reloaded_round_trips_through_json() {
assert_eq!(decoded.display.window_title, config.display.window_title);
assert_eq!(decoded.playlist.len(), config.playlist.len());
assert_eq!(decoded.remote_control.port, config.remote_control.port);
assert_eq!(decoded.source_path, config.source_path);
assert_eq!(decoded.source_dir, config.source_dir);
}
other => panic!("unexpected message after round trip: {:?}", other),
}
@@ -880,6 +1042,414 @@ fn handle_message_skips_disabled_plugins() {
assert!(!manager.plugin_states()[1].enabled);
}
#[test]
fn set_plugin_enabled_runs_full_lifecycle() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
let (plugin, _plan) =
LifecyclePlugin::new("beta", "base", events.clone(), PluginFailurePlan::default());
manager.register(Box::new(plugin));
manager.start_all().expect("start_all should succeed");
manager
.set_plugin_enabled("beta", false)
.expect("disable should stop plugin");
manager
.set_plugin_enabled("beta", true)
.expect("enable should re-init and restart plugin");
let events = non_plugin_state_events(&events);
assert_eq!(
events,
vec![
"init:beta:base",
"start:beta:base",
"stop:beta:base",
"init:beta:base",
"start:beta:base",
]
);
assert!(manager.plugin_states()[0].enabled);
}
#[test]
fn set_plugin_enabled_rolls_back_to_disabled_when_restart_fails() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
let (plugin, plan) =
LifecyclePlugin::new("beta", "base", events.clone(), PluginFailurePlan::default());
manager.register(Box::new(plugin));
manager.start_all().expect("start_all should succeed");
manager
.set_plugin_enabled("beta", false)
.expect("disable should succeed");
plan.lock().expect("plan mutex poisoned").fail_start = true;
let error = manager
.set_plugin_enabled("beta", true)
.expect_err("failed start should keep plugin disabled");
assert!(error.to_string().contains("failed to enable plugin 'beta'"));
let events = non_plugin_state_events(&events);
assert_eq!(
events,
vec![
"init:beta:base",
"start:beta:base",
"stop:beta:base",
"init:beta:base",
"start:beta:base",
"stop:beta:base",
]
);
assert!(!manager.plugin_states()[0].enabled);
}
#[test]
fn start_all_broadcasts_initial_plugin_states() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
manager.register(Box::new(TestPlugin::new("alpha", vec![], events.clone())));
manager.register(Box::new(TestPlugin::new("beta", vec![], events.clone())));
manager.start_all().expect("start_all should succeed");
let states = latest_plugin_states(&events, "alpha");
assert_eq!(states.len(), 2);
assert!(states.iter().all(|state| state.enabled));
}
#[test]
fn custom_plugin_disable_command_disables_plugin_and_broadcasts_states() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
manager.register(Box::new(TestPlugin::new("alpha", vec![], events.clone())));
manager.register(Box::new(
LifecyclePlugin::new("beta", "base", events.clone(), PluginFailurePlan::default()).0,
));
manager.start_all().expect("start_all should succeed");
clear_events(&events);
let sender = manager.sender();
sender
.send(Envelope {
from: "http".to_string(),
to: Destination::Manager,
message: Message::Custom {
kind: "plugin_disable".to_string(),
payload: "beta".to_string(),
},
})
.expect("disable should send");
sender
.send(Envelope {
from: "test".to_string(),
to: Destination::Manager,
message: Message::Shutdown,
})
.expect("shutdown should send");
manager.run().expect("run should succeed");
let states = latest_plugin_states(&events, "alpha");
let beta = states.iter().find(|state| state.id == "beta").unwrap();
assert!(!beta.enabled);
assert!(has_event(&events, "stop:beta:base"));
}
#[test]
fn custom_plugin_enable_command_enables_plugin_and_broadcasts_states() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
let (plugin, _plan) =
LifecyclePlugin::new("beta", "base", events.clone(), PluginFailurePlan::default());
manager.register(Box::new(TestPlugin::new("alpha", vec![], events.clone())));
manager.register(Box::new(plugin));
manager.start_all().expect("start_all should succeed");
manager
.set_plugin_enabled("beta", false)
.expect("disable should succeed");
clear_events(&events);
let sender = manager.sender();
sender
.send(Envelope {
from: "http".to_string(),
to: Destination::Manager,
message: Message::Custom {
kind: "plugin_enable".to_string(),
payload: "beta".to_string(),
},
})
.expect("enable should send");
sender
.send(Envelope {
from: "test".to_string(),
to: Destination::Manager,
message: Message::Shutdown,
})
.expect("shutdown should send");
manager.run().expect("run should succeed");
let states = latest_plugin_states(&events, "alpha");
let beta = states.iter().find(|state| state.id == "beta").unwrap();
assert!(beta.enabled);
assert!(has_event(&events, "init:beta:base"));
assert!(has_event(&events, "start:beta:base"));
}
#[test]
fn custom_plugin_rollback_command_rolls_back_registry_and_broadcasts_states() {
let tmp = unique_test_dir("custom_plugin_rollback");
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
manager.set_version_manager(setup_rollback_store(&tmp, "sensor"));
manager.register(Box::new(TestPlugin::new("alpha", vec![], events.clone())));
manager.register_dynamic(
Box::new(FailingPlugin::new("sensor", events.clone())),
ErrorPolicy::AutoRollback,
1,
);
manager.start_all().expect("start_all should succeed");
clear_events(&events);
let sender = manager.sender();
sender
.send(Envelope {
from: "http".to_string(),
to: Destination::Manager,
message: Message::Custom {
kind: "plugin_rollback".to_string(),
payload: "sensor".to_string(),
},
})
.expect("rollback should send");
sender
.send(Envelope {
from: "test".to_string(),
to: Destination::Manager,
message: Message::Shutdown,
})
.expect("shutdown should send");
manager.run().expect("run should succeed");
let registry = PluginLoader::new(&tmp)
.load_registry()
.expect("registry should load");
assert_eq!(registry.plugins["sensor"].active_version, "1.0.0");
let states = latest_plugin_states(&events, "alpha");
let sensor = states.iter().find(|state| state.id == "sensor").unwrap();
assert!(sensor.needs_rollback);
let _ = fs::remove_dir_all(&tmp);
}
#[test]
fn custom_plugin_switch_command_with_invalid_payload_still_broadcasts_states() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
manager.register(Box::new(TestPlugin::new("alpha", vec![], events.clone())));
manager.start_all().expect("start_all should succeed");
clear_events(&events);
let sender = manager.sender();
sender
.send(Envelope {
from: "http".to_string(),
to: Destination::Manager,
message: Message::Custom {
kind: "plugin_switch".to_string(),
payload: "not-json".to_string(),
},
})
.expect("switch should send");
sender
.send(Envelope {
from: "test".to_string(),
to: Destination::Manager,
message: Message::Shutdown,
})
.expect("shutdown should send");
manager.run().expect("run should succeed");
let states = latest_plugin_states(&events, "alpha");
assert_eq!(states.len(), 1);
assert_eq!(states[0].id, "alpha");
}
#[test]
fn custom_plugin_install_command_without_version_manager_still_broadcasts_states() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
manager.register(Box::new(TestPlugin::new("alpha", vec![], events.clone())));
manager.start_all().expect("start_all should succeed");
clear_events(&events);
let sender = manager.sender();
sender
.send(Envelope {
from: "http".to_string(),
to: Destination::Manager,
message: Message::Custom {
kind: "plugin_install".to_string(),
payload: serde_json::json!({ "id": "weather" }).to_string(),
},
})
.expect("install should send");
sender
.send(Envelope {
from: "test".to_string(),
to: Destination::Manager,
message: Message::Shutdown,
})
.expect("shutdown should send");
manager.run().expect("run should succeed");
let states = latest_plugin_states(&events, "alpha");
assert_eq!(states.len(), 1);
assert_eq!(states[0].id, "alpha");
}
#[test]
fn custom_plugin_check_updates_command_without_version_manager_still_broadcasts_states() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
manager.register(Box::new(TestPlugin::new("alpha", vec![], events.clone())));
manager.start_all().expect("start_all should succeed");
clear_events(&events);
let sender = manager.sender();
sender
.send(Envelope {
from: "http".to_string(),
to: Destination::Manager,
message: Message::Custom {
kind: "plugin_check_updates".to_string(),
payload: String::new(),
},
})
.expect("check updates should send");
sender
.send(Envelope {
from: "test".to_string(),
to: Destination::Manager,
message: Message::Shutdown,
})
.expect("shutdown should send");
manager.run().expect("run should succeed");
let states = latest_plugin_states(&events, "alpha");
assert_eq!(states.len(), 1);
assert_eq!(states[0].id, "alpha");
}
#[test]
fn replace_dynamic_plugin_stops_old_before_starting_new() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
let (old_plugin, _old_plan) = LifecyclePlugin::new(
"sensor",
"old",
events.clone(),
PluginFailurePlan::default(),
);
let (new_plugin, _new_plan) = LifecyclePlugin::new(
"sensor",
"new",
events.clone(),
PluginFailurePlan::default(),
);
manager.register_dynamic(Box::new(old_plugin), ErrorPolicy::DisableAndLog, 5);
manager.start_all().expect("start_all should succeed");
manager
.replace_dynamic_plugin(
"sensor",
Box::new(new_plugin),
ErrorPolicy::DisableAndLog,
5,
)
.expect("replace should succeed");
let events = non_plugin_state_events(&events);
assert_eq!(
events,
vec![
"init:sensor:old",
"start:sensor:old",
"stop:sensor:old",
"init:sensor:new",
"start:sensor:new",
]
);
}
#[test]
fn replace_dynamic_plugin_restores_old_plugin_when_new_start_fails() {
let events = Arc::new(Mutex::new(Vec::new()));
let mut manager = ServiceManager::new(test_config());
let (old_plugin, _old_plan) = LifecyclePlugin::new(
"sensor",
"old",
events.clone(),
PluginFailurePlan::default(),
);
let (new_plugin, _new_plan) = LifecyclePlugin::new(
"sensor",
"new",
events.clone(),
PluginFailurePlan {
fail_start: true,
..PluginFailurePlan::default()
},
);
manager.register_dynamic(Box::new(old_plugin), ErrorPolicy::DisableAndLog, 5);
manager.start_all().expect("start_all should succeed");
let error = manager
.replace_dynamic_plugin(
"sensor",
Box::new(new_plugin),
ErrorPolicy::DisableAndLog,
5,
)
.expect_err("failed replacement should restore old plugin");
assert!(error.to_string().contains("restored previous plugin"));
let events = non_plugin_state_events(&events);
assert_eq!(
events,
vec![
"init:sensor:old",
"start:sensor:old",
"stop:sensor:old",
"init:sensor:new",
"start:sensor:new",
"stop:sensor:new",
"init:sensor:old",
"start:sensor:old",
]
);
assert!(manager.plugin_states()[0].enabled);
}
#[test]
fn rollback_without_stable_version_returns_error_and_keeps_active_version() {
let tmp = unique_test_dir("rollback_without_stable");

130
src/core/version_manager.rs
View File

@@ -4,6 +4,7 @@
use crate::core::plugin_loader::PluginLoader;
use anyhow::{anyhow, Context, Result};
use std::collections::HashSet;
/// 版本管理器
pub struct VersionManager {
@@ -42,9 +43,7 @@ impl VersionManager {
entry.last_stable_version = Some(version.to_string());
self.loader.save_registry(&registry)?;
println!(
"[VersionManager] 插件 '{plugin_id}' v{version} 标记为稳定版本"
);
println!("[VersionManager] 插件 '{plugin_id}' v{version} 标记为稳定版本");
Ok(())
}
@@ -71,20 +70,14 @@ impl VersionManager {
entry.active_version = stable_version.clone();
self.loader.save_registry(&registry)?;
println!(
"[VersionManager] 插件 '{plugin_id}' 从 v{old_version} 回退到 v{stable_version}"
);
println!("[VersionManager] 插件 '{plugin_id}' 从 v{old_version} 回退到 v{stable_version}");
Ok(stable_version)
}
/// 切换到指定版本
pub fn switch_version(&self, plugin_id: &str, version: &str) -> Result<()> {
// 验证版本目录存在
let version_dir = self
.loader
.store_path()
.join(plugin_id)
.join(version);
let version_dir = self.loader.store_path().join(plugin_id).join(version);
if !version_dir.exists() {
return Err(anyhow!(
"version {version} not found for plugin '{plugin_id}'"
@@ -100,9 +93,7 @@ impl VersionManager {
entry.active_version = version.to_string();
self.loader.save_registry(&registry)?;
println!(
"[VersionManager] 插件 '{plugin_id}' 切换到 v{version}"
);
println!("[VersionManager] 插件 '{plugin_id}' 切换到 v{version}");
Ok(())
}
@@ -116,9 +107,8 @@ impl VersionManager {
.into_iter()
.map(|v| {
let is_active = entry.map_or(false, |e| e.active_version == v);
let is_stable = entry.map_or(false, |e| {
e.last_stable_version.as_deref() == Some(&v)
});
let is_stable =
entry.map_or(false, |e| e.last_stable_version.as_deref() == Some(&v));
VersionInfo {
version: v,
is_active,
@@ -137,6 +127,10 @@ impl VersionManager {
let active = entry.map(|e| e.active_version.as_str());
let stable = entry.and_then(|e| e.last_stable_version.as_deref());
let protected_count = IntoIterator::into_iter([active, stable])
.flatten()
.collect::<HashSet<_>>()
.len();
// 保护活跃版本和稳定版本
let mut deletable: Vec<&str> = versions
@@ -147,22 +141,14 @@ impl VersionManager {
// 保留最近的 keep 个（版本排在后面的更新）
let mut removed = Vec::new();
while deletable.len() + 2 > keep && !deletable.is_empty() {
// 2 是为活跃和稳定版本预留
while deletable.len() + protected_count > keep && !deletable.is_empty() {
let oldest = deletable.remove(0);
let version_dir = self
.loader
.store_path()
.join(plugin_id)
.join(oldest);
let version_dir = self.loader.store_path().join(plugin_id).join(oldest);
if version_dir.exists() {
std::fs::remove_dir_all(&version_dir).with_context(|| {
format!("failed to remove {}", version_dir.display())
})?;
std::fs::remove_dir_all(&version_dir)
.with_context(|| format!("failed to remove {}", version_dir.display()))?;
removed.push(oldest.to_string());
println!(
"[VersionManager] 已清理 '{plugin_id}' v{oldest}"
);
println!("[VersionManager] 已清理 '{plugin_id}' v{oldest}");
}
}
@@ -178,6 +164,19 @@ mod tests {
use std::path::Path;
fn setup(base: &Path) -> VersionManager {
setup_with_entry(
base,
PluginRegistryEntry {
active_version: "1.1.0".to_string(),
last_stable_version: Some("1.0.0".to_string()),
enabled: true,
error_policy: ErrorPolicy::AutoRollback,
max_errors: 5,
},
)
}
fn setup_with_entry(base: &Path, entry: PluginRegistryEntry) -> VersionManager {
let _ = fs::remove_dir_all(base);
fs::create_dir_all(base).unwrap();
@@ -190,16 +189,7 @@ mod tests {
// 写入注册表
let mut registry = PluginRegistry::default();
registry.plugins.insert(
"test-plugin".to_string(),
PluginRegistryEntry {
active_version: "1.1.0".to_string(),
last_stable_version: Some("1.0.0".to_string()),
enabled: true,
error_policy: ErrorPolicy::AutoRollback,
max_errors: 5,
},
);
registry.plugins.insert("test-plugin".to_string(), entry);
loader.save_registry(&registry).unwrap();
VersionManager::new(loader)
@@ -283,4 +273,64 @@ mod tests {
let _ = fs::remove_dir_all(&tmp);
}
#[test]
fn gc_keeps_overlap_once_when_active_equals_stable() {
let tmp = std::env::temp_dir().join("showen_test_gc_overlap");
let vm = setup_with_entry(
&tmp,
PluginRegistryEntry {
active_version: "1.1.0".to_string(),
last_stable_version: Some("1.1.0".to_string()),
enabled: true,
error_policy: ErrorPolicy::AutoRollback,
max_errors: 5,
},
);
let removed = vm.gc("test-plugin", 2).unwrap();
assert_eq!(removed, vec!["1.0.0"]);
let versions = vm.loader().list_versions("test-plugin").unwrap();
assert_eq!(versions, vec!["1.1.0", "2.0.0"]);
let _ = fs::remove_dir_all(&tmp);
}
#[test]
fn gc_keeps_active_when_stable_is_none() {
let tmp = std::env::temp_dir().join("showen_test_gc_no_stable");
let vm = setup_with_entry(
&tmp,
PluginRegistryEntry {
active_version: "1.1.0".to_string(),
last_stable_version: None,
enabled: true,
error_policy: ErrorPolicy::AutoRollback,
max_errors: 5,
},
);
let removed = vm.gc("test-plugin", 1).unwrap();
assert_eq!(removed, vec!["1.0.0", "2.0.0"]);
let versions = vm.loader().list_versions("test-plugin").unwrap();
assert_eq!(versions, vec!["1.1.0"]);
let _ = fs::remove_dir_all(&tmp);
}
#[test]
fn gc_does_not_remove_protected_versions_when_keep_is_smaller() {
let tmp = std::env::temp_dir().join("showen_test_gc_keep_smaller");
let vm = setup(&tmp);
let removed = vm.gc("test-plugin", 1).unwrap();
assert_eq!(removed, vec!["2.0.0"]);
let versions = vm.loader().list_versions("test-plugin").unwrap();
assert_eq!(versions, vec!["1.0.0", "1.1.0"]);
let _ = fs::remove_dir_all(&tmp);
}
}

226
src/plugins/ble/gatt.rs
View File

@@ -6,14 +6,14 @@ use dbus::blocking::stdintf::org_freedesktop_dbus::{ObjectManager, Properties};
use dbus::blocking::Connection;
use dbus::channel::MatchingReceiver;
use dbus::channel::Sender;
use dbus::message::MatchRule;
use dbus::message::{MatchRule, Message as DbusMessage, MessageType};
use dbus::Path;
use dbus_crossroads::{Crossroads, IfaceBuilder, IfaceToken, MethodErr};
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, TryRecvError};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use std::time::{Duration, Instant};
const BUS_NAME: &str = "io.showen.BleProvisioning";
const BLUEZ_SERVICE: &str = "org.bluez";
@@ -37,6 +37,14 @@ const PROXY_TIMEOUT: Duration = Duration::from_secs(10);
type ManagedObjects = HashMap<Path<'static>, HashMap<String, PropMap>>;
#[derive(Default)]
struct RegistrationReplies {
gatt_serial: Option<u32>,
advertisement_serial: Option<u32>,
gatt: Option<Result<()>>,
advertisement: Option<Result<()>>,
}
pub enum BleControl {
UpdateStatus(String),
}
@@ -171,8 +179,7 @@ pub fn run_ble_service(
let shared = SharedState::new(tx.clone());
eprintln!("[BLE] connecting to system bus...");
let conn =
Connection::new_system().context("failed to connect to system bus for BLE")?;
let conn = Connection::new_system().context("failed to connect to system bus for BLE")?;
conn.request_name(BUS_NAME, false, true, false)
.context("failed to request BLE D-Bus name")?;
eprintln!("[BLE] D-Bus name acquired");
@@ -273,6 +280,35 @@ pub fn run_ble_service(
}),
);
let registration_replies = Arc::new(Mutex::new(RegistrationReplies::default()));
let replies_for_success = Arc::clone(&registration_replies);
let mut success_rule = MatchRule::new();
success_rule.msg_type = Some(MessageType::MethodReturn);
conn.start_receive(
success_rule,
Box::new(move |msg, _conn| {
record_registration_reply(&replies_for_success, &msg, Ok(()));
true
}),
);
let replies_for_error = Arc::clone(&registration_replies);
let mut error_rule = MatchRule::new();
error_rule.msg_type = Some(MessageType::Error);
conn.start_receive(
error_rule,
Box::new(move |msg, _conn| {
let error_message = msg.get1::<String>().unwrap_or_default();
let error = if error_message.is_empty() {
anyhow!("{msg:?}")
} else {
anyhow!("{error_message}")
};
record_registration_reply(&replies_for_error, &msg, Err(error));
true
}),
);
// 配置 adapter
let adapter_path = find_adapter(&conn)?;
configure_adapter(&conn, &adapter_path, &device_name)?;
@@ -280,18 +316,23 @@ pub fn run_ble_service(
// 先尝试清理上一次进程残留的注册（防止崩溃后 BlueZ 状态残留）
let _ = unregister_ble_objects(&conn, &adapter_path);
// 非阻塞发送 RegisterApplication + RegisterAdvertisement
let _gatt_serial = send_register_gatt_app(&conn, &adapter_path)?;
let _ad_serial = send_register_advertisement(&conn, &adapter_path)?;
eprintln!("[BLE] registration requests sent, processing callbacks...");
let gatt_serial = send_register_gatt_app(&conn, &adapter_path)?;
let ad_serial = send_register_advertisement(&conn, &adapter_path)?;
if let Ok(mut replies) = registration_replies.lock() {
replies.gatt_serial = Some(gatt_serial);
replies.advertisement_serial = Some(ad_serial);
}
eprintln!("[BLE] registration requests sent, waiting for BlueZ replies...");
// 处理消息循环等待 BlueZ 回调 GetManagedObjects 并完成注册
// start_receive 会处理所有入站方法调用（包括 BlueZ 的回调），
// 注册回复也由 process() 内部分发，我们只需等待足够时间
let deadline = std::time::Instant::now() + Duration::from_secs(5);
while std::time::Instant::now() < deadline {
conn.process(Duration::from_millis(100))
.context("BLE connection process failed during registration")?;
if let Err(error) = wait_for_registration_replies(
&conn,
&registration_replies,
gatt_serial,
ad_serial,
Duration::from_secs(10),
) {
let _ = unregister_ble_objects(&conn, &adapter_path);
return Err(error);
}
eprintln!("[BLE] GATT application and advertisement registered");
@@ -504,6 +545,81 @@ fn build_managed_objects() -> ManagedObjects {
objects
}
fn record_registration_reply(
replies: &Arc<Mutex<RegistrationReplies>>,
msg: &DbusMessage,
result: Result<()>,
) {
let Some(reply_serial) = msg.get_reply_serial() else {
return;
};
if let Ok(mut replies) = replies.lock() {
match replies_for_serial(&mut replies, reply_serial) {
Some(slot) if slot.is_none() => *slot = Some(result),
_ => {}
}
}
}
fn replies_for_serial(
replies: &mut RegistrationReplies,
reply_serial: u32,
) -> Option<&mut Option<Result<()>>> {
if replies.gatt_serial == Some(reply_serial) {
Some(&mut replies.gatt)
} else if replies.advertisement_serial == Some(reply_serial) {
Some(&mut replies.advertisement)
} else {
None
}
}
fn wait_for_registration_replies(
conn: &Connection,
replies: &Arc<Mutex<RegistrationReplies>>,
gatt_serial: u32,
advertisement_serial: u32,
timeout: Duration,
) -> Result<()> {
let deadline = Instant::now() + timeout;
loop {
if let Ok(mut replies) = replies.lock() {
match reply_status(&mut replies.gatt, "RegisterApplication")? {
Some(()) => {}
None => {}
}
match reply_status(&mut replies.advertisement, "RegisterAdvertisement")? {
Some(()) => {}
None => {}
}
if replies.gatt.is_some() && replies.advertisement.is_some() {
return Ok(());
}
}
let now = Instant::now();
if now >= deadline {
return Err(anyhow!(
"timed out waiting for BLE registration reply (gatt_serial={gatt_serial}, advertisement_serial={advertisement_serial})"
));
}
conn.process(Duration::from_millis(100).min(deadline.saturating_duration_since(now)))
.context("BLE connection process failed during registration")?;
}
}
fn reply_status(reply: &mut Option<Result<()>>, operation: &str) -> Result<Option<()>> {
match reply {
Some(Ok(())) => Ok(Some(())),
Some(Err(error)) => Err(anyhow!("{operation} failed: {error}")),
None => Ok(None),
}
}
fn send_register_gatt_app(conn: &Connection, adapter_path: &str) -> Result<u32> {
let msg = dbus::Message::method_call(
&BLUEZ_SERVICE.into(),
@@ -594,8 +710,7 @@ fn unregister_ble_objects(conn: &Connection, adapter_path: &str) -> Result<()> {
fn bytes_to_string(value: &[u8]) -> String {
String::from_utf8_lossy(value)
.trim_end_matches('\0')
.trim()
.trim_matches(|c: char| c == '\0' || c.is_whitespace())
.to_string()
}
@@ -630,3 +745,80 @@ fn emit_status_notification(conn: &Connection, shared: &SharedState) -> Result<(
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::message::WifiCommand;
#[test]
fn bytes_to_string_trims_nulls_and_surrounding_whitespace() {
assert_eq!(bytes_to_string(b" demo-ssid\0\0 "), "demo-ssid");
}
#[test]
fn dispatch_command_uses_cached_wifi_credentials() {
let (tx, rx) = mpsc::channel();
let shared = SharedState::new(tx);
shared.set_ssid(br#"Cafe \"Guest\""#);
shared.set_password(br#"pa\\ss word"#);
shared
.dispatch_command(b"connect")
.expect("connect command should dispatch");
let envelope = rx.recv().expect("command should be forwarded to core");
match envelope.message {
Message::WifiCommand(WifiCommand::Connect { ssid, password }) => {
assert_eq!(ssid, r#"Cafe \"Guest\""#);
assert_eq!(password, r#"pa\\ss word"#);
}
other => panic!("unexpected forwarded BLE command: {:?}", other),
}
assert_eq!(
String::from_utf8(shared.read_status()).expect("status should be utf8"),
r#"{"ok":true,"action":"connect","state":"queued"}"#
);
}
#[test]
fn dispatch_command_reports_invalid_command_in_status() {
let (tx, _rx) = mpsc::channel();
let shared = SharedState::new(tx);
let error = shared
.dispatch_command(b"bad-command")
.expect_err("invalid command should fail");
assert!(error
.to_string()
.contains("unsupported command: bad-command"));
let status = String::from_utf8(shared.read_status()).expect("status should be utf8");
assert!(status.contains(r#""ok":false"#));
assert!(status.contains(r#""action":"bad-command""#));
assert!(status.contains(r#"unsupported command: bad-command"#));
}
#[test]
fn drain_control_messages_updates_status_without_dbus() {
let (tx, _rx) = mpsc::channel();
let shared = SharedState::new(tx);
let (control_tx, control_rx) = mpsc::channel();
control_tx
.send(BleControl::UpdateStatus(
r#"{"ok":true,"action":"status"}"#.to_string(),
))
.expect("status update should send");
drain_control_messages(&shared, &control_rx).expect("control queue should drain");
assert_eq!(
String::from_utf8(shared.read_status()).expect("status should be utf8"),
r#"{"ok":true,"action":"status"}"#
);
assert!(shared.take_pending_notification());
}
}

34
src/plugins/http/mod.rs
View File

@@ -11,7 +11,9 @@ use anyhow::{Context, Result};
use serde::Serialize;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::JoinHandle;
use tokio::sync::broadcast;
use tokio::sync::{oneshot, Mutex as AsyncMutex};
#[derive(Serialize)]
struct WsEvent<'a, T> {
@@ -36,6 +38,7 @@ struct PendingWifiResponse {
}
pub(crate) struct HttpState {
wifi_request_lock: AsyncMutex<()>,
wifi_response: Mutex<PendingWifiResponse>,
wifi_response_cv: Condvar,
last_wifi_result: Mutex<Option<String>>,
@@ -60,6 +63,7 @@ impl HttpState {
};
Self {
wifi_request_lock: AsyncMutex::new(()),
wifi_response: Mutex::new(PendingWifiResponse {
version: 0,
payload: None,
@@ -202,6 +206,8 @@ impl HttpState {
pub struct HttpPlugin {
ctx: Option<PluginContext>,
state: Option<Arc<HttpState>>,
shutdown_tx: Option<oneshot::Sender<()>>,
server_thread: Option<JoinHandle<()>>,
}
impl HttpPlugin {
@@ -209,6 +215,8 @@ impl HttpPlugin {
Self {
ctx: None,
state: None,
shutdown_tx: None,
server_thread: None,
}
}
}
@@ -244,6 +252,8 @@ impl Plugin for HttpPlugin {
}
fn start(&mut self) -> Result<()> {
self.stop()?;
let ctx = self
.ctx
.as_ref()
@@ -263,7 +273,9 @@ impl Plugin for HttpPlugin {
.context("http plugin state is not initialized")?,
);
std::thread::spawn(move || {
let (shutdown_tx, shutdown_rx) = oneshot::channel();
let server_thread = std::thread::spawn(move || {
let runtime = match tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
@@ -294,10 +306,18 @@ impl Plugin for HttpPlugin {
}
println!("[HttpPlugin] listening on http://{addr}");
warp::serve(routes).run(addr).await;
warp::serve(routes)
.bind_with_graceful_shutdown(addr, async move {
let _ = shutdown_rx.await;
})
.1
.await;
});
});
self.shutdown_tx = Some(shutdown_tx);
self.server_thread = Some(server_thread);
Ok(())
}
@@ -344,6 +364,16 @@ impl Plugin for HttpPlugin {
}
fn stop(&mut self) -> Result<()> {
if let Some(shutdown_tx) = self.shutdown_tx.take() {
let _ = shutdown_tx.send(());
}
if let Some(server_thread) = self.server_thread.take() {
server_thread
.join()
.map_err(|_| anyhow::anyhow!("http server thread panicked"))?;
}
Ok(())
}
}

119
src/plugins/http/routes.rs
View File

@@ -10,11 +10,16 @@ use serde_json::Value;
use std::convert::Infallible;
use std::path::{Path, PathBuf};
use std::sync::{mpsc, Arc};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tokio::io::AsyncWriteExt;
use warp::http::StatusCode;
use warp::multipart::FormData;
use warp::{Filter, Reply};
const MAX_UPLOAD_FILE_SIZE: u64 = 100 * 1024 * 1024;
static UPLOAD_TMP_COUNTER: AtomicU64 = AtomicU64::new(0);
#[derive(Deserialize)]
struct WifiConnectRequest {
ssid: String,
@@ -801,28 +806,13 @@ async fn handle_video_upload(
continue;
}
let data = match part
.stream()
.try_fold(Vec::new(), |mut acc, buf| async move {
acc.extend_from_slice(buf.chunk());
Ok(acc)
})
.await
{
Ok(data) => data,
Err(error) => {
return Ok(error_json(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("读取文件失败: {error}"),
));
}
};
if let Err(error) = std::fs::write(dir.join(&safe_name), &data) {
return Ok(error_json(
StatusCode::INTERNAL_SERVER_ERROR,
&format!("保存文件失败: {error}"),
));
if let Err(error) = stream_upload_part(part, &dir.join(&safe_name)).await {
let status = if error.contains("文件大小超过限制") {
StatusCode::PAYLOAD_TOO_LARGE
} else {
StatusCode::INTERNAL_SERVER_ERROR
};
return Ok(error_json(status, &error));
}
uploaded.push(safe_name);
@@ -959,6 +949,7 @@ async fn wifi_request(
state: Arc<HttpState>,
command: WifiCommand,
) -> Result<Value, warp::reply::Response> {
let _request_guard = state.wifi_request_lock.lock().await;
let version = match state.wifi_response.lock() {
Ok(guard) => guard.version,
Err(_) => {
@@ -1044,6 +1035,69 @@ async fn wifi_request(
Ok(payload)
}
async fn stream_upload_part(
part: warp::multipart::Part,
destination: &Path,
) -> Result<(), String> {
let parent = destination
.parent()
.ok_or_else(|| "上传目标目录无效".to_string())?;
let temp_path = parent.join(format!(
".upload-{}-{}.part",
std::process::id(),
UPLOAD_TMP_COUNTER.fetch_add(1, Ordering::Relaxed)
));
let mut file = match tokio::fs::File::create(&temp_path).await {
Ok(file) => file,
Err(error) => return Err(format!("创建临时文件失败: {error}")),
};
let mut total_size = 0u64;
let mut stream = part.stream();
while let Some(chunk) = match stream.try_next().await {
Ok(chunk) => chunk,
Err(error) => {
let _ = tokio::fs::remove_file(&temp_path).await;
return Err(format!("读取文件失败: {error}"));
}
} {
let chunk_size = chunk.remaining() as u64;
total_size = total_size.saturating_add(chunk_size);
if total_size > MAX_UPLOAD_FILE_SIZE {
let _ = file.flush().await;
drop(file);
let _ = tokio::fs::remove_file(&temp_path).await;
return Err(format!(
"文件大小超过限制: 单文件最大 {} MB",
MAX_UPLOAD_FILE_SIZE / 1024 / 1024
));
}
if let Err(error) = file.write_all(chunk.chunk()).await {
drop(file);
let _ = tokio::fs::remove_file(&temp_path).await;
return Err(format!("写入临时文件失败: {error}"));
}
}
if let Err(error) = file.flush().await {
drop(file);
let _ = tokio::fs::remove_file(&temp_path).await;
return Err(format!("刷新临时文件失败: {error}"));
}
drop(file);
if let Err(error) = tokio::fs::rename(&temp_path, destination).await {
let _ = tokio::fs::remove_file(&temp_path).await;
return Err(format!("保存文件失败: {error}"));
}
Ok(())
}
async fn websocket_session(
ws: warp::ws::WebSocket,
tx: mpsc::Sender<Envelope>,
@@ -1369,20 +1423,13 @@ fn file_upload_route(
continue;
}
let data = match part
.stream()
.try_fold(Vec::new(), |mut acc, buf| async move {
acc.extend_from_slice(buf.chunk());
Ok(acc)
})
.await
{
Ok(d) => d,
Err(e) => return Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("读取失败: {e}"))),
};
if let Err(e) = std::fs::write(&dest, &data) {
return Ok(error_json(StatusCode::INTERNAL_SERVER_ERROR, &format!("保存失败: {e}")));
if let Err(error) = stream_upload_part(part, &dest).await {
let status = if error.contains("文件大小超过限制") {
StatusCode::PAYLOAD_TOO_LARGE
} else {
StatusCode::INTERNAL_SERVER_ERROR
};
return Ok(error_json(status, &error));
}
uploaded.push(safe_name);
}

199
src/plugins/wifi/mod.rs
View File

@@ -36,7 +36,32 @@ impl WifiPlugin {
Self { ctx: None }
}
fn run_nmcli(args: &[&str]) -> Result<String> {
fn nmcli_args(parts: &[&str]) -> Vec<String> {
parts.iter().map(|part| (*part).to_string()).collect()
}
fn build_connect_args(ssid: &str, password: &str) -> Vec<String> {
let mut args = Self::nmcli_args(&["device", "wifi", "connect", ssid]);
if !password.trim().is_empty() {
args.push("password".to_string());
args.push(password.to_string());
}
args
}
fn build_hotspot_args(ssid: &str, password: &str) -> Vec<String> {
vec![
"device".to_string(),
"wifi".to_string(),
"hotspot".to_string(),
"ssid".to_string(),
ssid.to_string(),
"password".to_string(),
password.to_string(),
]
}
fn run_nmcli(args: &[String]) -> Result<String> {
let output = Command::new("nmcli")
.args(args)
.output()
@@ -52,6 +77,39 @@ impl WifiPlugin {
}
}
fn parse_nmcli_fields(line: &str, expected_fields: usize) -> Vec<String> {
let mut fields = Vec::with_capacity(expected_fields.max(1));
let mut current = String::new();
let mut escaped = false;
for ch in line.chars() {
if escaped {
current.push(ch);
escaped = false;
continue;
}
match ch {
'\\' => escaped = true,
':' if fields.len() + 1 < expected_fields => {
fields.push(current);
current = String::new();
}
_ => current.push(ch),
}
}
if escaped {
current.push('\\');
}
fields.push(current);
while fields.len() < expected_fields {
fields.push(String::new());
}
fields
}
fn send_result(&self, payload: String) -> Result<()> {
let ctx = self
.ctx
@@ -87,27 +145,30 @@ impl WifiPlugin {
}
fn scan_networks(&self) -> Result<serde_json::Value> {
Self::run_nmcli(&["device", "wifi", "rescan"])?;
Self::run_nmcli(&Self::nmcli_args(&["device", "wifi", "rescan"]))?;
thread::sleep(Duration::from_secs(2));
let output =
Self::run_nmcli(&["-t", "-f", "SSID,SIGNAL,SECURITY", "device", "wifi", "list"])?;
let output = Self::run_nmcli(&Self::nmcli_args(&[
"--terse",
"--escape",
"yes",
"-f",
"SSID,SIGNAL,SECURITY",
"device",
"wifi",
"list",
]))?;
let networks = output
.lines()
.filter(|line| !line.trim().is_empty())
.filter_map(|line| {
let mut parts = line.splitn(3, ':');
let ssid = parts.next().unwrap_or_default().trim().to_string();
let parts = Self::parse_nmcli_fields(line, 3);
let ssid = parts[0].trim().to_string();
if ssid.is_empty() {
return None;
}
let signal = parts
.next()
.unwrap_or_default()
.trim()
.parse::<i32>()
.unwrap_or_default();
let security = parts.next().unwrap_or_default().trim().to_string();
let signal = parts[1].trim().parse::<i32>().unwrap_or_default();
let security = parts[2].trim().to_string();
Some(WifiNetwork {
ssid,
@@ -134,11 +195,7 @@ impl WifiPlugin {
}
fn connect_network(&self, ssid: &str, password: &str) -> Result<serde_json::Value> {
let mut args = vec!["device", "wifi", "connect", ssid];
if !password.trim().is_empty() {
args.extend(["password", password]);
}
let output = Self::run_nmcli(&args)?;
let output = Self::run_nmcli(&Self::build_connect_args(ssid, password))?;
Ok(json!({
"ok": true,
@@ -149,20 +206,30 @@ impl WifiPlugin {
}
fn status(&self) -> Result<serde_json::Value> {
let device_output = Self::run_nmcli(&[
"-t",
let device_output = Self::run_nmcli(&Self::nmcli_args(&[
"--terse",
"--escape",
"yes",
"-f",
"DEVICE,TYPE,STATE,CONNECTION",
"device",
"status",
])?;
let ip_output = Self::run_nmcli(&["-t", "-f", "DEVICE,IP4.ADDRESS", "device", "show"])?;
]))?;
let ip_output = Self::run_nmcli(&Self::nmcli_args(&[
"--terse",
"--escape",
"yes",
"-f",
"DEVICE,IP4.ADDRESS",
"device",
"show",
]))?;
let mut ip_map: HashMap<String, Vec<String>> = HashMap::new();
for line in ip_output.lines().filter(|line| !line.trim().is_empty()) {
let mut parts = line.splitn(2, ':');
let device = parts.next().unwrap_or_default().trim();
let address = parts.next().unwrap_or_default().trim();
let parts = Self::parse_nmcli_fields(line, 2);
let device = parts[0].trim();
let address = parts[1].trim();
if device.is_empty() || address.is_empty() {
continue;
@@ -178,15 +245,15 @@ impl WifiPlugin {
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| {
let mut parts = line.splitn(4, ':');
let device = parts.next().unwrap_or_default().trim().to_string();
let parts = Self::parse_nmcli_fields(line, 4);
let device = parts[0].trim().to_string();
DeviceStatus {
ip4_addresses: ip_map.remove(&device).unwrap_or_default(),
device,
device_type: parts.next().unwrap_or_default().trim().to_string(),
state: parts.next().unwrap_or_default().trim().to_string(),
connection: parts.next().unwrap_or_default().trim().to_string(),
device_type: parts[1].trim().to_string(),
state: parts[2].trim().to_string(),
connection: parts[3].trim().to_string(),
}
})
.collect::<Vec<_>>();
@@ -199,9 +266,7 @@ impl WifiPlugin {
}
fn ap_start(&self, ssid: &str, password: &str) -> Result<serde_json::Value> {
let output = Self::run_nmcli(&[
"device", "wifi", "hotspot", "ssid", ssid, "password", password,
])?;
let output = Self::run_nmcli(&Self::build_hotspot_args(ssid, password))?;
Ok(json!({
"ok": true,
@@ -212,14 +277,23 @@ impl WifiPlugin {
}
fn ap_stop(&self) -> Result<serde_json::Value> {
let active = Self::run_nmcli(&["-t", "-f", "NAME", "connection", "show", "--active"])?;
let active = Self::run_nmcli(&Self::nmcli_args(&[
"--terse",
"--escape",
"yes",
"-f",
"NAME",
"connection",
"show",
"--active",
]))?;
let hotspot_name = active
.lines()
.map(str::trim)
.find(|name| *name == "hotspot")
.ok_or_else(|| anyhow!("active hotspot connection 'hotspot' not found"))?;
let output = Self::run_nmcli(&["connection", "down", hotspot_name])?;
let output = Self::run_nmcli(&Self::nmcli_args(&["connection", "down", hotspot_name]))?;
Ok(json!({
"ok": true,
@@ -276,3 +350,58 @@ impl Plugin for WifiPlugin {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::WifiPlugin;
#[test]
fn parse_nmcli_fields_unescapes_terse_output() {
let fields = WifiPlugin::parse_nmcli_fields(r#"Cafe\:Net:78:WPA2\\Enterprise"#, 3);
assert_eq!(fields, vec!["Cafe:Net", "78", r#"WPA2\Enterprise"#]);
}
#[test]
fn parse_nmcli_fields_keeps_colons_in_last_field() {
let fields = WifiPlugin::parse_nmcli_fields(r#"wlan0:wifi:connected:Office\:LAN"#, 4);
assert_eq!(fields, vec!["wlan0", "wifi", "connected", "Office:LAN"]);
}
#[test]
fn connect_args_preserve_special_characters() {
let args =
WifiPlugin::build_connect_args(r#"ssid \"qa\" demo"#, r#"p@ss\\word with spaces"#);
assert_eq!(
args,
vec![
"device",
"wifi",
"connect",
r#"ssid \"qa\" demo"#,
"password",
r#"p@ss\\word with spaces"#,
]
);
}
#[test]
fn hotspot_args_preserve_special_characters() {
let args = WifiPlugin::build_hotspot_args("Showen AP", r#"\\quoted pass\\"#);
assert_eq!(
args,
vec![
"device",
"wifi",
"hotspot",
"ssid",
"Showen AP",
"password",
r#"\\quoted pass\\"#,
]
);
}
}