sub2api/backend/internal/repository/intelligent_test_assessment_integration_test.go

package repository

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"github.com/Wei-Shaw/sub2api/internal/service"
	"github.com/google/uuid"
	"github.com/stretchr/testify/require"
)

func TestIntelligentAssessmentQueueActionsAndHistory(t *testing.T) {
	db := intelligentTestDB(t)
	repo := &intelligentTestRepository{db: db}
	svc := service.NewIntelligentTestService(repo, nil) // no runner: re-evaluation cannot use upstream
	ctx := context.Background()
	enqueue := func(model string) (*service.IntelligentTestEnqueued, error) {
		return svc.Enqueue(ctx, 1, service.IntelligentTestEnqueue{AccountIDs: []int64{10}, TestTypes: []string{"candy"}, Models: map[string]string{"candy": model}, IdempotencyKey: uuid.NewString()})
	}
	first, err := enqueue("gpt-5.4")
	require.NoError(t, err)
	require.Equal(t, 1, first.CreatedCount)
	again, err := enqueue("gpt-5.4")
	require.NoError(t, err)
	require.Equal(t, 1, again.ReusedCount)
	require.Zero(t, again.CreatedCount)
	_, err = enqueue("gpt-5.4-mini")
	require.ErrorContains(t, err, "不同模型或配置")
	_, err = svc.Cancel(ctx, 2, first.Records[0].ID)
	require.ErrorIs(t, err, service.ErrIntelligentTestForbidden)
	running, err := repo.Claim(ctx)
	require.NoError(t, err)
	require.NotNil(t, running)
	_, err = svc.Cancel(ctx, 1, running.ID)
	require.ErrorContains(t, err, "只能取消")
	require.NoError(t, repo.DeferForCapacity(ctx, running))
	waiting, err := repo.Get(ctx, running.ID)
	require.NoError(t, err)
	require.Equal(t, "queued", waiting.Status)
	require.Contains(t, waiting.QueueReason, "等待账号空闲")
	none, err := repo.Claim(ctx)
	require.NoError(t, err)
	require.Nil(t, none, "busy tasks do not monopolize claim cycles")
	cancelled, err := svc.Cancel(ctx, 1, running.ID)
	require.NoError(t, err)
	require.Equal(t, "cancelled", cancelled.Status)
	_, err = svc.Cancel(ctx, 1, running.ID)
	require.NoError(t, err)
	var audits int
	require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM audit_logs WHERE action='admin.intelligent_tests.cancel'`).Scan(&audits))
	require.Equal(t, 1, audits)

	// Construct a complete historical judgment with its original private snapshot.
	cfg := service.IntelligentTestConfig{Prompt: "original question", Model: "original-model", Evaluator: "exact_answer", ExpectedAnswer: "12", TimeoutSeconds: 120}
	encoded, err := json.Marshal(cfg)
	require.NoError(t, err)
	var oldID int64
	require.NoError(t, db.QueryRow(`INSERT INTO account_tests(account_id,test_type,status,score,result,input,raw_response,config_snapshot,evaluation,finished_at,anti_degradation,requested_by) VALUES(10,'candy','suspected_degradation',0,'12','original question','original raw',$1,'{"method":"exact_answer","reason":"legacy strict mismatch"}',NOW(),true,1) RETURNING id`, string(encoded)).Scan(&oldID))
	// Current settings differ: re-evaluation must use the historical snapshot.
	_, err = db.Exec(`UPDATE test_settings SET config=jsonb_set(config,'{expected_answer}','"999"'),user_visible=true WHERE test_type='candy'`)
	require.NoError(t, err)
	next, err := enqueue("gpt-5.4-mini")
	require.NoError(t, err)
	f := service.IntelligentTestFilter{AccountID: 10, TestType: "candy", Page: 1, PageSize: 12, OnlyAbnormal: true}
	before, err := repo.Accounts(ctx, f)
	require.NoError(t, err)
	require.EqualValues(t, 1, before.Total)
	require.EqualValues(t, 1, before.Overview.ReviewAccounts)
	require.Zero(t, before.Overview.AbnormalAccounts)
	require.Equal(t, next.Records[0].ID, before.Items[0].Tests[0].Latest.ID)
	require.Equal(t, oldID, before.Items[0].Tests[0].LatestCompleted.ID)
	require.Empty(t, before.Items[0].Tests[0].LatestCompleted.RawResponse)
	require.Empty(t, before.Items[0].Tests[0].Risk)
	_, err = svc.Reevaluate(ctx, 2, oldID)
	require.ErrorIs(t, err, service.ErrIntelligentTestForbidden)
	updated, err := svc.Reevaluate(ctx, 1, oldID)
	require.NoError(t, err)
	require.Equal(t, "completed", updated.Status)
	require.Equal(t, "correct", updated.Evaluation["answer_verdict"])
	require.Equal(t, "non_compliant", updated.Evaluation["format_verdict"])
	require.Equal(t, 100.0, *updated.Score)
	require.Equal(t, "12", updated.Result)
	require.Equal(t, "original raw", updated.RawResponse)
	require.Equal(t, "12", updated.ConfigSnapshot.ExpectedAnswer)
	require.Contains(t, updated.Evaluation, "original_judgment")
	_, err = svc.Reevaluate(ctx, 1, oldID)
	require.NoError(t, err)
	require.NoError(t, db.QueryRow(`SELECT COUNT(*) FROM audit_logs WHERE action='admin.intelligent_tests.reevaluate'`).Scan(&audits))
	require.Equal(t, 1, audits)
	public, err := repo.PublicGet(ctx, 2, oldID)
	require.NoError(t, err)
	require.Equal(t, "correct", public.Evaluation["answer_verdict"])
	require.NotContains(t, public.Evaluation, "expected_answer")
	require.NotContains(t, public.Evaluation, "original_judgment")
	after, err := repo.Accounts(ctx, f)
	require.NoError(t, err)
	require.Zero(t, after.Total)
	f.OnlyAbnormal = false
	after, err = repo.Accounts(ctx, f)
	require.NoError(t, err)
	require.EqualValues(t, 1, after.Overview.SuccessAccounts)
	require.Zero(t, after.Overview.ReviewAccounts)
	// Cancelled attempts also retain the last completed result.
	_, err = svc.Cancel(ctx, 1, next.Records[0].ID)
	require.NoError(t, err)
	after, err = repo.Accounts(ctx, f)
	require.NoError(t, err)
	require.Equal(t, oldID, after.Items[0].Tests[0].LatestCompleted.ID)
	require.EqualValues(t, 1, after.Overview.SuccessAccounts)
}

func TestIntelligentQueueDefersUntilCooldownAndAllowsCancellation(t *testing.T) {
	db := intelligentTestDB(t)
	repo := &intelligentTestRepository{db: db}
	ctx := context.Background()
	_, err := repo.Enqueue(ctx, 1, service.IntelligentTestEnqueue{AccountIDs: []int64{10}, TestTypes: []string{"candy"}, IdempotencyKey: uuid.NewString()})
	require.NoError(t, err)
	r, err := repo.Claim(ctx)
	require.NoError(t, err)
	require.NotNil(t, r)
	until := time.Now().Add(2 * time.Minute).UTC().Truncate(time.Second)
	r.AvailableAt = &until
	r.QueueReason = "模型冷却，等待后执行"
	require.NoError(t, repo.DeferForCapacity(ctx, r))
	saved, err := repo.Get(ctx, r.ID)
	require.NoError(t, err)
	require.True(t, until.Equal(*saved.AvailableAt), "queue timing is an instant, independent of database timezone")
	require.Equal(t, r.QueueReason, saved.QueueReason)
	next, err := repo.Claim(ctx)
	require.NoError(t, err)
	require.Nil(t, next)
	require.Error(t, repo.DeferForCapacity(ctx, r), "an expired worker cannot alter a queued/cancelled job")
	cancelled, err := repo.Cancel(ctx, 1, r.ID)
	require.NoError(t, err)
	require.Equal(t, "cancelled", cancelled.Status)
}

func TestIntelligentDeferredTaskDoesNotStarveAlreadyWaitingTask(t *testing.T) {
	db := intelligentTestDB(t)
	repo := &intelligentTestRepository{db: db}
	ctx := context.Background()
	_, err := repo.Enqueue(ctx, 1, service.IntelligentTestEnqueue{AccountIDs: []int64{10, 11}, TestTypes: []string{"candy"}, IdempotencyKey: uuid.NewString()})
	require.NoError(t, err)
	first, err := repo.Claim(ctx)
	require.NoError(t, err)
	require.EqualValues(t, 10, first.AccountID)
	require.NoError(t, repo.DeferForCapacity(ctx, first))
	_, err = db.Exec(`UPDATE account_tests SET available_at=CASE WHEN id=$1 THEN NOW()-INTERVAL '1 second' ELSE NOW()-INTERVAL '1 minute' END WHERE status='queued'`, first.ID)
	require.NoError(t, err)
	next, err := repo.Claim(ctx)
	require.NoError(t, err)
	require.EqualValues(t, 11, next.AccountID, "a retried low-ID task must yield to an older waiting task")
}