material_manage_system/internal/repository/company_isolation_integration_test.go
2026-10-08 17:04:09 +08:00

453 lines
20 KiB
Go

package repository
import (
"database/sql"
"fmt"
"os"
"path/filepath"
"reflect"
"strings"
"sync"
"testing"
"time"
"github.com/go-sql-driver/mysql"
"github.com/wmms/wmms/pkg/database"
)
// Each test creates its own disposable schema on the explicitly configured server.
func isolationDB(t *testing.T, legacy bool) *sql.DB {
t.Helper()
dsn := os.Getenv("TEST_DB_DSN")
if dsn == "" {
t.Skip("real MySQL integration requires TEST_DB_DSN")
}
cfg, err := mysql.ParseDSN(dsn)
if err != nil {
t.Fatal(err)
}
if !strings.HasPrefix(cfg.DBName, "company_isolation_test") {
t.Fatal("TEST_DB_DSN must name a company_isolation_test schema")
}
cfg.DBName = ""
cfg.MultiStatements = true
cfg.ParseTime = true
admin, err := sql.Open("mysql", cfg.FormatDSN())
if err != nil {
t.Fatal(err)
}
name := fmt.Sprintf("company_isolation_test_%d", time.Now().UnixNano())
if _, err = admin.Exec("CREATE DATABASE `" + name + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); err != nil {
admin.Close()
t.Fatal(err)
}
cfg.DBName = name
db, err := sql.Open("mysql", cfg.FormatDSN())
if err != nil {
t.Fatal(err)
}
original := database.DB
database.DB = db
t.Cleanup(func() { database.DB = original; db.Close(); admin.Exec("DROP DATABASE `" + name + "`"); admin.Close() })
path := filepath.Join("..", "..", "scripts", "init.sql")
if legacy {
path = filepath.Join("testdata", "company-isolation-legacy.sql")
}
content, err := os.ReadFile(path)
if err != nil {
t.Fatal(err)
}
var lines []string
for _, line := range strings.Split(strings.TrimPrefix(string(content), "\ufeff"), "\n") {
if strings.HasPrefix(line, "CREATE DATABASE") || strings.HasPrefix(line, "USE ") {
continue
}
lines = append(lines, line)
}
if _, err = db.Exec(strings.Join(lines, "\n")); err != nil {
t.Fatalf("initialize %s: %v", path, err)
}
return db
}
func mustSQL(t *testing.T, db *sql.DB, query string, args ...interface{}) {
t.Helper()
if _, err := db.Exec(query, args...); err != nil {
t.Fatalf("fixture SQL: %v\n%s", err, query)
}
}
// DELIMITER is a mysql-client directive, not server SQL. Execute the same file's
// statements with its declared delimiter, including the full stored procedure.
func runIsolationSQL(db *sql.DB, path string) error {
content, err := os.ReadFile(path)
if err != nil {
return err
}
delimiter := ";"
var statement strings.Builder
for _, line := range strings.Split(string(content), "\n") {
trimmed := strings.TrimSpace(line)
if strings.HasPrefix(trimmed, "DELIMITER ") {
delimiter = strings.TrimSpace(strings.TrimPrefix(trimmed, "DELIMITER "))
continue
}
if strings.HasPrefix(trimmed, "--") || trimmed == "" {
continue
}
statement.WriteString(line)
statement.WriteByte('\n')
accumulated := strings.TrimSpace(statement.String())
if strings.HasSuffix(accumulated, delimiter) {
accumulated = strings.TrimSuffix(accumulated, delimiter)
if _, err = db.Exec(accumulated); err != nil {
return err
}
statement.Reset()
}
}
if strings.TrimSpace(statement.String()) != "" {
return fmt.Errorf("unterminated SQL in %s", path)
}
return nil
}
func upgradeIsolation(db *sql.DB) error {
return runIsolationSQL(db, filepath.Join("..", "..", "scripts", "upgrades", "2026-10-08-company-master-data-isolation.sql"))
}
func seedIsolationHistory(t *testing.T, db *sql.DB) {
mustSQL(t, db, `INSERT INTO organizations(id,code,name) VALUES (1,'ONE','One'),(2,'TWO','Two');
INSERT INTO users(id,username,password) VALUES (1,'ordinary','x'),(2,'admin','x');
INSERT INTO roles(id,code,name) VALUES(1,'super_admin','Admin'); INSERT INTO user_roles VALUES(2,1,NOW());
INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(1,2,1);
INSERT INTO material_categories(id,organization_id,code,name) VALUES(1,1,'C','C'),(2,2,'C','C');
INSERT INTO units(id,organization_id,code,name) VALUES(1,1,'U','U'),(2,2,'U','U');
INSERT INTO materials(id,organization_id,material_code,name,category_id,base_unit_id) VALUES(1,1,'M','M',1,1),(2,2,'M','M',2,2);
INSERT INTO warehouses(id,organization_id,code,name) VALUES(1,1,'W','W'),(2,2,'W','W');
INSERT INTO warehouse_locations(id,warehouse_id,code,name) VALUES(1,1,'L','L'),(2,2,'L','L');
INSERT INTO inventory_balances(organization_id,warehouse_id,location_id,material_id,on_hand,reserved,locked,in_transit) VALUES(2,2,2,2,8,1,2,3);
INSERT INTO inbound_orders(id,organization_id,warehouse_id,location_id,inbound_date,handler) VALUES(1,2,2,2,NOW(),'H');
INSERT INTO inbound_items(inbound_order_id,material_id,quantity) VALUES(1,2,10);
INSERT INTO outbound_orders(id,organization_id,warehouse_id,location_id,outbound_date,handler,receiver) VALUES(1,2,2,2,NOW(),'H','R');
INSERT INTO outbound_items(outbound_order_id,material_id,quantity) VALUES(1,2,2);
INSERT INTO material_applications(id,user_id) VALUES(1,1);
INSERT INTO application_items(application_id,material_id,quantity) VALUES(1,2,2);
INSERT INTO stock_logs(material_id,type,quantity,related_order_type,related_order_id) VALUES(2,'in',10,'inbound',1),(2,'out',2,'outbound',1);`)
}
func assertIsolationConstraints(t *testing.T, db *sql.DB) {
t.Helper()
mustSQL(t, db, `INSERT INTO organizations(id,code,name) VALUES(10,'TEN','Ten'),(11,'ELEVEN','Eleven'),(12,'TWELVE','Twelve');
INSERT INTO users(id,username,password) VALUES(10,'constraint','x');
INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(10,10,1);`)
if _, err := db.Exec(`INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(10,11,1)`); err == nil {
t.Fatal("second default organization was accepted")
}
mustSQL(t, db, `INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(10,11,0),(10,12,0)`)
mustSQL(t, db, `INSERT INTO users(id,username,password) VALUES(11,'default-race','x')`)
var defaultWG sync.WaitGroup
defaultStart := make(chan struct{})
defaultResults := make(chan error, 2)
for _, organizationID := range []int64{10, 11} {
defaultWG.Add(1)
go func(id int64) {
defer defaultWG.Done()
<-defaultStart
_, err := db.Exec(`INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(11,?,1)`, id)
defaultResults <- err
}(organizationID)
}
close(defaultStart)
defaultWG.Wait()
close(defaultResults)
defaultSuccesses := 0
for err := range defaultResults {
if err == nil {
defaultSuccesses++
} else if me, ok := err.(*mysql.MySQLError); !ok || me.Number != 1062 {
t.Fatalf("unexpected default race error: %v", err)
}
}
if defaultSuccesses != 1 {
t.Fatalf("concurrent default successes=%d, want 1", defaultSuccesses)
}
mustSQL(t, db, `INSERT INTO warehouses(id,organization_id,code,name,status) VALUES(10,10,'W','W','inactive'),(11,11,'W','W','active');
INSERT INTO warehouse_locations(id,warehouse_id,code,name,status) VALUES(10,10,'L','L','inactive'),(11,11,'L','L','active');`)
for _, query := range []string{`INSERT INTO warehouses(organization_id,code,name) VALUES(10,'SECOND','Second')`, `INSERT INTO warehouse_locations(warehouse_id,code,name) VALUES(10,'SECOND','Second')`} {
if _, err := db.Exec(query); err == nil {
t.Fatalf("unique dimension bypassed: %s", query)
}
}
// Race two independent connections; only one creation may commit.
for _, query := range []string{`INSERT INTO warehouses(organization_id,code,name) VALUES(12,?,?)`, `INSERT INTO warehouse_locations(warehouse_id,code,name) SELECT id,?,? FROM warehouses WHERE organization_id=12`} {
var wg sync.WaitGroup
start := make(chan struct{})
results := make(chan error, 2)
for i := 0; i < 2; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
<-start
_, err := db.Exec(query, fmt.Sprint(i), fmt.Sprint(i))
results <- err
}(i)
}
close(start)
wg.Wait()
close(results)
successes := 0
for err := range results {
if err == nil {
successes++
} else if me, ok := err.(*mysql.MySQLError); !ok || me.Number != 1062 {
t.Fatalf("unexpected race failure: %v", err)
}
}
if successes != 1 {
t.Fatalf("concurrent creation successes=%d, want 1", successes)
}
}
}
func TestCompanyIsolationFreshInit(t *testing.T) {
db := isolationDB(t, false)
assertIsolationConstraints(t, db)
assertIsolationRequiredDimensions(t, db)
// Explicit dimensions are mandatory; absence must fail rather than company 1.
for _, query := range []string{`INSERT INTO inbound_orders(inbound_date,handler) VALUES(NOW(),'H')`, `INSERT INTO outbound_orders(outbound_date,handler,receiver) VALUES(NOW(),'H','R')`, `INSERT INTO material_applications(user_id) VALUES(10)`, `INSERT INTO stock_logs(material_id,type,quantity,related_order_type,related_order_id) VALUES(1,'in',1,'inbound',1)`} {
_, err := db.Exec(query)
if me, ok := err.(*mysql.MySQLError); !ok || me.Number != 1364 {
t.Fatalf("missing dimensions must fail with no default (1364), got %v: %s", err, query)
}
}
if err := upgradeIsolation(db); err == nil {
t.Fatal("fresh schema must already record the migration version")
}
}
func assertIsolationRequiredDimensions(t *testing.T, db *sql.DB) {
t.Helper()
for _, tc := range []struct {
table string
columns []string
}{
{"inbound_orders", []string{"organization_id", "warehouse_id", "location_id"}},
{"outbound_orders", []string{"organization_id", "warehouse_id", "location_id"}},
{"stock_logs", []string{"organization_id", "warehouse_id", "location_id"}},
{"material_applications", []string{"organization_id"}},
} {
for _, column := range tc.columns {
var nullable string
var defaultValue sql.NullString
if err := db.QueryRow(`SELECT is_nullable,column_default FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name=? AND column_name=?`, tc.table, column).Scan(&nullable, &defaultValue); err != nil {
t.Fatal(err)
}
if nullable != "NO" || defaultValue.Valid {
t.Fatalf("%s.%s nullable=%s default=%v", tc.table, column, nullable, defaultValue)
}
}
}
}
// Schema parity covers real column types/null/default/generated expressions,
// index column order/uniqueness, and foreign keys, ignoring names/order of FKs.
func isolationSchema(t *testing.T, db *sql.DB) []string {
t.Helper()
var signature []string
for _, query := range []string{
`SELECT CONCAT('column:',table_name,':',column_name,':',column_type,':',is_nullable,':',COALESCE(column_default,'<null>'),':',extra,':',generation_expression) FROM information_schema.columns WHERE table_schema=DATABASE() ORDER BY table_name,column_name`,
`SELECT CONCAT('index:',table_name,':',index_name,':',non_unique,':',GROUP_CONCAT(column_name ORDER BY seq_in_index)) FROM information_schema.statistics WHERE table_schema=DATABASE() GROUP BY table_name,index_name,non_unique ORDER BY table_name,index_name`,
`SELECT CONCAT('fk:',table_name,':',column_name,':',referenced_table_name,':',referenced_column_name) FROM information_schema.key_column_usage WHERE table_schema=DATABASE() AND referenced_table_name IS NOT NULL ORDER BY table_name,column_name`,
} {
rows, err := db.Query(query)
if err != nil {
t.Fatal(err)
}
for rows.Next() {
var value string
if err := rows.Scan(&value); err != nil {
rows.Close()
t.Fatal(err)
}
signature = append(signature, value)
}
if err := rows.Err(); err != nil {
t.Fatal(err)
}
rows.Close()
}
return signature
}
func TestCompanyIsolationFreshAndUpgradedSchemaAgree(t *testing.T) {
fresh := isolationDB(t, false)
want := isolationSchema(t, fresh)
upgraded := isolationDB(t, true)
seedIsolationHistory(t, upgraded)
if err := upgradeIsolation(upgraded); err != nil {
t.Fatal(err)
}
got := isolationSchema(t, upgraded)
if !reflect.DeepEqual(got, want) {
wanted := map[string]bool{}
actual := map[string]bool{}
for _, v := range want {
wanted[v] = true
}
for _, v := range got {
actual[v] = true
}
for _, v := range got {
if !wanted[v] {
t.Logf("upgrade only: %s", v)
}
}
for _, v := range want {
if !actual[v] {
t.Logf("fresh only: %s", v)
}
}
t.Fatal("fresh and upgraded schema differ")
}
}
func TestCompanyIsolationApplicationLogBackfill(t *testing.T) {
db := isolationDB(t, true)
seedIsolationHistory(t, db)
mustSQL(t, db, `UPDATE material_applications SET status='approved' WHERE id=1; UPDATE stock_logs SET related_order_type='application' WHERE id=2`)
if err := upgradeIsolation(db); err != nil {
t.Fatal(err)
}
assertIsolationRequiredDimensions(t, db)
var org, warehouse, location int64
if err := db.QueryRow(`SELECT organization_id,warehouse_id,location_id FROM stock_logs WHERE id=2`).Scan(&org, &warehouse, &location); err != nil {
t.Fatal(err)
}
if org != 2 || warehouse != 2 || location != 2 {
t.Fatalf("application log=%d/%d/%d, want 2/2/2", org, warehouse, location)
}
// After upgrading, checks must also reject stored dimensions that disagree
// with the verified source document, even if each standalone FK is valid.
mustSQL(t, db, `UPDATE stock_logs SET organization_id=1 WHERE id=2`)
if err := runIsolationSQL(db, filepath.Join("..", "..", "scripts", "checks", "company-master-data-integrity.sql")); err != nil {
t.Fatal(err)
}
var anomalies int
if err := db.QueryRow(`SELECT COUNT(*) FROM tmp_company_integrity_anomalies WHERE issue='stored_stock_log_dimensions' AND entity_id=2`).Scan(&anomalies); err != nil || anomalies != 1 {
t.Fatalf("stored dimension anomaly=%d err=%v", anomalies, err)
}
mustSQL(t, db, `UPDATE material_applications SET organization_id=1 WHERE id=1`)
if err := runIsolationSQL(db, filepath.Join("..", "..", "scripts", "checks", "company-master-data-integrity.sql")); err != nil {
t.Fatal(err)
}
if err := db.QueryRow(`SELECT COUNT(*) FROM tmp_company_integrity_anomalies WHERE issue='stored_application_dimensions' AND entity_id=1`).Scan(&anomalies); err != nil || anomalies != 1 {
t.Fatalf("stored application anomaly=%d err=%v", anomalies, err)
}
}
func TestCompanyIsolationApplicationLogsRejectUnverifiedSources(t *testing.T) {
for _, tc := range []struct{ name, query string }{
{"unapproved", `UPDATE stock_logs SET related_order_type='application' WHERE id=2`},
{"quantity mismatch", `UPDATE material_applications SET status='approved' WHERE id=1; UPDATE stock_logs SET related_order_type='application',quantity=999 WHERE id=2`},
{"missing inventory", `UPDATE material_applications SET status='approved' WHERE id=1; UPDATE stock_logs SET related_order_type='application' WHERE id=2; DELETE FROM inventory_balances`},
} {
t.Run(tc.name, func(t *testing.T) {
db := isolationDB(t, true)
seedIsolationHistory(t, db)
mustSQL(t, db, tc.query)
err := upgradeIsolation(db)
if err == nil || !strings.Contains(err.Error(), "integrity anomalies") {
t.Fatalf("unverified source accepted: %v", err)
}
})
}
}
func TestCompanyIsolationLegacyUpgrade(t *testing.T) {
db := isolationDB(t, true)
seedIsolationHistory(t, db)
if err := upgradeIsolation(db); err != nil {
t.Fatal(err)
}
var org, warehouse, location int64
if err := db.QueryRow(`SELECT organization_id,warehouse_id,location_id FROM stock_logs WHERE id=1`).Scan(&org, &warehouse, &location); err != nil {
t.Fatal(err)
}
if org != 2 || warehouse != 2 || location != 2 {
t.Fatalf("log dimensions=%d/%d/%d, want 2/2/2", org, warehouse, location)
}
if err := db.QueryRow(`SELECT organization_id FROM material_applications WHERE id=1`).Scan(&org); err != nil || org != 2 {
t.Fatalf("application org=%d err=%v", org, err)
}
var count int
if err := db.QueryRow(`SELECT COUNT(*) FROM schema_migration_reconciliations WHERE version='2026-10-08-company-master-data-isolation' AND phase IN ('before','after')`).Scan(&count); err != nil || count != 4 {
t.Fatalf("snapshots=%d err=%v", count, err)
}
var onHand string
if err := db.QueryRow(`SELECT on_hand FROM schema_migration_reconciliations WHERE organization_id=2 AND phase='after'`).Scan(&onHand); err != nil || onHand != "8.000000" {
t.Fatalf("stock reconciliation=%s err=%v", onHand, err)
}
for _, phase := range []string{"before", "after"} {
var reserved, locked, inTransit string
var inbound, outbound, applications, logs int64
if err := db.QueryRow(`SELECT reserved,locked,in_transit,inbound_count,outbound_count,application_count,stock_log_count FROM schema_migration_reconciliations WHERE organization_id=2 AND phase=?`, phase).Scan(&reserved, &locked, &inTransit, &inbound, &outbound, &applications, &logs); err != nil {
t.Fatal(err)
}
if reserved != "1.000000" || locked != "2.000000" || inTransit != "3.000000" || inbound != 1 || outbound != 1 || applications != 1 || logs != 2 {
t.Fatalf("%s reconciliation=%s/%s/%s/%d/%d/%d/%d", phase, reserved, locked, inTransit, inbound, outbound, applications, logs)
}
}
var status, version string
if err := db.QueryRow(`SELECT status,JSON_UNQUOTE(JSON_EXTRACT(details,'$.version')) FROM schema_migrations WHERE version='2026-10-08-company-master-data-isolation'`).Scan(&status, &version); err != nil || status != "completed" || version != "2026-10-08-company-master-data-isolation" {
t.Fatalf("version record=%s/%s err=%v", status, version, err)
}
if err := upgradeIsolation(db); err == nil {
t.Fatal("migration rerun accepted")
}
for _, state := range []string{"running", "failed"} {
mustSQL(t, db, `UPDATE schema_migrations SET status=? WHERE version='2026-10-08-company-master-data-isolation'`, state)
if err := upgradeIsolation(db); err == nil || !strings.Contains(err.Error(), "version already recorded") {
t.Fatalf("%s version was rerun: %v", state, err)
}
}
assertIsolationConstraints(t, db)
}
func TestCompanyIsolationInvalidHistoryBlocksUpgrade(t *testing.T) {
cases := []struct{ name, query string }{
{"duplicate defaults", `INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(1,1,1)`},
{"multiple warehouses", `INSERT INTO warehouses(organization_id,code,name) VALUES(2,'SECOND','Second')`},
{"multiple locations", `INSERT INTO warehouse_locations(warehouse_id,code,name) VALUES(2,'SECOND','Second')`},
{"cross-company parent", `UPDATE material_categories SET parent_id=1 WHERE id=2`},
{"cross-company category", `UPDATE materials SET category_id=1 WHERE id=2`},
{"cross-company unit", `UPDATE materials SET base_unit_id=1 WHERE id=2`},
{"rule category", `INSERT INTO code_rules(organization_id,category_id,prefix,sequence_length) VALUES(2,1,'X',4)`},
{"no effective grant", `DELETE FROM user_organizations WHERE user_id=1`},
{"empty application", `DELETE FROM application_items WHERE application_id=1`},
{"mixed application", `INSERT INTO application_items(application_id,material_id,quantity) VALUES(1,1,1)`},
{"orphan log", `UPDATE stock_logs SET related_order_id=999 WHERE id=1`},
{"wrong log material", `UPDATE stock_logs SET material_id=1 WHERE id=1`},
{"unknown log type", `UPDATE stock_logs SET related_order_type='unknown' WHERE id=1`},
{"balance dimensions", `UPDATE inventory_balances SET warehouse_id=1 WHERE material_id=2`},
{"warning dimensions", `INSERT INTO stock_warning_rules(organization_id,warehouse_id,material_id) VALUES(2,1,2)`},
{"order dimensions", `UPDATE inbound_orders SET location_id=1 WHERE id=1`},
{"order material", `UPDATE inbound_items SET material_id=1 WHERE inbound_order_id=1`},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
db := isolationDB(t, true)
seedIsolationHistory(t, db)
mustSQL(t, db, tc.query)
err := upgradeIsolation(db)
if err == nil || !strings.Contains(err.Error(), "integrity anomalies") {
t.Fatalf("invalid history must block with integrity error, got %v", err)
}
var count int
if err := db.QueryRow(`SELECT COUNT(*) FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='stock_logs' AND column_name='organization_id'`).Scan(&count); err != nil || count != 0 {
t.Fatalf("blocked upgrade changed business schema: %d %v", count, err)
}
})
}
}