姚章浩
8febe12d7d
真实本机 PG/ES/MinIO：v071 目录镜像 10 模式、结构引用/游标 15 模式、 8 表×2 外部范围身份矩阵、13 模式自然键（含软删除与合法同 parent 正例）、 备份期间三注入、收尾故障与授权撤销两个真实治理 PG 用例。 需隔离服务的用例在 test-scope.json 显式声明：普通 CI 跳过不算通过。
68 lines
3.0 KiB
Go
68 lines
3.0 KiB
Go
package httpapi
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"knowledge-governance/internal/governance"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type legacySourceAPIStore struct {
|
|
fakeLibraryStore
|
|
calls int
|
|
actor string
|
|
source governance.LibrarySourceIdentity
|
|
err error
|
|
}
|
|
|
|
func (s *legacySourceAPIStore) BindLegacyLibrarySource(_ context.Context, _ string, _ string, source governance.LibrarySourceIdentity, actor string, _ time.Time) error {
|
|
s.calls++
|
|
s.actor = actor
|
|
s.source = source
|
|
return s.err
|
|
}
|
|
func TestContractLegacySourceAssociationAPI(t *testing.T) {
|
|
for _, role := range []string{roleViewer, roleEngineer, roleAdmin} {
|
|
store := &legacySourceAPIStore{}
|
|
config := Config{Store: store, LibrarySourceProject: "configured-project", LibrarySourceApp: "configured-app", LibrarySourceTenantID: 10000, LibrarySourceInternalKB: "configured-kb"}
|
|
request := httptest.NewRequest(http.MethodPost, "/api/v1/library/source-associations", strings.NewReader(`{"scope":"internal","snapshot_id":"exact-snapshot"}`))
|
|
request = request.WithContext(context.WithValue(request.Context(), sessionKey{}, sessionIdentity{Username: "real-session-actor", Role: role}))
|
|
response := httptest.NewRecorder()
|
|
libraryLegacySourceBind(response, request, config)
|
|
if role == roleAdmin {
|
|
if response.Code != 200 || store.calls != 1 || store.actor != "real-session-actor" || store.source.Project != "configured-project" || store.source.KnowledgeBaseID != "configured-kb" {
|
|
t.Fatal("association contract", response.Code, response.Body.String())
|
|
}
|
|
} else if response.Code != 403 || store.calls != 0 {
|
|
t.Fatal("non-admin associated source")
|
|
}
|
|
}
|
|
if classifyRole("POST", "/api/v1/library/source-associations") != roleAdmin {
|
|
t.Fatal("route is not administrator only")
|
|
}
|
|
}
|
|
func TestContractLegacySourceAssociationAPIRejectsSpoofAndEvidenceFailure(t *testing.T) {
|
|
for _, body := range []string{`{"scope":"internal","snapshot_id":"x","actor":"forged"}`, `{"scope":"internal","snapshot_id":"x","source_identity":{"project":"forged"}}`, `{"scope":"internal","snapshot_id":"x"} {}`} {
|
|
store := &legacySourceAPIStore{}
|
|
req := httptest.NewRequest("POST", "/", strings.NewReader(body))
|
|
req = req.WithContext(context.WithValue(req.Context(), sessionKey{}, sessionIdentity{Username: "admin", Role: roleAdmin}))
|
|
out := httptest.NewRecorder()
|
|
libraryLegacySourceBind(out, req, Config{Store: store})
|
|
if out.Code != 400 || store.calls != 0 {
|
|
t.Fatal("spoofed or trailing payload accepted")
|
|
}
|
|
}
|
|
store := &legacySourceAPIStore{err: errors.New("original proof mismatch")}
|
|
req := httptest.NewRequest("POST", "/", strings.NewReader(`{"scope":"internal","snapshot_id":"x"}`))
|
|
req = req.WithContext(context.WithValue(req.Context(), sessionKey{}, sessionIdentity{Username: "admin", Role: roleAdmin}))
|
|
out := httptest.NewRecorder()
|
|
libraryLegacySourceBind(out, req, Config{Store: store, LibrarySourceProject: "p", LibrarySourceApp: "app", LibrarySourceTenantID: 1, LibrarySourceInternalKB: "kb"})
|
|
if out.Code != 409 {
|
|
t.Fatal("unproven association reported success")
|
|
}
|
|
}
|