test(library): 端到端编排在本机真实 PG/ES 上跑通

为什么：只验证单阶段写入还不够，完整链路（授权→预备份→四类写入→校验）也要在真实服务上跑。本机 WeKnora 测试栈一次性库与随机索引已通过，CI 无这些服务故在 test-scope 声明为可选。
2026-10-28 15:58:06 +08:00
parent 281699a431
commit 8da4666e76
4 changed files with 120 additions and 5 deletions
@@ -29,7 +29,9 @@

 - `TestLibraryProductionTargetIsolated` 在本机 WeKnora 测试栈真实 PG（一次性 `kg_library_test_*` 库）
  与真实 ES（随机索引）跑通：首次导入 → 校验 → 撤销传播（PG 行与 ES 文档都消失、跨租户不受影响）
-  → 独立预备份（3111e823…，626 字节）→ 注入故障（删行）→ 校验失败 → 恢复后校验通过。
+  → 独立预备份（626 字节）→ 注入故障（删行）→ 校验失败 → 恢复后校验通过。
+- `TestLibraryReleaseRunIsolated`：同一环境上跑完整编排（授权 → 预备份 → 数据库/资源/索引/Wiki →
+  校验），返回 succeeded，预备份快照 8e4ed5ba…（1074 字节）。

 ## 开发快照（生产发布授权 + 同步/恢复编排 + 存储管理真实化）

@@ -29,7 +29,9 @@ build-web / tests-go / tests-web / contracts / version / security-secrets / secu

 | 0.1.123 | 67a48faf23996b7cf55d2ed589e4235f9b4d4754 | 去掉 qawemail 单测的外部 DNS 依赖（run164 因此红） | 165 | **success** | 否 |

-0.1.124 起只是台账/进度补记（文档），候选以最后一次全绿 run 为准（见交付报告）。
+| 0.1.125 | 281699a43104af6ce5344b17375ac7102f3e563d | 生产整库真实读写/校验/恢复 + 本机真实 PG/ES 集成验证 | 167 | **success** | 否（未部署） |
+
+0.1.124 起只是台账/进度补记（文档）。当前候选 = 0.1.125（281699a），只推了 main，未切运行版。
 历史事实保留：0.1.120 run161、0.1.121 run162、0.1.122 run163 均 success，0.1.123 run165 success；
 run164 因 qawemail 依赖外网 DNS 失败，已修，不涂改。

@@ -39,8 +39,7 @@ func TestLibraryProductionTargetIsolated(t *testing.T) {
 		t.Fatalf("open isolated PG: %v", err)
 	}
 	defer pool.Close()
-	if _, err := pool.Exec(ctx, `DROP TABLE IF EXISTS knowledges,chunks,knowledge_tags,knowledge_tag_relations,
- resources,resource_bindings,wiki_pages,wiki_folders,wiki_page_issues,wiki_log_entries,knowledge_bases,schema_migrations CASCADE`); err != nil {
+	if _, err := pool.Exec(ctx, `DROP SCHEMA public CASCADE; CREATE SCHEMA public`); err != nil {
 		t.Fatalf("reset isolated schema: %v", err)
 	}
 	if _, err := pool.Exec(ctx, libraryTestSchema); err != nil {
@@ -198,10 +197,117 @@ func TestLibraryProductionTargetIsolated(t *testing.T) {
 }

 // fixtureObjectStore 只为让目标构造通过：本场景没有资源部件，不会被调用。
+// 端到端：授权 → 预备份 → 四类写入 → 校验，全走 libraryrelease.Run 的真实适配器。
+func TestLibraryReleaseRunIsolated(t *testing.T) {
+	dsn, base := os.Getenv("KG_LIBRARY_TEST_DATABASE_URL"), os.Getenv("KG_LIBRARY_TEST_ELASTIC_URL")
+	if dsn == "" || base == "" || os.Getenv("KG_LIBRARY_TEST_ISOLATED") != "1" {
+		t.Skip("requires isolated PG and ES opt-in")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	defer cancel()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("open isolated PG: %v", err)
+	}
+	defer pool.Close()
+	for _, statement := range []string{
+		`DROP SCHEMA public CASCADE; CREATE SCHEMA public`,
+		libraryTestSchema,
+		`ALTER TABLE wiki_pages ADD COLUMN status text DEFAULT 'published'`,
+		`ALTER TABLE knowledge_tag_relations ADD COLUMN tenant_id bigint`,
+		`ALTER TABLE resources ADD COLUMN provider text DEFAULT 's3', ADD COLUMN size bigint DEFAULT 8`,
+	} {
+		if _, err := pool.Exec(ctx, statement); err != nil {
+			t.Fatalf("fixture schema: %v", err)
+		}
+	}
+	token := make([]byte, 8)
+	if _, err := rand.Read(token); err != nil {
+		t.Fatal(err)
+	}
+	index := "kg_library_test_" + hex.EncodeToString(token)
+	client := &http.Client{Timeout: 20 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error {
+		return http.ErrUseLastResponse
+	}}
+	call := func(method, path, body string, want int) {
+		t.Helper()
+		request, err := http.NewRequestWithContext(ctx, method, base+path, strings.NewReader(body))
+		if err != nil {
+			t.Fatal(err)
+		}
+		request.Header.Set("Content-Type", "application/json")
+		response, err := client.Do(request)
+		if err != nil {
+			t.Fatalf("fixture HTTP: %v", err)
+		}
+		defer response.Body.Close()
+		if response.StatusCode != want {
+			t.Fatalf("fixture %s %s => %d, want %d", method, path, response.StatusCode, want)
+		}
+	}
+	call("PUT", "/"+index, `{"mappings":{"properties":{"tenant_id":{"type":"long"},"knowledge_base_id":{"type":"keyword"},"knowledge_id":{"type":"keyword"},"chunk_id":{"type":"keyword"},"content":{"type":"text"},"embedding":{"type":"dense_vector","dims":2}}}}`, 200)
+	defer func() {
+		call("DELETE", "/"+index, "", 200)
+		call("HEAD", "/"+index, "", 404)
+	}()
+	archiveRoot := t.TempDir()
+	if err := os.Chmod(archiveRoot, 0700); err != nil {
+		t.Fatal(err)
+	}
+	archive, err := librarysnapshot.Open(archiveRoot, 64*1024*1024)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer archive.Close()
+	snapshotID := sealTargetFixture(t, archive, "internal", []string{"doc-safe"}, []string{"chunk-safe"}, "wiki-safe")
+	// 生产"当前状态"也要自洽：库里有分块，索引里就必须有对应文档，
+	// 否则预备份的覆盖校验会（正确地）拒绝。
+	call("PUT", "/"+index+"/_doc/chunk-safe?refresh=true",
+		`{"tenant_id":10000,"knowledge_base_id":"kb-safe","knowledge_id":"doc-safe","chunk_id":"chunk-safe","source_id":"chunk-safe","content":"chunk","embedding":[0.1,0.2]}`, 201)
+	// 库里已有夹具种子行（doc-safe/chunk-safe），只需保证索引侧自洽。
+
+	target, err := NewProductionLibraryTarget(ctx, ProductionLibraryConfig{
+		Postgres: dsn, SearchURL: base, SearchIndex: index, Objects: fixtureObjectStore{},
+		ObjectsReader: fixtureObjectStore{}, ArchiveRoot: archiveRoot, MaxDocuments: 100, HTTPClient: client,
+	}, ProductionScope{TenantID: 10000, KnowledgeBaseID: "kb-safe"})
+	if err != nil {
+		t.Fatalf("target: %v", err)
+	}
+	defer target.Close()
+	mapping, err := target.Search().MappingSHA256(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	backups, err := NewProductionBackupStore(ProductionBackupConfig{
+		Pool: target.Pool(), Search: target.Search(), Objects: fixtureObjectStore{}, ArchiveRoot: archiveRoot,
+		Scope: "internal", Identity: target.Identity(), MaxDocuments: 100,
+		Compatibility: librarysnapshot.Compatibility{WeKnoraVersion: "v0.7.1", SchemaVersion: 74,
+			RetrieverEngine: "elasticsearch_v8", EmbeddingModel: "text-embedding-v4",
+			EmbeddingDimension: 1024, IndexMappingSHA256: mapping},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now().UTC()
+	request := libraryrelease.Request{Target: "internal", SnapshotID: snapshotID,
+		ParamsDigest: "digest-e2e", Now: now,
+		Grant: libraryrelease.Grant{Target: "internal", SnapshotID: snapshotID, ParamsDigest: "digest-e2e",
+			ExpiresAt: now.Add(10 * time.Minute), AuthorizedBy: "acceptance"}}
+	outcome := libraryrelease.Run(ctx, request, target, backups, nil)
+	if outcome.State != "succeeded" {
+		t.Fatalf("run outcome = %+v", outcome)
+	}
+	if outcome.Backup.SnapshotID == "" {
+		t.Fatalf("run must create a backup first: %+v", outcome)
+	}
+	t.Logf("e2e run succeeded snapshot=%s backup=%s bytes=%d", snapshotID, outcome.Backup.SnapshotID, outcome.Backup.Bytes)
+}
+
 type fixtureObjectStore struct{}

+// Open 返回与夹具资源行 size=8 一致的字节，供预备份读取对象。
 func (fixtureObjectStore) Open(context.Context, string, string) (io.ReadCloser, error) {
-	return nil, errors.New("fixture object store is read-only here")
+	return io.NopCloser(strings.NewReader("12345678")), nil
 }

 func (fixtureObjectStore) Put(context.Context, string, string, io.Reader, int64) error {
@@ -168,6 +168,11 @@
      "test": "TestLocalMirrorPeriodicSyncIsDurableAndDueOnly",
      "reason": "仅显式一次性隔离PG验证本地副本/持久同步或专门只读源夹具；默认CI不连接共享与生产库，另行实际执行隔离验收"
    },
+    {
+      "package": "knowledge-governance/internal/releaseadapter/v071",
+      "test": "TestLibraryReleaseRunIsolated",
+      "reason": "完整编排（授权→预备份→四类写入→校验）跑真实一次性隔离 PG 与随机 ES 索引；默认 CI 无这些服务，本机 WeKnora 测试栈已实跑通过"
+    },
    {
      "package": "knowledge-governance/internal/releaseadapter/v071",
      "test": "TestLibraryProductionTargetIsolated",