zhangwenrui/material_manage_system
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
f5e9029c46
material_manage_system/internal/handler/company_isolation_integration_test.go

736 lines
36 KiB
Go
Raw Blame History

package handler
import (
"database/sql"
"encoding/json"
"fmt"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
"github.com/wmms/wmms/internal/model"
"github.com/wmms/wmms/internal/testutil"
"golang.org/x/crypto/bcrypt"
)
// These tests add real route/auth/organization-boundary coverage. Existing
// repository/service suites own inventory concurrency and insufficient-stock
// rollback; this suite adds concurrent approval and mid-transaction failures.
// Do not parallelize: testutil swaps the process-global database.DB.
type companyHTTPFixture struct {
db *sql.DB
router *gin.Engine
tokens map[string]string
}
func companySQL(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", err)
}
}
func newCompanyHTTPFixture(t *testing.T) *companyHTTPFixture {
t.Helper()
// Requires TEST_DB_DSN with company_isolation_test prefix; creates and drops
// only a generated disposable schema. An unset DSN is a skip, not acceptance.
db := testutil.NewIsolationDB(t)
companySQL(t, db, `INSERT INTO organizations(id,code,name) VALUES(301,'HTTP_BJ','北京公司'),(302,'HTTP_WH','武汉公司');
INSERT INTO users(id,username,password) VALUES(301,'http-beijing','x'),(302,'http-wuhan','x'),(303,'http-both','x'),(304,'http-admin','x');
INSERT INTO user_roles(user_id,role_id) SELECT u.id,r.id FROM users u CROSS JOIN roles r WHERE u.id IN(301,302,303) AND r.code='warehouse_admin';
INSERT INTO user_roles(user_id,role_id) SELECT 304,id FROM roles WHERE code='super_admin';
INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(301,301,1),(302,302,1),(303,301,1),(303,302,0);
INSERT INTO material_categories(id,organization_id,code,name) VALUES(301,301,'BJ','北京机械'),(302,302,'WH','武汉电气');
INSERT INTO material_categories(id,organization_id,code,name,parent_id) VALUES(311,301,'BJ01','北京标准件',301),(312,302,'WH01','武汉元器件',302);
INSERT INTO units(id,organization_id,code,name) VALUES(301,301,'PCS','件'),(302,302,'KG','千克');
INSERT INTO code_rules(id,organization_id,category_id,prefix,sequence_length,next_sequence) VALUES(301,301,311,'BJ',4,1),(302,302,312,'WH',4,1);
INSERT INTO materials(id,organization_id,material_code,name,spec,unit,category_id,base_unit_id) VALUES(301,301,'SHARED','北京同编码物料','','件',311,301),(302,302,'SHARED','武汉同编码物料','','千克',312,302);
INSERT INTO warehouses(id,organization_id,code,name) VALUES(301,301,'BJ_W','北京仓库'),(302,302,'WH_W','武汉仓库');
INSERT INTO warehouse_locations(id,warehouse_id,code,name) VALUES(301,301,'BJ_L','北京库位'),(302,302,'WH_L','武汉库位');`)
password, err := bcrypt.GenerateFromPassword([]byte("CompanyTest123!"), bcrypt.MinCost)
if err != nil {
t.Fatal(err)
}
companySQL(t, db, `UPDATE users SET password=? WHERE id BETWEEN 301 AND 304`, string(password))
gin.SetMode(gin.TestMode)
f := &companyHTTPFixture{db: db, router: gin.New(), tokens: map[string]string{}}
NewHandler().RegisterRoutes(f.router)
for _, name := range []string{"beijing", "wuhan", "both", "admin"} {
var login struct {
Token string `json:"token"`
}
data := f.request(t, "", "", "POST", "/api/auth/login", fmt.Sprintf(`{"username":"http-%s","password":"CompanyTest123!"}`, name), 200)
companyDecode(t, data, &login)
if login.Token == "" {
t.Fatalf("login %s returned no token", name)
}
f.tokens[name] = login.Token
}
return f
}
func (f *companyHTTPFixture) serve(account, org, method, path, body string) *httptest.ResponseRecorder {
req := httptest.NewRequest(method, path, strings.NewReader(body))
if account != "" {
req.Header.Set("Authorization", "Bearer "+f.tokens[account])
}
if org != "" {
req.Header.Set("X-Organization-ID", org)
}
if body != "" {
req.Header.Set("Content-Type", "application/json")
}
w := httptest.NewRecorder()
f.router.ServeHTTP(w, req)
return w
}
func (f *companyHTTPFixture) request(t *testing.T, account, org, method, path, body string, want int) json.RawMessage {
t.Helper()
w := f.serve(account, org, method, path, body)
if w.Code != want {
t.Fatalf("%s %s account=%s org=%s: status=%d want=%d body=%s", method, path, account, org, w.Code, want, w.Body.String())
}
var envelope struct {
Code int `json:"code"`
Data json.RawMessage `json:"data"`
}
companyDecode(t, w.Body.Bytes(), &envelope)
if want < 300 && envelope.Code != 0 {
t.Fatalf("business error: %s", w.Body.String())
}
return envelope.Data
}
func companyDecode(t *testing.T, data []byte, value interface{}) {
t.Helper()
if err := json.Unmarshal(data, value); err != nil {
t.Fatalf("decode %s: %v", data, err)
}
}
type companyPage[T any] struct {
Items []T `json:"items"`
Total int64 `json:"total"`
HasMore bool `json:"has_more"`
}
func companyGetPage[T any](t *testing.T, f *companyHTTPFixture, account string, org int64, path string, total int64) companyPage[T] {
t.Helper()
var page companyPage[T]
companyDecode(t, f.request(t, account, fmt.Sprint(org), "GET", path, "", 200), &page)
if page.Total != total {
t.Fatalf("%s org=%d total=%d want=%d items=%+v", path, org, page.Total, total, page.Items)
}
return page
}
func TestCompanyIsolationAuthorizationMySQL(t *testing.T) {
f := newCompanyHTTPFixture(t)
for _, tc := range []struct {
account string
org int64
want int
}{
{"beijing", 301, 200}, {"beijing", 302, 403}, {"wuhan", 301, 403}, {"wuhan", 302, 200},
{"both", 301, 200}, {"both", 302, 200}, {"admin", 301, 200}, {"admin", 302, 200}, {"admin", 999999, 403},
} {
t.Run(fmt.Sprintf("%s/%d", tc.account, tc.org), func(t *testing.T) { f.request(t, tc.account, fmt.Sprint(tc.org), "GET", "/api/materials", "", tc.want) })
}
for _, header := range []string{"", "0", "-1", "abc", "301.5"} {
t.Run("invalid-header/"+header, func(t *testing.T) { f.request(t, "admin", header, "GET", "/api/units", "", 400) })
}
f.request(t, "", "301", "GET", "/api/materials", "", 401)
// Organization grants do not grant admin configuration or approval roles.
f.request(t, "both", "301", "POST", "/api/units", `{"code":"NEW","name":"新单位"}`, 403)
f.request(t, "both", "301", "GET", "/api/applications", "", 401)
for _, tc := range []struct {
account string
ids []int64
def int64
}{
{"beijing", []int64{301}, 301}, {"wuhan", []int64{302}, 302}, {"both", []int64{301, 302}, 301}, {"admin", []int64{1, 301, 302}, 0},
} {
var access model.OrganizationAccess
companyDecode(t, f.request(t, tc.account, "", "GET", "/api/auth/organizations", "", 200), &access)
if len(access.Organizations) != len(tc.ids) {
t.Fatalf("%s access=%+v", tc.account, access)
}
for _, want := range tc.ids {
found := false
for _, org := range access.Organizations {
found = found || org.ID == want
}
if !found {
t.Fatalf("%s missing company %d", tc.account, want)
}
}
if tc.def != 0 && (access.DefaultOrganizationID == nil || *access.DefaultOrganizationID != tc.def) {
t.Fatalf("%s default=%v", tc.account, access.DefaultOrganizationID)
}
}
for _, target := range []int64{302, 301} {
f.request(t, "both", "", "POST", "/api/auth/organization-context", fmt.Sprintf(`{"organization_id":%d,"previous_organization_id":%d}`, target, 603-target), 200)
}
var audits, defaults int
if err := f.db.QueryRow(`SELECT COUNT(*) FROM organization_audit_logs WHERE actor_user_id=303`).Scan(&audits); err != nil {
t.Fatal(err)
}
if err := f.db.QueryRow(`SELECT COUNT(*) FROM user_organizations WHERE user_id=303 AND organization_id=301 AND is_default=1`).Scan(&defaults); err != nil || defaults != 1 || audits != 2 {
t.Fatalf("switch audits=%d default=%d err=%v", audits, defaults, err)
}
// Database revocation and stale JWT roles remain in the existing context suite.
}
func TestCompanyIsolationMasterDataMySQL(t *testing.T) {
f := newCompanyHTTPFixture(t)
for _, org := range []int64{301, 302} {
for _, tc := range []struct {
path string
want int
}{{"/api/material-categories", 2}, {"/api/units", 1}, {"/api/code-rules", 1}} {
var items []struct {
OrganizationID int64 `json:"organization_id"`
}
companyDecode(t, f.request(t, "both", fmt.Sprint(org), "GET", tc.path, "", 200), &items)
if len(items) != tc.want {
t.Fatalf("%s count=%d want=%d", tc.path, len(items), tc.want)
}
for _, item := range items {
if item.OrganizationID != org {
t.Fatalf("%s leaked %+v", tc.path, item)
}
}
}
page := companyGetPage[model.Material](t, f, "both", org, "/api/materials?search=SHARED&limit=1", 1)
if len(page.Items) != 1 || page.Items[0].ID != org || page.Items[0].OrganizationID != org || page.HasMore {
t.Fatalf("same-code search org=%d: %+v", org, page)
}
var detail model.Material
companyDecode(t, f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/materials/%d", org), "", 200), &detail)
if detail.CategoryID == nil || *detail.CategoryID != org+10 || detail.BaseUnitID == nil || *detail.BaseUnitID != org {
t.Fatalf("references=%+v", detail)
}
f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/materials/%d", 603-org), "", 404)
}
preview := func(org int64, code string) {
t.Helper()
var next model.NextMaterialCode
companyDecode(t, f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/code-rules/next?category_id=%d", org+10), "", 200), &next)
if !next.Available || next.Code != code {
t.Fatalf("org=%d preview=%+v want=%s", org, next, code)
}
}
create := func(org int64, code string) model.Material {
t.Helper()
var m model.Material
body := fmt.Sprintf(`{"organization_id":%d,"material_code":%q,"name":"路由建档","category_id":%d,"base_unit_id":%d}`, 603-org, code, org+10, org)
companyDecode(t, f.request(t, "both", fmt.Sprint(org), "POST", "/api/materials", body, 201), &m)
if m.OrganizationID != org {
t.Fatalf("forged ownership=%+v", m)
}
return m
}
preview(301, "BJ0001")
preview(302, "WH0001")
a := create(301, "BJ0001")
preview(301, "BJ0002")
preview(302, "WH0001")
create(302, "WH0001")
preview(302, "WH0002")
create(301, "SHARED-NEW")
create(302, "SHARED-NEW")
f.request(t, "both", "301", "POST", "/api/materials", `{"material_code":"SHARED-NEW","name":"重复","category_id":311,"base_unit_id":301}`, 400)
page := companyGetPage[model.Material](t, f, "both", 301, "/api/materials?limit=1", 3)
if len(page.Items) != 1 || !page.HasMore {
t.Fatalf("page=%+v", page)
}
page = companyGetPage[model.Material](t, f, "both", 301, "/api/materials?limit=1&offset=3", 3)
if len(page.Items) != 0 || page.HasMore {
t.Fatalf("last page=%+v", page)
}
companyGetPage[model.Material](t, f, "both", 301, "/api/materials?search=武汉", 0)
for _, tc := range []struct{ path, body string }{
{"/api/materials", `{"material_code":"BJ0002","name":"坏分类","category_id":312,"base_unit_id":301}`},
{"/api/materials", `{"material_code":"BJ0002","name":"坏单位","category_id":311,"base_unit_id":302}`},
{"/api/material-categories", `{"code":"BAD","name":"坏父级","parent_id":302}`},
{"/api/code-rules", `{"category_id":312,"prefix":"BAD","sequence_length":4}`},
} {
f.request(t, "admin", "301", "POST", tc.path, tc.body, 400)
}
preview(301, "BJ0002")
preview(302, "WH0002")
companyGetPage[model.Material](t, f, "both", 301, "/api/materials", 3)
for _, body := range []string{`{"category_id":312,"name":"坏分类"}`, `{"category_id":311,"base_unit_id":302,"name":"坏单位"}`} {
f.request(t, "both", "301", "PUT", fmt.Sprintf("/api/materials/%d", a.ID), body, 400)
}
for _, tc := range []struct{ method, suffix, body string }{{"PUT", "", `{"category_id":311,"name":"恶意修改"}`}, {"POST", "/obsolete", ""}, {"POST", "/restore", ""}} {
f.request(t, "both", "301", tc.method, "/api/materials/302"+tc.suffix, tc.body, 404)
}
f.request(t, "both", "301", "PUT", fmt.Sprintf("/api/materials/%d", a.ID), `{"organization_id":302,"category_id":311,"base_unit_id":301,"name":"北京更新"}`, 200)
f.request(t, "both", "301", "POST", fmt.Sprintf("/api/materials/%d/obsolete", a.ID), "", 200)
companyGetPage[model.Material](t, f, "both", 301, "/api/materials?status=obsolete", 1)
companyGetPage[model.Material](t, f, "both", 302, "/api/materials?status=obsolete", 0)
f.request(t, "both", "301", "POST", fmt.Sprintf("/api/materials/%d/restore", a.ID), "", 200)
var updated model.Material
companyDecode(t, f.request(t, "both", "301", "GET", fmt.Sprintf("/api/materials/%d", a.ID), "", 200), &updated)
if updated.OrganizationID != 301 || updated.Name != "北京更新" || updated.Status != "active" || updated.Version != 4 {
t.Fatalf("CRUD=%+v", updated)
}
var logs []model.MaterialLog
companyDecode(t, f.request(t, "both", "301", "GET", fmt.Sprintf("/api/logs/materials/%d", a.ID), "", 200), &logs)
if len(logs) != 4 {
t.Fatalf("CRUD logs=%+v", logs)
}
for _, log := range logs {
if log.MaterialID != a.ID {
t.Fatalf("foreign log=%+v", log)
}
}
companyGetPage[model.MaterialLog](t, f, "both", 301, "/api/logs?material_id=302", 0)
f.request(t, "both", "301", "GET", "/api/logs/materials/302", "", 404)
companyGetPage[model.MaterialLog](t, f, "both", 301, "/api/logs?limit=1", 5)
companyGetPage[model.MaterialLog](t, f, "both", 302, "/api/logs", 2)
// Configuration CRUD also takes ownership from the selected company.
for _, org := range []int64{301, 302} {
header := fmt.Sprint(org)
var category model.MaterialCategory
companyDecode(t, f.request(t, "admin", header, "POST", "/api/material-categories",
fmt.Sprintf(`{"organization_id":%d,"code":"EXTRA","name":"新增小类","parent_id":%d}`, 603-org, org), 201), &category)
f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/material-categories/%d", category.ID),
fmt.Sprintf(`{"organization_id":%d,"name":"更新小类","parent_id":%d}`, 603-org, org), 200)
var unit model.Unit
companyDecode(t, f.request(t, "admin", header, "POST", "/api/units",
fmt.Sprintf(`{"organization_id":%d,"code":"EXTRA","name":"新增单位"}`, 603-org), 201), &unit)
// A no-op update is valid; MySQL must count the matched row as production does.
f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/units/%d", unit.ID),
`{"name":"新增单位","precision_scale":0,"status":"active"}`, 200)
f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/units/%d", unit.ID),
fmt.Sprintf(`{"organization_id":%d,"name":"更新单位","precision_scale":2}`, 603-org), 200)
var rule model.CodeRule
companyDecode(t, f.request(t, "admin", header, "POST", "/api/code-rules",
fmt.Sprintf(`{"organization_id":%d,"category_id":%d,"prefix":"EX","sequence_length":3}`, 603-org, category.ID), 201), &rule)
f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/code-rules/%d", rule.ID),
fmt.Sprintf(`{"organization_id":%d,"category_id":%d,"prefix":"EX","sequence_length":3,"next_sequence":9}`, 603-org, category.ID), 200)
for _, tc := range []struct {
table string
id int64
}{
{"material_categories", category.ID}, {"units", unit.ID}, {"code_rules", rule.ID},
} {
var owner int64
if err := f.db.QueryRow("SELECT organization_id FROM "+tc.table+" WHERE id=?", tc.id).Scan(&owner); err != nil || owner != org {
t.Fatalf("%s id=%d owner=%d want=%d err=%v", tc.table, tc.id, owner, org, err)
}
}
var savedRule model.NextMaterialCode
companyDecode(t, f.request(t, "both", header, "GET", fmt.Sprintf("/api/code-rules/next?category_id=%d", category.ID), "", 200), &savedRule)
if !savedRule.Available || savedRule.Code != "EX009" {
t.Fatalf("saved rule=%+v", savedRule)
}
}
for _, tc := range []struct{ path, body string }{
{"/api/material-categories/312", `{"name":"跨公司","parent_id":301}`},
{"/api/units/302", `{"name":"跨公司"}`},
{"/api/code-rules/302", `{"category_id":311,"prefix":"HACK","sequence_length":4}`},
} {
f.request(t, "admin", "301", "PUT", tc.path, tc.body, 404)
}
}
func TestCompanyIsolationWarehouseApplicationFlowMySQL(t *testing.T) {
f := newCompanyHTTPFixture(t)
type flowIDs struct{ inbound, outbound, application int64 }
flows := map[int64]flowIDs{}
for _, org := range []int64{301, 302} {
account, received, requested := "beijing", 10, 3
if org == 302 {
account, received, requested = "wuhan", 20, 4
}
header := fmt.Sprint(org)
var in model.InboundOrder
body := fmt.Sprintf(`{"organization_id":%d,"warehouse_id":%d,"location_id":%d,"handler":"经办","supplier":"供应商","items":[{"material_id":%d,"quantity":%d}]}`, 603-org, 603-org, 603-org, org, received)
companyDecode(t, f.request(t, account, header, "POST", "/api/inbound", body, 201), &in)
if in.OrganizationID != org || in.WarehouseID != org || in.LocationID != org {
t.Fatalf("inbound scope=%+v", in)
}
var app model.MaterialApplication
body = fmt.Sprintf(`{"organization_id":%d,"purpose":"领用","items":[{"material_id":%d,"quantity":%d}]}`, 603-org, org, requested)
companyDecode(t, f.request(t, account, header, "POST", "/api/applications", body, 201), &app)
if app.OrganizationID != org {
t.Fatalf("application scope=%+v", app)
}
f.request(t, account, header, "POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":true}`, 401)
f.request(t, "admin", header, "POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":true}`, 200)
page := companyGetPage[model.OutboundOrder](t, f, account, org, "/api/outbound?limit=1", 1)
if len(page.Items) != 1 || page.Items[0].OrganizationID != org || page.Items[0].WarehouseID != org || page.Items[0].LocationID != org {
t.Fatalf("auto outbound=%+v", page)
}
var detail model.OutboundOrder
companyDecode(t, f.request(t, account, header, "GET", fmt.Sprintf("/api/outbound/%d", page.Items[0].ID), "", 200), &detail)
if len(detail.Items) != 1 || detail.Items[0].MaterialID != org || detail.Items[0].Quantity != requested {
t.Fatalf("auto outbound items=%+v", detail)
}
flows[org] = flowIDs{in.ID, page.Items[0].ID, app.ID}
stocks := companyGetPage[model.Stock](t, f, account, org, "/api/stock?search=SHARED&limit=1", 1)
if len(stocks.Items) != 1 || stocks.Items[0].MaterialID != org || stocks.Items[0].TotalOnHand.String() != fmt.Sprint(received-requested) {
t.Fatalf("stock=%+v want=%d", stocks, received-requested)
}
companyGetPage[model.InboundOrder](t, f, account, org, "/api/inbound", 1)
companyGetPage[model.MaterialApplication](t, f, account, org, "/api/applications/my?status=approved", 1)
companyGetPage[model.MaterialApplication](t, f, "both", org, "/api/applications/my", 0)
companyGetPage[model.MaterialApplication](t, f, "admin", org, "/api/applications?status=approved", 1)
f.request(t, account, header, "PUT", fmt.Sprintf("/api/stock/%d/warning", org), `{"warning_stock":100}`, 200)
companyGetPage[model.Stock](t, f, account, org, "/api/stock/warning", 1)
logs := companyGetPage[model.StockLogResponse](t, f, account, org, "/api/stock/logs?limit=1", 2)
if len(logs.Items) != 1 || logs.Items[0].MaterialID != org || !logs.HasMore {
t.Fatalf("stock logs=%+v", logs)
}
companyGetPage[model.StockLogResponse](t, f, account, org, fmt.Sprintf("/api/stock/logs/%d", org), 2)
var dimensions []model.InventoryBalanceDetail
companyDecode(t, f.request(t, account, header, "GET", fmt.Sprintf("/api/stock/%d/details", org), "", 200), &dimensions)
if len(dimensions) != 1 || dimensions[0].OrganizationID != org || dimensions[0].WarehouseID != org || dimensions[0].LocationID != org {
t.Fatalf("inventory dimensions=%+v", dimensions)
}
category := "BJ"
if org == 302 {
category = "WH"
}
companyGetPage[model.Stock](t, f, account, org, "/api/stock?category="+category, 1)
companyGetPage[model.Stock](t, f, account, org, "/api/stock?category=OTHER", 0)
var mismatches int
if err := f.db.QueryRow(`SELECT COUNT(*) FROM stock_logs WHERE organization_id=? AND (warehouse_id<>? OR location_id<>? OR material_id<>?)`, org, org, org, org).Scan(&mismatches); err != nil || mismatches != 0 {
t.Fatalf("stock dimensions mismatch=%d err=%v", mismatches, err)
}
}
for _, org := range []int64{301, 302} {
other := 603 - org
ids := flows[other]
for _, path := range []string{
fmt.Sprintf("/api/materials/%d", other), fmt.Sprintf("/api/stock/%d", other), fmt.Sprintf("/api/stock/%d/details", other),
fmt.Sprintf("/api/stock/logs/%d", other), fmt.Sprintf("/api/inbound/%d", ids.inbound), fmt.Sprintf("/api/outbound/%d", ids.outbound),
fmt.Sprintf("/api/inbound/materials/%d", other), fmt.Sprintf("/api/outbound/materials/%d", other),
} {
f.request(t, "both", fmt.Sprint(org), "GET", path, "", 404)
}
f.request(t, "both", fmt.Sprint(org), "PUT", fmt.Sprintf("/api/stock/%d/warning", other), `{"warning_stock":99}`, 404)
for _, suffix := range []string{"approve", "reject", "cancel"} {
f.request(t, "admin", fmt.Sprint(org), "POST", fmt.Sprintf("/api/applications/%d/%s", ids.application, suffix), `{"auto_outbound":true,"reason":"跨公司"}`, 404)
}
for _, kind := range []string{"inbound", "outbound"} {
id := ids.inbound
if kind == "outbound" {
id = ids.outbound
}
for _, method := range []string{"PUT", "DELETE"} {
body := ""
if method == "PUT" {
body = fmt.Sprintf(`{"handler":"h","supplier":"s","receiver":"r","items":[{"material_id":%d,"quantity":1}]}`, org)
}
f.request(t, "both", fmt.Sprint(org), method, fmt.Sprintf("/api/%s/%d", kind, id), body, 404)
}
}
f.request(t, "both", fmt.Sprint(org), "POST", "/api/applications", fmt.Sprintf(`{"items":[{"material_id":%d,"quantity":1}]}`, other), 400)
for _, action := range []string{"cancel", "reject"} {
var app model.MaterialApplication
companyDecode(t, f.request(t, "both", fmt.Sprint(org), "POST", "/api/applications", fmt.Sprintf(`{"items":[{"material_id":%d,"quantity":1}]}`, org), 201), &app)
account := "both"
if action == "reject" {
account = "admin"
}
f.request(t, account, fmt.Sprint(org), "POST", fmt.Sprintf("/api/applications/%d/%s", app.ID, action), `{"reason":"不需要"}`, 200)
status := "cancelled"
if action == "reject" {
status = "rejected"
}
companyGetPage[model.MaterialApplication](t, f, "both", org, "/api/applications/my?status="+status, 1)
}
var balance string
want := "7.000000"
if org == 302 {
want = "16.000000"
}
if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=? AND warehouse_id=? AND location_id=? AND material_id=?`, org, org, org, org).Scan(&balance); err != nil || balance != want {
t.Fatalf("final org=%d balance=%s want=%s err=%v", org, balance, want, err)
}
companyGetPage[model.OutboundOrder](t, f, "both", org, "/api/outbound", 1)
// Existing order_flow integration owns inbound update/delete. Exercise the
// corresponding manual outbound reversal through the real HTTP route.
var manual model.OutboundOrder
companyDecode(t, f.request(t, "both", fmt.Sprint(org), "POST", "/api/outbound",
fmt.Sprintf(`{"receiver":"r","items":[{"material_id":%d,"quantity":1}]}`, org), 201), &manual)
f.request(t, "both", fmt.Sprint(org), "PUT", fmt.Sprintf("/api/outbound/%d", manual.ID),
fmt.Sprintf(`{"receiver":"r","items":[{"material_id":%d,"quantity":2}]}`, org), 200)
f.request(t, "both", fmt.Sprint(org), "DELETE", fmt.Sprintf("/api/outbound/%d", manual.ID), "", 200)
if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=? AND material_id=?`, org, org).Scan(&balance); err != nil || balance != want {
t.Fatalf("outbound reversal org=%d balance=%s want=%s err=%v", org, balance, want, err)
}
companyGetPage[model.OutboundOrder](t, f, "both", org, "/api/outbound", 1)
for _, state := range []struct{ setup, restore string }{
{"UPDATE warehouses SET status='inactive' WHERE id=?", "UPDATE warehouses SET status='active' WHERE id=?"},
{"UPDATE warehouse_locations SET status='inactive' WHERE id=?", "UPDATE warehouse_locations SET status='active' WHERE id=?"},
} {
companySQL(t, f.db, state.setup, org)
// Disabled dimensions still permit reading existing stock/documents.
companyGetPage[model.Stock](t, f, "both", org, "/api/stock", 1)
companyGetPage[model.InboundOrder](t, f, "both", org, "/api/inbound", 1)
companyGetPage[model.OutboundOrder](t, f, "both", org, "/api/outbound", 1)
f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/stock/%d/details", org), "", 200)
own := flows[org]
for _, req := range []struct{ method, path, body string }{
{"PUT", fmt.Sprintf("/api/inbound/%d", own.inbound), fmt.Sprintf(`{"handler":"h","supplier":"s","items":[{"material_id":%d,"quantity":10}]}`, org)},
{"DELETE", fmt.Sprintf("/api/inbound/%d", own.inbound), ""},
{"PUT", fmt.Sprintf("/api/outbound/%d", own.outbound), fmt.Sprintf(`{"receiver":"r","items":[{"material_id":%d,"quantity":1}]}`, org)},
{"DELETE", fmt.Sprintf("/api/outbound/%d", own.outbound), ""},
} {
f.request(t, "both", fmt.Sprint(org), req.method, req.path, req.body, 409)
}
companySQL(t, f.db, state.restore, org)
}
}
}
func TestCompanyIsolationInventoryConfigurationMySQL(t *testing.T) {
f := newCompanyHTTPFixture(t)
companySQL(t, f.db, `DELETE FROM warehouse_locations WHERE id=301; DELETE FROM warehouses WHERE id=301`)
for _, tc := range []struct{ name, setup, message string }{
{"no warehouse", "", "请先配置本公司仓库"},
{"no location", `INSERT INTO warehouses(id,organization_id,code,name) VALUES(301,301,'BJ_W','北京仓库')`, "请先配置本公司仓库的库位"},
{"inactive warehouse", `INSERT INTO warehouse_locations(id,warehouse_id,code,name) VALUES(301,301,'BJ_L','北京库位'); UPDATE warehouses SET status='inactive' WHERE id=301`, "本公司仓库已停用"},
{"inactive location", `UPDATE warehouses SET status='active' WHERE id=301; UPDATE warehouse_locations SET status='inactive' WHERE id=301`, "本公司库位已停用"},
} {
t.Run(tc.name, func(t *testing.T) {
if tc.setup != "" {
companySQL(t, f.db, tc.setup)
}
companyGetPage[model.Stock](t, f, "beijing", 301, "/api/stock", 0)
f.request(t, "beijing", "301", "GET", "/api/materials/301", "", 200)
var app model.MaterialApplication
companyDecode(t, f.request(t, "beijing", "301", "POST", "/api/applications", `{"items":[{"material_id":301,"quantity":1}]}`, 201), &app)
for _, req := range []struct{ method, path, body string }{
{"POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":301,"quantity":1}]}`},
{"POST", "/api/outbound", `{"receiver":"r","items":[{"material_id":301,"quantity":1}]}`},
{"PUT", "/api/stock/301/warning", `{"warning_stock":1}`},
{"POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":true}`},
} {
w := f.serve("admin", "301", req.method, req.path, req.body)
if w.Code != 409 || !strings.Contains(w.Body.String(), tc.message) {
t.Fatalf("%s %s status=%d body=%s want409/%s", req.method, req.path, w.Code, w.Body.String(), tc.message)
}
}
// Ordinary approval does not require active stock dimensions.
f.request(t, "admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":false}`, 200)
companyGetPage[model.OutboundOrder](t, f, "beijing", 301, "/api/outbound", 0)
})
}
var writes int
if err := f.db.QueryRow(`SELECT (SELECT COUNT(*) FROM inventory_balances WHERE organization_id=301)+(SELECT COUNT(*) FROM inbound_orders WHERE organization_id=301)+(SELECT COUNT(*) FROM outbound_orders WHERE organization_id=301)+(SELECT COUNT(*) FROM stock_logs WHERE organization_id=301)`).Scan(&writes); err != nil || writes != 0 {
t.Fatalf("configuration failures left writes=%d err=%v", writes, err)
}
f.request(t, "wuhan", "302", "POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":302,"quantity":1}]}`, 201)
}
func TestCompanyIsolationApprovalConcurrencyAndFaultMySQL(t *testing.T) {
f := newCompanyHTTPFixture(t)
f.request(t, "both", "301", "POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":301,"quantity":10}]}`, 201)
f.request(t, "both", "302", "POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":302,"quantity":20}]}`, 201)
newApplication := func() int64 {
t.Helper()
var app model.MaterialApplication
companyDecode(t, f.request(t, "both", "301", "POST", "/api/applications", `{"items":[{"material_id":301,"quantity":3}]}`, 201), &app)
return app.ID
}
assertState := func(id int64, status, balance string, outbounds, logs int) {
t.Helper()
var gotStatus, gotBalance, otherBalance string
var gotOrders, gotLogs int
if err := f.db.QueryRow(`SELECT status FROM material_applications WHERE id=? AND organization_id=301`, id).Scan(&gotStatus); err != nil {
t.Fatal(err)
}
if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=301 AND material_id=301`).Scan(&gotBalance); err != nil {
t.Fatal(err)
}
if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=302 AND material_id=302`).Scan(&otherBalance); err != nil {
t.Fatal(err)
}
if err := f.db.QueryRow(`SELECT COUNT(*) FROM outbound_orders WHERE organization_id=301`).Scan(&gotOrders); err != nil {
t.Fatal(err)
}
if err := f.db.QueryRow(`SELECT COUNT(*) FROM stock_logs WHERE organization_id=301`).Scan(&gotLogs); err != nil {
t.Fatal(err)
}
if gotStatus != status || gotBalance != balance || otherBalance != "20.000000" || gotOrders != outbounds || gotLogs != logs {
t.Fatalf("state=%s/%s/%s/%d/%d want=%s/%s/20/%d/%d", gotStatus, gotBalance, otherBalance, gotOrders, gotLogs, status, balance, outbounds, logs)
}
}
id := newApplication()
start := make(chan struct{})
results := make(chan *httptest.ResponseRecorder, 2)
var wg sync.WaitGroup
for i := 0; i < 2; i++ {
wg.Add(1)
go func() {
defer wg.Done()
<-start
results <- f.serve("admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", id), `{"auto_outbound":true}`)
}()
}
close(start)
wg.Wait()
close(results)
successes, conflicts := 0, 0
for w := range results {
switch w.Code {
case 200:
successes++
case 400:
conflicts++
default:
t.Fatalf("concurrent approval status=%d body=%s", w.Code, w.Body.String())
}
}
if successes != 1 || conflicts != 1 {
t.Fatalf("concurrent successes=%d conflicts=%d", successes, conflicts)
}
assertState(id, "approved", "7.000000", 1, 2)
// Force failure after stock deduction and outbound insertion, at stock log
// insertion. The transaction must restore all four artifacts.
id = newApplication()
companySQL(t, f.db, `CREATE TRIGGER fail_company_application_log BEFORE INSERT ON stock_logs FOR EACH ROW BEGIN IF NEW.organization_id=301 AND NEW.related_order_type='application' THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='integration injected log failure'; END IF; END`)
f.request(t, "admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", id), `{"auto_outbound":true}`, 500)
assertState(id, "pending", "7.000000", 1, 2)
companySQL(t, f.db, `DROP TRIGGER fail_company_application_log`)
f.request(t, "admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", id), `{"auto_outbound":true}`, 200)
assertState(id, "approved", "4.000000", 2, 3)
}
func TestCompanyIsolationWebSocketRevocationMySQL(t *testing.T) {
f := newCompanyHTTPFixture(t)
withTicketService(t, 30*time.Second)
f.router.GET("/ws", HandleWebSocket)
server := httptest.NewServer(f.router)
t.Cleanup(server.Close)
ticket := func(account, org string) string {
t.Helper()
var result struct {
Ticket string `json:"ticket"`
}
companyDecode(t, f.request(t, account, org, "POST", "/api/auth/websocket-ticket", "", 200), &result)
if result.Ticket == "" {
t.Fatal("empty WebSocket ticket")
}
return result.Ticket
}
type socketResult struct {
event StockUpdateEvent
err error
}
connect := func(account, org string) <-chan socketResult {
t.Helper()
url := "ws" + strings.TrimPrefix(server.URL, "http") + "/ws?ticket=" + ticket(account, org)
conn, response, err := websocket.DefaultDialer.Dial(url, nil)
if err != nil {
t.Fatalf("connect %s/%s: %v response=%v", account, org, err, response)
}
t.Cleanup(func() { conn.Close() })
results := make(chan socketResult, 8)
go func() {
for {
var event StockUpdateEvent
err := conn.ReadJSON(&event)
results <- socketResult{event, err}
if err != nil {
return
}
}
}()
return results
}
expectEvent := func(results <-chan socketResult, org int64, order int64) {
t.Helper()
select {
case result := <-results:
if result.err != nil || result.event.OrganizationID != org || result.event.OrderID != order {
t.Fatalf("event=%+v want=%d/%d", result, org, order)
}
case <-time.After(3 * time.Second):
t.Fatalf("no event for company %d", org)
}
}
expectClosed := func(results <-chan socketResult) {
t.Helper()
select {
case result := <-results:
if result.err == nil {
t.Fatalf("revoked socket received event=%+v", result.event)
}
case <-time.After(3 * time.Second):
t.Fatal("revoked socket remained open")
}
}
beijing := connect("both", "301")
wuhan := connect("wuhan", "302")
admin := connect("admin", "302")
disabledCompany := connect("beijing", "301")
// Both organizations are connected concurrently through real TCP handshakes.
GetHub().BroadcastStockUpdate(301, "inbound", 701, nil)
GetHub().BroadcastStockUpdate(302, "inbound", 702, nil)
expectEvent(beijing, 301, 701)
expectEvent(wuhan, 302, 702)
expectEvent(admin, 302, 702)
expectEvent(disabledCompany, 301, 701)
for _, results := range []<-chan socketResult{beijing, wuhan, admin, disabledCompany} {
select {
case extra := <-results:
t.Fatalf("cross-company/extra event=%+v", extra)
case <-time.After(50 * time.Millisecond):
}
}
// Recheck the database at delivery, rather than trusting handshake grants.
beforeGrantRevocation := ticket("both", "301")
companySQL(t, f.db, `DELETE FROM user_organizations WHERE user_id=303 AND organization_id=301`)
f.request(t, "", "", "GET", "/ws?ticket="+beforeGrantRevocation, "", 403)
GetHub().BroadcastStockUpdate(301, "outbound", 703, nil)
expectClosed(beijing)
expectEvent(disabledCompany, 301, 703)
f.request(t, "both", "301", "GET", "/api/materials", "", 403)
beforeRoleRevocation := ticket("admin", "302")
companySQL(t, f.db, `DELETE FROM user_roles WHERE user_id=304; INSERT INTO user_roles(user_id,role_id) SELECT 304,id FROM roles WHERE code='warehouse_admin'`)
GetHub().BroadcastStockUpdate(302, "outbound", 704, nil)
expectClosed(admin)
expectEvent(wuhan, 302, 704)
f.request(t, "", "", "GET", "/ws?ticket="+beforeRoleRevocation, "", 403)
f.request(t, "admin", "302", "GET", "/api/materials", "", 403)
companySQL(t, f.db, `UPDATE organizations SET status='inactive' WHERE id=301`)
GetHub().BroadcastStockUpdate(301, "outbound", 706, nil)
expectClosed(disabledCompany)
f.request(t, "beijing", "301", "GET", "/api/materials", "", 403)
// A ticket issued before logout cannot start a new connection, and the live
// socket cannot receive the next event. The old token also fails HTTP auth.
pending := ticket("wuhan", "302")
f.request(t, "wuhan", "", "POST", "/api/auth/logout", "", 200)
f.request(t, "wuhan", "302", "GET", "/api/materials", "", 401)
f.request(t, "wuhan", "302", "POST", "/api/auth/websocket-ticket", "", 401)
f.request(t, "", "", "GET", "/ws?ticket="+pending, "", 401)
GetHub().BroadcastStockUpdate(302, "outbound", 705, nil)
expectClosed(wuhan)
var revoked int
if err := f.db.QueryRow(`SELECT COUNT(*) FROM revoked_tokens WHERE user_id=302`).Scan(&revoked); err != nil || revoked != 1 {
t.Fatalf("logout persisted revocations=%d err=%v", revoked, err)
}
oldToken := f.tokens["wuhan"]
var login struct {
Token string `json:"token"`
}
companyDecode(t, f.request(t, "", "", "POST", "/api/auth/login", `{"username":"http-wuhan","password":"CompanyTest123!"}`, 200), &login)
if login.Token == "" || login.Token == oldToken {
t.Fatal("new login did not create a distinct session")
}
f.tokens["wuhan"] = login.Token
f.request(t, "wuhan", "302", "GET", "/api/materials", "", 200)
}
Reference in New Issue View Git Blame Copy Permalink