zheromiku
dcdc8d1ddd
全栈知识治理平台首次入库（P1-P10 + 发布链路模拟验证闭环）： - 治理核心：源目录/变更集/人工审核/发布批次/发布执行（fail-closed 闸口） - 测试通道：WeKnora 测试 KB 上传/轮询/验收；poll 完成时写入 parser_generation/index_generation，修复 manual/seed 路径发布闸口永久拦截 - 发布适配器 v071（本地 source=测试角色 / 云端 target=生产角色 模拟验证通过）： * ES 导出/导入 mapping 自适应（.keyword 子字段、可选 tenant_id） * OSS locator 支持嵌套 scheme；chunks.seq_id 由目标序列分配 * bundle 归档同身份重试原子替换；verifier 按 bundle chunk_type 计数 * 回滚 job payload 支持 prior_execution_id（strict 解码） - 三通道发布+回滚演练实测：publish 4/4 verified，检索命中， rollback 4/4 rolled_back，目标端 PG is_enabled=false、ES is_enabled=false - 前端 A·Data-console 风格全站统一，120 用例通过 - 门禁：go vet / go test 全包通过，gofmt 干净
396 lines
15 KiB
Go
396 lines
15 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
|
|
"knowledge-governance/internal/governance"
|
|
)
|
|
|
|
func TestTriggerPPSourceScanAtomicallyEnqueuesScanJob(t *testing.T) {
|
|
now := time.Date(2026, 9, 6, 4, 0, 0, 0, time.UTC)
|
|
db := &recordingDB{row: sourceScanRow{id: "scan-fixed", connectionID: "conn-pp", sourceType: "pp", state: "pending", createdAt: now}}
|
|
store := NewStore(db)
|
|
|
|
scan, err := store.TriggerPPSourceScan(context.Background(), "conn-pp", now)
|
|
if err != nil {
|
|
t.Fatalf("TriggerPPSourceScan() error = %v", err)
|
|
}
|
|
if scan.ID != "scan-fixed" || scan.SourceConnectionID != "conn-pp" || scan.SourceType != governance.SourceTypePP || scan.State != governance.SourceScanPending {
|
|
t.Fatalf("scan = %#v", scan)
|
|
}
|
|
sql := strings.ToLower(db.queryRowSQL)
|
|
for _, required := range []string{
|
|
"insert into source_scans",
|
|
"insert into jobs",
|
|
"'source.scan.pp'",
|
|
"'pending'",
|
|
"not exists",
|
|
"returning",
|
|
} {
|
|
if !strings.Contains(sql, required) {
|
|
t.Errorf("trigger SQL missing %q", required)
|
|
}
|
|
}
|
|
if strings.Index(sql, "insert into source_scans") > strings.Index(sql, "insert into jobs") {
|
|
t.Fatal("scan must be created before its job")
|
|
}
|
|
if len(db.queryRowArgs) < 4 {
|
|
t.Fatalf("query args = %#v", db.queryRowArgs)
|
|
}
|
|
payload, ok := db.queryRowArgs[2].([]byte)
|
|
if !ok {
|
|
t.Fatalf("payload arg type = %T", db.queryRowArgs[2])
|
|
}
|
|
var decoded map[string]string
|
|
if err := json.Unmarshal(payload, &decoded); err != nil || decoded["scan_id"] == "" {
|
|
t.Fatalf("payload = %s err=%v", payload, err)
|
|
}
|
|
}
|
|
|
|
func TestTriggerPPSourceScanDistinguishesAlreadyActive(t *testing.T) {
|
|
db := &sourceScanSequenceDB{
|
|
recordingDB: &recordingDB{},
|
|
rows: []pgx.Row{sourceScanErrorRow{err: pgx.ErrNoRows}, sourceScanActiveRow{sourceType: "pp", active: true}},
|
|
}
|
|
_, err := NewStore(db).TriggerPPSourceScan(context.Background(), "conn-pp", time.Now())
|
|
if !errors.Is(err, ErrSourceScanAlreadyActive) {
|
|
t.Fatalf("TriggerPPSourceScan() error = %v, want ErrSourceScanAlreadyActive", err)
|
|
}
|
|
}
|
|
|
|
func TestStartSourceScanTransitionsOrReturnsCompletedForRecovery(t *testing.T) {
|
|
now := time.Date(2026, 9, 6, 4, 10, 0, 0, time.UTC)
|
|
db := &recordingDB{row: sourceScanRow{id: "scan-1", connectionID: "conn-pp", sourceType: "pp", state: "running", createdAt: now.Add(-time.Minute)}}
|
|
scan, err := NewStore(db).StartSourceScan(context.Background(), "scan-1", now)
|
|
if err != nil {
|
|
t.Fatalf("StartSourceScan() error = %v", err)
|
|
}
|
|
if scan.State != governance.SourceScanRunning {
|
|
t.Fatalf("state = %q", scan.State)
|
|
}
|
|
sql := strings.ToLower(db.queryRowSQL)
|
|
for _, required := range []string{"update source_scans", "state in ('pending', 'running')", "coalesce(started_at", "state = 'completed'", "union all"} {
|
|
if !strings.Contains(sql, required) {
|
|
t.Errorf("start SQL missing %q", required)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestStoreLoadsOnlyPPStableIdentityAndRevision(t *testing.T) {
|
|
db := &recordingDB{rows: &ppCurrentRows{values: [][2]string{{"source-a", "revision-a"}, {"source-b", "revision-b"}}}}
|
|
current, err := NewStore(db).LoadPPCurrentItems(context.Background(), "conn-pp")
|
|
if err != nil {
|
|
t.Fatalf("LoadPPCurrentItems() error = %v", err)
|
|
}
|
|
if len(current) != 2 || current["source-a"].Revision != "revision-a" || current["source-b"].Revision != "revision-b" {
|
|
t.Fatalf("current = %#v", current)
|
|
}
|
|
sql := strings.ToLower(db.querySQL)
|
|
if !strings.Contains(sql, "select stable_external_id, revision") || strings.Contains(sql, "normalized_path") || strings.Contains(sql, "raw_record_summary") {
|
|
t.Fatalf("unexpected PP current query: %s", sql)
|
|
}
|
|
}
|
|
|
|
func TestCompleteSourceScanAtomicallyPersistsChangesAndHealth(t *testing.T) {
|
|
now := time.Date(2026, 9, 6, 4, 20, 0, 0, time.UTC)
|
|
db := &recordingDB{row: sourceScanCompletionRow{completed: 1, changes: 2, health: 1}}
|
|
result := governance.SourceScanResult{
|
|
SnapshotSHA256: strings.Repeat("f", 64), ObservedItems: 613, UnchangedItems: 558,
|
|
UpdatedItems: 1, OverLimitItems: 54, MissingItems: 1, PlannedUploads: 1,
|
|
Changes: []governance.SourceScanChange{
|
|
{StableExternalID: strings.Repeat("a", 64), ChangeType: governance.SourceScanUpdated, PreviousRevision: "old", ObservedRevision: "new", ObservedSizeBytes: scanInt64Pointer(10)},
|
|
{StableExternalID: strings.Repeat("b", 64), ChangeType: governance.SourceScanUnavailable, PreviousRevision: "old"},
|
|
},
|
|
}
|
|
if err := NewStore(db).CompleteSourceScan(context.Background(), "scan-1", result, now); err != nil {
|
|
t.Fatalf("CompleteSourceScan() error = %v", err)
|
|
}
|
|
sql := strings.ToLower(db.queryRowSQL)
|
|
for _, required := range []string{
|
|
"update source_scans", "state = 'running'", "state = 'completed'",
|
|
"insert into source_scan_changes", "jsonb_to_recordset", "left join source_items",
|
|
"update source_connections", "jsonb_build_object", "'healthy'",
|
|
} {
|
|
if !strings.Contains(sql, required) {
|
|
t.Errorf("complete SQL missing %q", required)
|
|
}
|
|
}
|
|
for _, forbidden := range []string{"normalized_path", "raw_record_summary", "base_url", "secret_ref"} {
|
|
if strings.Contains(sql, forbidden) {
|
|
t.Errorf("complete SQL includes forbidden field %q", forbidden)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestFailSourceScanStoresOnlyFixedCodeAndUnhealthyState(t *testing.T) {
|
|
now := time.Date(2026, 9, 6, 4, 30, 0, 0, time.UTC)
|
|
db := &recordingDB{row: sourceScanFailureRow{failed: 1, health: 1}}
|
|
if err := NewStore(db).FailSourceScan(context.Background(), "scan-1", "pp_snapshot_invalid", now); err != nil {
|
|
t.Fatalf("FailSourceScan() error = %v", err)
|
|
}
|
|
sql := strings.ToLower(db.queryRowSQL)
|
|
for _, required := range []string{"update source_scans", "state = 'running'", "state = 'failed'", "error_code = $2", "update source_connections", "'unhealthy'"} {
|
|
if !strings.Contains(sql, required) {
|
|
t.Errorf("failure SQL missing %q", required)
|
|
}
|
|
}
|
|
if len(db.queryRowArgs) != 3 || db.queryRowArgs[1] != "pp_snapshot_invalid" {
|
|
t.Fatalf("failure args = %#v", db.queryRowArgs)
|
|
}
|
|
}
|
|
|
|
func TestListSourceScansReturnsOnlySafeSummaryColumns(t *testing.T) {
|
|
now := time.Date(2026, 9, 6, 6, 0, 0, 0, time.UTC)
|
|
db := &recordingDB{rows: &sourceScanSummaryRows{values: []sourceScanSummaryRow{{
|
|
id: "scan-safe", state: "completed", observed: 613, unchanged: 559,
|
|
overLimit: 54, createdAt: now.Add(-time.Minute), completedAt: &now,
|
|
}}}}
|
|
page, err := NewStore(db).ListSourceScans(context.Background(), "conn-pp", 20, 0)
|
|
if err != nil {
|
|
t.Fatalf("ListSourceScans() error = %v", err)
|
|
}
|
|
if page.Total != 1 || len(page.Items) != 1 || page.Items[0].ID != "scan-safe" || page.Items[0].OverLimitItems != 54 {
|
|
t.Fatalf("page = %#v", page)
|
|
}
|
|
sql := strings.ToLower(db.querySQL)
|
|
for _, required := range []string{"from source_scans", "source_connection_id = $1", "count(*) over()", "limit $2 offset $3"} {
|
|
if !strings.Contains(sql, required) {
|
|
t.Errorf("list SQL missing %q", required)
|
|
}
|
|
}
|
|
for _, forbidden := range []string{"snapshot_sha256", "stable_external_id", "previous_revision", "observed_revision"} {
|
|
if strings.Contains(sql, forbidden) {
|
|
t.Errorf("list SQL includes forbidden field %q", forbidden)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestGetSourceScanDetailReturnsSafeChangeProjection(t *testing.T) {
|
|
now := time.Date(2026, 9, 6, 6, 10, 0, 0, time.UTC)
|
|
db := &sourceScanReadDB{
|
|
recordingDB: &recordingDB{},
|
|
row: sourceScanDetailRow{id: "scan-safe", state: "completed", observed: 2, added: 1, missing: 1, plannedUploads: 1, createdAt: now, completedAt: &now},
|
|
rows: &sourceScanChangeRows{values: []sourceScanChangeRow{
|
|
{id: "chg-1", sourceItemID: scanStringPointer("si-safe"), changeType: "updated", observedSize: scanInt64Pointer(42), createdAt: now},
|
|
{id: "chg-2", changeType: "unavailable", createdAt: now},
|
|
}},
|
|
}
|
|
detail, err := NewStore(db).GetSourceScanDetail(context.Background(), "conn-pp", "scan-safe")
|
|
if err != nil {
|
|
t.Fatalf("GetSourceScanDetail() error = %v", err)
|
|
}
|
|
if detail.ID != "scan-safe" || len(detail.Changes) != 2 || detail.Changes[0].SourceItemID == nil || *detail.Changes[0].SourceItemID != "si-safe" {
|
|
t.Fatalf("detail = %#v", detail)
|
|
}
|
|
joined := strings.ToLower(db.queryRowSQL + "\n" + db.querySQL)
|
|
for _, required := range []string{"source_connection_id = $1", "id = $2", "from source_scan_changes", "source_scan_id = $2"} {
|
|
if !strings.Contains(joined, required) {
|
|
t.Errorf("detail SQL missing %q", required)
|
|
}
|
|
}
|
|
for _, forbidden := range []string{"snapshot_sha256", "stable_external_id", "previous_revision", "observed_revision"} {
|
|
if strings.Contains(joined, forbidden) {
|
|
t.Errorf("detail SQL includes forbidden field %q", forbidden)
|
|
}
|
|
}
|
|
}
|
|
|
|
func scanInt64Pointer(value int64) *int64 { return &value }
|
|
func scanStringPointer(value string) *string { return &value }
|
|
|
|
type sourceScanSequenceDB struct {
|
|
*recordingDB
|
|
rows []pgx.Row
|
|
index int
|
|
}
|
|
|
|
func (d *sourceScanSequenceDB) QueryRow(_ context.Context, _ string, _ ...any) pgx.Row {
|
|
row := d.rows[d.index]
|
|
d.index++
|
|
return row
|
|
}
|
|
|
|
type sourceScanErrorRow struct{ err error }
|
|
|
|
func (r sourceScanErrorRow) Scan(...any) error { return r.err }
|
|
|
|
type sourceScanActiveRow struct {
|
|
sourceType string
|
|
active bool
|
|
}
|
|
|
|
func (r sourceScanActiveRow) Scan(dest ...any) error {
|
|
*dest[0].(*string) = r.sourceType
|
|
*dest[1].(*bool) = r.active
|
|
return nil
|
|
}
|
|
|
|
type sourceScanRow struct {
|
|
id string
|
|
connectionID string
|
|
sourceType string
|
|
state string
|
|
createdAt time.Time
|
|
}
|
|
|
|
func (r sourceScanRow) Scan(dest ...any) error {
|
|
*dest[0].(*string) = r.id
|
|
*dest[1].(*string) = r.connectionID
|
|
*dest[2].(*string) = r.sourceType
|
|
*dest[3].(*string) = r.state
|
|
*dest[4].(*time.Time) = r.createdAt
|
|
return nil
|
|
}
|
|
|
|
type sourceScanCompletionRow struct{ completed, changes, health int }
|
|
|
|
func (r sourceScanCompletionRow) Scan(dest ...any) error {
|
|
*dest[0].(*int) = r.completed
|
|
*dest[1].(*int) = r.changes
|
|
*dest[2].(*int) = r.health
|
|
return nil
|
|
}
|
|
|
|
type sourceScanFailureRow struct{ failed, health int }
|
|
|
|
func (r sourceScanFailureRow) Scan(dest ...any) error {
|
|
*dest[0].(*int) = r.failed
|
|
*dest[1].(*int) = r.health
|
|
return nil
|
|
}
|
|
|
|
type ppCurrentRows struct {
|
|
values [][2]string
|
|
index int
|
|
}
|
|
|
|
func (r *ppCurrentRows) Close() {}
|
|
func (r *ppCurrentRows) Err() error { return nil }
|
|
func (r *ppCurrentRows) CommandTag() pgconn.CommandTag { return pgconn.CommandTag{} }
|
|
func (r *ppCurrentRows) FieldDescriptions() []pgconn.FieldDescription { return nil }
|
|
func (r *ppCurrentRows) RawValues() [][]byte { return nil }
|
|
func (r *ppCurrentRows) Conn() *pgx.Conn { return nil }
|
|
func (r *ppCurrentRows) Next() bool {
|
|
if r.index >= len(r.values) {
|
|
return false
|
|
}
|
|
r.index++
|
|
return true
|
|
}
|
|
func (r *ppCurrentRows) Scan(dest ...any) error {
|
|
if r.index == 0 || r.index > len(r.values) {
|
|
return errors.New("scan outside row")
|
|
}
|
|
*dest[0].(*string) = r.values[r.index-1][0]
|
|
*dest[1].(*string) = r.values[r.index-1][1]
|
|
return nil
|
|
}
|
|
func (r *ppCurrentRows) Values() ([]any, error) { return nil, nil }
|
|
|
|
type sourceScanSummaryRow struct {
|
|
id, state string
|
|
observed, unchanged, added, updated, overLimit, missing int
|
|
plannedUploads int
|
|
errorCode string
|
|
createdAt time.Time
|
|
completedAt *time.Time
|
|
}
|
|
|
|
type sourceScanSummaryRows struct {
|
|
values []sourceScanSummaryRow
|
|
index int
|
|
}
|
|
|
|
func (r *sourceScanSummaryRows) Close() {}
|
|
func (r *sourceScanSummaryRows) Err() error { return nil }
|
|
func (r *sourceScanSummaryRows) CommandTag() pgconn.CommandTag { return pgconn.CommandTag{} }
|
|
func (r *sourceScanSummaryRows) FieldDescriptions() []pgconn.FieldDescription { return nil }
|
|
func (r *sourceScanSummaryRows) RawValues() [][]byte { return nil }
|
|
func (r *sourceScanSummaryRows) Conn() *pgx.Conn { return nil }
|
|
func (r *sourceScanSummaryRows) Next() bool {
|
|
if r.index >= len(r.values) {
|
|
return false
|
|
}
|
|
r.index++
|
|
return true
|
|
}
|
|
func (r *sourceScanSummaryRows) Scan(dest ...any) error {
|
|
v := r.values[r.index-1]
|
|
*dest[0].(*string), *dest[1].(*string) = v.id, v.state
|
|
*dest[2].(*int), *dest[3].(*int), *dest[4].(*int), *dest[5].(*int) = v.observed, v.unchanged, v.added, v.updated
|
|
*dest[6].(*int), *dest[7].(*int), *dest[8].(*int) = v.overLimit, v.missing, v.plannedUploads
|
|
*dest[9].(*string), *dest[10].(*time.Time), *dest[11].(**time.Time) = v.errorCode, v.createdAt, v.completedAt
|
|
*dest[12].(*int) = len(r.values)
|
|
return nil
|
|
}
|
|
func (r *sourceScanSummaryRows) Values() ([]any, error) { return nil, nil }
|
|
|
|
type sourceScanDetailRow sourceScanSummaryRow
|
|
|
|
func (r sourceScanDetailRow) Scan(dest ...any) error {
|
|
v := sourceScanSummaryRow(r)
|
|
*dest[0].(*string), *dest[1].(*string) = v.id, v.state
|
|
*dest[2].(*int), *dest[3].(*int), *dest[4].(*int), *dest[5].(*int) = v.observed, v.unchanged, v.added, v.updated
|
|
*dest[6].(*int), *dest[7].(*int), *dest[8].(*int) = v.overLimit, v.missing, v.plannedUploads
|
|
*dest[9].(*string), *dest[10].(*time.Time), *dest[11].(**time.Time) = v.errorCode, v.createdAt, v.completedAt
|
|
return nil
|
|
}
|
|
|
|
type sourceScanChangeRow struct {
|
|
id string
|
|
sourceItemID *string
|
|
changeType string
|
|
observedSize *int64
|
|
createdAt time.Time
|
|
}
|
|
|
|
type sourceScanChangeRows struct {
|
|
values []sourceScanChangeRow
|
|
index int
|
|
}
|
|
|
|
func (r *sourceScanChangeRows) Close() {}
|
|
func (r *sourceScanChangeRows) Err() error { return nil }
|
|
func (r *sourceScanChangeRows) CommandTag() pgconn.CommandTag { return pgconn.CommandTag{} }
|
|
func (r *sourceScanChangeRows) FieldDescriptions() []pgconn.FieldDescription { return nil }
|
|
func (r *sourceScanChangeRows) RawValues() [][]byte { return nil }
|
|
func (r *sourceScanChangeRows) Conn() *pgx.Conn { return nil }
|
|
func (r *sourceScanChangeRows) Next() bool {
|
|
if r.index >= len(r.values) {
|
|
return false
|
|
}
|
|
r.index++
|
|
return true
|
|
}
|
|
func (r *sourceScanChangeRows) Scan(dest ...any) error {
|
|
v := r.values[r.index-1]
|
|
*dest[0].(*string), *dest[1].(**string), *dest[2].(*string) = v.id, v.sourceItemID, v.changeType
|
|
*dest[3].(**int64), *dest[4].(*time.Time) = v.observedSize, v.createdAt
|
|
return nil
|
|
}
|
|
func (r *sourceScanChangeRows) Values() ([]any, error) { return nil, nil }
|
|
|
|
type sourceScanReadDB struct {
|
|
*recordingDB
|
|
row pgx.Row
|
|
rows pgx.Rows
|
|
}
|
|
|
|
func (d *sourceScanReadDB) QueryRow(_ context.Context, sql string, args ...any) pgx.Row {
|
|
d.queryRowSQL, d.queryRowArgs = sql, args
|
|
return d.row
|
|
}
|
|
|
|
func (d *sourceScanReadDB) Query(_ context.Context, sql string, args ...any) (pgx.Rows, error) {
|
|
d.querySQL, d.queryArgs = sql, args
|
|
return d.rows, nil
|
|
}
|