package service import ( "encoding/base64" "encoding/json" "testing" "time" "github.com/1024XEngineer/xinfra/server/internal/model" ) func TestValidateDeliveryInput(t *testing.T) { dataDisks := []string{"/data", "/disk1"} valid := MySQLDeliveryInput{BusinessLineID: 1, TargetID: 1, Namespace: "team-a", InstanceName: "mysql-01", CPUCores: 2, MemoryGB: 4, StorageGB: 50} normalizeMySQLDeliveryInput(&valid) if err := validateDeliveryInput(valid, dataDisks); err != nil { t.Fatalf("valid input rejected: %v", err) } full := valid full.MySQLVersion = "8.0" full.Topology = "standalone" full.MySQLPort = 13306 full.DataDisk = "/var/lib/mysql01" full.TargetHost = "k8s-server-03" full.CPUMilli = 2000 full.MemoryMi = 8192 full.StorageGi = 2000 full.Timezone = "+08:00" lowerCaseZero := 0 full.LowerCaseTableNames = &lowerCaseZero full.CharacterSet = "utf8mb4" full.Collation = "utf8mb4_general_ci" full.MaxConnections = "auto" full.InnodbRedoLogCapacity = "256M" flushLog := 2 full.InnodbFlushLogAtTrxCommit = &flushLog syncBinlog := 0 full.SyncBinlog = &syncBinlog full.InnodbIOCapacity = 2000 full.LongQueryTime = 0.5 full.BinlogExpireLogsSeconds = 604800 full.MaxBinlogSize = "512M" if err := validateDeliveryInput(full, dataDisks); err != nil { t.Fatalf("valid full input rejected: %v", err) } namedZone := valid namedZone.Timezone = "Asia/Shanghai" if err := validateDeliveryInput(namedZone, dataDisks); err != nil { t.Fatalf("named timezone rejected: %v", err) } lts := valid lts.MySQLVersion = "8.4" if err := validateDeliveryInput(lts, dataDisks); err != nil { t.Fatalf("8.4 LTS rejected: %v", err) } for name, mutate := range map[string]func(*MySQLDeliveryInput){ "uppercase namespace": func(in *MySQLDeliveryInput) { in.Namespace = "Team-A" }, "bad instance": func(in *MySQLDeliveryInput) { in.InstanceName = "mysql_01" }, "too little cpu": func(in *MySQLDeliveryInput) { in.CPUMilli = 50 }, "too much cpu": func(in *MySQLDeliveryInput) { in.CPUMilli = 65000 }, "too little memory": func(in *MySQLDeliveryInput) { in.MemoryMi = 1024 }, "too much memory": func(in *MySQLDeliveryInput) { in.MemoryMi = 131072 }, "too little storage": func(in *MySQLDeliveryInput) { in.StorageGi = 10 }, "too much storage": func(in *MySQLDeliveryInput) { in.StorageGi = 4000 }, "unsupported version": func(in *MySQLDeliveryInput) { in.MySQLVersion = "5.7" }, "eol version": func(in *MySQLDeliveryInput) { in.MySQLVersion = "5.6" }, "unsupported topology": func(in *MySQLDeliveryInput) { in.Topology = "mgr_3" }, "port below pool": func(in *MySQLDeliveryInput) { in.MySQLPort = 3307 }, "port above pool": func(in *MySQLDeliveryInput) { in.MySQLPort = 14000 }, "bad target host": func(in *MySQLDeliveryInput) { in.TargetHost = "-bad-host" }, "bad timezone": func(in *MySQLDeliveryInput) { in.Timezone = "UTC+8" }, "bad lower case": func(in *MySQLDeliveryInput) { v := 2; in.LowerCaseTableNames = &v }, "bad charset": func(in *MySQLDeliveryInput) { in.CharacterSet = "big5" }, "collation mismatch": func(in *MySQLDeliveryInput) { in.CharacterSet = "gbk"; in.Collation = "utf8mb4_general_ci" }, "bad max connections": func(in *MySQLDeliveryInput) { in.MaxConnections = "300" }, "bad redo capacity": func(in *MySQLDeliveryInput) { in.InnodbRedoLogCapacity = "2G" }, "bad flush log": func(in *MySQLDeliveryInput) { v := 3; in.InnodbFlushLogAtTrxCommit = &v }, "bad sync binlog": func(in *MySQLDeliveryInput) { v := 2; in.SyncBinlog = &v }, "bad io capacity": func(in *MySQLDeliveryInput) { in.InnodbIOCapacity = 500 }, "bad long query time": func(in *MySQLDeliveryInput) { in.LongQueryTime = 3 }, "bad binlog expire": func(in *MySQLDeliveryInput) { in.BinlogExpireLogsSeconds = 3600 }, "bad max binlog size": func(in *MySQLDeliveryInput) { in.MaxBinlogSize = "64M" }, } { input := valid mutate(&input) if err := validateDeliveryInput(input, dataDisks); err == nil { t.Errorf("%s was accepted", name) } } } func TestParseMySQLInspectResults(t *testing.T) { payload := `[{"id":7,"task_id":"task-1","instance_name":"mysql-01","host":"k8s-01","status":"running","port_listening":true}]` stdout := `ok: [k8s-01] => {"msg": "XINFRA_MYSQL_INSPECT_RESULT_B64=` + base64.StdEncoding.EncodeToString([]byte(payload)) + `"}` items := parseMySQLInspectResults(stdout) if len(items) != 1 { t.Fatalf("len(items) = %d, want 1", len(items)) } if items[0].ID != 7 || items[0].Status != "running" || !items[0].PortListening { t.Fatalf("unexpected inspect result: %+v", items[0]) } } func TestFirstFreeHost(t *testing.T) { hosts := []targetHost{{Name: "node-a"}, {Name: "node-b"}} if h := firstFreeHost(hosts, nil, 1); h == nil || h.Name != "node-a" { t.Fatalf("expected node-a on empty occupancy, got %+v", h) } if h := firstFreeHost(hosts, []string{"node-a"}, 1); h == nil || h.Name != "node-b" { t.Fatalf("expected node-b when node-a is full at limit 1, got %+v", h) } if h := firstFreeHost(hosts, []string{"node-a", "node-b"}, 1); h != nil { t.Fatalf("limit 1 with all hosts taken should return nil, got %+v", h) } if h := firstFreeHost(hosts, []string{"node-a", "node-b"}, 2); h == nil || h.Name != "node-a" { t.Fatalf("expected node-a for second round at limit 2, got %+v", h) } if h := firstFreeHost(hosts, []string{"node-a", "node-a", "node-b", "node-b"}, 2); h != nil { t.Fatalf("limit 2 with all hosts saturated should return nil, got %+v", h) } if h := firstFreeHost(hosts, []string{"node-a"}, 0); h == nil || h.Name != "node-b" { t.Fatalf("limit 0 should degrade to 1, got %+v", h) } } func TestAllocatePort(t *testing.T) { if port, err := allocatePort(0, nil); err != nil || port != mysqlPortPoolStart { t.Fatalf("expected first pool port %d, got %d err=%v", mysqlPortPoolStart, port, err) } if port, err := allocatePort(0, []int{13306, 13307}); err != nil || port != 13308 { t.Fatalf("expected 13308 skipping occupied, got %d err=%v", port, err) } if port, err := allocatePort(13400, []int{13306}); err != nil || port != 13400 { t.Fatalf("expected requested port 13400, got %d err=%v", port, err) } if _, err := allocatePort(13306, []int{13306}); err == nil { t.Fatal("requested occupied port was accepted") } used := make([]int, 0, mysqlPortPoolEnd-mysqlPortPoolStart+1) for p := mysqlPortPoolStart; p <= mysqlPortPoolEnd; p++ { used = append(used, p) } if _, err := allocatePort(0, used); err == nil { t.Fatal("exhausted pool still allocated a port") } } func TestBuildCloudDMRegisterRequest(t *testing.T) { req := buildCloudDMRegisterRequest( model.DeploymentResult{ID: 42, InstanceName: "mysql-payment-prod", Host: "10.0.0.10", Port: 3306, Version: "8.4"}, deliveryPayload{MySQLDeliveryInput: MySQLDeliveryInput{InstanceDesc: "支付生产 MySQL", Timezone: "Asia/Shanghai"}}, "secret", ) if req.SourceSystem != "xinfra" || req.ResourceType != "MYSQL_INSTANCE" { t.Fatalf("unexpected request metadata: %#v", req) } if req.ExternalResourceID != "mysql-instance:42" { t.Fatalf("unexpected externalResourceId: %q", req.ExternalResourceID) } if req.DataSource.Host != "10.0.0.10:3306" || req.DataSource.Password != "secret" { t.Fatalf("unexpected data source fields: %#v", req.DataSource) } if req.DataSource.ClusterID != nil { t.Fatalf("cluster_id must be nullable when xinfra cannot resolve it, got %#v", req.DataSource.ClusterID) } if req.DataSource.MySQLVersion != "8.4" { t.Fatalf("unexpected mysql version: %#v", req.DataSource.MySQLVersion) } if req.DataSource.ClientTimeZone != "Asia/Shanghai" || req.DataSource.ConnectionCharset != "utf8" { t.Fatalf("unexpected time zone or charset: %#v", req.DataSource) } raw, err := json.Marshal(req) if err != nil { t.Fatalf("marshal CloudDM request: %v", err) } var decoded map[string]any if err := json.Unmarshal(raw, &decoded); err != nil { t.Fatalf("unmarshal CloudDM request: %v", err) } dataSource := decoded["dataSource"].(map[string]any) if value, ok := dataSource["defaultSchema"]; !ok || value != nil { t.Fatalf("defaultSchema must be present as null, got %#v", dataSource["defaultSchema"]) } if value, ok := dataSource["cluster_id"]; !ok || value != nil { t.Fatalf("cluster_id must be present as null, got %#v", dataSource["cluster_id"]) } if value, ok := dataSource["mysql_version"]; !ok || value != "8.4" { t.Fatalf("mysql_version must be propagated from deployed instance, got %#v", value) } } func TestBuildCloudDMDeleteRequest(t *testing.T) { req := buildCloudDMDeleteRequest(model.DeploymentResult{ID: 42}) if req.SourceSystem != "xinfra" || req.ResourceType != "MYSQL_INSTANCE" { t.Fatalf("unexpected request metadata: %#v", req) } if req.ExternalResourceID != "mysql-instance:42" { t.Fatalf("unexpected externalResourceId: %q", req.ExternalResourceID) } } func TestCloudDMDeleteInfoAndMetadata(t *testing.T) { meta := map[string]any{ "clouddm": map[string]any{ "data_source_id": 123.0, "delete_status": "Failed", "delete_error": "boom", "deleted_at": "", "external_resource_id": "mysql-instance:42", }, } dataSourceID, deleteStatus := cloudDMDeleteInfo(meta) if dataSourceID != 123 || deleteStatus != "failed" { t.Fatalf("unexpected delete info: %d, %q", dataSourceID, deleteStatus) } updated := updateCloudDMDeleteMetadata(meta, "deleted", "", "2026-07-30T12:00:00Z") clouddm := updated["clouddm"].(map[string]any) if clouddm["delete_status"] != "deleted" || clouddm["delete_error"] != "" || clouddm["deleted_at"] != "2026-07-30T12:00:00Z" { t.Fatalf("unexpected updated metadata: %#v", clouddm) } } func TestCloudDMDataSourceIDFromResponse(t *testing.T) { tests := []struct { name string body string want uint64 ok bool }{ {name: "direct data", body: `{"data":123}`, want: 123, ok: true}, {name: "nested data source id", body: `{"code":0,"data":{"dataSourceId":456}}`, want: 456, ok: true}, {name: "nested id", body: `{"success":true,"data":{"id":789}}`, want: 789, ok: true}, {name: "string ds id", body: `{"dsId":"321"}`, want: 321, ok: true}, {name: "missing", body: `{"code":0,"message":"ok"}`, want: 0, ok: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { got, ok := cloudDMDataSourceIDFromResponse([]byte(tt.body)) if got != tt.want || ok != tt.ok { t.Fatalf("cloudDMDataSourceIDFromResponse() = %d, %v; want %d, %v", got, ok, tt.want, tt.ok) } }) } } func TestRollbackExtraVarsTargetsOnlyTheAllocatedInstance(t *testing.T) { task := &model.DeliveryTask{ID: "task-1", TargetHost: "db-01"} payload := deliveryPayload{MySQLDeliveryInput: MySQLDeliveryInput{InstanceName: "mysql-a", DataDisk: "/disk1"}} vars := rollbackExtraVars(task, payload) if vars["target_hosts"] != "db-01" || vars["instance_name"] != "mysql-a" || vars["data_disk"] != "/disk1" { t.Fatalf("rollback vars target the wrong instance: %#v", vars) } if vars["rollback"] != true { t.Fatalf("rollback marker missing: %#v", vars) } } func TestDirectoryPathsFromAWXStdout(t *testing.T) { stdout := ` TASK [Show directory completions] ********************************************** ok: [db-01] => { "stdout": "XINFRA_PATH_COMPLETIONS_JSON=[{\"path\":\"/a/bc\",\"available_gi\":12},{\"path\":\"/a/bb\",\"available_gi\":8}]" } ` items := directoryPathsFromAWXStdout(stdout) if len(items) != 2 { t.Fatalf("expected two completion items, got %#v", items) } if items[0].Path != "/a/bb" || items[1].Path != "/a/bc" { t.Fatalf("items should be parsed and sorted by path: %#v", items) } } func TestValidateDirectoryLookupPrefix(t *testing.T) { for _, path := range []string{"/", "/a", "/a/b", "/lib/data"} { if err := validateDirectoryLookupPrefix(path); err != nil { t.Fatalf("valid path prefix %q rejected: %v", path, err) } } for _, path := range []string{"a", "relative/path"} { if err := validateDirectoryLookupPrefix(path); err == nil { t.Fatalf("invalid path prefix %q was accepted", path) } } } func TestRegisterFailedIsProtectedFromRollback(t *testing.T) { if !rollbackProtectedStatus(model.TaskRegisterFailed) { t.Fatal("register_failed must preserve the healthy instance and resource usage") } if rollbackProtectedStatus(model.TaskValidationFailed) { t.Fatal("validation_failed must still be eligible for cleanup rollback") } } func TestRollbackLaunchExpired(t *testing.T) { now := time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC) started := now.Add(-rollbackLaunchTimeout - time.Second) if !rollbackLaunchExpired(model.RollbackJob{Status: "launching", StartedAt: &started}, now) { t.Fatal("stale launching rollback job must be recoverable") } if rollbackLaunchExpired(model.RollbackJob{Status: "launching", StartedAt: ptrTime(now.Add(-rollbackLaunchTimeout + time.Second))}, now) { t.Fatal("recent launching rollback job must remain pending") } if rollbackLaunchExpired(model.RollbackJob{Status: "running", StartedAt: &started}, now) { t.Fatal("running rollback job is not a launch timeout") } } func ptrTime(v time.Time) *time.Time { return &v }