package repository import ( "database/sql" "fmt" "os" "strings" "sync" "testing" "cmautobuy/admin/config" "cmautobuy/admin/model" "cmautobuy/admin/spec" ) // TestMySQLMigrate_真实MySQL8 只在显式提供隔离测试库时运行。 // 库名必须以 _test 结尾,防止测试清理误碰生产库。 func TestMySQLMigrate_真实MySQL8(t *testing.T) { if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" { t.Skip("未启用真实 MySQL 8 集成测试") } cfg, err := config.LoadDatabaseFromEnv() if err != nil { t.Fatal(err) } if !strings.HasSuffix(cfg.Name, "_test") { t.Fatalf("拒绝清理非测试数据库 %q:库名必须以 _test 结尾", cfg.Name) } db, err := OpenMySQL(cfg) if err != nil { t.Fatal(err) } defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) if err := MigrateMySQL(db); err != nil { t.Fatalf("首次建立 MySQL schema 失败: %v", err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("重复迁移应该无副作用: %v", err) } var version int if err := db.QueryRow(`SELECT MAX(version) FROM schema_migrations`).Scan(&version); err != nil { t.Fatal(err) } if version != mysqlSchemaVersion { t.Fatalf("schema 版本=%d,期望 %d", version, mysqlSchemaVersion) } } func TestMySQLMigrate_V2升级V3并转换真实主链路(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV2(t, db) now := "2026-08-10T03:00:00Z" mustExec(t, db, `INSERT INTO shopee_products (goods_id,title,created_at,updated_at) VALUES ('REPORT-1','报表商品',?,?)`, now, now) mustExec(t, db, `INSERT INTO shopee_skus (sku_id,goods_id,spec_raw,created_at,updated_at) VALUES ('SKU-1','REPORT-1','黑色, M',?,?)`, now, now) mustExec(t, db, `INSERT INTO sku_mappings (shopee_sku_id,pdd_goods_id,pdd_option_key,pdd_options,goods_id,mapped_at) VALUES ('SKU-1','PDD-1','color=黑色&size=M','color=黑色&size=M','REPORT-1',?)`, now) mustExec(t, db, `INSERT INTO syb_orders (syb_id,order_no,title,shopee_goods_id,product_spec,quantity,syb_data,created_at,updated_at) VALUES ('SYB-1','ORDER-1','报表商品','REPORT-1',' 黑色, M ',1,'{}',?,?), ('SYB-2','ORDER-2','骨架商品','SKELETON-1','白色,L',1,'{}',?,?), ('SYB-3','ORDER-3','空规格','EMPTY-1',NULL,1,'{}',?,?)`, now, now, now, now, now, now) if err := MigrateMySQL(db); err != nil { t.Fatalf("v2→v3 失败: %v", err) } var key string if err := db.QueryRow(`SELECT spec_key FROM syb_orders WHERE syb_id='SYB-1'`).Scan(&key); err != nil || key != "黑色, M" { t.Fatalf("规格回填=%q err=%v", key, err) } var empty sql.NullString if err := db.QueryRow(`SELECT spec_key FROM syb_orders WHERE syb_id='SYB-3'`).Scan(&empty); err != nil || empty.Valid { t.Fatalf("空规格必须保持 NULL: %+v err=%v", empty, err) } var source string if err := db.QueryRow(`SELECT source FROM shopee_products WHERE goods_id='SKELETON-1'`).Scan(&source); err != nil || source != "syb" { t.Fatalf("骨架来源=%q err=%v", source, err) } var mappingCount int if err := db.QueryRow(`SELECT COUNT(*) FROM spec_mappings WHERE shopee_goods_id='REPORT-1' AND spec_key='黑色, M' AND pdd_goods_id='PDD-1'`).Scan(&mappingCount); err != nil || mappingCount != 1 { t.Fatalf("旧映射转换数量=%d err=%v", mappingCount, err) } if exists, _ := mysqlTableExists(db, "sku_mappings"); exists { t.Fatal("旧 sku_mappings 应已删除") } if exists, _ := mysqlTableExists(db, "sku_mappings_v3_backup"); !exists { t.Fatal("删除旧表前必须保留受限备份表") } } func TestMySQLMigrate_V3三个中断点重跑收敛(t *testing.T) { for _, tc := range []struct { name string prepare func(*testing.T, *sql.DB) }{ {"DDL完成回填未开始", func(t *testing.T, db *sql.DB) { if err := ensureMySQLV3Columns(db); err != nil { t.Fatal(err) } mustExec(t, db, mysqlSchemaV3SpecMappings) }}, {"回填到一半", func(t *testing.T, db *sql.DB) { if err := ensureMySQLV3Columns(db); err != nil { t.Fatal(err) } mustExec(t, db, mysqlSchemaV3SpecMappings) key, _ := spec.SpecKey("黑色,M") mustExec(t, db, `UPDATE syb_orders SET spec_key=? WHERE syb_id='SYB-A'`, key) }}, {"全部完成版本未记录", func(t *testing.T, db *sql.DB) { if err := migrateMySQLV3(db); err != nil { t.Fatal(err) } }}, } { t.Run(tc.name, func(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV2(t, db) now := "2026-08-10T03:00:00Z" mustExec(t, db, `INSERT INTO syb_orders (syb_id,order_no,title,shopee_goods_id,product_spec,quantity,syb_data,created_at,updated_at) VALUES ('SYB-A','A','A','G-A','黑色,M',1,'{}',?,?),('SYB-B','B','B','G-B','白色,L',1,'{}',?,?)`, now, now, now, now) tc.prepare(t, db) if err := MigrateMySQL(db); err != nil { t.Fatalf("重跑失败: %v", err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("再次重跑失败: %v", err) } var count int if err := db.QueryRow(`SELECT COUNT(*) FROM syb_orders WHERE spec_key IS NOT NULL`).Scan(&count); err != nil || count != 2 { t.Fatalf("回填未收敛 count=%d err=%v", count, err) } }) } } func TestMySQLMigrate_V3字段存在但约束缺失可收敛(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV2(t, db) mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,created_at,updated_at) VALUES ('EXISTING-1','存量商品','2026-08-10T00:00:00Z','2026-08-10T00:00:00Z')`) mustExec(t, db, `ALTER TABLE shopee_products ADD COLUMN source VARCHAR(16) NULL`) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } var existingSource string if err := db.QueryRow(`SELECT source FROM shopee_products WHERE goods_id='EXISTING-1'`).Scan(&existingSource); err != nil || existingSource != "report" { t.Fatalf("可空 source 中断点的存量行未收敛: source=%q err=%v", existingSource, err) } mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) VALUES ('CHECK-1','检查','report','2026-08-10T00:00:00Z','2026-08-10T00:00:00Z')`) if _, err := db.Exec(`UPDATE shopee_products SET source='typo'`); err == nil { t.Fatal("source CHECK 必须拒绝非法值") } } func TestMySQLMigrate_V3形状自检失败不记版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV2(t, db) mustExec(t, db, `CREATE TABLE spec_mappings ( shopee_goods_id VARCHAR(191) NOT NULL, spec_key VARCHAR(191) NOT NULL, pdd_goods_id VARCHAR(191) NOT NULL, pdd_option_key VARCHAR(191) NOT NULL, pdd_options LONGTEXT NOT NULL, spec_raw TEXT NOT NULL, mapped_at VARCHAR(35) NOT NULL, PRIMARY KEY (shopee_goods_id, pdd_goods_id, spec_key) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`) if err := MigrateMySQL(db); err == nil { t.Fatal("错误主键顺序必须让 v3 形状自检失败") } var count int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=3`).Scan(&count); err != nil || count != 0 { t.Fatalf("自检失败时不得记录 v3: count=%d err=%v", count, err) } } func TestMySQLMigrate_V3升级V4且断点重跑(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV2(t, db) if err := migrateMySQLV3(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (3,'2026-08-10T00:00:00Z')`) // 模拟 DDL 已完成、版本未记录。 mustExec(t, db, mysqlSchemaV4Decisions) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("重跑失败: %v", err) } var versions int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=4`).Scan(&versions); err != nil || versions != 1 { t.Fatalf("v4=%d err=%v", versions, err) } mustExec(t, db, `INSERT INTO spec_mapping_decisions(shopee_goods_id,spec_key,pdd_goods_id,rules_version,chosen_option_key,accepted,decided_at) VALUES('S','K','P','rules_v1','O',1,'2026-08-10T00:00:00Z')`) if _, err := db.Exec(`UPDATE spec_mapping_decisions SET accepted=2`); err == nil { t.Fatal("accepted CHECK 必须拒绝 2") } } func TestMySQLMigrate_V4形状错误不记版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV2(t, db) if err := migrateMySQLV3(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (3,'2026-08-10T00:00:00Z')`) mustExec(t, db, strings.Replace(mysqlSchemaV4Decisions, "CHECK (accepted IN (0,1))", "CHECK (accepted IN (0,1,2))", 1)) if err := MigrateMySQL(db); err == nil { t.Fatal("错误 CHECK 必须阻止 v4") } var count int db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=4`).Scan(&count) if count != 0 { t.Fatal("自检失败不得记 v4") } } func TestMySQLMigrate_V4升级V5且断点重跑(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV4(t, db) // 模拟全部 DDL 已完成但版本尚未记录。 if err := migrateMySQLV5(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v5 重跑失败: %v", err) } var versions int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=5`).Scan(&versions); err != nil || versions != 1 { t.Fatalf("v5=%d err=%v", versions, err) } now := "2026-08-10T00:00:00Z" mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,created_at,updated_at) VALUES('OLD','collect','pending','https://example.invalid',?,?)`, now, now) var mode string if err := db.QueryRow(`SELECT execution_mode FROM tasks WHERE task_id='OLD'`).Scan(&mode); err != nil || mode != "dry_run" { t.Fatalf("历史/缺省任务模式=%q err=%v", mode, err) } if _, err := db.Exec(`UPDATE tasks SET execution_mode='invalid' WHERE task_id='OLD'`); err == nil { t.Fatal("执行模式 CHECK 必须拒绝非法值") } if _, err := db.Exec(`UPDATE tasks SET execution_mode='live' WHERE task_id='OLD'`); err == nil { t.Fatal("缺少确认审计的 live 必须被拒绝") } if _, err := db.Exec(`UPDATE tasks SET live_confirmed_by='U',live_confirmed_at=? WHERE task_id='OLD'`, now); err == nil { t.Fatal("dry_run 不得携带 live 确认审计") } } func TestMySQLMigrate_V5形状错误不记版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV4(t, db) mustExec(t, db, `ALTER TABLE tasks ADD COLUMN execution_mode VARCHAR(8) NOT NULL DEFAULT 'dry_run'`) if err := MigrateMySQL(db); err == nil { t.Fatal("错误 execution_mode 形状必须阻止 v5") } var count int db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=5`).Scan(&count) if count != 0 { t.Fatal("v5 自检失败不得记录版本") } } func TestMySQLMigrate_V5升级V6且断点重跑(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV4(t, db) if err := migrateMySQLV5(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (5,'2026-08-10T00:00:00Z')`) now := "2026-08-10T00:00:00Z" mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,created_at,updated_at) VALUES('HISTORY','collect','pending','https://example.invalid',?,?)`, now, now) // 模拟 DDL 已提交但版本号尚未写入,再启动必须能够收敛。 if err := migrateMySQLV6(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v6 重跑失败: %v", err) } var versions int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=6`).Scan(&versions); err != nil || versions != 1 { t.Fatalf("v6=%d err=%v", versions, err) } var creator sql.NullString if err := db.QueryRow(`SELECT created_by_user_id FROM tasks WHERE task_id='HISTORY'`).Scan(&creator); err != nil || creator.Valid { t.Fatalf("存量任务应保持历史任务 NULL: creator=%+v err=%v", creator, err) } } func TestMySQLMigrate_V6形状错误不记版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV4(t, db) if err := migrateMySQLV5(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (5,'2026-08-10T00:00:00Z')`) mustExec(t, db, `ALTER TABLE tasks ADD COLUMN created_by_user_id VARCHAR(32) NULL`) if err := MigrateMySQL(db); err == nil { t.Fatal("错误 created_by_user_id 形状必须阻止 v6") } var count int db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=6`).Scan(&count) if count != 0 { t.Fatal("v6 自检失败不得记录版本") } } func TestMySQLMigrate_V6升级V7且断点重跑(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV6(t, db) // 模拟 MySQL DDL 已经提交、版本号尚未记录的中断状态。 if err := migrateMySQLV7(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v7 重跑失败: %v", err) } var versions int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=7`).Scan(&versions); err != nil || versions != 1 { t.Fatalf("v7=%d err=%v", versions, err) } mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) VALUES('API-1','接口商品','api','2026-08-11T00:00:00Z','2026-08-11T00:00:00Z')`) } func TestMySQLMigrate_V7形状错误不记版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV6(t, db) mustExec(t, db, `ALTER TABLE shopee_products ADD COLUMN source_observed_at VARCHAR(10) NULL`) if err := MigrateMySQL(db); err == nil { t.Fatal("错误 source_observed_at 形状必须阻止 v7") } var count int db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=7`).Scan(&count) if count != 0 { t.Fatal("v7 自检失败不得记录版本") } } func TestMySQLMigrate_V7升级V8保留内部引用且可重放(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV7(t, db) now := "2026-08-11T00:00:00Z" mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) VALUES('S-1','商品','report',?,?)`, now, now) mustExec(t, db, `INSERT INTO shopee_skus(sku_id,goods_id,spec_raw,created_at,updated_at) VALUES('REAL-1','S-1',' 黑色, M ',?,?)`, now, now) // 模拟全部 DDL/回填已提交,但 schema_migrations 尚未记录 v8。 if err := migrateMySQLV8(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v8 重跑失败: %v", err) } var internalID, externalID, key string if err := db.QueryRow(`SELECT sku_id,shopee_sku_id,spec_key FROM shopee_skus`).Scan(&internalID, &externalID, &key); err != nil { t.Fatal(err) } if internalID != "REAL-1" || externalID != "REAL-1" || key != "黑色, M" { t.Fatalf("历史身份未保留:internal=%q external=%q key=%q", internalID, externalID, key) } } func TestMySQLMigrate_V8重复规格停止且不记版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV7(t, db) now := "2026-08-11T00:00:00Z" mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) VALUES('S-1','商品','report',?,?)`, now, now) mustExec(t, db, `INSERT INTO shopee_skus(sku_id,goods_id,spec_raw,created_at,updated_at) VALUES('A','S-1','黑色,M',?,?),('B','S-1','黑色,M',?,?)`, now, now, now, now) if err := MigrateMySQL(db); err == nil || !strings.Contains(err.Error(), "重复规格") { t.Fatalf("应明确阻止冲突迁移:%v", err) } var count int db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=8`).Scan(&count) if count != 0 { t.Fatal("冲突迁移不得记录 v8") } } func TestMySQLMigrate_V8升级V9且断点重放(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV7(t, db) if err := migrateMySQLV8(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES(8,'2026-08-11T00:00:00Z')`) if err := migrateMySQLV9(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v9 重放失败: %v", err) } mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,image_url,shopee_shop_name,image_is_manual,shop_name_is_manual,created_at,updated_at) VALUES('S','商品','api','https://example.com/a.jpg','店铺',0,0,'2026-08-11T00:00:00Z','2026-08-11T00:00:00Z')`) } func TestMySQLMigrate_V9形状错误不记版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV8(t, db) mustExec(t, db, `ALTER TABLE shopee_products ADD COLUMN image_url VARCHAR(10) NULL`) if err := MigrateMySQL(db); err == nil { t.Fatal("错误 image_url 形状必须阻止 v9") } var count int db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=9`).Scan(&count) if count != 0 { t.Fatal("v9 自检失败不得记录版本") } } func TestMySQLMigrate_已记录V9但缺少V8V9结构时由V10修复(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV7(t, db) now := "2026-08-11T00:00:00Z" mustExec(t, db, `INSERT INTO shopee_products(goods_id,title,source,created_at,updated_at) VALUES('S-1','商品','report',?,?)`, now, now) mustExec(t, db, `INSERT INTO shopee_skus(sku_id,goods_id,spec_raw,color,size,advice,parse_ok,sku_code,is_manual,source_observed_at,created_at,updated_at) VALUES('REAL-1','S-1',' 黑色, M ','黑色','M','建议',1,'CODE-1',1,?,?,?)`, now, now, now) if err := migrateMySQLV8(db); err != nil { t.Fatal(err) } // 模拟 #141 开发中间状态:版本已继续推进到 v9,但两个后加入的 v8 字段不存在。 mustExec(t, db, `ALTER TABLE shopee_skus DROP COLUMN field_observed_at, DROP COLUMN field_sources`) if err := migrateMySQLV9(db); err != nil { t.Fatal(err) } // 同时模拟 v9 已记录但两个图片/店铺人工标记约束尚未建立。 mustExec(t, db, `ALTER TABLE shopee_products DROP CHECK chk_shopee_products_image_manual`) mustExec(t, db, `ALTER TABLE shopee_products DROP CHECK chk_shopee_products_shop_manual`) mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (8,'2026-08-11T00:00:00Z'),(9,'2026-08-11T00:00:00Z')`) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v10 重放失败: %v", err) } var internalID, externalID, key, color, size, advice, skuCode, colorSource, colorObserved string var manual int err := db.QueryRow(`SELECT sku_id,shopee_sku_id,spec_key,color,size,advice,sku_code,is_manual, JSON_UNQUOTE(JSON_EXTRACT(field_sources,'$.color')), JSON_UNQUOTE(JSON_EXTRACT(field_observed_at,'$.color')) FROM shopee_skus WHERE sku_id='REAL-1'`). Scan(&internalID, &externalID, &key, &color, &size, &advice, &skuCode, &manual, &colorSource, &colorObserved) if err != nil { t.Fatal(err) } if internalID != "REAL-1" || externalID != "REAL-1" || key != "黑色, M" || color != "黑色" || size != "M" || advice != "建议" || skuCode != "CODE-1" || manual != 1 { t.Fatalf("v10 修改了既有 SKU 业务字段:id=%q external=%q key=%q color=%q size=%q advice=%q code=%q manual=%d", internalID, externalID, key, color, size, advice, skuCode, manual) } if colorSource != "report" || colorObserved != now { t.Fatalf("v10 来源回填不正确:source=%q observed=%q", colorSource, colorObserved) } var versionCount int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=10`).Scan(&versionCount); err != nil { t.Fatal(err) } if versionCount != 1 { t.Fatalf("v10 应只记录一次,实际 %d", versionCount) } } func TestMySQLMigrate_V10执行后未记版本可继续收敛(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) prepareMySQLV8(t, db) if err := migrateMySQLV9(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES(9,'2026-08-11T00:00:00Z')`) // 模拟 v10 的 DDL/回填已隐式提交、但版本号尚未记录时进程退出。 if err := migrateMySQLV10(db); err != nil { t.Fatal(err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v10 中断后重跑失败: %v", err) } var versionCount int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=10`).Scan(&versionCount); err != nil { t.Fatal(err) } if versionCount != 1 { t.Fatalf("v10 应只记录一次,实际 %d", versionCount) } } func TestMySQLMigrate_V11回填顺运宝店铺且可重放(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } mustExec(t, db, `DELETE FROM schema_migrations WHERE version=11`) mustExec(t, db, `ALTER TABLE syb_orders DROP COLUMN shop_name`) now := model.NowISO() mustExec(t, db, `INSERT INTO syb_orders (syb_id,order_no,title,product_spec,quantity,syb_data,created_at,updated_at) VALUES ('S1','O1','商品','黑色,M',1,'{"stock":{"shopName":" 测试店铺 "}}',?,?), ('S2','O2','商品','白色,L',1,'{not-json',?,?)`, now, now, now, now) if err := MigrateMySQL(db); err != nil { t.Fatalf("v10 升级 v11 失败: %v", err) } var shop sql.NullString if err := db.QueryRow(`SELECT shop_name FROM syb_orders WHERE syb_id='S1'`).Scan(&shop); err != nil || !shop.Valid || shop.String != "测试店铺" { t.Fatalf("MySQL 店铺回填错误: %+v err=%v", shop, err) } if err := db.QueryRow(`SELECT shop_name FROM syb_orders WHERE syb_id='S2'`).Scan(&shop); err != nil || shop.Valid { t.Fatalf("无效 JSON 应保持 NULL: %+v err=%v", shop, err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v11 重放失败: %v", err) } var count int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=11`).Scan(&count); err != nil || count != 1 { t.Fatalf("v11 版本记录错误: count=%d err=%v", count, err) } } func TestMySQLMigrate_V11升级V12并修复孤儿采集中状态(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } // 模拟已在 v11 的生产库:去掉 v12 版本及其附加表,保留全部 v1-v11 结构。 mustExec(t, db, `DELETE FROM schema_migrations WHERE version=12`) mustExec(t, db, `DROP TABLE task_syb_sources`) now := model.NowISO() mustExec(t, db, `INSERT INTO pdd_products(goods_id,url,collect_status,created_at,updated_at) VALUES ('ORPHAN','https://mobile.yangkeduo.com/goods.html?goods_id=1','collecting',?,?), ('ACTIVE','https://mobile.yangkeduo.com/goods.html?goods_id=2','collecting',?,?)`, now, now, now, now) mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,pdd_goods_id,created_at,updated_at) VALUES('COL-ACTIVE','collect','pending','https://mobile.yangkeduo.com/goods.html?goods_id=2','ACTIVE',?,?)`, now, now) if err := MigrateMySQL(db); err != nil { t.Fatalf("v11 升级 v12 失败: %v", err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v12 重放失败: %v", err) } var orphanStatus, activeStatus string if err := db.QueryRow(`SELECT collect_status FROM pdd_products WHERE goods_id='ORPHAN'`).Scan(&orphanStatus); err != nil { t.Fatal(err) } if err := db.QueryRow(`SELECT collect_status FROM pdd_products WHERE goods_id='ACTIVE'`).Scan(&activeStatus); err != nil { t.Fatal(err) } if orphanStatus != "pending" || activeStatus != "collecting" { t.Fatalf("v12 状态修复错误:orphan=%s active=%s", orphanStatus, activeStatus) } var versionCount int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=12`).Scan(&versionCount); err != nil || versionCount != 1 { t.Fatalf("v12 版本记录错误: count=%d err=%v", versionCount, err) } } func TestMySQLMigrate_V12形状错误不记录版本(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } mustExec(t, db, `DELETE FROM schema_migrations WHERE version=12`) mustExec(t, db, `DROP TABLE task_syb_sources`) mustExec(t, db, `CREATE TABLE task_syb_sources ( task_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, syb_id VARCHAR(191) COLLATE utf8mb4_bin NOT NULL, created_at VARCHAR(35) NOT NULL, PRIMARY KEY(syb_id,task_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci`) if err := MigrateMySQL(db); err == nil { t.Fatal("错误主键和缺失外键必须阻止 v12") } var versionCount int if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migrations WHERE version=12`).Scan(&versionCount); err != nil || versionCount != 0 { t.Fatalf("v12 自检失败时不得记录版本:count=%d err=%v", versionCount, err) } } func TestMySQLMigrate_V12升级V13迁移任务主键和关联(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } // 模拟生产 v12:移除 v13 表、版本和新增的级联更新外键。 mustExec(t, db, `DELETE FROM schema_migrations WHERE version=13`) mustExec(t, db, `DROP TABLE task_sequences`) mustExec(t, db, `ALTER TABLE task_claims DROP FOREIGN KEY fk_task_claims_task`) mustExec(t, db, `ALTER TABLE task_syb_sources DROP FOREIGN KEY fk_task_syb_sources_task`) mustExec(t, db, `ALTER TABLE task_syb_sources ADD CONSTRAINT fk_task_syb_sources_task FOREIGN KEY(task_id) REFERENCES tasks(task_id) ON DELETE CASCADE`) mustExec(t, db, `INSERT INTO syb_orders(syb_id,order_no,title,product_spec,quantity,syb_data,created_at,updated_at) VALUES('SYB-V13','ORDER-SOURCE','商品','黑色,M',1,'{}','2026-08-12T00:00:00Z','2026-08-12T00:00:00Z')`) mustExec(t, db, `INSERT INTO tasks(task_id,task_type,status,pdd_goods_url,pdd_goods_id,order_no,result_data,created_at,updated_at) VALUES ('OLD-C-LATE','collect','succeeded','https://example.invalid/c2','C-LATE',NULL,NULL,'2026-08-12T02:00:00Z','2026-08-12T02:00:00Z'), ('OLD-C-FIRST','collect','assigned','https://example.invalid/c1','C-FIRST',NULL,NULL,'2026-08-12T01:00:00Z','2026-08-12T01:00:00Z'), ('OLD-P-LATE','purchase','succeeded','https://example.invalid/p2','P-LATE','ORDER-P-LATE','{\"purchase\":{\"order_no\":\"PDD-2\",\"ordered_at\":\"2026-08-12T02:00:00Z\"}}','2026-08-12T04:00:00Z','2026-08-12T04:00:00Z'), ('OLD-P-FIRST','purchase','succeeded','https://example.invalid/p1','P-FIRST','ORDER-P-FIRST','{\"purchase\":{\"order_no\":\"PDD-1\",\"ordered_at\":\"2026-08-12T01:00:00Z\"}}','2026-08-12T03:00:00Z','2026-08-12T03:00:00Z')`) mustExec(t, db, `INSERT INTO task_claims(task_id,client_id,claimed_at) VALUES ('OLD-C-FIRST','CLIENT-1','2026-08-12T01:10:00Z'), ('OLD-P-FIRST','CLIENT-1','2026-08-12T03:10:00Z'), ('DELETED-TASK','CLIENT-OLD','2026-08-01T00:00:00Z')`) mustExec(t, db, `INSERT INTO task_syb_sources(task_id,syb_id,created_at) VALUES('OLD-C-FIRST','SYB-V13','2026-08-12T01:00:00Z')`) if err := MigrateMySQL(db); err != nil { t.Fatalf("v12 升级 v13 失败: %v", err) } wants := map[string]string{"C-FIRST": "cj1", "C-LATE": "cj2", "P-FIRST": "cg1", "P-LATE": "cg2"} for goodsID, wantID := range wants { var taskID string if err := db.QueryRow(`SELECT task_id FROM tasks WHERE pdd_goods_id=?`, goodsID).Scan(&taskID); err != nil || taskID != wantID { t.Fatalf("商品 %s 任务编号=%q err=%v,期望 %s", goodsID, taskID, err, wantID) } } var claimCount, orphanCount, sourceCount int if err := db.QueryRow(`SELECT COUNT(*) FROM task_claims WHERE task_id IN ('cj1','cg1')`).Scan(&claimCount); err != nil || claimCount != 2 { t.Fatalf("有效领取历史未迁移: count=%d err=%v", claimCount, err) } if err := db.QueryRow(`SELECT COUNT(*) FROM task_claims c LEFT JOIN tasks t ON t.task_id=c.task_id WHERE t.task_id IS NULL`).Scan(&orphanCount); err != nil || orphanCount != 0 { t.Fatalf("孤儿领取历史未清理: count=%d err=%v", orphanCount, err) } if err := db.QueryRow(`SELECT COUNT(*) FROM task_syb_sources WHERE task_id='cj1' AND syb_id='SYB-V13'`).Scan(&sourceCount); err != nil || sourceCount != 1 { t.Fatalf("顺运宝来源未迁移: count=%d err=%v", sourceCount, err) } var resultData string if err := db.QueryRow(`SELECT result_data FROM tasks WHERE task_id='cg1'`).Scan(&resultData); err != nil || !strings.Contains(resultData, `"order_no":"PDD-1"`) { t.Fatalf("采购结果不得变化: result=%q err=%v", resultData, err) } for taskType, want := range map[model.TaskType]int64{model.TaskCollect: 2, model.TaskPurchase: 2} { var last int64 if err := db.QueryRow(`SELECT current_value FROM task_sequences WHERE task_type=?`, taskType).Scan(&last); err != nil || last != want { t.Fatalf("%s 序列=%d err=%v,期望 %d", taskType, last, err, want) } } if err := MigrateMySQL(db); err != nil { t.Fatalf("v13 重放失败: %v", err) } var stableID string if err := db.QueryRow(`SELECT task_id FROM tasks WHERE pdd_goods_id='C-FIRST'`).Scan(&stableID); err != nil || stableID != "cj1" { t.Fatalf("重放不得二次改号: id=%s err=%v", stableID, err) } } func TestNextTaskID_MySQL并发独立递增且回滚不耗号(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } const workers = 16 ids := make(map[string]bool, workers) var mu sync.Mutex errCh := make(chan error, workers) var wg sync.WaitGroup for i := 0; i < workers; i++ { wg.Add(1) go func() { defer wg.Done() tx, err := db.Begin() if err != nil { errCh <- err return } id, err := NextTaskID(tx, model.TaskCollect) if err == nil { err = tx.Commit() } else { tx.Rollback() } if err != nil { errCh <- err return } mu.Lock() ids[id] = true mu.Unlock() }() } wg.Wait() close(errCh) for err := range errCh { t.Fatal(err) } if len(ids) != workers { t.Fatalf("并发分配得到 %d 个唯一编号,期望 %d", len(ids), workers) } for i := 1; i <= workers; i++ { if !ids[fmt.Sprintf("cj%d", i)] { t.Fatalf("并发编号缺少 cj%d: %+v", i, ids) } } rollbackTx, err := db.Begin() if err != nil { t.Fatal(err) } rolledBackID, err := NextTaskID(rollbackTx, model.TaskPurchase) if err != nil { t.Fatal(err) } if err := rollbackTx.Rollback(); err != nil { t.Fatal(err) } commitTx, err := db.Begin() if err != nil { t.Fatal(err) } committedID, err := NextTaskID(commitTx, model.TaskPurchase) if err != nil { t.Fatal(err) } if err := commitTx.Commit(); err != nil { t.Fatal(err) } if rolledBackID != "cg1" || committedID != "cg1" { t.Fatalf("回滚不应消耗采购序号: rollback=%s commit=%s", rolledBackID, committedID) } } func TestMySQLMigrate_V13升级V14并可重放(t *testing.T) { db := openMySQLMigrationTestDB(t) defer db.Close() cleanMySQLTestSchema(t, db) defer cleanMySQLTestSchema(t, db) if err := MigrateMySQL(db); err != nil { t.Fatal(err) } // 模拟生产 v13,并保留一条旧同步记录核对回填。 mustExec(t, db, `DELETE FROM syb_allowed_shops`) mustExec(t, db, `ALTER TABLE syb_sync_runs DROP COLUMN shop_filter_hash`) mustExec(t, db, `ALTER TABLE syb_sync_runs DROP COLUMN shop_skipped_count`) mustExec(t, db, `ALTER TABLE syb_sync_runs DROP COLUMN accepted_stock_count`) mustExec(t, db, `DROP TABLE syb_allowed_shops`) mustExec(t, db, `DELETE FROM schema_migrations WHERE version=14`) mustExec(t, db, `INSERT INTO users(user_id,username,password_hash,role,status,password_changed_at,created_at,updated_at) VALUES('V14-USER','v14-user','x','admin','active','2026-08-12T00:00:00Z','2026-08-12T00:00:00Z','2026-08-12T00:00:00Z')`) mustExec(t, db, `INSERT INTO syb_sync_runs(run_id,user_id,date_from,date_to,status,stock_count,started_at,finished_at) VALUES('V14-RUN','V14-USER','2026-08-11','2026-08-11','succeeded',7,'2026-08-12T00:00:00Z','2026-08-12T00:01:00Z')`) if err := MigrateMySQL(db); err != nil { t.Fatalf("v13 升级 v14 失败: %v", err) } var accepted, skipped int if err := db.QueryRow(`SELECT accepted_stock_count,shop_skipped_count FROM syb_sync_runs WHERE run_id='V14-RUN'`).Scan(&accepted, &skipped); err != nil || accepted != 7 || skipped != 0 { t.Fatalf("旧同步记录回填错误: accepted=%d skipped=%d err=%v", accepted, skipped, err) } if err := MigrateMySQL(db); err != nil { t.Fatalf("v14 重放失败: %v", err) } if err := checkMySQLV14Shape(db); err != nil { t.Fatalf("v14 自检失败: %v", err) } } func openMySQLMigrationTestDB(t *testing.T) *sql.DB { t.Helper() if os.Getenv("CMAUTOBUY_MYSQL_TEST") != "1" { t.Skip("未启用真实 MySQL 8 集成测试") } cfg, err := config.LoadDatabaseFromEnv() if err != nil { t.Fatal(err) } if !strings.HasSuffix(cfg.Name, "_test") { t.Fatalf("拒绝使用非 _test 数据库 %q", cfg.Name) } db, err := OpenMySQL(cfg) if err != nil { t.Fatal(err) } return db } func prepareMySQLV2(t *testing.T, db *sql.DB) { t.Helper() mustExec(t, db, `CREATE TABLE schema_migrations (version INT PRIMARY KEY, applied_at VARCHAR(35) NOT NULL) ENGINE=InnoDB`) for _, statement := range mysqlSchemaV1 { mustExec(t, db, statement) } for _, statement := range mysqlSchemaV2 { mustExec(t, db, statement) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (1,'2026-08-10T00:00:00Z'),(2,'2026-08-10T00:00:00Z')`) } func prepareMySQLV4(t *testing.T, db *sql.DB) { t.Helper() prepareMySQLV2(t, db) if err := migrateMySQLV3(db); err != nil { t.Fatal(err) } mustExec(t, db, mysqlSchemaV4Decisions) mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (3,'2026-08-10T00:00:00Z'),(4,'2026-08-10T00:00:00Z')`) } func prepareMySQLV6(t *testing.T, db *sql.DB) { t.Helper() prepareMySQLV4(t, db) if err := migrateMySQLV5(db); err != nil { t.Fatal(err) } if err := migrateMySQLV6(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (5,'2026-08-10T00:00:00Z'),(6,'2026-08-10T00:00:00Z')`) } func prepareMySQLV7(t *testing.T, db *sql.DB) { t.Helper() prepareMySQLV6(t, db) if err := migrateMySQLV7(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES (7,'2026-08-11T00:00:00Z')`) } func prepareMySQLV8(t *testing.T, db *sql.DB) { t.Helper() prepareMySQLV7(t, db) if err := migrateMySQLV8(db); err != nil { t.Fatal(err) } mustExec(t, db, `INSERT INTO schema_migrations(version,applied_at) VALUES(8,'2026-08-11T00:00:00Z')`) } func mustExec(t *testing.T, db *sql.DB, query string, args ...any) { t.Helper() if _, err := db.Exec(query, args...); err != nil { t.Fatalf("执行 SQL 失败: %v", err) } } func cleanMySQLTestSchema(t *testing.T, db *sql.DB) { t.Helper() rows, err := db.Query(`SELECT table_name FROM information_schema.tables WHERE table_schema = DATABASE() AND table_type = 'BASE TABLE'`) if err != nil { t.Fatal(err) } var tables []string for rows.Next() { var table string if err := rows.Scan(&table); err != nil { rows.Close() t.Fatal(err) } tables = append(tables, table) } if err := rows.Close(); err != nil { t.Fatal(err) } if _, err := db.Exec(`SET FOREIGN_KEY_CHECKS = 0`); err != nil { t.Fatal(err) } defer db.Exec(`SET FOREIGN_KEY_CHECKS = 1`) for _, table := range tables { // 表名只来自当前 _test 数据库的 information_schema,并对反引号转义。 quoted := "`" + strings.ReplaceAll(table, "`", "``") + "`" if _, err := db.Exec("DROP TABLE " + quoted); err != nil { t.Fatal(fmt.Errorf("清理 MySQL 测试表 %s 失败: %w", table, err)) } } }