This commit is contained in:
MengMengCode
2026-08-09 14:43:23 +08:00
parent d22246ee92
commit 9ca25383bc
3 changed files with 0 additions and 209 deletions
-52
View File
@@ -1,52 +0,0 @@
#!/usr/bin/env python3
"""Deploy the freshly built vocat binary to the test machine and restart the service."""
import time
import paramiko
HOST, USER, PASSWORD = "192.168.2.222", "root", "lammy520"
LOCAL_BINARY = r"E:\tool\vocat\build\vocat-linux-amd64"
REMOTE_TMP = "/opt/vocat/bin/vocat.new"
REMOTE_BIN = "/opt/vocat/bin/vocat"
def run(client, command, timeout=60):
_, stdout, stderr = client.exec_command(command, timeout=timeout)
out = stdout.read().decode("utf-8", "replace")
err = stderr.read().decode("utf-8", "replace")
code = stdout.channel.recv_exit_status()
if out.strip():
print(out)
if err.strip():
print(err)
if code != 0:
raise SystemExit(f"command failed ({code}): {command}")
def main():
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(HOST, username=USER, password=PASSWORD, timeout=10)
stamp = time.strftime("%Y%m%d-%H%M%S")
# Use SQLite's online backup command so WAL contents are included.
database_backup = f"/opt/vocat/data/vocat.db.bak-deploy-{stamp}"
run(client, f'''sqlite3 /opt/vocat/data/vocat.db ".backup '{database_backup}'"''')
sftp = client.open_sftp()
print(f"uploading {LOCAL_BINARY} -> {REMOTE_TMP} ...")
sftp.put(LOCAL_BINARY, REMOTE_TMP)
sftp.close()
run(client, f"chmod 0755 {REMOTE_TMP}")
run(client, f"cp -a {REMOTE_BIN} {REMOTE_BIN}.bak-deploy-{stamp}")
run(client, f"mv {REMOTE_TMP} {REMOTE_BIN}")
run(client, "systemctl restart vocat")
time.sleep(2)
run(client, "systemctl is-active vocat && systemctl show vocat -p MainPID --value")
run(client, "curl -fsS http://127.0.0.1:7575/api/health || true")
client.close()
print("deploy OK")
if __name__ == "__main__":
main()
-126
View File
@@ -1,126 +0,0 @@
#!/usr/bin/env python3
"""Verify the current vocat deployment on 192.168.2.222:
1. notification channels trimmed to telegram/email/webhook/bark/pushplus
2. removed channels (feishu/qq/weixin) rejected
3. the new unsaved-config front-proxy probe endpoint (UDP Associate for VoWiFi)
4. the served frontend bundle reflects the changes (probe button present, API
docs button and Feishu/QQ tabs gone)
"""
import http.client
import http.cookiejar
import json
import re
import time
import urllib.error
import urllib.request
BASE = "http://192.168.2.222:7575"
USERNAME = "admin"
PASSWORD = "vocat520"
jar = http.cookiejar.CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
results = []
def _once(method, path, body, csrf, timeout, raw):
req = urllib.request.Request(BASE + path, method=method)
data = None
if body is not None:
req.add_header("Content-Type", "application/json")
data = json.dumps(body).encode()
if csrf:
req.add_header("X-CSRF-Token", csrf)
req.add_header("Connection", "close")
try:
with opener.open(req, data=data, timeout=timeout) as resp:
payload = resp.read().decode("utf-8", "replace")
return (resp.status, payload) if raw else (resp.status, json.loads(payload) if payload else {})
except urllib.error.HTTPError as exc:
text = exc.read().decode("utf-8", "replace")
if raw:
return exc.code, text
try:
return exc.code, json.loads(text)
except json.JSONDecodeError:
return exc.code, {"raw": text[:200]}
def call(method, path, body=None, csrf=None, timeout=40, raw=False):
# The Windows TCP stack sometimes reports a RST right after the server has
# already sent a complete response; retry those client-side resets.
last = None
for attempt in range(4):
try:
return _once(method, path, body, csrf, timeout, raw)
except (ConnectionResetError, urllib.error.URLError, http.client.RemoteDisconnected) as exc:
last = exc
time.sleep(0.4 * (attempt + 1))
raise last
def check(name, cond, detail=""):
results.append((name, bool(cond)))
print(("PASS " if cond else "FAIL ") + name + (f" | {detail}" if detail and not cond else ""))
def bundle_has(bundle, s):
# bundle may keep Chinese literally or \uXXXX-escaped depending on charset
if s in bundle:
return True
esc = "".join("\\u%04x" % ord(c) for c in s)
return esc.lower() in bundle.lower()
# 1. login
st, login = call("POST", "/api/auth/login", {"username": USERNAME, "password": PASSWORD})
csrf = (login.get("data") or {}).get("csrf_token", "")
check("登录 admin", st == 200 and bool(csrf), json.dumps(login, ensure_ascii=False)[:160])
# 2. notifications trimmed to exactly the 5 kept channels
st, notif = call("GET", "/api/settings/notifications")
data = notif.get("data") or {}
channels = set(data.keys())
expect = {"telegram", "email", "webhook", "bark", "pushplus"}
check("通知渠道恰好 5 个", st == 200 and channels == expect, ",".join(sorted(channels)))
check("feishu/qq/weixin 已移除", not ({"feishu", "qq", "weixin"} & channels), ",".join(sorted(channels)))
# 3. removed channels rejected by both PUT and the test endpoint
st, resp = call("PUT", "/api/settings/notifications", {"feishu": {"enabled": False}}, csrf)
check("PUT feishu 配置被拒(400)", st == 400, f"status={st} {json.dumps(resp)[:120]}")
for removed in ("feishu", "qq", "weixin"):
st, _ = call("POST", f"/api/settings/notifications/{removed}/test", {}, csrf)
check(f"{removed} 测试端点 404", st == 404, f"status={st}")
# 4. new unsaved-config front-proxy probe endpoint (UDP Associate for VoWiFi)
st, probe = call("POST", "/api/upstream-proxy-probe", {"addr": "127.0.0.1:9"}, csrf, timeout=20)
pdata = probe.get("data") or {}
presult = pdata.get("probe") or {}
check("探测端点存在且返回 probe 结构", st == 200 and "probe" in pdata, f"status={st} {json.dumps(probe, ensure_ascii=False)[:200]}")
check(
"不可达地址正确判定 reachable/udp=false",
presult.get("reachable") is False and presult.get("udp_associate_ok") is False,
json.dumps(presult, ensure_ascii=False)[:200],
)
st, _ = call("POST", "/api/upstream-proxy-probe", {"addr": ""}, csrf, timeout=20)
check("空地址探测返回 400", st == 400, f"status={st}")
# 5. frontend bundle reflects the changes (the app code lives in /assets/index-*.js)
st, index = call("GET", "/", raw=True)
m = re.search(r'src="(/assets/index-[^"]+\.js)"', index)
bundle = ""
if m:
_, bundle = call("GET", m.group(1), raw=True)
check("获取前端 JS bundle", bool(bundle), "no bundle url in index.html")
if bundle:
check("bundle 含「检测连通性」按钮", bundle_has(bundle, "检测连通性"))
check("bundle 含 UDP Associate(VoWiFi) 探测", bundle_has(bundle, "UDP Associate"))
check("bundle 已无「API 文档」按钮", not bundle_has(bundle, "API 文档"))
check("bundle 已无飞书", not bundle_has(bundle, "飞书"))
check("bundle 已无 QQ 渠道", not bundle_has(bundle, "QQ 机器人") and not bundle_has(bundle, "QQ Bot"))
failed = [n for n, ok in results if not ok]
print(f"\n==== {len(results) - len(failed)}/{len(results)} 通过 ====")
if failed:
print("失败项:", "; ".join(failed))
raise SystemExit(1)
-31
View File
@@ -1,31 +0,0 @@
#!/usr/bin/env python3
"""Run a command on the vocat test machine over SSH (paramiko, password auth)."""
import base64
import sys
import paramiko
HOST, USER, PASSWORD = "192.168.2.222", "root", "lammy520"
def main() -> int:
if len(sys.argv) > 2 and sys.argv[1] == "--base64":
command = base64.b64decode(sys.argv[2]).decode("utf-8")
else:
command = sys.argv[1] if len(sys.argv) > 1 else "uname -a"
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(HOST, username=USER, password=PASSWORD, timeout=10)
_, stdout, stderr = client.exec_command(command, timeout=60)
out = stdout.read().decode("utf-8", "replace")
err = stderr.read().decode("utf-8", "replace")
code = stdout.channel.recv_exit_status()
client.close()
sys.stdout.write(out)
if err:
sys.stderr.write(err)
return code
if __name__ == "__main__":
sys.exit(main())