Files
knowledge-governance/internal/httpapi/server_test.go
姚章浩 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

2054 lines
109 KiB
Go

package httpapi
import (
"bytes"
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"net/url"
"slices"
"strings"
"testing"
"time"
"knowledge-governance/internal/governance"
"knowledge-governance/internal/jobs"
"knowledge-governance/internal/libraryexecutor"
postgresstore "knowledge-governance/internal/platform/postgres"
)
func TestHandlerHealth(t *testing.T) {
handler, err := NewHandler(Config{Store: &fakeJobStore{}})
if err != nil {
t.Fatalf("NewHandler() error = %v", err)
}
request := httptest.NewRequest(http.MethodGet, "/health", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", response.Code)
}
if got := response.Header().Get("Content-Type"); got != "application/json" {
t.Fatalf("content type = %q", got)
}
if got := response.Body.String(); got != "{\"status\":\"ok\"}\n" {
t.Fatalf("body = %q", got)
}
}
func TestHandlerCreatesJob(t *testing.T) {
now := time.Date(2026, time.September, 3, 13, 0, 0, 0, time.UTC)
store := &fakeJobStore{enqueueResult: &jobs.Job{ID: "job-fixed", Type: "demo", State: jobs.StatePending}}
handler, err := NewHandler(Config{
Store: store,
Now: func() time.Time { return now },
NewID: func() (string, error) { return "job-fixed", nil },
})
if err != nil {
t.Fatalf("NewHandler() error = %v", err)
}
body := bytes.NewBufferString(`{"type":"demo","payload":{"message":"hello"},"idempotency_key":"demo:fixed"}`)
request := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", body)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusCreated {
t.Fatalf("status = %d, want 201; body=%s", response.Code, response.Body.String())
}
if store.enqueued.ID != "job-fixed" || store.enqueued.Type != "demo" || store.enqueued.MaxAttempts != 5 {
t.Fatalf("enqueued = %#v", store.enqueued)
}
if store.enqueued.IdempotencyKey != "demo:fixed" || !store.enqueued.AvailableAt.Equal(now) {
t.Fatalf("enqueued idempotency/time = %#v", store.enqueued)
}
}
func TestHandlerTestPublicationQueueTargetScopeFilter(t *testing.T) {
store := &fakeJobStore{publicationResult: governance.TestPublicationPage{Total: 0}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/test-publications?target_scope=internal", nil)
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, request)
if recorder.Code != http.StatusOK {
t.Fatalf("filter status = %d body %s", recorder.Code, recorder.Body.String())
}
if store.publicationFilter.TargetScope != "internal" {
t.Fatalf("filter = %+v", store.publicationFilter)
}
invalid := httptest.NewRequest(http.MethodGet, "/api/v1/test-publications?target_scope=prod", nil)
bad := httptest.NewRecorder()
handler.ServeHTTP(bad, invalid)
if bad.Code != http.StatusBadRequest {
t.Fatalf("invalid scope status = %d", bad.Code)
}
}
func TestHandlerListsJobFailuresAndRunsServerSideBatchRetry(t *testing.T) {
store := &fakeJobStore{
jobFailurePage: governance.JobFailurePage{
Items: []governance.JobFailure{{ID: "jf-1", JobID: "job-9", JobType: "test.upload", PublicationID: "tp-1", ErrorCode: "remote_permanent", Attempt: 5, MaxAttempts: 5, State: "open"}},
Total: 1, Limit: 50,
},
retryBatchResult: governance.RetryPublicationsBatchResult{
BatchKey: "op-batch-1", Total: 2, Started: 1, AlreadyRunning: 1,
Outcomes: []governance.RetryOutcome{{PublicationID: "tp-1", Result: "started"}, {PublicationID: "tp-2", Result: "already_running"}},
},
}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return time.Date(2026, time.September, 3, 9, 0, 0, 0, time.UTC) }})
if err != nil {
t.Fatal(err)
}
recorder := httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/job-failures?state=open&limit=50", nil))
if recorder.Code != http.StatusOK {
t.Fatalf("list failures status = %d body %s", recorder.Code, recorder.Body.String())
}
if store.jobFailureFilter.State != "open" || store.jobFailureFilter.Limit != 50 {
t.Fatalf("filter = %+v", store.jobFailureFilter)
}
if !strings.Contains(recorder.Body.String(), "\"data\"") {
t.Fatalf("job failures must be an envelope: %s", recorder.Body.String())
}
recorder = httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/retry-batch",
strings.NewReader(`{"publication_ids":["tp-1","tp-2"],"reason":"解析服务恢复"}`))
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Idempotency-Key", "op-batch-1")
handler.ServeHTTP(recorder, request)
if recorder.Code != http.StatusOK {
t.Fatalf("retry-batch status = %d body %s", recorder.Code, recorder.Body.String())
}
var envelope struct {
Data governance.RetryPublicationsBatchResult `json:"data"`
Operation string `json:"operation"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatal(err)
}
if envelope.Operation != "created" || envelope.Data.Started != 1 || envelope.Data.AlreadyRunning != 1 {
t.Fatalf("envelope = %+v", envelope)
}
if store.retryBatchOperation != "op-batch-1" || store.retryBatchRequest == nil || len(store.retryBatchRequest.PublicationIDs) != 2 {
t.Fatalf("batch request = %+v op=%q", store.retryBatchRequest, store.retryBatchOperation)
}
if store.retryBatchRequest.Now.IsZero() {
t.Fatal("retry batch request now not set")
}
// Missing Idempotency-Key is rejected before touching the store.
recorder = httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/retry-batch",
strings.NewReader(`{"publication_ids":["tp-1"],"reason":"r"}`)))
if recorder.Code != http.StatusBadRequest {
t.Fatalf("missing key status = %d", recorder.Code)
}
// Duplicate ids rejected.
recorder = httptest.NewRecorder()
dup := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/retry-batch",
strings.NewReader(`{"publication_ids":["tp-1","tp-1"],"reason":"r"}`))
dup.Header.Set("Idempotency-Key", "op-dup")
handler.ServeHTTP(recorder, dup)
if recorder.Code != http.StatusBadRequest {
t.Fatalf("duplicate ids status = %d", recorder.Code)
}
}
func TestHandlerRunsServerSideBatchCancel(t *testing.T) {
store := &fakeJobStore{
cancelBatchResult: governance.RetryPublicationsBatchResult{
BatchKey: "op-cancel-1", Total: 2, Started: 1, NotRetryable: 1,
Outcomes: []governance.RetryOutcome{{PublicationID: "tp-1", Result: "cancelled"}, {PublicationID: "tp-x", Result: "not_retryable", Detail: "publication not found"}},
},
}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return time.Date(2026, time.September, 3, 9, 0, 0, 0, time.UTC) }})
if err != nil {
t.Fatal(err)
}
recorder := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/cancel-batch",
strings.NewReader(`{"publication_ids":["tp-1","tp-x"],"reason":"源头为空文档,不再重试"}`))
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Idempotency-Key", "op-cancel-1")
handler.ServeHTTP(recorder, request)
if recorder.Code != http.StatusOK {
t.Fatalf("cancel-batch status = %d body %s", recorder.Code, recorder.Body.String())
}
if store.cancelBatchOperation != "op-cancel-1" || store.cancelBatchRequest == nil ||
len(store.cancelBatchRequest.PublicationIDs) != 2 ||
store.cancelBatchRequest.Reason != "源头为空文档,不再重试" || store.cancelBatchRequest.Now.IsZero() {
t.Fatalf("cancel request = %+v op=%q", store.cancelBatchRequest, store.cancelBatchOperation)
}
// Missing Idempotency-Key and missing reason are rejected before touching the store.
recorder = httptest.NewRecorder()
handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/cancel-batch",
strings.NewReader(`{"publication_ids":["tp-1"],"reason":"r"}`)))
if recorder.Code != http.StatusBadRequest {
t.Fatalf("missing key status = %d", recorder.Code)
}
recorder = httptest.NewRecorder()
noReason := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/cancel-batch",
strings.NewReader(`{"publication_ids":["tp-1"],"reason":""}`))
noReason.Header.Set("Idempotency-Key", "op-cancel-2")
handler.ServeHTTP(recorder, noReason)
if recorder.Code != http.StatusBadRequest {
t.Fatalf("missing reason status = %d", recorder.Code)
}
}
func TestHandlerListsJobsWithJSONPayload(t *testing.T) {
store := &fakeJobStore{listResult: []jobs.Job{{
ID: "job-6",
Type: "demo",
State: jobs.StateCompleted,
Payload: []byte(`{"message":"done"}`),
}}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatalf("NewHandler() error = %v", err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/jobs?limit=10", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d, want 200; body=%s", response.Code, response.Body.String())
}
want := `"id":"job-6"`
if !bytes.Contains(response.Body.Bytes(), []byte(want)) {
t.Fatalf("body missing %s: %s", want, response.Body.String())
}
if !bytes.Contains(response.Body.Bytes(), []byte(`"payload":{"message":"done"}`)) {
t.Fatalf("payload is not embedded JSON: %s", response.Body.String())
}
if store.listLimit != 10 {
t.Fatalf("list limit = %d, want 10", store.listLimit)
}
}
func TestHandlerListsSourceAssetsWithFilterAndSummary(t *testing.T) {
store := &fakeJobStore{assetResult: governance.SourceAssetPage{
Items: []governance.SourceAsset{{ID: "si-safe", SourceType: governance.SourceTypePP, SizeBytes: 52428801, ContentState: governance.ContentOverLimit, LegacyStatus: "over_limit"}},
Summary: governance.SourceAssetSummary{Total: 613, Completed: 530, ContentAlias: 29, OverLimit: 54},
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/source-items?state=over_limit&limit=25&offset=50", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d; body=%s", response.Code, response.Body.String())
}
if !bytes.Contains(response.Body.Bytes(), []byte(`"total":613`)) || !bytes.Contains(response.Body.Bytes(), []byte(`"content_state":"over_limit"`)) {
t.Fatalf("body = %s", response.Body.String())
}
if store.assetFilter != (governance.SourceAssetFilter{State: "over_limit", Limit: 25, Offset: 50}) {
t.Fatalf("filter = %#v", store.assetFilter)
}
}
func TestHandlerSetsListsAndClearsAuditedScopeRules(t *testing.T) {
now := time.Date(2026, 9, 8, 2, 0, 0, 0, time.UTC)
store := &fakeJobStore{scopeRuleResult: governance.ScopeRule{TargetType: governance.ScopeTargetDirectory, ConnectionID: "conn-safe", DirectoryPath: "产品", Visibility: governance.VisibilityInternal, UpdatedBy: "engineer-1", Reason: "内部资料", UpdatedAt: now}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }})
if err != nil {
t.Fatal(err)
}
set := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPut, "/api/v1/scope-rules", bytes.NewBufferString(`{"target_type":"directory","connection_id":"conn-safe","directory_path":"产品","visibility":"internal","reason":"内部资料"}`))
request.Header.Set("Idempotency-Key", "op-scope-set")
handler.ServeHTTP(set, request)
if set.Code != http.StatusOK || store.scopeRuleRequest.ConnectionID != "conn-safe" || store.scopeRuleOperationID != "op-scope-set" || store.scopeRuleRequest.Clear || store.scopeRuleRequest.Now != now || !strings.Contains(set.Body.String(), `"operation":"created"`) {
t.Fatalf("set=%d request=%#v body=%s", set.Code, store.scopeRuleRequest, set.Body.String())
}
store.scopeRules = []governance.ScopeRule{store.scopeRuleResult}
list := httptest.NewRecorder()
handler.ServeHTTP(list, httptest.NewRequest(http.MethodGet, "/api/v1/scope-rules?connection_id=conn-safe", nil))
if list.Code != http.StatusOK || store.scopeRuleConnectionID != "conn-safe" || !strings.Contains(list.Body.String(), `"directory_path":"产品"`) {
t.Fatalf("list=%d id=%q body=%s", list.Code, store.scopeRuleConnectionID, list.Body.String())
}
clear := httptest.NewRecorder()
request = httptest.NewRequest(http.MethodDelete, "/api/v1/scope-rules", bytes.NewBufferString(`{"target_type":"directory","connection_id":"conn-safe","directory_path":"产品","reason":"恢复来源设置"}`))
request.Header.Set("Idempotency-Key", "op-scope-clear")
handler.ServeHTTP(clear, request)
if clear.Code != http.StatusOK || !store.scopeRuleRequest.Clear || store.scopeRuleRequest.Visibility != "" || !strings.Contains(clear.Body.String(), `"operation":"created"`) {
t.Fatalf("clear=%d request=%#v body=%s", clear.Code, store.scopeRuleRequest, clear.Body.String())
}
}
func TestHandlerRejectsUnsafeScopeRuleBeforeStore(t *testing.T) {
store := &fakeJobStore{}
handler, _ := NewHandler(Config{Store: store, Now: time.Now})
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPut, "/api/v1/scope-rules", bytes.NewBufferString(`{"target_type":"directory","connection_id":"conn-safe","directory_path":"../private","visibility":"external","reason":"bad"}`))
handler.ServeHTTP(response, request)
if response.Code != http.StatusBadRequest || store.scopeRuleRequest.TargetType != "" {
t.Fatalf("status=%d request=%#v", response.Code, store.scopeRuleRequest)
}
}
func TestHandlerNormalizesAndPersistsFilenameFilters(t *testing.T) {
now := time.Date(2026, 9, 15, 2, 0, 0, 0, time.UTC)
store := &fakeJobStore{filenameFilterResult: governance.FilenameFilterRule{
ConnectionID: "conn-safe", DirectoryPath: "固件", ExcludedExtensions: []string{"mp4", "rar", "zip"}, UpdatedBy: "engineer-1", UpdatedAt: now,
}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPut, "/api/v1/filename-filters", bytes.NewBufferString(`{"connection_id":"conn-safe","directory_path":"固件","excluded_extensions":[".ZIP","rar","Mp4","zip"]}`))
request.Header.Set("Idempotency-Key", "op-filter-set")
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK || store.filenameFilterOperationID != "op-filter-set" ||
!slices.Equal(store.filenameFilterRequest.ExcludedExtensions, []string{"mp4", "rar", "zip"}) ||
!strings.Contains(response.Body.String(), `"excluded_extensions":["mp4","rar","zip"]`) {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.filenameFilterRequest, response.Body.String())
}
allow := httptest.NewRecorder()
request = httptest.NewRequest(http.MethodPut, "/api/v1/filename-filters", bytes.NewBufferString(`{"connection_id":"conn-safe","directory_path":"","mode":"allow","extensions":["PDF","docx"]}`))
request.Header.Set("Idempotency-Key", "op-filter-allow")
handler.ServeHTTP(allow, request)
if allow.Code != http.StatusOK || store.filenameFilterRequest.Mode != "allow" ||
!slices.Equal(store.filenameFilterRequest.ExcludedExtensions, []string{"docx", "pdf"}) {
t.Fatalf("allow status=%d request=%#v", allow.Code, store.filenameFilterRequest)
}
}
func TestHandlerListsSnapshotHistoryAndReadsHistoricalNodes(t *testing.T) {
now := time.Date(2026, 9, 11, 14, 0, 0, 0, time.UTC)
hash := strings.Repeat("c", 64)
store := &fakeJobStore{
snapshotPageResult: governance.CatalogSnapshotPage{Items: []governance.CatalogSnapshot{{SnapshotID: "ss-a", ConnectionID: "conn-cat", RootNodeID: "root", SourceName: "PP", NodeCount: 2, FolderCount: 1, FileCount: 1, ObservedAt: now, CreatedAt: now, IsHead: true}}, Total: 1, Limit: 20},
snapshotNodePageResult: governance.CatalogNodePage{Current: governance.CatalogNode{NodeID: "root", NodeType: "folder", Name: "PP", Visibility: "unknown"}, Items: []governance.CatalogNode{{NodeID: "file-1", ParentNodeID: "root", NodeType: "file", Name: "手册.pdf", Visibility: "unknown", ContentSHA256: &hash}}, Total: 1, Limit: 50},
}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
list := httptest.NewRecorder()
handler.ServeHTTP(list, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/snapshots?connection_id=conn-cat", nil))
if list.Code != http.StatusOK || store.snapshotConnectionID != "conn-cat" || !strings.Contains(list.Body.String(), `"is_head":true`) {
t.Fatalf("list=%d id=%q body=%s", list.Code, store.snapshotConnectionID, list.Body.String())
}
if strings.Contains(list.Body.String(), "snapshot_sha256") {
t.Fatalf("snapshot history leaked internal hash: %s", list.Body.String())
}
nodes := httptest.NewRecorder()
handler.ServeHTTP(nodes, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/snapshots/ss-old/nodes?connection_id=conn-cat&parent_id=root", nil))
if nodes.Code != http.StatusOK || store.snapshotNodeFilter.SnapshotID != "ss-old" || store.snapshotNodeFilter.ParentNodeID != "root" || !strings.Contains(nodes.Body.String(), `"content_sha256":"`) {
t.Fatalf("nodes=%d filter=%#v body=%s", nodes.Code, store.snapshotNodeFilter, nodes.Body.String())
}
bad := httptest.NewRecorder()
handler.ServeHTTP(bad, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/snapshots/..%2Fprivate/nodes?connection_id=conn-cat&parent_id=root", nil))
if bad.Code != http.StatusBadRequest {
t.Fatalf("bad snapshot id status=%d", bad.Code)
}
}
func TestHandlerBrowsesCatalogSourcesAndOnlyDirectChildren(t *testing.T) {
now := time.Date(2026, 9, 8, 10, 0, 0, 0, time.UTC)
store := &fakeJobStore{
catalogSources: []governance.CatalogSource{{ConnectionID: "conn-cat", SourceType: governance.SourceTypeDingTalk, Name: "公司级技术支持知识库", RootNodeID: "root-cat", FolderCount: 8, FileCount: 3468, ObservedAt: now, SnapshotOnly: true}},
catalogPage: governance.CatalogNodePage{Current: governance.CatalogNode{NodeID: "root-cat", NodeType: "folder", Name: "公司级技术支持知识库"}, Breadcrumbs: []governance.CatalogBreadcrumb{{NodeID: "root-cat", Name: "公司级技术支持知识库"}}, Items: []governance.CatalogNode{{NodeID: "folder-cat", ParentNodeID: "root-cat", NodeType: "folder", Name: "01 产品总览与定位", SubtreeFileCount: 10, Visibility: "internal", ImportedCount: 8, ParsedCount: 7}}, Total: 1, Limit: 50},
}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
sources := httptest.NewRecorder()
handler.ServeHTTP(sources, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/sources", nil))
if sources.Code != http.StatusOK || !strings.Contains(sources.Body.String(), `"name":"公司级技术支持知识库"`) {
t.Fatalf("sources=%d body=%s", sources.Code, sources.Body.String())
}
nodes := httptest.NewRecorder()
handler.ServeHTTP(nodes, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/nodes?connection_id=conn-cat&parent_id=root-cat&limit=50&offset=0", nil))
if nodes.Code != http.StatusOK || store.catalogFilter != (governance.CatalogNodeFilter{ConnectionID: "conn-cat", ParentNodeID: "root-cat", Limit: 50, Offset: 0}) || !strings.Contains(nodes.Body.String(), `"name":"01 产品总览与定位"`) {
t.Fatalf("nodes=%d filter=%#v body=%s", nodes.Code, store.catalogFilter, nodes.Body.String())
}
for _, forbidden := range []string{"relative_path", "source_locator_hash", "snapshot_sha256", "stable_external_id"} {
if strings.Contains(nodes.Body.String()+sources.Body.String(), forbidden) {
t.Fatalf("catalog API leaked %s", forbidden)
}
}
}
func TestHandlerPreviewsLatestPersistedCatalogChangeSet(t *testing.T) {
now := time.Date(2026, 9, 10, 12, 0, 0, 0, time.UTC)
store := &fakeJobStore{catalogChangeSetResult: governance.CatalogChangeSetDetail{
CatalogChangeSet: governance.CatalogChangeSet{ID: "cs-safe", ConnectionID: "conn-cat", SourceName: "国内工业", State: "planned", AddCount: 2, UpdateCount: 1, MoveScopeCount: 1, DeleteCount: 1, NoChangeCount: 10, CreatedAt: now},
Items: []governance.CatalogChangeItem{{Action: "delete", Title: "旧手册.pdf", OldPath: "旧目录/旧手册.pdf"}}, Total: 1, Limit: 20,
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/change-sets/latest?connection_id=conn-cat&action=delete&limit=20&offset=0", nil))
if response.Code != http.StatusOK || store.catalogChangeSetFilter.ConnectionID != "conn-cat" || store.catalogChangeSetFilter.Action != "delete" || !strings.Contains(response.Body.String(), `"title":"旧手册.pdf"`) || !strings.Contains(response.Body.String(), `"add_count":2`) {
t.Fatalf("status=%d filter=%#v body=%s", response.Code, store.catalogChangeSetFilter, response.Body.String())
}
for _, forbidden := range []string{"source_locator_hash", "snapshot_sha256", "secret_ref"} {
if strings.Contains(response.Body.String(), forbidden) {
t.Fatalf("change-set response leaked %s", forbidden)
}
}
}
func TestHandlerRejectsInvalidCatalogParentBeforeStore(t *testing.T) {
store := &fakeJobStore{}
handler, _ := NewHandler(Config{Store: store})
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/nodes?connection_id=conn-cat&parent_id=../private", nil))
if response.Code != http.StatusBadRequest || store.catalogFilter.ParentNodeID != "" {
t.Fatalf("status=%d filter=%#v", response.Code, store.catalogFilter)
}
}
func TestHandlerRoutesSelectedCatalogFilesToServerBoundScopeKB(t *testing.T) {
now := time.Date(2026, 9, 9, 1, 0, 0, 0, time.UTC)
store := &fakeJobStore{catalogRouteResult: governance.CatalogRouteResult{SelectedFiles: 2, PlannedPublications: 2}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, InternalRoutingKnowledgeBaseID: "kb-internal", ExternalRoutingKnowledgeBaseID: "kb-external"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/route", bytes.NewBufferString(`{"connection_id":"conn-cat","node_ids":["file-a","file-b"],"visibility":"internal","reason":"当前页批量标注"}`))
request.Header.Set("Idempotency-Key", "op-route-internal")
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted || store.catalogRouteRequest.TargetKnowledgeBaseID != "kb-internal" || store.catalogRouteRequest.Now != now || store.catalogRouteRequest.Visibility != governance.VisibilityInternal {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.catalogRouteRequest, response.Body.String())
}
if strings.Contains(response.Body.String(), "kb-internal") || !strings.Contains(response.Body.String(), `"planned_publications":2`) {
t.Fatalf("unsafe route response=%s", response.Body.String())
}
capabilities := httptest.NewRecorder()
handler.ServeHTTP(capabilities, httptest.NewRequest(http.MethodGet, "/api/v1/source-catalog/routing-capabilities", nil))
if capabilities.Code != http.StatusOK || !strings.Contains(capabilities.Body.String(), `"internal_enabled":true`) || strings.Contains(capabilities.Body.String(), "kb-") {
t.Fatalf("capabilities=%s", capabilities.Body.String())
}
}
func TestHandlerRoutesBothScopesInOneAtomicRequest(t *testing.T) {
now := time.Date(2026, 9, 9, 1, 30, 0, 0, time.UTC)
store := &fakeJobStore{catalogRouteResult: governance.CatalogRouteResult{SelectedFiles: 1, PlannedPublications: 1}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, InternalRoutingKnowledgeBaseID: "kb-internal", ExternalRoutingKnowledgeBaseID: "kb-external"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/route", bytes.NewBufferString(`{"connection_id":"conn-cat","node_ids":["file-a"],"policy":"both","reason":"同时供内外知识库使用"}`))
request.Header.Set("Idempotency-Key", "op-route-both")
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted || store.catalogRouteRequest.Visibility != governance.VisibilityBoth {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.catalogRouteRequest, response.Body.String())
}
if got := store.catalogRouteRequest.TargetKnowledgeBaseIDs; len(got) != 2 || got[0] != "kb-internal" || got[1] != "kb-external" {
t.Fatalf("target knowledge bases = %#v", got)
}
if strings.Contains(response.Body.String(), "kb-internal") || strings.Contains(response.Body.String(), "kb-external") {
t.Fatalf("route response leaked target identity: %s", response.Body.String())
}
}
func TestHandlerAcceptsBlockedPolicyWithoutAConfiguredTarget(t *testing.T) {
store := &fakeJobStore{catalogRouteResult: governance.CatalogRouteResult{SelectedFiles: 1, BlockedItems: 1}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/route", bytes.NewBufferString(`{"connection_id":"conn-cat","node_ids":["file-a"],"policy":"blocked","reason":"资料已禁止导入"}`))
request.Header.Set("Idempotency-Key", "op-route-blocked")
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted || store.catalogRouteRequest.Visibility != governance.VisibilityBlocked || len(store.catalogRouteRequest.TargetKnowledgeBaseIDs) != 0 {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.catalogRouteRequest, response.Body.String())
}
if !strings.Contains(response.Body.String(), `"blocked_items":1`) {
t.Fatalf("blocked route response=%s", response.Body.String())
}
}
func TestHandlerEnqueuesCatalogRefreshWithSafeSourceIdentity(t *testing.T) {
now := time.Date(2026, 9, 9, 2, 0, 0, 0, time.UTC)
store := &fakeJobStore{
catalogSources: []governance.CatalogSource{{ConnectionID: "conn-cat", SourceType: governance.SourceTypePP, Name: "PP资料下载中心", RootNodeID: "root-cat"}},
catalogRefreshResult: jobs.Job{ID: "job-refresh", Type: "source.catalog.refresh", State: jobs.StatePending, Payload: json.RawMessage(`{"connection_id":"conn-cat","source_type":"pp","source_name":"PP资料下载中心","stage":"queued","progress":0,"message":"等待后台刷新"}`)},
}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/sources/conn-cat/refresh", nil)
request.Header.Set("Idempotency-Key", "op-refresh-safe")
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted || store.catalogRefreshSource.ConnectionID != "conn-cat" || store.catalogRefreshOperationID != "op-refresh-safe" || !strings.Contains(response.Body.String(), `"operation":"created"`) {
t.Fatalf("status=%d source=%#v operation_id=%q body=%s", response.Code, store.catalogRefreshSource, store.catalogRefreshOperationID, response.Body.String())
}
for _, forbidden := range []string{"base_url", "secret", "capture", "token"} {
if strings.Contains(strings.ToLower(response.Body.String()), forbidden) {
t.Fatalf("refresh response leaked %s", forbidden)
}
}
}
func TestHandlerSourceSettingsRevisionEndpoints(t *testing.T) {
store := &fakeJobStore{}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return time.Date(2026, 9, 12, 9, 0, 0, 0, time.UTC) }})
if err != nil {
t.Fatalf("NewHandler() error = %v", err)
}
forbidden := httptest.NewRecorder()
handler.ServeHTTP(forbidden, httptest.NewRequest("POST", "/api/v1/source-connections/conn-1/settings/revisions", strings.NewReader(`{"source_type":"alist","name":"n","schedule":{}}`)))
if forbidden.Code != http.StatusForbidden {
t.Fatalf("intent header gate = %d, want 403", forbidden.Code)
}
missingKey := httptest.NewRecorder()
request := httptest.NewRequest("POST", "/api/v1/source-connections/conn-1/settings/revisions", strings.NewReader(`{"source_type":"alist","name":"国内工业AList","enabled":true,"schedule":{"refresh_interval_seconds":3600}}`))
request.Header.Set("X-Governance-Action", "source-settings")
handler.ServeHTTP(missingKey, request)
if missingKey.Code != http.StatusBadRequest {
t.Fatalf("missing key = %d, want 400", missingKey.Code)
}
created := httptest.NewRecorder()
request = httptest.NewRequest("POST", "/api/v1/source-connections/conn-1/settings/revisions", strings.NewReader(`{"source_type":"alist","name":"国内工业AList","enabled":true,"schedule":{"refresh_interval_seconds":3600}}`))
request.Header.Set("X-Governance-Action", "source-settings")
request.Header.Set("Idempotency-Key", "create-v1")
handler.ServeHTTP(created, request)
if created.Code != http.StatusCreated {
t.Fatalf("create = %d body=%s", created.Code, created.Body.String())
}
if store.createSettingsConnectionID != "conn-1" || store.createSettingsOperationID != "create-v1" || store.createSettingsDraft.Name != "国内工业AList" || store.createSettingsDraft.Operator != "system" {
t.Fatalf("create request = %s %+v op=%s", store.createSettingsConnectionID, store.createSettingsDraft, store.createSettingsOperationID)
}
if !strings.Contains(created.Body.String(), "scr-fake") || !strings.Contains(created.Body.String(), "created") {
t.Fatalf("create body=%s", created.Body.String())
}
store.createSettingsErr = postgresstore.ErrInvalidSourceSettings
rejected := httptest.NewRecorder()
request = httptest.NewRequest("POST", "/api/v1/source-connections/conn-1/settings/revisions", strings.NewReader(`{"source_type":"ftp","name":"x","schedule":{}}`))
request.Header.Set("X-Governance-Action", "source-settings")
request.Header.Set("Idempotency-Key", "create-v2")
handler.ServeHTTP(rejected, request)
if rejected.Code != http.StatusBadRequest {
t.Fatalf("validation = %d want 400", rejected.Code)
}
store.createSettingsErr = postgresstore.ErrOperationRequestConflict
conflicted := httptest.NewRecorder()
request = httptest.NewRequest("POST", "/api/v1/source-connections/conn-1/settings/revisions", strings.NewReader(`{"source_type":"alist","name":"x","schedule":{}}`))
request.Header.Set("X-Governance-Action", "source-settings")
request.Header.Set("Idempotency-Key", "create-v3")
handler.ServeHTTP(conflicted, request)
if conflicted.Code != http.StatusConflict {
t.Fatalf("conflict = %d want 409", conflicted.Code)
}
store.createSettingsErr = nil
activated := httptest.NewRecorder()
request = httptest.NewRequest("POST", "/api/v1/source-connections/conn-1/settings/revisions/scr-1/activate", strings.NewReader(`{}`))
request.Header.Set("X-Governance-Action", "source-settings")
request.Header.Set("Idempotency-Key", "activate-v1")
handler.ServeHTTP(activated, request)
if activated.Code != http.StatusOK || store.activateSettingsConnectionID != "conn-1" || store.activateSettingsRevisionID != "scr-1" || store.activateSettingsOperationID != "activate-v1" {
t.Fatalf("activate = %d conn=%s rev=%s op=%s", activated.Code, store.activateSettingsConnectionID, store.activateSettingsRevisionID, store.activateSettingsOperationID)
}
store.activateSettingsErr = postgresstore.ErrInvalidSourceSettings
badActivate := httptest.NewRecorder()
request = httptest.NewRequest("POST", "/api/v1/source-connections/conn-1/settings/revisions/scr-9/activate", strings.NewReader(`{}`))
request.Header.Set("X-Governance-Action", "source-settings")
request.Header.Set("Idempotency-Key", "activate-v2")
handler.ServeHTTP(badActivate, request)
if badActivate.Code != http.StatusBadRequest {
t.Fatalf("bad activate = %d want 400", badActivate.Code)
}
store.activateSettingsErr = nil
store.settingsRevisions = []governance.SourceConnectionRevision{{ID: "scr-2", RevisionNumber: 2, State: "active", Name: "国内工业AList", SecretRef: "alist-file", Enabled: true, RefreshIntervalSeconds: "3600", CreatedBy: "zherom"}}
history := httptest.NewRecorder()
handler.ServeHTTP(history, httptest.NewRequest("GET", "/api/v1/source-connections/conn-1/settings/revisions", nil))
if history.Code != http.StatusOK {
t.Fatalf("history = %d", history.Code)
}
body := history.Body.String()
if !strings.Contains(body, "alist-file") || !strings.Contains(body, "3600") {
t.Fatalf("history body=%s", body)
}
}
func TestHandlerReleaseAuditChain(t *testing.T) {
store := &fakeJobStore{auditChain: governance.ReleaseAuditChain{
ExecutionID: "rx-1", Operation: "publish", State: "completed",
BatchID: "rb-1", BatchName: "9月批次", ReleaseChangeSetID: "rcs-1",
ChangeSetName: "9月发布", ManifestSHA256: strings.Repeat("a", 64),
Snapshots: []governance.AuditSnapshot{{ID: "snap-1", SourceName: "PP资料下载中心", NodeCount: 66, FileCount: 621}},
CatalogChangeSets: []governance.AuditCatalogSet{{ID: "cs-1", State: "applied", AddCount: 2, MoveCount: 1}},
Items: []governance.AuditChainItem{{PublicationID: "tp-1", Title: "手册.pdf", Action: "add", ItemState: "verified", ReviewID: "rev-1", AcceptanceRunID: "run-1", SourceItemID: "si-1", SourceVersionID: "siv-1", ProductionRevision: "rev-tp-1"}},
}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return time.Now() }})
if err != nil {
t.Fatal(err)
}
rec := httptest.NewRecorder()
handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/release-executions/rx-1/audit-chain", nil))
if rec.Code != http.StatusOK || store.auditChainID != "rx-1" {
t.Fatalf("audit chain = %d %q body=%s", rec.Code, store.auditChainID, rec.Body.String())
}
body := rec.Body.String()
if !strings.Contains(body, `"release_batch_name":"9月批次"`) || !strings.Contains(body, `"snapshots"`) || !strings.Contains(body, `"items"`) || !strings.Contains(body, `"catalog_change_sets":[{"id":"cs-1"`) {
t.Fatalf("incomplete chain body=%s", body)
}
// R6: the projection must never leak bundle paths, base URLs, credentials.
for _, forbidden := range []string{"spool/", "http://", "https://", "bundle_path", "api_key", "secret_ref", "cookie", "bearer", "password"} {
if strings.Contains(strings.ToLower(body), forbidden) {
t.Fatalf("audit chain leaks %q body=%s", forbidden, body)
}
}
// 404 path.
store.auditChainErr = postgresstore.ErrReleaseAuditChainNotFound
missing := httptest.NewRecorder()
handler.ServeHTTP(missing, httptest.NewRequest(http.MethodGet, "/api/v1/release-executions/rx-missing/audit-chain", nil))
if missing.Code != http.StatusNotFound {
t.Fatalf("missing = %d %s", missing.Code, missing.Body.String())
}
}
func TestHandlerReleaseChangeSetEndpoints(t *testing.T) {
store := &fakeJobStore{generateChangeSetResult: governance.ReleaseChangeSet{ID: "rcs-1", ManifestSHA256: strings.Repeat("a", 64)}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return time.Date(2026, 9, 12, 12, 0, 0, 0, time.UTC) }})
if err != nil {
t.Fatal(err)
}
body := strings.NewReader(`{"name":"9月发布"}`)
generated := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/release-change-sets", body)
request.Header.Set("Idempotency-Key", "11111111-1111-4111-8111-111111111111")
handler.ServeHTTP(generated, request)
if generated.Code != http.StatusCreated || store.generateChangeSetRequest.Name != "9月发布" || store.generateChangeSetOperationID == "" {
t.Fatalf("generate = %d %#v %q body=%s", generated.Code, store.generateChangeSetRequest, store.generateChangeSetOperationID, generated.Body.String())
}
// Missing header → 400 (C4 contract).
noKey := httptest.NewRecorder()
handler.ServeHTTP(noKey, httptest.NewRequest(http.MethodPost, "/api/v1/release-change-sets", strings.NewReader(`{"name":"x"}`)))
if noKey.Code != http.StatusBadRequest {
t.Fatalf("generate without key = %d", noKey.Code)
}
// Empty plan → 422 honest reason.
store.generateChangeSetErr = postgresstore.ErrReleaseChangeSetEmpty
empty := httptest.NewRecorder()
request = httptest.NewRequest(http.MethodPost, "/api/v1/release-change-sets", strings.NewReader(`{"name":"x"}`))
request.Header.Set("Idempotency-Key", "22222222-2222-4222-8222-222222222222")
handler.ServeHTTP(empty, request)
if empty.Code != http.StatusUnprocessableEntity {
t.Fatalf("empty plan = %d %s", empty.Code, empty.Body.String())
}
store.generateChangeSetErr = nil
// Confirm path with envelope.
store.confirmReleaseSetResult = governance.ReleaseChangeSet{ID: "rcs-1", State: "confirmed"}
confirmed := httptest.NewRecorder()
request = httptest.NewRequest(http.MethodPost, "/api/v1/release-change-sets/rcs-1/confirm", strings.NewReader(`{}`))
request.Header.Set("Idempotency-Key", "33333333-3333-4333-8333-333333333333")
handler.ServeHTTP(confirmed, request)
if confirmed.Code != http.StatusOK || store.confirmReleaseSetRequest.ChangeSetID != "rcs-1" || !strings.Contains(confirmed.Body.String(), `"operation":"created"`) {
t.Fatalf("confirm = %d %#v body=%s", confirmed.Code, store.confirmReleaseSetRequest, confirmed.Body.String())
}
// State conflict → 409.
store.confirmReleaseSetErr = postgresstore.ErrReleaseChangeSetStateConflict
conflict := httptest.NewRecorder()
request = httptest.NewRequest(http.MethodPost, "/api/v1/release-change-sets/rcs-1/confirm", strings.NewReader(`{}`))
request.Header.Set("Idempotency-Key", "44444444-4444-4444-8444-444444444444")
handler.ServeHTTP(conflict, request)
if conflict.Code != http.StatusConflict {
t.Fatalf("confirm conflict = %d", conflict.Code)
}
// Detail endpoints.
store.changeSetDetail = governance.ReleaseChangeSetDetail{ReleaseChangeSet: governance.ReleaseChangeSet{ID: "rcs-1"}, Items: []governance.ReleaseChangeSetItem{{Seq: 1, PublicationID: "tp-1", Action: "add", CanRelease: true}}}
detail := httptest.NewRecorder()
handler.ServeHTTP(detail, httptest.NewRequest(http.MethodGet, "/api/v1/release-change-sets/rcs-1", nil))
if detail.Code != http.StatusOK || !strings.Contains(detail.Body.String(), `"publication_id":"tp-1"`) {
t.Fatalf("detail = %d %s", detail.Code, detail.Body.String())
}
latest := httptest.NewRecorder()
handler.ServeHTTP(latest, httptest.NewRequest(http.MethodGet, "/api/v1/release-change-sets/latest", nil))
if latest.Code != http.StatusOK {
t.Fatalf("latest = %d %s", latest.Code, latest.Body.String())
}
store.changeSetDetailErr = postgresstore.ErrReleaseChangeSetNotFound
missing := httptest.NewRecorder()
handler.ServeHTTP(missing, httptest.NewRequest(http.MethodGet, "/api/v1/release-change-sets/rcs-404", nil))
if missing.Code != http.StatusNotFound {
t.Fatalf("missing = %d", missing.Code)
}
// Batch citing a change set passes the binding through; unconfirmed → 422.
store.createBatchResult = governance.ReleaseBatch{ID: "rb-1"}
batch := httptest.NewRecorder()
request = httptest.NewRequest(http.MethodPost, "/api/v1/release-batches", strings.NewReader(`{"name":"批次","publication_ids":["tp-1"],"delete_publication_ids":[],"release_change_set_id":"rcs-1"}`))
request.Header.Set("Idempotency-Key", "55555555-5555-4555-8555-555555555555")
handler.ServeHTTP(batch, request)
if batch.Code != http.StatusCreated || store.createBatchRequest.ReleaseChangeSetID != "rcs-1" {
t.Fatalf("batch = %d %#v", batch.Code, store.createBatchRequest)
}
store.createBatchErr = postgresstore.ErrReleaseChangeSetUnconfirmed
unconf := httptest.NewRecorder()
request = httptest.NewRequest(http.MethodPost, "/api/v1/release-batches", strings.NewReader(`{"name":"批次","publication_ids":["tp-1"],"delete_publication_ids":[],"release_change_set_id":"rcs-9"}`))
request.Header.Set("Idempotency-Key", "66666666-6666-4666-8666-666666666666")
handler.ServeHTTP(unconf, request)
if unconf.Code != http.StatusUnprocessableEntity || !strings.Contains(unconf.Body.String(), "confirmed") {
t.Fatalf("unconfirmed batch = %d %s", unconf.Code, unconf.Body.String())
}
}
func TestHandlerLivenessReadinessAreIndependent(t *testing.T) {
now := time.Date(2026, 9, 12, 10, 0, 0, 0, time.UTC)
healthy := governance.ServiceHealth{Key: "postgres", Name: "PostgreSQL", Liveness: "up", Readiness: "ready", Configured: true, Applicable: true}
degraded := governance.ServiceHealth{Key: "worker", Name: "解析Worker", Liveness: "down", Readiness: "blocked", Reason: "队列无新鲜心跳", Configured: true, Applicable: true}
unconfigured := governance.ServiceHealth{Key: "mineru", Name: "MinerU", Liveness: "unknown", Readiness: "not_configured", Configured: false, Applicable: true}
notDeployed := governance.ServiceHealth{Key: "redis", Name: "Redis / MQ", Liveness: "unknown", Readiness: "not_applicable", Configured: false, Applicable: false}
store := &fakeJobStore{}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, Monitor: &fakeMonitor{services: []governance.ServiceHealth{healthy, unconfigured, notDeployed}}})
if err != nil {
t.Fatal(err)
}
live := httptest.NewRecorder()
handler.ServeHTTP(live, httptest.NewRequest(http.MethodGet, "/livez", nil))
if live.Code != http.StatusOK || !strings.Contains(live.Body.String(), "live") {
t.Fatalf("livez = %d %s", live.Code, live.Body.String())
}
ready := httptest.NewRecorder()
handler.ServeHTTP(ready, httptest.NewRequest(http.MethodGet, "/readyz", nil))
if ready.Code != http.StatusOK || !strings.Contains(ready.Body.String(), `"not_configured":["mineru"]`) {
t.Fatalf("readyz = %d %s", ready.Code, ready.Body.String())
}
if strings.Contains(ready.Body.String(), "redis") {
t.Fatal("not-applicable service must not appear in readiness")
}
// A worker outage degrades the matrix but keeps the API itself in rotation
// (admin UI must stay reachable to fix it) and never touches liveness.
handler, err = NewHandler(Config{Store: store, Now: func() time.Time { return now }, Monitor: &fakeMonitor{services: []governance.ServiceHealth{healthy, degraded}}})
if err != nil {
t.Fatal(err)
}
live = httptest.NewRecorder()
handler.ServeHTTP(live, httptest.NewRequest(http.MethodGet, "/livez", nil))
if live.Code != http.StatusOK {
t.Fatalf("livez during dependency outage = %d", live.Code)
}
degradedRec := httptest.NewRecorder()
handler.ServeHTTP(degradedRec, httptest.NewRequest(http.MethodGet, "/readyz", nil))
if degradedRec.Code != http.StatusOK || !strings.Contains(degradedRec.Body.String(), `"degraded":["worker"]`) {
t.Fatalf("readyz worker outage = %d %s", degradedRec.Code, degradedRec.Body.String())
}
if strings.Contains(degradedRec.Body.String(), `"blocked":["worker"]`) {
t.Fatal("worker outage must not block API readiness")
}
// PostgreSQL failure blocks readiness (hard dependency).
pgDown := governance.ServiceHealth{Key: "postgres", Name: "PostgreSQL", Liveness: "down", Readiness: "blocked", Reason: "网络不可达", Configured: true, Applicable: true}
handler, err = NewHandler(Config{Store: store, Now: func() time.Time { return now }, Monitor: &fakeMonitor{services: []governance.ServiceHealth{pgDown}}})
if err != nil {
t.Fatal(err)
}
blocked := httptest.NewRecorder()
handler.ServeHTTP(blocked, httptest.NewRequest(http.MethodGet, "/readyz", nil))
if blocked.Code != http.StatusServiceUnavailable || !strings.Contains(blocked.Body.String(), `"blocked":["postgres"]`) {
t.Fatalf("readyz postgres outage = %d %s", blocked.Code, blocked.Body.String())
}
// weknora gating is opt-in: configured-but-blocked weknora only blocks when
// RequireWeKnoraReady is set.
wkDown := governance.ServiceHealth{Key: "weknora", Name: "WeKnora API", Liveness: "up", Readiness: "blocked", Reason: "服务可达但目标知识库无读取权限", Configured: true, Applicable: true}
handler, err = NewHandler(Config{Store: store, Now: func() time.Time { return now }, RequireWeKnoraReady: true, Monitor: &fakeMonitor{services: []governance.ServiceHealth{healthy, wkDown}}})
if err != nil {
t.Fatal(err)
}
blocked = httptest.NewRecorder()
handler.ServeHTTP(blocked, httptest.NewRequest(http.MethodGet, "/readyz", nil))
if blocked.Code != http.StatusServiceUnavailable || !strings.Contains(blocked.Body.String(), `"blocked":["weknora"]`) {
t.Fatalf("readyz gated weknora = %d %s", blocked.Code, blocked.Body.String())
}
}
func TestHandlerServesExplicitlyLabelledMonitorSimulation(t *testing.T) {
now := time.Date(2026, 9, 8, 12, 0, 0, 0, time.UTC)
monitor := &fakeMonitor{
overview: governance.MonitorOverview{Mode: "simulation", Simulated: true, ObservedAt: now, Host: governance.MonitorHost{CPUPercent: 42, Simulated: true}, Services: []governance.MonitorService{{Key: "mineru", Name: "MinerU", State: "healthy", Detail: "模拟", ObservedAt: now, Simulated: true}}, KnowledgeBases: []governance.KnowledgeBaseProgress{{ID: "kb-sim", Name: "公司级技术支持知识库", Imported: 5750, Completed: 4530}}},
files: governance.KnowledgeMonitorFilePage{Items: []governance.KnowledgeMonitorFile{{Name: "[模拟] 文件.pdf", Status: "failed"}}, Total: 1, Limit: 20, SampleOnly: true},
}
handler, err := NewHandler(Config{Store: &fakeJobStore{}, Monitor: monitor})
if err != nil {
t.Fatal(err)
}
overview := httptest.NewRecorder()
handler.ServeHTTP(overview, httptest.NewRequest(http.MethodGet, "/api/v1/monitor/overview", nil))
if overview.Code != http.StatusOK || !strings.Contains(overview.Body.String(), `"mode":"simulation"`) || !strings.Contains(overview.Body.String(), `"simulated":true`) {
t.Fatalf("overview=%d body=%s", overview.Code, overview.Body.String())
}
metricsReq := httptest.NewRecorder()
monitor.metrics = governance.RunMetrics{ObservedAt: now, WorkerSource: "pg_lease", ActiveWorkers: 2,
Queue: governance.QueueGauges{Pending: 3, Running: 2, Failed: 1}}
handler.ServeHTTP(metricsReq, httptest.NewRequest(http.MethodGet, "/api/v1/monitor/metrics", nil))
if metricsReq.Code != http.StatusOK || !strings.Contains(metricsReq.Body.String(), `"worker_source":"pg_lease"`) || !strings.Contains(metricsReq.Body.String(), `"success_rate":null`) {
t.Fatalf("metrics=%d body=%s", metricsReq.Code, metricsReq.Body.String())
}
metricsDown := httptest.NewRecorder()
monitor.metricsErr = errors.New("db down")
handler.ServeHTTP(metricsDown, httptest.NewRequest(http.MethodGet, "/api/v1/monitor/metrics", nil))
if metricsDown.Code != http.StatusServiceUnavailable {
t.Fatalf("metrics down status = %d", metricsDown.Code)
}
monitor.metricsErr = nil
files := httptest.NewRecorder()
handler.ServeHTTP(files, httptest.NewRequest(http.MethodGet, "/api/v1/monitor/knowledge-bases/kb-sim/files?status=failed&limit=20&offset=0", nil))
if files.Code != http.StatusOK || monitor.knowledgeBaseID != "kb-sim" || monitor.status != "failed" || !strings.Contains(files.Body.String(), `"sample_only":true`) {
t.Fatalf("files=%d monitor=%#v body=%s", files.Code, monitor, files.Body.String())
}
for _, forbidden := range []string{"api_key", "base_url", "authorization", "secret"} {
if strings.Contains(strings.ToLower(overview.Body.String()+files.Body.String()), forbidden) {
t.Fatalf("monitor response leaked %s", forbidden)
}
}
}
func TestHandlerListsSourceConnectionsSafeProjection(t *testing.T) {
store := &fakeJobStore{connectionResult: governance.SourceConnectionPage{
Items: []governance.SourceConnectionView{{
ID: "conn-safe", SourceType: governance.SourceTypeAList, Name: "AList只读源",
Enabled: true, ScheduleConfigured: true, HealthStatus: governance.SourceHealthHealthy, ItemCount: 42,
}},
Summary: governance.SourceConnectionSummary{Total: 1, Enabled: 1, Healthy: 1},
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/source-connections", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d; body=%s", response.Code, response.Body.String())
}
for _, expected := range []string{`"id":"conn-safe"`, `"source_type":"alist"`, `"item_count":42`, `"total":1`} {
if !bytes.Contains(response.Body.Bytes(), []byte(expected)) {
t.Fatalf("body missing %s: %s", expected, response.Body.String())
}
}
for _, forbidden := range []string{"base_url", "secret_ref", "https://", "api_key"} {
if bytes.Contains(bytes.ToLower(response.Body.Bytes()), []byte(forbidden)) {
t.Fatalf("body leaked %q: %s", forbidden, response.Body.String())
}
}
}
func TestHandlerTriggersPPSourceScanWithSafeAcceptedResponse(t *testing.T) {
now := time.Date(2026, 9, 6, 5, 0, 0, 0, time.UTC)
store := &fakeJobStore{triggerScanResult: governance.SourceScan{
ID: "scan-safe", SourceConnectionID: "conn-pp", SourceType: governance.SourceTypePP,
State: governance.SourceScanPending, CreatedAt: now,
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodPost, "/api/v1/source-connections/conn-pp/scan", nil)
request.Header.Set("Idempotency-Key", "op-scan-1")
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
if store.triggerScanConnectionID != "conn-pp" {
t.Fatalf("connection ID = %q", store.triggerScanConnectionID)
}
for _, expected := range []string{`"id":"scan-safe"`, `"source_connection_id":"conn-pp"`, `"state":"pending"`} {
if !bytes.Contains(response.Body.Bytes(), []byte(expected)) {
t.Errorf("response missing %s: %s", expected, response.Body.String())
}
}
for _, forbidden := range []string{"snapshot_sha256", "stable_external_id", "base_url", "secret_ref"} {
if bytes.Contains(response.Body.Bytes(), []byte(forbidden)) {
t.Errorf("response leaked %q: %s", forbidden, response.Body.String())
}
}
}
func TestHandlerListsSourceScanHistoryWithSafeProjection(t *testing.T) {
now := time.Date(2026, 9, 6, 6, 30, 0, 0, time.UTC)
store := &fakeJobStore{scanPageResult: governance.SourceScanPage{
Items: []governance.SourceScanSummaryView{{
ID: "scan-safe", State: governance.SourceScanCompleted, ObservedItems: 613,
UnchangedItems: 559, OverLimitItems: 54, CreatedAt: now, CompletedAt: &now,
}}, Total: 5, Limit: 2, Offset: 1,
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/source-connections/conn-pp/scans?limit=2&offset=1", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
}
if store.scanPageConnectionID != "conn-pp" || store.scanPageLimit != 2 || store.scanPageOffset != 1 {
t.Fatalf("list args=%q/%d/%d", store.scanPageConnectionID, store.scanPageLimit, store.scanPageOffset)
}
for _, expected := range []string{`"id":"scan-safe"`, `"observed_items":613`, `"over_limit_items":54`, `"total":5`, `"limit":2`, `"offset":1`} {
if !bytes.Contains(response.Body.Bytes(), []byte(expected)) {
t.Errorf("body missing %s: %s", expected, response.Body.String())
}
}
assertNoSourceScanSecrets(t, response.Body.Bytes())
}
func TestHandlerGetsSourceScanDetailWithSafeChanges(t *testing.T) {
now := time.Date(2026, 9, 6, 6, 40, 0, 0, time.UTC)
itemID := "si-safe"
size := int64(42)
store := &fakeJobStore{scanDetailResult: governance.SourceScanDetailView{
SourceScanSummaryView: governance.SourceScanSummaryView{ID: "scan-safe", State: governance.SourceScanCompleted, ObservedItems: 1, UpdatedItems: 1, PlannedUploads: 1, CreatedAt: now, CompletedAt: &now},
Changes: []governance.SourceScanChangeView{{ID: "chg-safe", SourceItemID: &itemID, ChangeType: governance.SourceScanUpdated, ObservedSizeBytes: &size, CreatedAt: now}},
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/source-connections/conn-pp/scans/scan-safe", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK || store.scanDetailConnectionID != "conn-pp" || store.scanDetailID != "scan-safe" {
t.Fatalf("status=%d args=%q/%q body=%s", response.Code, store.scanDetailConnectionID, store.scanDetailID, response.Body.String())
}
for _, expected := range []string{`"id":"chg-safe"`, `"source_item_id":"si-safe"`, `"change_type":"updated"`, `"observed_size_bytes":42`} {
if !bytes.Contains(response.Body.Bytes(), []byte(expected)) {
t.Errorf("body missing %s: %s", expected, response.Body.String())
}
}
assertNoSourceScanSecrets(t, response.Body.Bytes())
}
func TestHandlerReturnsNotFoundForUnknownSourceScanDetail(t *testing.T) {
store := &fakeJobStore{scanDetailErr: postgresstore.ErrSourceScanNotFound}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/source-connections/conn-pp/scans/missing", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusNotFound || response.Body.String() != "{\"error\":\"source scan not found\"}\n" {
t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
}
}
func TestHandlerListsSafeReviewCandidates(t *testing.T) {
now := time.Date(2026, 9, 6, 10, 0, 0, 0, time.UTC)
store := &fakeJobStore{reviewPageResult: governance.ReviewCandidatePage{
Items: []governance.ReviewCandidate{{PublicationID: "tp-safe", Title: "Synthetic", SourceType: governance.SourceTypeManual, CurrentVisibility: governance.VisibilityInternal, AcceptanceRunID: "hidden-run", KnowledgeExists: governance.AcceptancePassed, ChunksPresent: governance.AcceptancePassed, RetrievalHit: governance.AcceptanceManualRequired, GateStatus: governance.ReviewGateReady, GateCode: "ready", ReviewState: governance.ReviewPending, PublicationUpdated: now}},
Total: 1, Limit: 20,
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/review-candidates?review_state=pending&limit=20&offset=0", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK || store.reviewFilter.ReviewState != "pending" {
t.Fatalf("status=%d filter=%#v body=%s", response.Code, store.reviewFilter, response.Body.String())
}
for _, expected := range []string{`"publication_id":"tp-safe"`, `"gate_status":"ready"`, `"retrieval_hit":"manual_required"`, `"total":1`} {
if !bytes.Contains(response.Body.Bytes(), []byte(expected)) {
t.Errorf("missing %s", expected)
}
}
for _, forbidden := range []string{"hidden-run", "acceptance_run_id", "publication_updated_at", "knowledge_id", "test_kb_id"} {
if bytes.Contains(response.Body.Bytes(), []byte(forbidden)) {
t.Errorf("response leaked %q", forbidden)
}
}
}
func TestHandlerRecordsPublicationReview(t *testing.T) {
now := time.Date(2026, 9, 6, 10, 10, 0, 0, time.UTC)
store := &fakeJobStore{reviewResult: governance.PublicationReview{ID: "rv-safe", PublicationID: "tp-safe", Decision: governance.ReviewApproved, DecidedVisibility: governance.VisibilityInternal, ReviewerRef: "engineer-1", DecisionCode: "approved_for_release", AcceptanceRunID: "hidden-run", CreatedAt: now}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodPost, "/api/v1/review-candidates/tp-safe/decision", bytes.NewBufferString(`{"decision":"approved","decided_visibility":"internal","decision_code":"approved_for_release"}`))
request.Header.Set("Idempotency-Key", "op-review-1")
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusCreated || store.reviewRequest.PublicationID != "tp-safe" || store.reviewRequest.Now != now {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.reviewRequest, response.Body.String())
}
if bytes.Contains(response.Body.Bytes(), []byte("hidden-run")) || bytes.Contains(response.Body.Bytes(), []byte("acceptance_run_id")) {
t.Fatalf("review response leaked internal gate identity: %s", response.Body.String())
}
}
func TestHandlerVersionsAndActivatesUISettings(t *testing.T) {
now := time.Date(2026, 9, 10, 8, 0, 0, 0, time.UTC)
activeAt := now.Add(time.Second)
store := &fakeJobStore{
activeUISettingsResult: governance.UISettings{ID: "settings-default", State: "active", DocumentRefreshSeconds: 15, ResourceRefreshSeconds: 30, StalledAfterSeconds: 1800, FinalizingAfterSeconds: 7200, UpdatedBy: "system:migration", CreatedAt: now, ActivatedAt: &now},
candidateUISettingsResult: governance.UISettings{ID: "uis-candidate", State: "validated", DocumentRefreshSeconds: 30, ResourceRefreshSeconds: 60, StalledAfterSeconds: 3600, FinalizingAfterSeconds: 14400, UpdatedBy: "engineer-1", CreatedAt: now},
activatedUISettingsResult: governance.UISettings{ID: "uis-candidate", State: "active", DocumentRefreshSeconds: 30, ResourceRefreshSeconds: 60, StalledAfterSeconds: 3600, FinalizingAfterSeconds: 14400, UpdatedBy: "engineer-1", CreatedAt: now, ActivatedAt: &activeAt},
}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }})
if err != nil {
t.Fatal(err)
}
get := httptest.NewRecorder()
handler.ServeHTTP(get, httptest.NewRequest(http.MethodGet, "/api/v1/ui-settings", nil))
if get.Code != http.StatusOK || !strings.Contains(get.Body.String(), `"state":"active"`) {
t.Fatalf("get status=%d body=%s", get.Code, get.Body.String())
}
noIntent := httptest.NewRecorder()
handler.ServeHTTP(noIntent, httptest.NewRequest(http.MethodPost, "/api/v1/ui-settings/candidates", bytes.NewBufferString(`{"document_refresh_seconds":30,"resource_refresh_seconds":60,"stalled_after_seconds":3600,"finalizing_after_seconds":14400,"ingestion_concurrency":2}`)))
if noIntent.Code != http.StatusForbidden {
t.Fatalf("missing intent status=%d body=%s", noIntent.Code, noIntent.Body.String())
}
create := httptest.NewRecorder()
createRequest := httptest.NewRequest(http.MethodPost, "/api/v1/ui-settings/candidates", bytes.NewBufferString(`{"document_refresh_seconds":30,"resource_refresh_seconds":60,"stalled_after_seconds":3600,"finalizing_after_seconds":14400,"ingestion_concurrency":2}`))
createRequest.Header.Set("X-Governance-Action", "ui-settings")
createRequest.Header.Set("Idempotency-Key", "op-uis-create")
handler.ServeHTTP(create, createRequest)
if create.Code != http.StatusCreated || store.createUISettingsRequest.Now != now || store.createUISettingsRequest.UpdatedBy != "system" || !strings.Contains(create.Body.String(), `"operation":"created"`) {
t.Fatalf("create status=%d request=%#v body=%s", create.Code, store.createUISettingsRequest, create.Body.String())
}
unknown := httptest.NewRecorder()
unknownRequest := httptest.NewRequest(http.MethodPost, "/api/v1/ui-settings/candidates", bytes.NewBufferString(`{"document_refresh_seconds":30,"resource_refresh_seconds":60,"stalled_after_seconds":3600,"finalizing_after_seconds":14400,"secret":"must-reject"}`))
unknownRequest.Header.Set("X-Governance-Action", "ui-settings")
handler.ServeHTTP(unknown, unknownRequest)
if unknown.Code != http.StatusBadRequest {
t.Fatalf("unknown field status=%d body=%s", unknown.Code, unknown.Body.String())
}
activate := httptest.NewRecorder()
activateRequest := httptest.NewRequest(http.MethodPost, "/api/v1/ui-settings/uis-candidate/activate", bytes.NewBufferString(`{}`))
activateRequest.Header.Set("X-Governance-Action", "ui-settings")
activateRequest.Header.Set("Idempotency-Key", "op-uis-activate")
handler.ServeHTTP(activate, activateRequest)
if activate.Code != http.StatusOK || store.activateUISettingsID != "uis-candidate" || !strings.Contains(activate.Body.String(), `"state":"active"`) {
t.Fatalf("activate status=%d id=%q body=%s", activate.Code, store.activateUISettingsID, activate.Body.String())
}
}
func TestHandlerCreatesAndSealsReleaseBatchWithoutReleaseJob(t *testing.T) {
now := time.Date(2026, 9, 6, 10, 20, 0, 0, time.UTC)
store := &fakeJobStore{
createBatchResult: governance.ReleaseBatch{ID: "rb-safe", Name: "Batch", State: governance.ReleaseBatchDraft, PolicyVersion: "release-gate-v1", CreatedBy: "engineer-1", ItemCount: 1, CreatedAt: now},
sealBatchResult: governance.ReleaseBatch{ID: "rb-safe", Name: "Batch", State: governance.ReleaseBatchSealed, PolicyVersion: "release-gate-v1", CreatedBy: "engineer-1", SealedBy: "engineer-2", ItemCount: 1, CreatedAt: now, SealedAt: &now},
}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }})
if err != nil {
t.Fatal(err)
}
createRequest := httptest.NewRequest(http.MethodPost, "/api/v1/release-batches", bytes.NewBufferString(`{"name":"Batch","publication_ids":["tp-safe"],"delete_publication_ids":["tp-delete"]}`))
createRequest.Header.Set("Idempotency-Key", "op-batch-create")
createResponse := httptest.NewRecorder()
handler.ServeHTTP(createResponse, createRequest)
if createResponse.Code != http.StatusCreated || len(store.createBatchRequest.PublicationIDs) != 1 || len(store.createBatchRequest.DeletePublicationIDs) != 1 || store.createBatchRequest.DeletePublicationIDs[0] != "tp-delete" {
t.Fatalf("create status=%d request=%#v body=%s", createResponse.Code, store.createBatchRequest, createResponse.Body.String())
}
sealRequest := httptest.NewRequest(http.MethodPost, "/api/v1/release-batches/rb-safe/seal", bytes.NewBufferString(`{}`))
sealRequest.Header.Set("Idempotency-Key", "op-batch-seal")
sealResponse := httptest.NewRecorder()
handler.ServeHTTP(sealResponse, sealRequest)
if sealResponse.Code != http.StatusOK || store.sealBatchRequest.BatchID != "rb-safe" || store.sealBatchRequest.SealedBy != "system" {
t.Fatalf("seal status=%d request=%#v body=%s", sealResponse.Code, store.sealBatchRequest, sealResponse.Body.String())
}
if store.enqueued.Type != "" {
t.Fatalf("P5 API enqueued unexpected job: %#v", store.enqueued)
}
}
func TestHandlerListsSafeSourceTruthReleasePlan(t *testing.T) {
store := &fakeJobStore{releasePlanResult: governance.ReleasePlan{
Items: []governance.ReleasePlanItem{{PublicationID: "tp-safe", Title: "B.pdf", TargetScope: "external", Action: "add", TestState: "ready", ProductionState: "absent", CanRelease: true}, {Title: "未就绪.pdf", TargetScope: "internal", Action: "blocked", TestState: "not_ready", ProductionState: "absent", Blocker: "test_not_ready"}},
Summary: governance.ReleasePlanSummary{Add: 1},
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v1/release-plan", nil))
if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), `"action":"add"`) || !strings.Contains(response.Body.String(), `"add":1`) || !strings.Contains(response.Body.String(), `"publication_id":"","title":"未就绪.pdf"`) {
t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
}
for _, forbidden := range []string{"production_revision", "test_kb_id", "knowledge_id", "bundle_sha256"} {
if strings.Contains(response.Body.String(), forbidden) {
t.Fatalf("release plan leaked %q: %s", forbidden, response.Body.String())
}
}
}
func TestHandlerListsAndGetsReleaseBatches(t *testing.T) {
now := time.Date(2026, 9, 6, 10, 30, 0, 0, time.UTC)
batch := governance.ReleaseBatch{ID: "rb-safe", Name: "Batch", State: governance.ReleaseBatchSealed, PolicyVersion: "release-gate-v1", CreatedBy: "engineer-1", SealedBy: "engineer-2", ItemCount: 1, CreatedAt: now, SealedAt: &now}
store := &fakeJobStore{batchPageResult: governance.ReleaseBatchPage{Items: []governance.ReleaseBatch{batch}, Total: 1, Limit: 20}, batchDetailResult: governance.ReleaseBatchDetail{ReleaseBatch: batch, Items: []governance.ReleaseBatchItemView{{PublicationID: "tp-safe", Title: "Synthetic", DecidedVisibility: governance.VisibilityInternal}}}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
listResponse := httptest.NewRecorder()
handler.ServeHTTP(listResponse, httptest.NewRequest(http.MethodGet, "/api/v1/release-batches?state=sealed&limit=20", nil))
detailResponse := httptest.NewRecorder()
handler.ServeHTTP(detailResponse, httptest.NewRequest(http.MethodGet, "/api/v1/release-batches/rb-safe", nil))
if listResponse.Code != http.StatusOK || detailResponse.Code != http.StatusOK || store.batchID != "rb-safe" {
t.Fatalf("list=%d detail=%d batchID=%q", listResponse.Code, detailResponse.Code, store.batchID)
}
for _, body := range [][]byte{listResponse.Body.Bytes(), detailResponse.Body.Bytes()} {
for _, forbidden := range []string{"knowledge_id", "test_kb_id", "acceptance_run_id", "publication_updated_at"} {
if bytes.Contains(body, []byte(forbidden)) {
t.Errorf("batch response leaked %q", forbidden)
}
}
}
}
func TestHandlerBlocksProductionReleaseAuthorizationByDefault(t *testing.T) {
store := &fakeJobStore{}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/release-batches/rb-safe/authorize-production", bytes.NewBufferString(`{"authorized_by":"release-owner","confirmation":"publish-to-production"}`))
request.Header.Set("Content-Type", "application/json")
handler.ServeHTTP(response, request)
if response.Code != http.StatusForbidden || store.authorizeReleaseCalled {
t.Fatalf("status=%d called=%v body=%s", response.Code, store.authorizeReleaseCalled, response.Body.String())
}
}
func TestHandlerBlocksProductionAuthorizationUntilBackupRestoreIsVerified(t *testing.T) {
store := &fakeJobStore{}
handler, err := NewHandler(Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionReleaseAuthorizerRef: "release-owner", ProductionReleaseAuthorizationToken: "safe-token"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/release-batches/rb-safe/authorize-production", bytes.NewBufferString(`{"confirmation":"publish-to-production"}`))
request.Header.Set("Authorization", "Bearer safe-token")
handler.ServeHTTP(response, request)
if response.Code != http.StatusPreconditionFailed || store.authorizeReleaseCalled || !strings.Contains(response.Body.String(), "production backup restore verification is required") {
t.Fatalf("status=%d called=%v body=%s", response.Code, store.authorizeReleaseCalled, response.Body.String())
}
capabilities := httptest.NewRecorder()
handler.ServeHTTP(capabilities, httptest.NewRequest(http.MethodGet, "/api/v1/release-capabilities", nil))
if !strings.Contains(capabilities.Body.String(), `"production_backup_restore_verified":false`) {
t.Fatalf("capabilities=%s", capabilities.Body.String())
}
}
func TestHandlerAuthorizesProductionReleaseOnlyBehindExplicitGate(t *testing.T) {
now := time.Date(2026, 9, 6, 5, 0, 0, 0, time.UTC)
store := &fakeJobStore{authorizeReleaseResult: governance.ReleaseExecution{
ID: "rx-safe", ReleaseBatchID: "rb-safe", Operation: governance.ReleaseOperationPublish,
State: governance.ReleaseExecutionPending, TargetEnvironment: "production", AuthorizedBy: "release-owner",
AuthorizationCode: "production_release_authorized", JobID: "job-safe", ItemCount: 1, CreatedAt: now,
}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true, ProductionReleaseAuthorizerRef: "release-owner", ProductionReleaseAuthorizationToken: "safe-token"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/release-batches/rb-safe/authorize-production", bytes.NewBufferString(`{"confirmation":"publish-to-production"}`))
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Authorization", "Bearer safe-token")
request.Header.Set("Idempotency-Key", "op-authz-publish")
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted {
t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
}
if store.authorizeReleaseRequest.BatchID != "rb-safe" || store.authorizeReleaseRequest.AuthorizedBy != "release-owner" || store.authorizeReleaseRequest.AuthorizationCode != "production_release_authorized" {
t.Fatalf("request=%#v", store.authorizeReleaseRequest)
}
body := response.Body.String()
if strings.Contains(body, "manifest_sha256") || strings.Contains(body, "production_database") {
t.Fatalf("unsafe authorization response: %s", body)
}
}
func TestHandlerAuthorizesRollbackOnlyBehindExplicitGate(t *testing.T) {
now := time.Date(2026, 9, 6, 6, 0, 0, 0, time.UTC)
store := &fakeJobStore{authorizeRollbackResult: governance.ReleaseExecution{
ID: "rx-rollback", ReleaseBatchID: "rb-safe", Operation: governance.ReleaseOperationRollback,
State: governance.ReleaseExecutionPending, TargetEnvironment: "production", AuthorizedBy: "release-owner",
AuthorizationCode: "production_rollback_authorized", JobID: "job-rollback", ItemCount: 1, CreatedAt: now,
}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, ProductionReleaseRequestsEnabled: true, ProductionReleaseAuthorizerRef: "release-owner", ProductionReleaseAuthorizationToken: "safe-token"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/release-executions/rx-publish/authorize-rollback", bytes.NewBufferString(`{"confirmation":"rollback-production-release"}`))
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Authorization", "Bearer safe-token")
request.Header.Set("Idempotency-Key", "op-authz-rollback")
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted || store.authorizeRollbackRequest.PriorExecutionID != "rx-publish" || store.authorizeRollbackRequest.AuthorizationCode != "production_rollback_authorized" {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.authorizeRollbackRequest, response.Body.String())
}
}
func TestHandlerRetriesFailedReleaseOnlyBehindExplicitGate(t *testing.T) {
now := time.Date(2026, 9, 6, 17, 0, 0, 0, time.UTC)
store := &fakeJobStore{retryReleaseResult: governance.ReleaseExecution{ID: "rx-failed", ReleaseBatchID: "rb-safe", Operation: governance.ReleaseOperationPublish, State: governance.ReleaseExecutionPending, TargetEnvironment: "production", AuthorizedBy: "release-owner", AuthorizationCode: "production_release_authorized", JobID: "job-safe", ItemCount: 1, CreatedAt: now}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, ProductionReleaseRequestsEnabled: true, ProductionReleaseAuthorizerRef: "release-owner", ProductionReleaseAuthorizationToken: "safe-token"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/release-executions/rx-failed/retry", bytes.NewBufferString(`{"confirmation":"retry-production-release"}`))
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Authorization", "Bearer safe-token")
request.Header.Set("Idempotency-Key", "op-retry-release")
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted || store.retryReleaseRequest.ExecutionID != "rx-failed" {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.retryReleaseRequest, response.Body.String())
}
if body := response.Body.String(); strings.Contains(body, "authorization_code") || strings.Contains(body, "manifest_sha256") {
t.Fatalf("unsafe retry response: %s", body)
}
}
func TestHandlerRejectsProductionAuthorizationWithoutServerIdentityAndToken(t *testing.T) {
for _, config := range []Config{
{Store: &fakeJobStore{}, ProductionReleaseRequestsEnabled: true},
{Store: &fakeJobStore{}, ProductionReleaseRequestsEnabled: true, ProductionReleaseAuthorizerRef: "release-owner"},
} {
if _, err := NewHandler(config); err == nil {
t.Fatal("NewHandler() error=nil")
}
}
}
func TestHandlerRejectsMissingProductionAuthorizationBearer(t *testing.T) {
store := &fakeJobStore{}
handler, err := NewHandler(Config{Store: store, ProductionReleaseRequestsEnabled: true, ProductionBackupRestoreVerified: true, ProductionReleaseAuthorizerRef: "release-owner", ProductionReleaseAuthorizationToken: "safe-token"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/release-batches/rb-safe/authorize-production", bytes.NewBufferString(`{"confirmation":"publish-to-production"}`))
request.Header.Set("Content-Type", "application/json")
handler.ServeHTTP(response, request)
if response.Code != http.StatusUnauthorized || store.authorizeReleaseCalled {
t.Fatalf("status=%d called=%v", response.Code, store.authorizeReleaseCalled)
}
}
func TestHandlerListsAndGetsSafeReleaseExecutions(t *testing.T) {
now := time.Date(2026, 9, 6, 9, 0, 0, 0, time.UTC)
execution := governance.ReleaseExecution{
ID: "rx-safe", ReleaseBatchID: "rb-safe", Operation: governance.ReleaseOperationPublish,
State: governance.ReleaseExecutionCompleted, TargetEnvironment: "production", AuthorizedBy: "owner",
AuthorizationCode: "hidden-auth-code", JobID: "job-safe", ManifestSHA256: strings.Repeat("a", 64),
ItemCount: 1, CompletedItemCount: 1, CreatedAt: now, CompletedAt: &now,
}
store := &fakeJobStore{
releaseExecutionPageResult: governance.ReleaseExecutionPage{Items: []governance.ReleaseExecution{execution}, Total: 1, Limit: 20},
releaseExecutionDetailResult: governance.ReleaseExecutionDetail{ReleaseExecution: execution, Items: []governance.ReleaseItemExecutionView{{PublicationID: "tp-safe", Title: "Synthetic", State: "verified", ProductionRevision: "prod-v1", CreatedAt: now, CompletedAt: &now}}},
}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
list := httptest.NewRecorder()
handler.ServeHTTP(list, httptest.NewRequest(http.MethodGet, "/api/v1/release-executions?operation=publish&state=completed&limit=20", nil))
detail := httptest.NewRecorder()
handler.ServeHTTP(detail, httptest.NewRequest(http.MethodGet, "/api/v1/release-executions/rx-safe", nil))
if list.Code != http.StatusOK || detail.Code != http.StatusOK || store.releaseExecutionID != "rx-safe" {
t.Fatalf("list=%d detail=%d id=%q", list.Code, detail.Code, store.releaseExecutionID)
}
serialized := strings.ToLower(list.Body.String() + detail.Body.String())
for _, forbidden := range []string{"authorization_code", "manifest_sha256", "knowledge_id", "test_kb_id", "database_url", "secret"} {
if strings.Contains(serialized, forbidden) {
t.Fatalf("release execution response leaked %q", forbidden)
}
}
}
func TestHandlerReportsProductionReleaseCapabilityDisabledByDefault(t *testing.T) {
handler, err := NewHandler(Config{Store: &fakeJobStore{}})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v1/release-capabilities", nil))
if response.Code != http.StatusOK || !strings.Contains(response.Body.String(), `"production_release_requests_enabled":false`) {
t.Fatalf("status=%d body=%s", response.Code, response.Body.String())
}
}
func assertNoSourceScanSecrets(t *testing.T, body []byte) {
t.Helper()
for _, forbidden := range []string{"snapshot_sha256", "stable_external_id", "previous_revision", "observed_revision", "base_url", "secret_ref", "local_path"} {
if bytes.Contains(bytes.ToLower(body), []byte(forbidden)) {
t.Errorf("response leaked %q: %s", forbidden, body)
}
}
}
func TestHandlerListsTestPublicationsSafeProjection(t *testing.T) {
store := &fakeJobStore{publicationResult: governance.TestPublicationPage{
Items: []governance.TestPublication{{ID: "pub-safe", Title: "Safe title", ProcessingState: "failed", LastError: "remote_parse_failed", CanRetry: true}},
Total: 1,
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/test-publications?processing_state=failed&limit=25", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d; body=%s", response.Code, response.Body.String())
}
if !bytes.Contains(response.Body.Bytes(), []byte(`"id":"pub-safe"`)) || !bytes.Contains(response.Body.Bytes(), []byte(`"last_error":"remote_parse_failed"`)) {
t.Fatalf("body = %s", response.Body.String())
}
if store.publicationFilter != (governance.TestPublicationFilter{ProcessingState: "failed", Limit: 25, Offset: 0}) {
t.Fatalf("filter = %#v", store.publicationFilter)
}
}
func TestHandlerAcceptsMultiStateAttentionFilter(t *testing.T) {
store := &fakeJobStore{}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v1/test-publications?processing_state=failed,stalled&limit=100", nil))
if response.Code != http.StatusOK {
t.Fatalf("status = %d; body=%s", response.Code, response.Body.String())
}
if store.publicationFilter.ProcessingState != "failed,stalled" {
t.Fatalf("filter = %#v", store.publicationFilter)
}
// One bad entry anywhere in the list is still rejected.
for _, value := range []string{"failed,bogus", "failed,", ",stalled"} {
response = httptest.NewRecorder()
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, "/api/v1/test-publications?processing_state="+url.QueryEscape(value), nil))
if response.Code != http.StatusBadRequest {
t.Fatalf("processing_state=%q status = %d; want 400", value, response.Code)
}
}
}
func TestHandlerConfirmsChangeSetItems(t *testing.T) {
now := time.Date(2026, 9, 12, 5, 0, 0, 0, time.UTC)
store := &fakeJobStore{confirmChangeSetResult: governance.ConfirmCatalogChangeSetResult{ChangeSetID: "cs-1", ConfirmedItems: 2, UploadsCreated: 2}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, InternalRoutingKnowledgeBaseID: "kb-internal", ExternalRoutingKnowledgeBaseID: "kb-external"})
if err != nil {
t.Fatal(err)
}
response := httptest.NewRecorder()
request := httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/change-sets/cs-1/confirm", bytes.NewBufferString(`{"item_seqs":[4,9],"reason":"确认新增与更新"}`))
request.Header.Set("Idempotency-Key", "op-confirm-1")
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK || store.confirmChangeSetOperationID != "op-confirm-1" || len(store.confirmChangeSetRequest.ItemSeqs) != 2 || store.confirmChangeSetRequest.InternalTargetKBID != "kb-internal" || store.confirmChangeSetRequest.Now != now {
t.Fatalf("status=%d id=%q request=%#v body=%s", response.Code, store.confirmChangeSetOperationID, store.confirmChangeSetRequest, response.Body.String())
}
if !strings.Contains(response.Body.String(), `"operation":"created"`) || strings.Contains(response.Body.String(), "source_locator_hash") {
t.Fatalf("unsafe confirm response=%s", response.Body.String())
}
missing := httptest.NewRecorder()
store.confirmChangeSetErr = postgresstore.ErrCatalogChangeSetNotFound
missingRequest := httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/change-sets/cs-nope/confirm", bytes.NewBufferString(`{"item_seqs":[1],"reason":"确认"}`))
missingRequest.Header.Set("Idempotency-Key", "op-confirm-missing")
handler.ServeHTTP(missing, missingRequest)
if missing.Code != http.StatusNotFound {
t.Fatalf("missing change set status=%d", missing.Code)
}
bad := httptest.NewRecorder()
handler.ServeHTTP(bad, httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/change-sets/cs-1/confirm", bytes.NewBufferString(`{"item_seqs":[],"reason":"确认"}`)))
if bad.Code != http.StatusBadRequest {
t.Fatalf("empty selection status=%d", bad.Code)
}
noKey := httptest.NewRecorder()
handler.ServeHTTP(noKey, httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/change-sets/cs-1/confirm", bytes.NewBufferString(`{"item_seqs":[1],"reason":"确认"}`)))
if noKey.Code != http.StatusBadRequest {
t.Fatalf("missing idempotency key status=%d", noKey.Code)
}
}
func TestHandlerStartsConfirmedTestPublications(t *testing.T) {
now := time.Date(2026, 9, 9, 3, 0, 0, 0, time.UTC)
store := &fakeJobStore{startPublicationsResult: governance.StartTestPublicationsResult{Selected: 2, Started: 1, AlreadyStarted: 1}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/start", bytes.NewBufferString(`{"publication_ids":["pub-a","pub-b"],"reason":"确认差异后开始处理"}`))
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Idempotency-Key", "op-processing-start")
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusAccepted || store.startPublicationsRequest.Now != now || len(store.startPublicationsRequest.PublicationIDs) != 2 || store.startPublicationsRequest.OperatorRef != "system" || !strings.Contains(response.Body.String(), `"operation":"created"`) {
t.Fatalf("status=%d request=%#v body=%s", response.Code, store.startPublicationsRequest, response.Body.String())
}
if !strings.Contains(response.Body.String(), `"started":1`) || strings.Contains(response.Body.String(), "test_kb_id") {
t.Fatalf("unsafe response=%s", response.Body.String())
}
}
func TestHandlerListsAcceptanceResults(t *testing.T) {
store := &fakeJobStore{acceptanceResult: []governance.AcceptanceRecord{
{CheckType: "chunks_present", State: "passed", EvidenceSummary: json.RawMessage(`{"chunk_total":2}`)},
}}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodGet, "/api/v1/test-publications/pub-1/acceptance", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK || store.acceptanceID != "pub-1" {
t.Fatalf("status=%d id=%q", response.Code, store.acceptanceID)
}
if !bytes.Contains(response.Body.Bytes(), []byte(`"chunks_present"`)) || !bytes.Contains(response.Body.Bytes(), []byte(`{"chunk_total":2}`)) {
t.Fatalf("body = %s", response.Body.String())
}
}
func TestHandlerRetriesPublication(t *testing.T) {
store := &fakeJobStore{}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/pub-1/retry", nil)
request.Header.Set("Idempotency-Key", "op-retry-1")
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK || store.retryID != "pub-1" {
t.Fatalf("status=%d id=%q body=%s", response.Code, store.retryID, response.Body.String())
}
}
func TestHandlerRetryReturnsConflictWhenNotRetryable(t *testing.T) {
store := &fakeJobStore{retryErr: postgresstore.ErrPublicationStateConflict}
handler, err := NewHandler(Config{Store: store})
if err != nil {
t.Fatal(err)
}
request := httptest.NewRequest(http.MethodPost, "/api/v1/test-publications/pub-1/retry", nil)
request.Header.Set("Idempotency-Key", "op-retry-1")
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusConflict {
t.Fatalf("status = %d, want 409; body=%s", response.Code, response.Body.String())
}
}
func TestHandlerRejectsCatalogMutationsWithoutOperationID(t *testing.T) {
now := time.Date(2026, 9, 11, 5, 0, 0, 0, time.UTC)
store := &fakeJobStore{catalogSources: []governance.CatalogSource{{ConnectionID: "conn-cat", SourceType: governance.SourceTypePP, Name: "PP资料下载中心", RootNodeID: "root-cat"}}}
handler, err := NewHandler(Config{Store: store, Now: func() time.Time { return now }, InternalRoutingKnowledgeBaseID: "kb-internal"})
if err != nil {
t.Fatal(err)
}
for _, request := range []*http.Request{
httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/route", bytes.NewBufferString(`{"connection_id":"conn-cat","node_ids":["file-a"],"policy":"internal","reason":"人工标注"}`)),
httptest.NewRequest(http.MethodPost, "/api/v1/source-catalog/sources/conn-cat/refresh", nil),
} {
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusBadRequest || !strings.Contains(response.Body.String(), "Idempotency-Key") {
t.Fatalf("%s %s status=%d body=%s", request.Method, request.URL.Path, response.Code, response.Body.String())
}
}
}
type fakeJobStore struct {
createSettingsConnectionID string
createSettingsDraft postgresstore.SourceSettingsDraft
createSettingsOperationID string
createSettingsErr error
createSettingsReplayed bool
activateSettingsConnectionID string
activateSettingsRevisionID string
activateSettingsOperationID string
activateSettingsReplayed bool
activateSettingsErr error
// enqueueEchoRequestedID:幂等"首次入队"语义 —— 返回调用方请求的那个 id
// (默认 false 时返回 enqueueResult,即"复用已有任务"的路径)。
enqueueEchoRequestedID bool
listSettingsConnectionID string
listSettingsErr error
settingsRevisions []governance.SourceConnectionRevision
jobFailureFilter governance.JobFailureFilter
jobFailurePage governance.JobFailurePage
jobFailureErr error
retryBatchRequest *governance.RetryPublicationsBatchRequest
retryBatchOperation string
retryBatchResult governance.RetryPublicationsBatchResult
cancelBatchRequest *governance.CancelPublicationsBatchRequest
cancelBatchOperation string
cancelBatchResult governance.RetryPublicationsBatchResult
cancelBatchReplayed bool
cancelBatchErr error
retryBatchReplayed bool
retryBatchErr error
requeueFailureRequest *governance.RequeueArtifactFetchRequest
requeueFailureOperation string
requeueFailureResult governance.RequeueArtifactFetchResult
requeueFailureReplayed bool
requeueFailureErr error
enqueued jobs.EnqueueRequest
enqueueResult *jobs.Job
// executorStatus 是封存执行器的"最近一次观测";nil 表示还没有任何上报。
executorStatus *libraryexecutor.Observation
listResult []jobs.Job
listType string
listByTypeResult []jobs.Job
listByTypeLimit int
listByTypeErr error
listLimit int
assetResult governance.SourceAssetPage
assetFilter governance.SourceAssetFilter
connectionResult governance.SourceConnectionPage
triggerScanResult governance.SourceScan
triggerScanConnectionID string
triggerScanOperationID string
triggerScanErr error
scanPageResult governance.SourceScanPage
scanPageConnectionID string
scanPageLimit int
scanPageOffset int
scanDetailResult governance.SourceScanDetailView
scanDetailErr error
scanDetailConnectionID string
scanDetailID string
reviewPageResult governance.ReviewCandidatePage
reviewFilter governance.ReviewCandidateFilter
reviewResult governance.PublicationReview
reviewRequest governance.ReviewPublicationRequest
createBatchResult governance.ReleaseBatch
createBatchRequest governance.CreateReleaseBatchRequest
auditChainID string
auditChain governance.ReleaseAuditChain
auditChainErr error
generateChangeSetRequest governance.GenerateReleaseChangeSetRequest
generateChangeSetOperationID string
generateChangeSetResult governance.ReleaseChangeSet
generateChangeSetReplayed bool
generateChangeSetErr error
confirmReleaseSetRequest governance.ConfirmReleaseChangeSetRequest
confirmReleaseSetOperationID string
confirmReleaseSetResult governance.ReleaseChangeSet
confirmReleaseSetReplayed bool
confirmReleaseSetErr error
listChangeSetsFilter governance.ReleaseBatchFilter
changeSetPage governance.ReleaseChangeSetPage
getChangeSetID string
changeSetDetail governance.ReleaseChangeSetDetail
changeSetDetailErr error
createBatchOperationID string
createBatchErr error
sealBatchResult governance.ReleaseBatch
sealBatchRequest governance.SealReleaseBatchRequest
sealBatchOperationID string
sealBatchErr error
batchPageResult governance.ReleaseBatchPage
batchFilter governance.ReleaseBatchFilter
batchDetailResult governance.ReleaseBatchDetail
batchID string
authorizeReleaseResult governance.ReleaseExecution
authorizeReleaseRequest governance.AuthorizeReleaseBatchRequest
authorizeReleaseOperationID string
authorizeReleaseErr error
authorizeReleaseCalled bool
authorizeRollbackResult governance.ReleaseExecution
authorizeRollbackRequest governance.AuthorizeReleaseRollbackRequest
authorizeRollbackOperationID string
authorizeRollbackErr error
retryReleaseResult governance.ReleaseExecution
retryReleaseRequest governance.RetryReleaseExecutionRequest
retryReleaseOperationID string
retryReleaseErr error
releaseExecutionPageResult governance.ReleaseExecutionPage
releaseExecutionFilter governance.ReleaseExecutionFilter
releaseExecutionDetailResult governance.ReleaseExecutionDetail
releaseExecutionID string
releasePlanResult governance.ReleasePlan
activeUISettingsResult governance.UISettings
candidateUISettingsResult governance.UISettings
activatedUISettingsResult governance.UISettings
revisionUISettingsResult governance.UISettings
createUISettingsRequest governance.CreateUISettingsCandidateRequest
createUISettingsOperationID string
createUISettingsErr error
activateUISettingsID string
activationRevisionID string
activateUISettingsOperationID string
activateUISettingsErr error
publicationResult governance.TestPublicationPage
publicationFilter governance.TestPublicationFilter
confirmChangeSetResult governance.ConfirmCatalogChangeSetResult
confirmChangeSetRequest governance.ConfirmCatalogChangeSetRequest
confirmChangeSetOperationID string
confirmChangeSetReplayed bool
confirmChangeSetErr error
startPublicationsResult governance.StartTestPublicationsResult
startPublicationsRequest governance.StartTestPublicationsRequest
startPublicationsOperationID string
startPublicationsReplayed bool
acceptanceResult []governance.AcceptanceRecord
acceptanceID string
retryErr error
retryID string
retryOperationID string
retryReplayed bool
scopeRules []governance.ScopeRule
scopeRuleConnectionID string
scopeRuleResult governance.ScopeRule
scopeRuleRequest governance.ScopeRuleMutation
scopeRuleOperationID string
scopeRuleReplayed bool
scopeRuleErr error
filenameFilters []governance.FilenameFilterRule
filenameFilterResult governance.FilenameFilterRule
filenameFilterRequest governance.FilenameFilterMutation
filenameFilterOperationID string
filenameFilterReplayed bool
filenameFilterErr error
catalogSources []governance.CatalogSource
catalogPage governance.CatalogNodePage
catalogFilter governance.CatalogNodeFilter
catalogErr error
catalogRouteRequest governance.CatalogRouteRequest
catalogRouteOperationID string
catalogRouteReplayed bool
catalogRouteResult governance.CatalogRouteResult
catalogRouteErr error
catalogRefreshSource governance.CatalogSource
catalogRefreshOperationID string
catalogRefreshReplayed bool
catalogRefreshResult jobs.Job
catalogRefreshErr error
snapshotPageResult governance.CatalogSnapshotPage
snapshotNodePageResult governance.CatalogNodePage
snapshotConnectionID string
snapshotNodeFilter governance.CatalogSnapshotNodeFilter
snapshotErr error
catalogChangeSetResult governance.CatalogChangeSetDetail
catalogChangeSetFilter governance.CatalogChangeSetFilter
}
func (s *fakeJobStore) ListSourceAssets(_ context.Context, filter governance.SourceAssetFilter) (governance.SourceAssetPage, error) {
s.assetFilter = filter
return s.assetResult, nil
}
func (s *fakeJobStore) ListSourceConnections(context.Context) (governance.SourceConnectionPage, error) {
return s.connectionResult, nil
}
func (s *fakeJobStore) CreateSourceSettingsRevisionWithOperation(_ context.Context, connectionID string, draft postgresstore.SourceSettingsDraft, operationID string, _ time.Time) (string, bool, error) {
s.createSettingsConnectionID = connectionID
s.createSettingsDraft = draft
s.createSettingsOperationID = operationID
if s.createSettingsErr != nil {
return "", false, s.createSettingsErr
}
return "scr-fake", s.createSettingsReplayed, nil
}
func (s *fakeJobStore) ActivateSourceSettingsRevisionWithOperation(_ context.Context, connectionID, revisionID, operationID string, _ time.Time) (bool, error) {
s.activateSettingsConnectionID = connectionID
s.activateSettingsRevisionID = revisionID
s.activateSettingsOperationID = operationID
return s.activateSettingsReplayed, s.activateSettingsErr
}
func (s *fakeJobStore) ListSourceConnectionRevisions(_ context.Context, connectionID string) ([]governance.SourceConnectionRevision, error) {
s.listSettingsConnectionID = connectionID
return s.settingsRevisions, s.listSettingsErr
}
func (s *fakeJobStore) TriggerPPSourceScan(_ context.Context, connectionID string, _ time.Time) (governance.SourceScan, error) {
s.triggerScanConnectionID = connectionID
return s.triggerScanResult, nil
}
func (s *fakeJobStore) TriggerPPSourceScanWithOperation(_ context.Context, connectionID string, _ time.Time, operationID string) (governance.SourceScan, bool, error) {
s.triggerScanConnectionID = connectionID
s.triggerScanOperationID = operationID
return s.triggerScanResult, false, s.triggerScanErr
}
func (s *fakeJobStore) ListSourceScans(_ context.Context, connectionID string, limit, offset int) (governance.SourceScanPage, error) {
s.scanPageConnectionID, s.scanPageLimit, s.scanPageOffset = connectionID, limit, offset
return s.scanPageResult, nil
}
func (s *fakeJobStore) GetSourceScanDetail(_ context.Context, connectionID, scanID string) (governance.SourceScanDetailView, error) {
s.scanDetailConnectionID, s.scanDetailID = connectionID, scanID
return s.scanDetailResult, s.scanDetailErr
}
func (s *fakeJobStore) ListReviewCandidates(_ context.Context, filter governance.ReviewCandidateFilter) (governance.ReviewCandidatePage, error) {
s.reviewFilter = filter
return s.reviewPageResult, nil
}
func (s *fakeJobStore) ReviewPublication(_ context.Context, request governance.ReviewPublicationRequest) (governance.PublicationReview, error) {
s.reviewRequest = request
return s.reviewResult, nil
}
func (s *fakeJobStore) ReviewPublicationWithOperation(_ context.Context, request governance.ReviewPublicationRequest, _ string) (governance.PublicationReview, bool, error) {
s.reviewRequest = request
return s.reviewResult, false, nil
}
func (s *fakeJobStore) CreateReleaseBatch(_ context.Context, request governance.CreateReleaseBatchRequest) (governance.ReleaseBatch, error) {
s.createBatchRequest = request
return s.createBatchResult, nil
}
func (s *fakeJobStore) CreateReleaseBatchWithOperation(_ context.Context, request governance.CreateReleaseBatchRequest, operationID string) (governance.ReleaseBatch, bool, error) {
s.createBatchRequest = request
s.createBatchOperationID = operationID
return s.createBatchResult, false, s.createBatchErr
}
func (s *fakeJobStore) GenerateReleaseChangeSetWithOperation(_ context.Context, request governance.GenerateReleaseChangeSetRequest, operationID string) (governance.ReleaseChangeSet, bool, error) {
s.generateChangeSetRequest = request
s.generateChangeSetOperationID = operationID
return s.generateChangeSetResult, s.generateChangeSetReplayed, s.generateChangeSetErr
}
func (s *fakeJobStore) ConfirmReleaseChangeSetWithOperation(_ context.Context, request governance.ConfirmReleaseChangeSetRequest, operationID string) (governance.ReleaseChangeSet, bool, error) {
s.confirmReleaseSetRequest = request
s.confirmReleaseSetOperationID = operationID
return s.confirmReleaseSetResult, s.confirmReleaseSetReplayed, s.confirmReleaseSetErr
}
func (s *fakeJobStore) ListReleaseChangeSets(_ context.Context, filter governance.ReleaseBatchFilter) (governance.ReleaseChangeSetPage, error) {
s.listChangeSetsFilter = filter
return s.changeSetPage, nil
}
func (s *fakeJobStore) GetReleaseChangeSet(_ context.Context, id string) (governance.ReleaseChangeSetDetail, error) {
s.getChangeSetID = id
return s.changeSetDetail, s.changeSetDetailErr
}
func (s *fakeJobStore) LatestReleaseChangeSet(_ context.Context) (governance.ReleaseChangeSetDetail, error) {
return s.changeSetDetail, s.changeSetDetailErr
}
func (s *fakeJobStore) SealReleaseBatch(_ context.Context, request governance.SealReleaseBatchRequest) (governance.ReleaseBatch, error) {
s.sealBatchRequest = request
return s.sealBatchResult, nil
}
func (s *fakeJobStore) SealReleaseBatchWithOperation(_ context.Context, request governance.SealReleaseBatchRequest, operationID string) (governance.ReleaseBatch, bool, error) {
s.sealBatchRequest = request
s.sealBatchOperationID = operationID
return s.sealBatchResult, false, s.sealBatchErr
}
func (s *fakeJobStore) ListReleaseBatches(_ context.Context, filter governance.ReleaseBatchFilter) (governance.ReleaseBatchPage, error) {
s.batchFilter = filter
return s.batchPageResult, nil
}
func (s *fakeJobStore) GetReleaseBatch(_ context.Context, batchID string) (governance.ReleaseBatchDetail, error) {
s.batchID = batchID
return s.batchDetailResult, nil
}
func (s *fakeJobStore) AuthorizeReleaseBatch(_ context.Context, request governance.AuthorizeReleaseBatchRequest) (governance.ReleaseExecution, error) {
s.authorizeReleaseCalled = true
s.authorizeReleaseRequest = request
return s.authorizeReleaseResult, nil
}
func (s *fakeJobStore) AuthorizeReleaseBatchWithOperation(_ context.Context, request governance.AuthorizeReleaseBatchRequest, operationID string) (governance.ReleaseExecution, bool, error) {
s.authorizeReleaseCalled = true
s.authorizeReleaseRequest = request
s.authorizeReleaseOperationID = operationID
return s.authorizeReleaseResult, false, s.authorizeReleaseErr
}
func (s *fakeJobStore) AuthorizeReleaseRollback(_ context.Context, request governance.AuthorizeReleaseRollbackRequest) (governance.ReleaseExecution, error) {
s.authorizeRollbackRequest = request
return s.authorizeRollbackResult, nil
}
func (s *fakeJobStore) AuthorizeReleaseRollbackWithOperation(_ context.Context, request governance.AuthorizeReleaseRollbackRequest, operationID string) (governance.ReleaseExecution, bool, error) {
s.authorizeRollbackRequest = request
s.authorizeRollbackOperationID = operationID
return s.authorizeRollbackResult, false, s.authorizeRollbackErr
}
func (s *fakeJobStore) RetryReleaseExecution(_ context.Context, request governance.RetryReleaseExecutionRequest) (governance.ReleaseExecution, error) {
s.retryReleaseRequest = request
return s.retryReleaseResult, nil
}
func (s *fakeJobStore) RetryReleaseExecutionWithOperation(_ context.Context, request governance.RetryReleaseExecutionRequest, operationID string) (governance.ReleaseExecution, bool, error) {
s.retryReleaseRequest = request
s.retryReleaseOperationID = operationID
return s.retryReleaseResult, false, s.retryReleaseErr
}
func (s *fakeJobStore) ListReleaseExecutions(_ context.Context, filter governance.ReleaseExecutionFilter) (governance.ReleaseExecutionPage, error) {
s.releaseExecutionFilter = filter
return s.releaseExecutionPageResult, nil
}
func (s *fakeJobStore) GetReleaseAuditChain(_ context.Context, executionID string) (governance.ReleaseAuditChain, error) {
s.auditChainID = executionID
return s.auditChain, s.auditChainErr
}
func (s *fakeJobStore) GetReleaseExecution(_ context.Context, executionID string) (governance.ReleaseExecutionDetail, error) {
s.releaseExecutionID = executionID
return s.releaseExecutionDetailResult, nil
}
func (s *fakeJobStore) ListReleasePlan(context.Context) (governance.ReleasePlan, error) {
return s.releasePlanResult, nil
}
func (s *fakeJobStore) GetActiveUISettings(context.Context) (governance.UISettings, error) {
return s.activeUISettingsResult, nil
}
func (s *fakeJobStore) GetUISettingsRevision(_ context.Context, revisionID string) (governance.UISettings, error) {
s.activationRevisionID = revisionID
return s.revisionUISettingsResult, nil
}
func (s *fakeJobStore) CreateUISettingsCandidate(_ context.Context, request governance.CreateUISettingsCandidateRequest) (governance.UISettings, error) {
s.createUISettingsRequest = request
return s.candidateUISettingsResult, nil
}
func (s *fakeJobStore) CreateUISettingsCandidateWithOperation(_ context.Context, request governance.CreateUISettingsCandidateRequest, operationID string) (governance.UISettings, bool, error) {
s.createUISettingsRequest = request
s.createUISettingsOperationID = operationID
return s.candidateUISettingsResult, false, s.createUISettingsErr
}
func (s *fakeJobStore) ActivateUISettings(_ context.Context, revisionID string, _ time.Time) (governance.UISettings, error) {
s.activateUISettingsID = revisionID
return s.activatedUISettingsResult, nil
}
func (s *fakeJobStore) ActivateUISettingsWithOperation(_ context.Context, revisionID string, _ time.Time, operationID string) (governance.UISettings, bool, error) {
s.activateUISettingsID = revisionID
s.activateUISettingsOperationID = operationID
return s.activatedUISettingsResult, false, s.activateUISettingsErr
}
func (s *fakeJobStore) Enqueue(_ context.Context, request jobs.EnqueueRequest, _ time.Time) (*jobs.Job, error) {
s.enqueued = request
if s.enqueueEchoRequestedID {
// The real store returns the stored row, including its payload: a reused
// idempotency key must be judged by that payload, not by the request.
return &jobs.Job{ID: request.ID, Type: request.Type, State: jobs.StatePending, Payload: request.Payload}, nil
}
return s.enqueueResult, nil
}
func (s *fakeJobStore) List(_ context.Context, limit int) ([]jobs.Job, error) {
s.listLimit = limit
return s.listResult, nil
}
func (s *fakeJobStore) ListByType(_ context.Context, jobType string, limit int) ([]jobs.Job, error) {
s.listType = jobType
s.listByTypeLimit = limit
return s.listByTypeResult, s.listByTypeErr
}
func (s *fakeJobStore) ListTestPublications(_ context.Context, filter governance.TestPublicationFilter) (governance.TestPublicationPage, error) {
s.publicationFilter = filter
return s.publicationResult, nil
}
func (s *fakeJobStore) ConfirmCatalogChangeSetWithOperation(_ context.Context, request governance.ConfirmCatalogChangeSetRequest, operationID string) (governance.ConfirmCatalogChangeSetResult, bool, error) {
s.confirmChangeSetRequest = request
s.confirmChangeSetOperationID = operationID
return s.confirmChangeSetResult, s.confirmChangeSetReplayed, s.confirmChangeSetErr
}
func (s *fakeJobStore) StartTestPublications(_ context.Context, request governance.StartTestPublicationsRequest) (governance.StartTestPublicationsResult, error) {
s.startPublicationsRequest = request
return s.startPublicationsResult, nil
}
func (s *fakeJobStore) StartTestPublicationsWithOperation(_ context.Context, request governance.StartTestPublicationsRequest, operationID string) (governance.StartTestPublicationsResult, bool, error) {
s.startPublicationsRequest = request
s.startPublicationsOperationID = operationID
return s.startPublicationsResult, s.startPublicationsReplayed, nil
}
func (s *fakeJobStore) ListAcceptanceResults(_ context.Context, id string) ([]governance.AcceptanceRecord, error) {
s.acceptanceID = id
return s.acceptanceResult, nil
}
func (s *fakeJobStore) RetryPublication(_ context.Context, id string, _ time.Time) error {
s.retryID = id
return s.retryErr
}
func (s *fakeJobStore) RetryPublicationWithOperation(_ context.Context, id string, _ time.Time, operationID string) (bool, error) {
s.retryID = id
s.retryOperationID = operationID
return s.retryReplayed, s.retryErr
}
func (s *fakeJobStore) ListJobFailures(_ context.Context, filter governance.JobFailureFilter) (governance.JobFailurePage, error) {
s.jobFailureFilter = filter
return s.jobFailurePage, s.jobFailureErr
}
func (s *fakeJobStore) RetryPublicationsBatchWithOperation(_ context.Context, request governance.RetryPublicationsBatchRequest, operationID string) (governance.RetryPublicationsBatchResult, bool, error) {
s.retryBatchRequest = &request
s.retryBatchOperation = operationID
return s.retryBatchResult, s.retryBatchReplayed, s.retryBatchErr
}
func (s *fakeJobStore) CancelPublicationsBatchWithOperation(_ context.Context, request governance.CancelPublicationsBatchRequest, operationID string) (governance.RetryPublicationsBatchResult, bool, error) {
s.cancelBatchRequest = &request
s.cancelBatchOperation = operationID
return s.cancelBatchResult, s.cancelBatchReplayed, s.cancelBatchErr
}
func (s *fakeJobStore) RequeueArtifactFetchWithOperation(_ context.Context, request governance.RequeueArtifactFetchRequest, operationID string) (governance.RequeueArtifactFetchResult, bool, error) {
s.requeueFailureRequest = &request
s.requeueFailureOperation = operationID
return s.requeueFailureResult, s.requeueFailureReplayed, s.requeueFailureErr
}
func (s *fakeJobStore) ListScopeRules(_ context.Context, connectionID string) ([]governance.ScopeRule, error) {
s.scopeRuleConnectionID = connectionID
return s.scopeRules, s.scopeRuleErr
}
func (s *fakeJobStore) MutateScopeRule(_ context.Context, request governance.ScopeRuleMutation) (governance.ScopeRule, error) {
s.scopeRuleRequest = request
return s.scopeRuleResult, s.scopeRuleErr
}
func (s *fakeJobStore) MutateScopeRuleWithOperation(_ context.Context, request governance.ScopeRuleMutation, operationID string) (governance.ScopeRule, bool, error) {
s.scopeRuleRequest = request
s.scopeRuleOperationID = operationID
return s.scopeRuleResult, s.scopeRuleReplayed, s.scopeRuleErr
}
func (s *fakeJobStore) ListFilenameFilterRules(_ context.Context, _ string) ([]governance.FilenameFilterRule, error) {
return s.filenameFilters, s.filenameFilterErr
}
func (s *fakeJobStore) MutateFilenameFilterRuleWithOperation(_ context.Context, request governance.FilenameFilterMutation, operationID string) (governance.FilenameFilterRule, bool, error) {
s.filenameFilterRequest = request
s.filenameFilterOperationID = operationID
return s.filenameFilterResult, s.filenameFilterReplayed, s.filenameFilterErr
}
func (s *fakeJobStore) ListCatalogSources(context.Context) ([]governance.CatalogSource, error) {
return s.catalogSources, s.catalogErr
}
func (s *fakeJobStore) ListCatalogNodes(_ context.Context, filter governance.CatalogNodeFilter) (governance.CatalogNodePage, error) {
s.catalogFilter = filter
return s.catalogPage, s.catalogErr
}
func (s *fakeJobStore) ListCatalogSnapshots(_ context.Context, connectionID string, limit, offset int) (governance.CatalogSnapshotPage, error) {
s.snapshotConnectionID = connectionID
return s.snapshotPageResult, s.snapshotErr
}
func (s *fakeJobStore) ListCatalogNodesInSnapshot(_ context.Context, filter governance.CatalogSnapshotNodeFilter) (governance.CatalogNodePage, error) {
s.snapshotNodeFilter = filter
return s.snapshotNodePageResult, s.snapshotErr
}
func (s *fakeJobStore) RouteCatalogFiles(_ context.Context, request governance.CatalogRouteRequest) (governance.CatalogRouteResult, error) {
s.catalogRouteRequest = request
return s.catalogRouteResult, s.catalogRouteErr
}
func (s *fakeJobStore) RouteCatalogFilesWithOperation(_ context.Context, request governance.CatalogRouteRequest, operationID string) (governance.CatalogRouteResult, bool, error) {
s.catalogRouteRequest = request
s.catalogRouteOperationID = operationID
return s.catalogRouteResult, s.catalogRouteReplayed, s.catalogRouteErr
}
func (s *fakeJobStore) EnqueueCatalogRefreshWithOperation(_ context.Context, source governance.CatalogSource, operationID string, _ time.Time) (jobs.Job, bool, error) {
s.catalogRefreshSource = source
s.catalogRefreshOperationID = operationID
return s.catalogRefreshResult, s.catalogRefreshReplayed, s.catalogRefreshErr
}
func (s *fakeJobStore) GetLatestCatalogChangeSet(_ context.Context, filter governance.CatalogChangeSetFilter) (governance.CatalogChangeSetDetail, error) {
s.catalogChangeSetFilter = filter
return s.catalogChangeSetResult, nil
}
type fakeMonitor struct {
overview governance.MonitorOverview
files governance.KnowledgeMonitorFilePage
knowledgeBaseID string
status string
metrics governance.RunMetrics
metricsErr error
services []governance.ServiceHealth
}
func (monitor *fakeMonitor) Overview(context.Context) (governance.MonitorOverview, error) {
return monitor.overview, nil
}
func (monitor *fakeMonitor) Services(context.Context) []governance.ServiceHealth {
return monitor.services
}
func (monitor *fakeMonitor) Metrics(context.Context) (governance.RunMetrics, error) {
return monitor.metrics, monitor.metricsErr
}
func (monitor *fakeMonitor) ListFiles(_ context.Context, knowledgeBaseID, status string, _, _ int) (governance.KnowledgeMonitorFilePage, error) {
monitor.knowledgeBaseID = knowledgeBaseID
monitor.status = status
return monitor.files, nil
}