knowledge-governance/internal/workflows/catalogrefresh/handler_test.go

package catalogrefresh

import (
	"context"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"knowledge-governance/internal/jobs"
)

type fakeStore struct{ updates []Progress }

func (s *fakeStore) UpdateCatalogRefreshProgress(_ context.Context, _ string, _ string, p Progress, _ time.Time) error {
	s.updates = append(s.updates, p)
	return nil
}

func TestHandlerClampsMonotonicProgressAndKeepsFailuresVisible(t *testing.T) {
	store := &fakeStore{}
	handler, err := NewHandler(Config{Store: store, Runner: fakeRunner{}, Now: func() time.Time { return time.Now() }})
	if err != nil {
		t.Fatal(err)
	}
	runner := func(ctx context.Context, _ Request, report func(Progress) error) error {
		_ = report(Progress{Stage: "scanning", Progress: 40, Message: "扫描中", Processed: 60, Total: 100})
		// Discovered more directories than estimated: total must not shrink, percent must not regress.
		_ = report(Progress{Stage: "scanning", Progress: 30, Message: "发现更多目录", Processed: 70, Total: 50})
		return report(Progress{Stage: "failed", Progress: 0, Message: "来源不可用"})
	}
	custom := *handler
	custom.runner = runnerFunc(runner)
	job := jobs.Job{ID: "job-1", LeaseOwner: "worker-1", Payload: []byte(`{"connection_id":"conn-1","source_type":"alist","source_name":"国内工业","stage":"queued","progress":0,"message":"等待后台刷新","processed":0,"total":0}`)}
	if err = custom.Handle(context.Background(), job); err != nil {
		t.Fatal(err)
	}
	if len(store.updates) != 3 {
		t.Fatalf("updates=%#v", store.updates)
	}
	if store.updates[1].Progress < 40 || store.updates[1].Total < 70 {
		t.Fatalf("regression not clamped: %#v", store.updates[1])
	}
	if store.updates[2].Stage != "failed" || store.updates[2].Progress != 0 {
		t.Fatalf("failure must keep its own state: %#v", store.updates[2])
	}
}

func TestValidProgressRejectsControlCharsURLsAndCredentials(t *testing.T) {
	base := Progress{Stage: "scanning", Progress: 50, Message: "正在扫描", Processed: 10, Total: 20}
	if !validProgress(base) {
		t.Fatal("baseline rejected")
	}
	for name, mutate := range map[string]func(*Progress){
		"control":     func(p *Progress) { p.CurrentDirectory = "/产品\a资料" },
		"url":         func(p *Progress) { p.CurrentDirectory = "https://evil.invalid/x" },
		"url-message": func(p *Progress) { p.Message = "访问 https://evil.invalid 失败" },
		"token":       func(p *Progress) { p.Message = "token=abc123 已过期" },
		"too-long":    func(p *Progress) { p.CurrentDirectory = strings.Repeat("目", 1100) },
		"negative":    func(p *Progress) { p.ScannedFiles = -1 },
		"secret-dir":  func(p *Progress) { p.CurrentDirectory = "/run/source-secrets/pp" },
	} {
		item := base
		mutate(&item)
		if validProgress(item) {
			t.Errorf("%s accepted", name)
		}
	}
}

func TestCommandRunnerRejectsUnknownProgressFields(t *testing.T) {
	python, err := exec.LookPath("python3")
	if err != nil {
		t.Skip("python3 unavailable")
	}
	runner, err := NewCommandRunner(python, writeFakeProgressScript(t))
	if err != nil {
		t.Fatal(err)
	}
	err = runner.Run(context.Background(), Request{ConnectionID: "c", SourceType: "pp", SourceName: "n"}, func(Progress) error { return nil })
	if err == nil || err.Error() != "catalog_refresh_invalid_output" {
		t.Fatalf("unknown field must fail: %v", err)
	}
}

func writeFakeProgressScript(t *testing.T) string {
	t.Helper()
	directory := t.TempDir()
	path := filepath.Join(directory, "progress.py")
	body := "import json\nprint(json.dumps({'stage':'scanning','progress':10,'message':'ok','processed':1,'total':2,'unexpected_field':'leak'}),flush=True)\n"
	if err := os.WriteFile(path, []byte(body), 0o644); err != nil {
		t.Fatal(err)
	}
	return path
}

type runnerFunc func(context.Context, Request, func(Progress) error) error

func (f runnerFunc) Run(ctx context.Context, request Request, report func(Progress) error) error {
	return f(ctx, request, report)
}

type fakeRunner struct{}

func (fakeRunner) Run(_ context.Context, _ Request, report func(Progress) error) error {
	if err := report(Progress{Stage: "scanning", Progress: 50, Message: "正在扫描"}); err != nil {
		return err
	}
	return report(Progress{Stage: "completed", Progress: 100, Message: "完成"})
}
func TestHandlerAcceptsPersistedScanCountersOnRetry(t *testing.T) {
	store := &fakeStore{}
	handler, _ := NewHandler(Config{Store: store, Runner: fakeRunner{}})
	job := jobs.Job{ID: "retry", LeaseOwner: "worker", Payload: []byte(`{"connection_id":"c","source_type":"dingtalk","source_name":"wiki","current_directory":"目录","scanned_directories":2,"discovered_directories":3,"scanned_files":4,"estimated_total_files":5,"total_is_estimate":true}`)}
	if err := handler.Handle(context.Background(), job); err != nil {
		t.Fatal(err)
	}
	if len(store.updates) != 2 {
		t.Fatal("runner was not called")
	}
}

func TestHandlerRunsLiveRefreshAndPersistsProgress(t *testing.T) {
	store := &fakeStore{}
	handler, err := NewHandler(Config{Store: store, Runner: fakeRunner{}, Now: func() time.Time { return time.Now() }})
	if err != nil {
		t.Fatal(err)
	}
	job := jobs.Job{ID: "job-1", LeaseOwner: "worker-1", Payload: []byte(`{"connection_id":"conn-1","source_type":"pp","source_name":"PP资料下载中心","stage":"queued","progress":0,"message":"等待后台刷新","processed":0,"total":0}`)}
	if err = handler.Handle(context.Background(), job); err != nil {
		t.Fatal(err)
	}
	if len(store.updates) != 2 || store.updates[0].Stage != "scanning" || store.updates[1].Progress != 100 {
		t.Fatalf("updates=%#v", store.updates)
	}
}