清理旧版无用接口,优化测试

This commit is contained in:
imengying committed 2026-09-08 21:30:41 +08:00
1 parent 304b98ed69
commit a7194dd287
11 files changed
+244 -142

No files matched your search

+3 -2
View File
@@ -119,9 +119,10 @@ if ($Uninstall) {
Stop-ScheduledTask -TaskName $TaskName -ErrorAction SilentlyContinue
Unregister-ScheduledTask -TaskName $TaskName -Confirm:$false -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $AgentFile -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $LauncherFile -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $StateDir -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $DataDir -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $InstallDir -Force -ErrorAction SilentlyContinue
try { [IO.Directory]::Delete($DataDir, $false) } catch [IO.IOException] { }
try { [IO.Directory]::Delete($InstallDir, $false) } catch [IO.IOException] { }
Write-Host "Agent 已卸载"
exit 0
}
+107 -37
View File
@@ -512,26 +512,6 @@ pub async fn reorder_servers(db: &Database, ids: &[String]) -> Result<()> {
Ok(())
}
pub async fn rotate_server_token(db: &Database, id: &str) -> Result<Option<String>> {
let token = auth::random_token(32);
let token_hash = auth::token_hash(&token);
let mut transaction = db.pool().begin().await?;
let result = sqlx::query(db.sql("UPDATE servers SET token_hash=?, updated_at=? WHERE id=?"))
.bind(&token_hash)
.bind(now())
.bind(id)
.execute(&mut *transaction)
.await?;
if result.rows_affected() > 0 {
sqlx::query(db.sql("DELETE FROM server_install_tokens WHERE server_id=?"))
.bind(id)
.execute(&mut *transaction)
.await?;
}
transaction.commit().await?;
Ok((result.rows_affected() > 0).then_some(token))
}
pub async fn agent_install_token(db: &Database, id: &str) -> Result<Option<String>> {
let token = auth::random_token(32);
let exists = sqlx::query_scalar::<_, i64>(db.sql("SELECT COUNT(*) FROM servers WHERE id=?"))
@@ -1478,6 +1458,19 @@ pub async fn update_latency_task(
) -> Result<bool> {
let timestamp = now();
let mut transaction = db.pool().begin().await?;
// Only a different probe destination starts a new history segment.
sqlx::query(db.sql(
"UPDATE latency_task_servers SET assigned_at=? WHERE task_id=? AND EXISTS (\
SELECT 1 FROM latency_tasks WHERE id=latency_task_servers.task_id AND \
(task_type<>? OR target<>? OR COALESCE(port, 0)<>?))",
))
.bind(timestamp)
.bind(id)
.bind(&input.task_type)
.bind(input.target.trim())
.bind(input.port.unwrap_or(0))
.execute(&mut *transaction)
.await?;
let result = sqlx::query(db.sql(
"UPDATE latency_tasks SET name=?, task_type=?, target=?, port=?, interval_seconds=?, \
default_enabled=?, updated_at=? WHERE id=?",
@@ -1508,6 +1501,20 @@ async fn replace_task_servers(
server_ids: &[String],
timestamp: i64,
) -> Result<()> {
let existing = sqlx::query(
db.sql("SELECT server_id, assigned_at FROM latency_task_servers WHERE task_id=?"),
)
.bind(task_id)
.fetch_all(&mut **transaction)
.await?
.into_iter()
.map(|row| {
Ok::<_, sqlx::Error>((
row.try_get::<String, _>("server_id")?,
row.try_get::<i64, _>("assigned_at")?,
))
})
.collect::<std::result::Result<HashMap<_, _>, _>>()?;
sqlx::query(db.sql("DELETE FROM latency_task_servers WHERE task_id=?"))
.bind(task_id)
.execute(&mut **transaction)
@@ -1518,7 +1525,7 @@ async fn replace_task_servers(
))
.bind(task_id)
.bind(server_id)
.bind(timestamp)
.bind(existing.get(server_id).copied().unwrap_or(timestamp))
.execute(&mut **transaction)
.await?;
}
@@ -2635,22 +2642,6 @@ mod tests {
.is_some()
);
assert_ne!(first_install_token, second_install_token);
let rotated_token = rotate_server_token(&db, &id).await.unwrap().unwrap();
assert!(agent_identity(&db, &primary_token).await.unwrap().is_none());
assert!(
agent_identity(&db, &first_install_token)
.await
.unwrap()
.is_none()
);
assert!(
agent_identity(&db, &second_install_token)
.await
.unwrap()
.is_none()
);
assert!(agent_identity(&db, &rotated_token).await.unwrap().is_some());
}
fn sample(timestamp: i64, cpu: f64) -> AgentReport {
@@ -3152,6 +3143,85 @@ mod tests {
assert_eq!(points[0].packet_loss, 100.0);
}
#[tokio::test]
async fn latency_history_survives_metadata_edits_but_not_probe_or_assignment_changes() {
for change in ["metadata", "target", "port", "type", "reassign"] {
let db = super::super::connect("sqlite::memory:").await.unwrap();
db.migrate().await.unwrap();
let (server_id, _) = create_server(&db, &server_input(0)).await.unwrap();
let mut input = LatencyTaskInput {
name: "Persistent latency".to_string(),
task_type: "tcp".to_string(),
target: "example.com".to_string(),
port: Some(443),
interval_seconds: 60,
default_enabled: false,
server_ids: vec![server_id.clone()],
};
let task_id = create_latency_task(&db, &input).await.unwrap();
let assigned_at = now() - 300;
sqlx::query(db.sql(
"UPDATE latency_task_servers SET assigned_at=? WHERE task_id=? AND server_id=?",
))
.bind(assigned_at)
.bind(&task_id)
.bind(&server_id)
.execute(db.pool())
.await
.unwrap();
sqlx::query(db.sql(
"INSERT INTO latency_results(task_id, server_id, timestamp, latency_ms, packet_loss) \
VALUES (?, ?, ?, 12.0, 0.0)",
))
.bind(&task_id)
.bind(&server_id)
.bind(assigned_at + 60)
.execute(db.pool())
.await
.unwrap();
assert_eq!(
latency_history(&db, &server_id, 1).await.unwrap().1.len(),
1
);
match change {
"metadata" => {
input.name = "Renamed latency".to_string();
input.interval_seconds = 120;
input.default_enabled = true;
let (additional, _) = create_server(&db, &server_input(0)).await.unwrap();
input.server_ids.push(additional);
}
"target" => input.target = "example.org".to_string(),
"port" => input.port = Some(80),
"type" => {
input.task_type = "icmp".to_string();
input.port = None;
}
"reassign" => {
input.server_ids.clear();
update_latency_task(&db, &task_id, &input).await.unwrap();
assert!(
latency_history(&db, &server_id, 1)
.await
.unwrap()
.1
.is_empty()
);
input.server_ids.push(server_id.clone());
}
_ => unreachable!(),
}
update_latency_task(&db, &task_id, &input).await.unwrap();
let (_, points) = latency_history(&db, &server_id, 1).await.unwrap();
assert_eq!(
points.len(),
if change == "metadata" { 1 } else { 0 },
"{change}"
);
}
}
#[tokio::test]
async fn database_cleanup_expires_tasks_and_removes_retained_data() {
let db = super::super::connect("sqlite::memory:").await.unwrap();
-5
View File
@@ -252,10 +252,6 @@ async fn main() -> Result<()> {
"/api/admin/servers/order",
patch(routes::admin::servers_order),
)
.route(
"/api/admin/servers/{id}/token",
post(routes::admin::server_token_rotate),
)
.route(
"/api/admin/servers/{id}/agent-token",
post(routes::admin::server_agent_token),
@@ -351,7 +347,6 @@ async fn main() -> Result<()> {
let app = Router::new()
.route("/api/bootstrap", get(routes::public::bootstrap))
.route("/api/config", get(routes::public::config))
.route("/api/servers", get(routes::public::servers))
.route("/api/history/{id}", get(routes::public::history))
.route("/api/latency/{id}", get(routes::public::latency_history))
.route("/api/exchange-rates", get(routes::public::exchange_rates))
-13
View File
@@ -120,19 +120,6 @@ pub async fn servers_order(
Ok(StatusCode::NO_CONTENT.into_response())
}
pub async fn server_token_rotate(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Result<Response, ApiResponse> {
validate_id(&id)?;
let token = crate::db::queries::rotate_server_token(&state.db, &id)
.await
.map_err(ApiResponse::internal)?
.ok_or_else(|| ApiResponse::not_found("节点不存在"))?;
state.disconnect_agent(&id).await;
Ok(Json(serde_json::json!({"agent_token": token})).into_response())
}
pub async fn server_agent_token(
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
-14
View File
@@ -122,20 +122,6 @@ pub async fn config(State(state): State<Arc<AppState>>) -> Result<Json<PublicCon
Ok(Json(public_config_with_totp(&state, &settings).await?))
}
pub async fn servers(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
) -> Result<Response, ApiResponse> {
require_dashboard(&state, &headers).await?;
let servers = crate::db::queries::list_servers(&state.db, false)
.await
.map_err(ApiResponse::internal)?
.into_iter()
.map(public_server)
.collect::<Vec<_>>();
Ok(Json(serde_json::json!({"servers": servers})).into_response())
}
pub async fn history(
State(state): State<Arc<AppState>>,
headers: HeaderMap,
+1 -1
View File
@@ -106,7 +106,7 @@ async fn run(
let _ = previous.sender.try_send(AgentCommand::Close);
}
// Close the token-rotation race between authentication and registration.
// Recheck the token after registration in case the node changed meanwhile.
let still_authorized = matches!(
crate::db::queries::agent_identity(&state.db, &token).await,
Ok(Some(current)) if current.server_id == identity.server_id
+96 -4
View File
@@ -1,7 +1,7 @@
import { describe, expect, test } from "bun:test";
import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { delimiter, join } from "node:path";
import { delimiter, dirname, join } from "node:path";
import { spawnSync } from "node:child_process";
const installer = readFileSync(new URL("../../install.sh", import.meta.url), "utf8");
@@ -70,10 +70,74 @@ describe("server uninstall ownership", () => {
}
});
function start(scenario: string) {
describe("Windows uninstall ownership", () => {
const powershell = Bun.which("pwsh");
for (const kind of ["server", "agent"] as const) {
const source = readFileSync(new URL(kind === "server" ? "../../install.ps1" : "../../agent/install.ps1", import.meta.url), "utf8");
const branch = kind === "server"
? source.match(/^if \(\$Mode -eq "Uninstall"\) \{[\s\S]*?^\}/m)?.[0]
: source.match(/^if \(\$Uninstall\) \{[\s\S]*?^\}/m)?.[0];
if (!branch) throw new Error(`Missing Windows ${kind} uninstall branch`);
test.skipIf(!powershell)(`${kind} keeps the other service's shared files and data`, () => {
const directory = mkdtempSync(join(tmpdir(), "nodeflare-windows-uninstall-"));
const files = {
"install/nodeflare.exe": "server-binary",
"install/share/frontend/index.html": "server-frontend",
"install/agent.exe": "agent-binary",
"install/run-agent.ps1": "agent-launcher",
"data/Server/config.toml": "server-data",
"data/Agent/config.json": "agent-data",
};
try {
for (const [path, content] of Object.entries(files)) {
mkdirSync(dirname(join(directory, path)), { recursive: true });
writeFileSync(join(directory, path), content);
}
const result = spawnSync(powershell!, ["-NoLogo", "-NoProfile", "-NonInteractive", "-Command", `
$ErrorActionPreference = "Stop"
function Write-Step { }
function Stop-ScheduledTask { }
function Unregister-ScheduledTask { }
$Mode = "Uninstall"
$Uninstall = $true
$Purge = $false
$TaskName = "unused"
$InstallDir = Join-Path $env:TEST_DIRECTORY "install"
$ShareDir = Join-Path $InstallDir "share"
$ServerFile = Join-Path $InstallDir "nodeflare.exe"
$AgentFile = Join-Path $InstallDir "agent.exe"
$LauncherFile = Join-Path $InstallDir "run-agent.ps1"
$DataDir = Join-Path $env:TEST_DIRECTORY "data"
$StateDir = Join-Path $DataDir "Agent"
if ($env:TEST_KIND -eq "server") { $DataDir = Join-Path $DataDir "Server" }
${branch}
`], { env: { ...process.env, TEST_DIRECTORY: directory, TEST_KIND: kind }, encoding: "utf8", timeout: 10_000 });
expect(result.stderr).toBe("");
expect(result.status).toBe(0);
const peerPaths = kind === "server"
? ["install/agent.exe", "install/run-agent.ps1", "data/Agent/config.json"]
: ["install/nodeflare.exe", "install/share/frontend/index.html", "data/Server/config.toml"];
for (const path of peerPaths) {
expect(readFileSync(join(directory, path), "utf8")).toBe(files[path as keyof typeof files]);
}
expect(existsSync(join(directory, kind === "server" ? "install/nodeflare.exe" : "install/agent.exe"))).toBe(false);
if (kind === "agent") {
expect(existsSync(join(directory, "install/run-agent.ps1"))).toBe(false);
expect(existsSync(join(directory, "data/Agent"))).toBe(false);
}
} finally {
rmSync(directory, { recursive: true, force: true });
}
});
}
});
function start(scenario: string, init = "systemd") {
return spawnSync("sh", ["-c", `
set -eu
init_system=systemd
init_system=$TEST_INIT
launchd_file=/unused/nodeflare.plist
tick=0
sleep() { tick=$((tick + 1)); }
systemctl() {
@@ -90,9 +154,24 @@ function start(scenario: string) {
*) return 99 ;;
esac
}
launchctl() {
case "$1" in
bootstrap) [ "$SCENARIO" != command-failure ] ;;
print)
[ "$SCENARIO" != query-failure ] || return 1
if [ "$SCENARIO" = waiting ] || { [ "$SCENARIO" = exits ] && [ "$tick" -ge 2 ]; }; then
printf 'state = waiting\nlast exit code = 1\n'
elif [ "$SCENARIO" = restarts ] && [ "$tick" -ge 2 ]; then
printf 'state = running\npid = 456\n'
else
printf 'state = running\npid = 123\n'
fi ;;
*) return 99 ;;
esac
}
${startFunction}
if start_server; then printf 'ready:%s\n' "$tick"; else exit 1; fi
`], { env: { ...process.env, SCENARIO: scenario }, encoding: "utf8" });
`], { env: { ...process.env, SCENARIO: scenario, TEST_INIT: init }, encoding: "utf8" });
}
describe("installer startup verification", () => {
@@ -108,6 +187,19 @@ describe("installer startup verification", () => {
}
});
describe("launchd installer startup verification", () => {
test("waits for the same process to remain active", () => {
const result = start("healthy", "launchd");
expect(result.status).toBe(0);
expect(result.stdout).toBe("ready:10\n");
});
for (const scenario of ["exits", "restarts", "command-failure", "waiting", "query-failure"]) {
test(`rejects ${scenario}`, () => {
expect(start(scenario, "launchd").status).toBe(1);
});
}
});
describe("installer port selection", () => {
test("writes the default or chosen port to a valid TOML configuration", () => {
const directory = mkdtempSync(join(tmpdir(), "nodeflare-install-test-"));
+1 -1
View File
@@ -182,7 +182,7 @@ if ($Mode -eq "Uninstall") {
Unregister-ScheduledTask -TaskName $TaskName -Confirm:$false -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $ServerFile -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $ShareDir -Recurse -Force -ErrorAction SilentlyContinue
Remove-Item -LiteralPath $InstallDir -Force -ErrorAction SilentlyContinue
try { [IO.Directory]::Delete($InstallDir, $false) } catch [IO.IOException] { }
if ($Purge) {
Remove-Item -LiteralPath $DataDir -Recurse -Force -ErrorAction SilentlyContinue
Write-Step "已删除配置和全部服务端持久数据"
+15 -1
View File
@@ -585,7 +585,21 @@ start_server() {
;;
launchd)
launchctl bootstrap system "$launchd_file" || return 1
launchctl print system/nodeflare >/dev/null
started_pid=""
attempt=1
while [ "$attempt" -le 10 ]; do
sleep 1
launchd_status=$(launchctl print system/nodeflare) || return 1
printf '%s\n' "$launchd_status" | grep -Eq '^[[:space:]]*state = running$' || return 1
current_pid=$(printf '%s\n' "$launchd_status" | awk '$1 == "pid" && $2 == "=" { print $3; exit }')
case "$current_pid" in ''|*[!0-9]*|0) return 1 ;; esac
if [ -z "$started_pid" ]; then
started_pid=$current_pid
else
[ "$current_pid" = "$started_pid" ] || return 1
fi
attempt=$((attempt + 1))
done
;;
freebsd)
sysrc nodeflare_enable=YES >/dev/null || return 1
-11
View File
@@ -243,17 +243,6 @@ MONITOR_BASE_URL="$MONITOR_BASE_URL" MONITOR_ADMIN_TOKEN="$admin_token" \
MONITOR_AGENT_TOKEN="$agent_token" MONITOR_SERVER_ID="$server_id" \
MONITOR_LATENCY_TASK_ID="$latency_task_id" \
node scripts/websocket-smoke.mjs
token_without_admin_status=$(monitor_curl --silent --output /dev/null --write-out '%{http_code}' \
-X POST \
"$MONITOR_BASE_URL/api/admin/servers/$server_id/token")
[ "$token_without_admin_status" = "401" ]
rotated_agent_token=$(MONITOR_BASE_URL="$MONITOR_BASE_URL" MONITOR_ADMIN_TOKEN="$admin_token" \
MONITOR_AGENT_TOKEN="$agent_token" MONITOR_SERVER_ID="$server_id" \
MONITOR_LATENCY_TASK_ID="$latency_task_id" MONITOR_CONFIG_ONLY=1 \
MONITOR_ROTATE_AGENT_TOKEN=1 node scripts/websocket-smoke.mjs)
[ "$rotated_agent_token" != "$agent_token" ]
agent_token=$rotated_agent_token
step "persisted metrics: admin IP metadata"
request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/admin/servers" | \
jq -e --arg id "$server_id" '.servers | any(.id == $id and .last_ip == "8.8.8.8")' >/dev/null
+21 -53
View File
@@ -11,15 +11,9 @@ const serverId = process.env.MONITOR_SERVER_ID;
const latencyTaskId = process.env.MONITOR_LATENCY_TASK_ID;
const expectTaskAssigned = process.env.MONITOR_EXPECT_TASK_ASSIGNED !== "0";
const configOnly = process.env.MONITOR_CONFIG_ONLY === "1";
const expectAgentRejected = process.env.MONITOR_EXPECT_AGENT_REJECTED === "1";
const rotateAgentToken = process.env.MONITOR_ROTATE_AGENT_TOKEN === "1";
if (!baseUrl || !agentToken || (!expectAgentRejected && (!adminToken || !serverId || !latencyTaskId))) {
throw new Error(
expectAgentRejected
? "MONITOR_BASE_URL and MONITOR_AGENT_TOKEN are required"
: "MONITOR_BASE_URL, MONITOR_ADMIN_TOKEN, MONITOR_AGENT_TOKEN, MONITOR_SERVER_ID and MONITOR_LATENCY_TASK_ID are required",
);
if (!baseUrl || !adminToken || !agentToken || !serverId || !latencyTaskId) {
throw new Error("MONITOR_BASE_URL, MONITOR_ADMIN_TOKEN, MONITOR_AGENT_TOKEN, MONITOR_SERVER_ID and MONITOR_LATENCY_TASK_ID are required");
}
function websocketUrl(path) {
@@ -279,11 +273,6 @@ function waitForSocketClose(socket, expected, timeoutMs = 5_000) {
await expectSocketStatus("/api/agent/ws", "invalid-agent-token", 401);
await expectSocketStatus("/api/agent/ws", "", 401);
await expectSocketStatus("/api/agent/ws", "x".repeat(513), 401);
if (expectAgentRejected) {
await expectSocketStatus("/api/agent/ws", agentToken, 401);
process.exit(0);
}
const dashboard = configOnly ? null : await openSocket("/api/ws", adminToken);
let agent;
let enabledTotpSecret = "";
@@ -325,48 +314,27 @@ try {
) {
throw new Error(`Invalid Agent config: ${JSON.stringify(config)}`);
}
if (rotateAgentToken) {
const firstSocket = agent;
const firstClosed = waitForSocketClose(firstSocket, "a duplicate Agent connected");
const replacement = await openSocket("/api/agent/ws", agentToken);
await waitForJsonMessage(
replacement,
"replacement Agent config",
(message) => message.type === "config" && message.config,
);
await firstClosed;
agent = replacement;
if (!configOnly) {
const installUrl = new URL(`/api/admin/servers/${encodeURIComponent(serverId)}/agent-token`, baseUrl);
assert.equal((await fetch(installUrl, { method: "POST" })).status, 401);
const installResponse = await fetch(installUrl, {
method: "POST", headers: { Authorization: `Bearer ${adminToken}` },
});
assert(installResponse.ok, await installResponse.clone().text());
const installToken = (await installResponse.json()).agent_token;
assert(installToken && installToken !== agentToken);
const originalPong = once(agent, "pong", { signal: AbortSignal.timeout(5_000) });
agent.ping();
await originalPong;
const rotatedClosed = waitForSocketClose(agent, "the Agent Token rotated");
const rotateResponse = await fetch(
new URL(`/api/admin/servers/${encodeURIComponent(serverId)}/token`, baseUrl),
{
method: "POST",
headers: { Authorization: `Bearer ${adminToken}` },
},
);
if (!rotateResponse.ok) {
throw new Error(
`Agent Token rotation returned HTTP ${rotateResponse.status}: ${await rotateResponse.text()}`,
);
}
const rotated = await rotateResponse.json();
const rotatedToken = rotated.agent_token;
if (!rotatedToken || rotatedToken === agentToken) {
throw new Error("Agent Token rotation did not return a new token");
}
await rotatedClosed;
await expectSocketStatus("/api/agent/ws", agentToken, 401);
const originalClosed = waitForSocketClose(agent, "a replacement Agent connected");
agent = await openSocket("/api/agent/ws", installToken);
await waitForJsonMessage(agent, "installed Agent config", (message) => message.type === "config");
await originalClosed;
await closeSocket(agent);
agent = await openSocket("/api/agent/ws", agentToken, { "X-Forwarded-For": "8.8.8.8" });
await waitForJsonMessage(agent, "primary Agent config", (message) => message.type === "config");
const rotatedSocket = await openSocket("/api/agent/ws", rotatedToken);
await waitForJsonMessage(
rotatedSocket,
"rotated Agent config",
(message) => message.type === "config" && message.config,
);
await closeSocket(rotatedSocket);
process.stdout.write(`${rotatedToken}\n`);
} else if (!configOnly) {
if (!passwordDerived) throw new Error("MONITOR_PASSWORD_DERIVED is required for TOTP setup");
const setupResponse = await fetch(new URL("/api/admin/2fa/setup", baseUrl), {
method: "POST",