package chaos_test import ( "net" "net/http" "os" "os/exec" "path/filepath" "strings" "testing" "time" "git.asio.asia/nixevol/NixMsg/test/accept" "git.asio.asia/nixevol/NixMsg/test/chaos" "git.asio.asia/nixevol/NixMsg/test/harness" ) const ( q3ComposeProject = "q3-chaos" q3ProxyName = "q3-nixmsg" epPassword = "password1234" ) // TestQ3ToxiproxyOfflineKeepDelivered:延迟 + 断开后,离线保留期内消息最终送达。 func TestQ3ToxiproxyOfflineKeepDelivered(t *testing.T) { if _, err := exec.LookPath("docker"); err != nil { t.Skip("docker 不可用,跳过 toxiproxy 弱网") } composeFile := filepath.Join(moduleRoot(t), "test", "chaos", "compose.yml") down := func() { cmd := exec.Command("docker", "compose", "-p", q3ComposeProject, "-f", composeFile, "down", "-v", "--remove-orphans") _ = cmd.Run() } down() t.Cleanup(down) up := exec.Command("docker", "compose", "-p", q3ComposeProject, "-f", composeFile, "up", "-d") if out, err := up.CombinedOutput(); err != nil { t.Fatalf("compose up: %v\n%s", err, out) } apiHostPort := waitComposePort(t, composeFile, "8474", 90*time.Second) listenHostPort := waitComposePort(t, composeFile, "8475", 90*time.Second) apiURL := "http://127.0.0.1:" + apiHostPort waitHTTP(t, apiURL+"/version", 90*time.Second) srv, err := harness.Start(harness.Options{}) if err != nil { t.Fatal(err) } defer func() { _ = srv.Stop() }() _, nixPort, err := net.SplitHostPort(srv.Addr) if err != nil { t.Fatal(err) } c := chaos.NewClient(apiURL) _ = c.DeleteProxy(q3ProxyName) if _, err := c.CreateProxy(q3ProxyName, "0.0.0.0:8475", "host.docker.internal:"+nixPort); err != nil { t.Fatalf("CreateProxy: %v", err) } t.Cleanup(func() { _ = c.DeleteProxy(q3ProxyName) }) proxyBase := "http://127.0.0.1:" + listenHostPort if _, err := c.AddLatency(q3ProxyName, "q3-lag", 150, 50, ""); err != nil { t.Fatalf("AddLatency: %v", err) } ac := accept.AdminLogin(t, srv) accept.CreateEndpoint(t, ac, "q3alice01", epPassword) accept.CreateEndpoint(t, ac, "q3bob0001", epPassword) alice := accept.MQTTLogin(t, proxyBase, "q3alice01", epPassword) defer alice.Close() sendOff := alice.Request(t, map[string]any{ "v": 1, "type": "send", "rid": "q3s1", "id": "q3-off-keep", "to": map[string]any{"kind": "endpoint", "id": "q3bob0001"}, "body": map[string]any{"enc": "utf8", "data": "keep-me"}, "delay_ms": int64(0), "offline": map[string]any{"keep": true, "ttl_seconds": int64(86400)}, }) if !sendOff.OK { t.Fatalf("send offline keep: %+v", sendOff) } if _, err := c.AddResetPeer(q3ProxyName, "q3-rst", 0, ""); err != nil { t.Fatalf("AddResetPeer: %v", err) } time.Sleep(400 * time.Millisecond) _ = c.RemoveToxic(q3ProxyName, "q3-rst") _ = c.RemoveToxic(q3ProxyName, "q3-lag") bob := accept.MQTTLogin(t, proxyBase, "q3bob0001", epPassword) defer bob.Close() msg := bob.WaitType(t, "msg", 20*time.Second) if msg["id"] != "q3-off-keep" { t.Fatalf("want q3-off-keep got %v", msg) } bob.Request(t, map[string]any{"v": 1, "type": "ack", "rid": "q3a1", "from": "q3alice01", "id": "q3-off-keep"}) } // TestQ3NetemUntested 记录本机 Windows 上 netem 20% 丢包未测原因。 func TestQ3NetemUntested(t *testing.T) { t.Log("未测:Linux netem 20% 丢包。本机为 Windows;无法在宿主直接用 tc/netem。" + "若只对 toxiproxy 容器网卡挂 netshoot,不等于 NixMsg 业务端到端丢包验收。" + "本波用 toxiproxy 延迟/断开覆盖弱网;netem 需 Linux 宿主或把服务放进同网络 Linux 容器后再测。") } func moduleRoot(t *testing.T) string { t.Helper() dir, err := os.Getwd() if err != nil { t.Fatal(err) } for { if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { return dir } parent := filepath.Dir(dir) if parent == dir { t.Fatal("go.mod not found") } dir = parent } } func waitComposePort(t *testing.T, composeFile, containerPort string, timeout time.Duration) string { t.Helper() deadline := time.Now().Add(timeout) var last string for time.Now().Before(deadline) { cmd := exec.Command("docker", "compose", "-p", q3ComposeProject, "-f", composeFile, "port", "toxiproxy", containerPort) out, err := cmd.CombinedOutput() last = strings.TrimSpace(string(out)) if err == nil && last != "" { _, port, perr := net.SplitHostPort(normalizeComposeAddr(last)) if perr == nil && port != "" { return port } } time.Sleep(400 * time.Millisecond) } t.Fatalf("compose port %s timeout, last=%q", containerPort, last) return "" } func normalizeComposeAddr(s string) string { s = strings.TrimSpace(s) if strings.HasPrefix(s, "[::]") { return "127.0.0.1" + s[len("[::]"):] } if strings.HasPrefix(s, "0.0.0.0:") { return "127.0.0.1" + s[len("0.0.0.0"):] } return s } func waitHTTP(t *testing.T, url string, timeout time.Duration) { t.Helper() client := &http.Client{Timeout: 2 * time.Second} deadline := time.Now().Add(timeout) var last error for time.Now().Before(deadline) { resp, err := client.Get(url) if err == nil { _ = resp.Body.Close() if resp.StatusCode < 500 { return } last = err } else { last = err } time.Sleep(400 * time.Millisecond) } t.Fatalf("wait HTTP %s: %v", url, last) }