Files
knowledge-governance/internal/libraryexecutor/executor_test.go
T
姚章浩 47eb23cf68 feat(library): 封存执行器可用性与还原条件来自真实观测
页面上的"可发起封存"原来只判 LIBRARY_AGENT_SOCKET 非空:正式栈的 API 根本没挂
那个 socket,而 agent 曾顶着正式单元名冻结候选实例——配置非空、页面可点、真发起
却封错库。

现在:agent 新增 health 动作(不占封存操作锁、不跑子进程,只回答身份/降级/忙),
worker 每 30 秒探测并写入迁移 074 的 library_executor_status;overview、快照列表与
POST /library/captures 共用同一份判定(过期、不可达、实例不符、降级、正忙都拒绝并
给原因)。封存结果由 agent 附上自报身份,worker 在停写入者之前再探测一次、执行后
用结果复核,不一致就不登记测试封存——心跳只能证明"刚才"。

期望身份用 LIBRARY_SOURCE_*(封存源),不是 LIBRARY_TARGET_*(同步目标),否则
正确的执行器会被判成不匹配。另加 GET /library/restore-availability:恢复能否执行由
双闸、该快照的恢复授权与实例维护状态决定,页面不再写死 disabled。

真跑(候选隔离栈,两处 socket 隔离不变):overview 返回 executor
{kgmaint-weknora/kgmaint-app};停 agent 后转不可达并 503 带原因;重启后恢复。
2026-10-29 13:01:37 +08:00

134 lines
4.7 KiB
Go

package libraryexecutor
import (
"context"
"encoding/json"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
func shortSocketDir(t *testing.T) string {
t.Helper()
// unix socket 路径上限 ~104 字节,TMPDIR(本机是 .hermes/cache/scratch)太长,
// 测试里用一个短目录并立刻清理。
dir, err := os.MkdirTemp("/tmp", "kgex-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
// 可用性必须来自运行时观测,而不是"配置里写了 socket 路径"。
func TestProbeReadsIdentityFromTheAgentOverTheSocket(t *testing.T) {
root := shortSocketDir(t)
socket := filepath.Join(root, "agent.sock")
listener, err := net.Listen("unix", socket)
if err != nil {
t.Fatal(err)
}
defer listener.Close()
go func() {
for {
conn, acceptErr := listener.Accept()
if acceptErr != nil {
return
}
var request map[string]any
_ = json.NewDecoder(conn).Decode(&request)
payload, _ := json.Marshal(map[string]any{
"project": "kgmaint-weknora", "archive_root": "/tmp/kg-e2e/archive",
"journal_root": "/tmp/kg-e2e/journal", "source_app": "kgmaint-app",
"source_pg": "kgmaint-postgres", "source_es": "WKNFull-elasticsearch",
"writers": "app", "busy": true, "degraded": "",
})
_ = json.NewEncoder(conn).Encode(map[string]any{"ok": true, "action": "health", "payload": json.RawMessage(payload)})
conn.Close()
}
}()
observation := Probe(context.Background(), "kgmaint-weknora", socket, 2*time.Second, nil)
if !observation.Reachable {
t.Fatalf("observation = %+v", observation)
}
if observation.Identity.Project != "kgmaint-weknora" || observation.Identity.SourceApp != "kgmaint-app" || !observation.Busy {
t.Fatalf("identity = %+v", observation.Identity)
}
if observation.Instance != "kgmaint-weknora" || observation.ObservedAt.IsZero() {
t.Fatalf("observation = %+v", observation)
}
}
func TestProbeReportsUnreachableAgentWithoutFakingSuccess(t *testing.T) {
observation := Probe(context.Background(), "kgmaint-weknora", filepath.Join(t.TempDir(), "absent.sock"), time.Second, nil)
if observation.Reachable || observation.Error == "" {
t.Fatalf("observation = %+v", observation)
}
available, reason := Evaluate(observation, Expectation{Instance: "kgmaint-weknora", Project: "kgmaint-weknora"},
time.Minute, time.Now().UTC())
if available || !strings.Contains(reason, "不可达") {
t.Fatalf("available=%v reason=%q", available, reason)
}
}
func TestEvaluateRejectsEveryIdentityAndFreshnessProblem(t *testing.T) {
now := time.Date(2026, 10, 29, 8, 0, 0, 0, time.UTC)
healthy := Observation{
Instance: "kgmaint-weknora", Reachable: true, ObservedAt: now,
Identity: Identity{Project: "kgmaint-weknora", SourceApp: "kgmaint-app"},
}
want := Expectation{Instance: "kgmaint-weknora", Project: "kgmaint-weknora"}
if available, reason := Evaluate(healthy, want, 2*time.Minute, now); !available || reason != "" {
t.Fatalf("healthy executor rejected: %v %q", available, reason)
}
tests := []struct {
name string
mut func(*Observation)
want func(Expectation) Expectation
text string
}{
{name: "missing instance config", want: func(Expectation) Expectation { return expectEmptyInstance },
text: "LIBRARY_SOURCE_INSTANCE"},
{name: "row from another instance", mut: func(o *Observation) { o.Instance = "someone-else" },
text: "另一个实例"},
{name: "stale", mut: func(o *Observation) { o.ObservedAt = now.Add(-10 * time.Minute) },
text: "已过期"},
{name: "unreachable", mut: func(o *Observation) { o.Reachable = false; o.Error = "socket missing" },
text: "不可达:socket missing"},
{name: "project not configured", want: func(Expectation) Expectation { return Expectation{Instance: "kgmaint-weknora"} },
text: "LIBRARY_SOURCE_PROJECT"},
{name: "wrong source project", mut: func(o *Observation) { o.Identity.Project = "weknora071full" },
text: "指向的实例不匹配"},
{name: "degraded", mut: func(o *Observation) { o.Degraded = "写入者恢复失败" },
text: "降级状态"},
{name: "busy", mut: func(o *Observation) { o.Busy = true },
text: "正在执行另一个封存"},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
observation := healthy
if test.mut != nil {
test.mut(&observation)
}
expectation := want
if test.want != nil {
expectation = test.want(want)
}
available, reason := Evaluate(observation, expectation, 2*time.Minute, now)
if available {
t.Fatalf("accepted: %+v", observation)
}
if !strings.Contains(reason, test.text) {
t.Fatalf("reason = %q, want it to mention %q", reason, test.text)
}
})
}
}
var expectEmptyInstance = Expectation{Project: "kgmaint-weknora"}