151 lines
6.4 KiB
Go
151 lines
6.4 KiB
Go
package integration
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/apperror"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/model"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/repository"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/secret"
|
|
"git.sundynix.cn/Blizzard/Veyra/internal/service"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
func testCandidateAI(t *testing.T, ctx context.Context, pool *pgxpool.Pool, admin model.User) {
|
|
// Start with isolated candidates to avoid consuming other scenario fixtures.
|
|
_, err := pool.Exec(ctx, `UPDATE veyra_candidate_revision SET valid_until=now()-interval '1 second' WHERE status='PENDING_REVIEW'`)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
providers := repository.NewProviderRepository(pool)
|
|
configs := map[string]model.ProviderConfig{}
|
|
for _, key := range []string{"model_primary", "model_review"} {
|
|
previous, e := providers.Get(ctx, key)
|
|
if e != nil && !errors.Is(e, apperror.ErrNotFound) {
|
|
t.Fatal(e)
|
|
}
|
|
p := model.ProviderConfig{ConfigKey: key, Name: "Fixture", BaseURL: "https://example.com/v1", ModelName: "fixture", Enabled: true, TimeoutSeconds: 15, LastTestStatus: "NOT_TESTED", WireAPI: "responses", DisableResponseStorage: true}
|
|
saved, e := providers.Save(ctx, p, previous.Version, admin.ID, uuid.NewString(), false)
|
|
if e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
configs[key] = saved
|
|
}
|
|
repo := repository.NewCandidateAnalysisRepository(pool)
|
|
policy, err := repo.Policy(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
policy.Enabled = true
|
|
policy.AutoPublish = true
|
|
policy, err = repo.SavePolicy(ctx, policy, admin.ID, uuid.NewString())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
candidateRepo := repository.NewCandidateRepository(pool)
|
|
create := func() model.CandidateRevision {
|
|
input := model.CandidateDraftInput{Content: model.CandidateContent{Symbol: "000042", Exchange: "SZSE", Name: "Fixture", StrategyCode: "trend_following", Title: "Fixture", Thesis: "Daily crossing", Evidence: []model.CandidateEvidence{{SourceURL: "https://example.com/bars", Locator: "bar:today", Summary: "Daily input"}}, CounterEvidence: "Historical data", Invalidation: "Trend invalid", ReferencePlan: model.CandidateReferencePlan{EntryLow: "10.00", EntryHigh: "11.00", StopLoss: "9.00", TargetPrice: "12.00"}}, ObservedAt: time.Now().Add(-time.Hour), ValidUntil: time.Now().Add(time.Hour)}
|
|
c, e := candidateRepo.Create(ctx, input, admin.ID, uuid.NewString(), "", 0)
|
|
if e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
c, e = candidateRepo.Transition(ctx, c.ID, "SUBMIT", model.CandidateActionInput{ExpectedVersion: c.Version, Reason: "fixture"}, admin.ID, uuid.NewString())
|
|
if e != nil {
|
|
t.Fatal(e)
|
|
}
|
|
return c
|
|
}
|
|
c := create()
|
|
claimed, token, err := repo.Claim(ctx, policy)
|
|
if err != nil || claimed.ID != c.ID {
|
|
t.Fatal("claim failed", err)
|
|
}
|
|
if _, _, err := repo.Claim(ctx, policy); !errors.Is(err, apperror.ErrNotFound) {
|
|
t.Fatal("duplicate claim allowed", err)
|
|
}
|
|
approved := model.AIConclusion{SchemaVersion: "candidate_analysis.v1", Symbol: c.Snapshot.Symbol, Exchange: c.Snapshot.Exchange, Decision: "APPROVE", Summary: "Evidence supported", Risks: "Historical sample", EvidenceIndices: []int{0}, MissingInformation: []string{"No fundamentals"}}
|
|
if err = repo.Complete(ctx, policy, claimed, token, configs["model_primary"], configs["model_review"], &approved, &approved, ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = repo.Complete(ctx, policy, claimed, token, configs["model_primary"], configs["model_review"], &approved, &approved, ""); !errors.Is(err, apperror.ErrConflict) {
|
|
t.Fatal("completed claim was replayed", err)
|
|
}
|
|
detail, err := candidateRepo.Detail(ctx, c.ID, false)
|
|
if err != nil || detail.Status != "PUBLISHED" || detail.AIAnalysis == nil || detail.AIAnalysis.Status != "PUBLISHED" {
|
|
t.Fatal("automatic publication/analysis missing", err)
|
|
}
|
|
if _, _, err := repo.Claim(ctx, policy); !errors.Is(err, apperror.ErrNotFound) {
|
|
t.Fatal("published task reclaimed")
|
|
}
|
|
// Changing policy during inference invalidates the output, even when both models approve.
|
|
c = create()
|
|
claimed, token, err = repo.Claim(ctx, policy)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
oldPolicy := policy
|
|
policy.AutoPublish = false
|
|
policy, err = repo.SavePolicy(ctx, policy, admin.ID, uuid.NewString())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = repo.Complete(ctx, oldPolicy, claimed, token, configs["model_primary"], configs["model_review"], &approved, &approved, ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
detail, err = candidateRepo.Detail(ctx, c.ID, true)
|
|
if err != nil || detail.Status != "PENDING_REVIEW" {
|
|
t.Fatal("stale policy published")
|
|
}
|
|
// Manual mode stores approved analysis but never publishes.
|
|
claimed, token, err = repo.Claim(ctx, policy)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = repo.Complete(ctx, policy, claimed, token, configs["model_primary"], configs["model_review"], &approved, &approved, ""); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
detail, err = candidateRepo.Detail(ctx, c.ID, true)
|
|
if err != nil || detail.Status != "PENDING_REVIEW" || detail.AIAnalysis == nil {
|
|
t.Fatal("manual mode violated")
|
|
}
|
|
// Worker makes two independent calls; a rejection cannot auto publish.
|
|
c = create()
|
|
vault, _ := secret.NewVault("")
|
|
fake := &rejectModel{}
|
|
worker := service.NewCandidateAnalysisService(repo, providers, vault, fake, true)
|
|
if err = worker.ProcessOne(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
detail, err = candidateRepo.Detail(ctx, c.ID, true)
|
|
if err != nil || detail.Status != "PENDING_REVIEW" || fake.calls != 2 {
|
|
t.Fatal("model rejection not honored")
|
|
}
|
|
results, err := repo.List(ctx)
|
|
if err != nil || len(results) != 4 {
|
|
t.Fatal("durable results missing", err, len(results))
|
|
}
|
|
for _, result := range results {
|
|
if result.Name != "Fixture" || result.Symbol != "000042" || result.Exchange != "SZSE" {
|
|
t.Fatal("analysis task lost candidate identity")
|
|
}
|
|
}
|
|
}
|
|
|
|
type rejectModel struct{ calls int }
|
|
|
|
func (f *rejectModel) Analyze(_ context.Context, _ model.ProviderConfig, _ string, input json.RawMessage) (model.AIConclusion, error) {
|
|
f.calls++
|
|
var payload struct {
|
|
Content model.CandidateContent `json:"content"`
|
|
}
|
|
if err := json.Unmarshal(input, &payload); err != nil {
|
|
return model.AIConclusion{}, err
|
|
}
|
|
return model.AIConclusion{SchemaVersion: "candidate_analysis.v1", Symbol: payload.Content.Symbol, Exchange: payload.Content.Exchange, Decision: "REJECT", Summary: "Insufficient evidence", Risks: "Missing data", EvidenceIndices: []int{}, MissingInformation: []string{"Context"}}, nil
|
|
}
|