feat: 迁移日汇总任务并修复商务菜单图标
This commit is contained in:
Yvan
@@ -0,0 +1,150 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/glebarez/sqlite"
|
||||
"gorm.io/gorm"
|
||||
businessbiz "kra/internal/biz/business"
|
||||
taskbiz "kra/internal/biz/task"
|
||||
businessdata "kra/internal/data/business"
|
||||
taskdata "kra/internal/data/task"
|
||||
"kra/internal/server/handler"
|
||||
"kra/internal/server/router"
|
||||
taskservice "kra/internal/service/task"
|
||||
"kra/internal/worker"
|
||||
platformtask "kra/pkg/task"
|
||||
)
|
||||
|
||||
type reportTaskHTTPData struct{ businessHTTPData }
|
||||
|
||||
func (reportTaskHTTPData) DatabaseReady() bool { return true }
|
||||
|
||||
func TestReportBuildThroughNativeTaskHTTP(t *testing.T) {
|
||||
db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "report-task.db")), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pool, _ := db.DB()
|
||||
t.Cleanup(func() { _ = pool.Close() })
|
||||
for _, step := range append(taskdata.Migrations(), businessdata.Migrations()...) {
|
||||
if err := step.Migrate(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, sql := range []string{
|
||||
"CREATE TABLE tab_user(id INTEGER,source_db TEXT,puid INTEGER,promote_id INTEGER,fgame_id INTEGER,idcard TEXT,register_time TEXT)",
|
||||
"CREATE TABLE tab_user_day_login(user_id INTEGER,source_db TEXT,promote_id INTEGER,game_id INTEGER,login_time INTEGER)",
|
||||
"CREATE TABLE tab_user_play_info(user_id INTEGER,source_db TEXT,promote_id INTEGER,game_id INTEGER,server_id TEXT,create_time INTEGER,play_time INTEGER)",
|
||||
"CREATE TABLE tab_spend(user_id INTEGER,source_db TEXT,promote_id INTEGER,game_id INTEGER,server_id TEXT,pay_time INTEGER,pay_status INTEGER,pay_amount DECIMAL(20,2))",
|
||||
"CREATE TABLE tab_game(id INTEGER,source_db TEXT,game_name TEXT)",
|
||||
"INSERT INTO tab_game VALUES(1,'lx','核验游戏')",
|
||||
"CREATE TABLE tab_sw_report_day(id INTEGER PRIMARY KEY,stat_date DATE,source_db TEXT,promote_id INTEGER,game_id INTEGER,game_name TEXT,dedup_mode INTEGER,new_register_cnt INTEGER,active_cnt INTEGER,pay_cnt INTEGER,new_pay_cnt INTEGER,total_pay DECIMAL(20,2))",
|
||||
"CREATE TABLE tab_sw_report_day_key(id INTEGER PRIMARY KEY,stat_date DATE,source_db TEXT,promote_id INTEGER,game_id INTEGER,game_name TEXT,dedup_mode INTEGER,metric INTEGER,identity_key TEXT)",
|
||||
"CREATE TABLE tab_sw_report_server_day(id INTEGER PRIMARY KEY,stat_date DATE,source_db TEXT,promote_id INTEGER,game_id INTEGER,game_name TEXT,server_num TEXT,dedup_mode INTEGER,new_register_cnt INTEGER,active_cnt INTEGER,pay_cnt INTEGER,total_pay DECIMAL(20,2))",
|
||||
"CREATE TABLE tab_sw_report_server_day_key(id INTEGER PRIMARY KEY,stat_date DATE,source_db TEXT,promote_id INTEGER,game_id INTEGER,game_name TEXT,server_num TEXT,dedup_mode INTEGER,metric INTEGER,identity_key TEXT)",
|
||||
} {
|
||||
if err := db.Exec(sql).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
stamp := time.Date(2026, 11, 4, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).Unix()
|
||||
for _, sql := range []string{"INSERT INTO tab_user VALUES(1,'lx',0,10,1,'aX',?)", "INSERT INTO tab_user_day_login VALUES(1,'lx',10,1,?)", "INSERT INTO tab_spend VALUES(1,'lx',10,1,'001',?,1,12.50)"} {
|
||||
if err := db.Exec(sql, stamp).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := db.Exec("INSERT INTO tab_user_play_info VALUES(1,'lx',10,1,'001',?,?)", stamp, stamp).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
provider := businessHTTPData{db}
|
||||
registry := platformtask.NewRegistry()
|
||||
tasks := taskbiz.NewTaskUsecaseWithRegistry(taskdata.NewTaskRepo(reportTaskHTTPData{provider}), registry)
|
||||
reports := businessbiz.NewReportBuildUsecase(businessdata.NewReportBuildRepo(provider))
|
||||
worker.NewTaskMethods(tasks, nil, nil, nil, reports).RegisterTasks(registry)
|
||||
scheduler := worker.NewTaskScheduler(tasks, nil, worker.NewTaskExecutorWithRegistry(tasks, registry), nil)
|
||||
defer func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
_ = scheduler.Stop(ctx)
|
||||
}()
|
||||
h := handler.NewTask(taskservice.NewTaskService(taskbiz.NewTaskApplicationUsecase(tasks, scheduler)))
|
||||
engine := gin.New()
|
||||
router.RegisterTask(engine.Group(""), h)
|
||||
srv := httptest.NewServer(engine)
|
||||
defer srv.Close()
|
||||
call := func(method, path string, body any) map[string]any {
|
||||
t.Helper()
|
||||
raw, _ := json.Marshal(body)
|
||||
req, _ := http.NewRequest(method, srv.URL+"/timedTask/"+path, bytes.NewReader(raw))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
res, err := srv.Client().Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
var value map[string]any
|
||||
if err := json.NewDecoder(res.Body).Decode(&value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if value["code"] != float64(0) {
|
||||
t.Fatalf("原生任务接口失败: %+v", value)
|
||||
}
|
||||
return value["data"].(map[string]any)
|
||||
}
|
||||
methods := call("GET", "getRegisteredMethods", nil)["methods"].([]any)
|
||||
if len(methods) != 8 {
|
||||
t.Fatalf("六个商务汇总方法未注册: %d", len(methods))
|
||||
}
|
||||
for index, method := range []string{"swReportDayBackfillTask", "swReportServerDayBackfillTask"} {
|
||||
name := fmt.Sprintf("隔离汇总任务%d", index)
|
||||
call("POST", "createTimedTask", map[string]any{"name": name, "spec": "@daily", "executorType": "method", "methodName": method, "params": map[string]any{"startDate": "2026-11-04", "endDate": "2026-11-04"}, "enabled": false})
|
||||
var id uint
|
||||
if err := db.Table("sys_timed_tasks").Select("id").Where("name=?", name).Scan(&id).Error; err != nil || id == 0 {
|
||||
t.Fatal("任务创建未落库")
|
||||
}
|
||||
call("POST", "triggerTimedTask", map[string]any{"id": id})
|
||||
deadline := time.Now().Add(8 * time.Second)
|
||||
done := false
|
||||
for time.Now().Before(deadline) {
|
||||
logs := call("GET", fmt.Sprintf("getTimedTaskLogList?taskId=%d&page=1&pageSize=10", id), nil)["list"].([]any)
|
||||
if len(logs) > 0 {
|
||||
if logs[0].(map[string]any)["status"] != "success" {
|
||||
t.Fatalf("汇总任务失败: %+v", logs)
|
||||
}
|
||||
done = true
|
||||
break
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
if !done {
|
||||
t.Fatal("未等到汇总任务完成")
|
||||
}
|
||||
}
|
||||
for _, table := range []string{"tab_sw_report_day", "tab_sw_report_server_day"} {
|
||||
var rows []struct {
|
||||
TotalPay string
|
||||
NewRegisterCnt, ActiveCnt, PayCnt int
|
||||
}
|
||||
if err := db.Table(table).Scan(&rows).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 2 || rows[0].TotalPay != "12.5" || rows[0].NewRegisterCnt != 1 || rows[0].ActiveCnt != 1 || rows[0].PayCnt != 1 {
|
||||
t.Fatalf("任务没有实际生成双口径汇总: %+v", rows)
|
||||
}
|
||||
}
|
||||
var enabled int64
|
||||
db.Table("sys_timed_tasks").Where("enabled=1").Count(&enabled)
|
||||
if enabled != 0 {
|
||||
t.Fatal("手动回跑不应启用自动计划")
|
||||
}
|
||||
}
|
||||
Generated
+3
-1
@@ -288,7 +288,9 @@ func wireApp(configServer *config.Server, store *config.Store, logger *slog.Logg
|
||||
routes := router.NewRoutes(set)
|
||||
maintenanceRepo := system.NewMaintenanceRepo(dataData)
|
||||
maintenanceUsecase := system2.NewMaintenanceUsecase(maintenanceRepo)
|
||||
taskMethods := worker.NewTaskMethods(taskUsecase, maintenanceUsecase, mediaUsecase, store)
|
||||
reportBuildRepo := business.NewReportBuildRepo(dataData)
|
||||
reportBuildUsecase := business2.NewReportBuildUsecase(reportBuildRepo)
|
||||
taskMethods := worker.NewTaskMethods(taskUsecase, maintenanceUsecase, mediaUsecase, store, reportBuildUsecase)
|
||||
routeRegistrar := buildRuntime(registry, routes, taskMethods)
|
||||
websocketServer, cleanup2, err := websocket.New(runtimeconfigStore)
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user