127 lines
4.2 KiB
Go
127 lines
4.2 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
omni "git.yuntuo.hk/k2safe/go-omni"
|
|
"git.yuntuo.hk/k2safe/go-omni/internal/testutil"
|
|
)
|
|
|
|
// Exercise the composed feature chain through real sockets and all runners.
|
|
func TestModulesConfiguredTransports(t *testing.T) {
|
|
for _, transport := range []string{omni.TransportNetHTTP, omni.TransportNative, omni.TransportHTTP2} {
|
|
t.Run(transport, func(t *testing.T) {
|
|
var loads atomic.Int32
|
|
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
loads.Add(1)
|
|
if r.Header.Get("X-Request-ID") == "" {
|
|
t.Error("missing outbound correlation")
|
|
}
|
|
io.WriteString(w, `{"name":"upstream catalog"}`)
|
|
}))
|
|
defer upstream.Close()
|
|
settings, err := omni.LoadConfig("MODULES_TEST_")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cert, key, tlsClient := testutil.CertificateFiles(t)
|
|
settings.Transport = transport
|
|
settings.TLS.CertFile = cert
|
|
settings.TLS.KeyFile = key
|
|
application, err := newApp(settings, "demo", upstream.URL)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- application.ServeContext(ctx, listener) }()
|
|
defer func() {
|
|
cancel()
|
|
select {
|
|
case err := <-done:
|
|
if err != nil {
|
|
t.Error(err)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
listener.Close()
|
|
t.Error("server failed to stop")
|
|
}
|
|
}()
|
|
client := &http.Client{Timeout: 3 * time.Second, Transport: &http.Transport{TLSClientConfig: tlsClient, ForceAttemptHTTP2: true}}
|
|
defer client.CloseIdleConnections()
|
|
request := func(method, path, body, key, requestID, token string) (*http.Response, string) {
|
|
r, err := http.NewRequest(method, "https://"+listener.Addr().String()+path, strings.NewReader(body))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
r.Header.Set("Content-Type", "application/json")
|
|
r.Header.Set("Idempotency-Key", key)
|
|
r.Header.Set("X-Request-ID", requestID)
|
|
r.Header.Set("Authorization", "Bearer "+token)
|
|
response, err := client.Do(r)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer response.Body.Close()
|
|
data, err := io.ReadAll(response.Body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if response.TLS == nil || len(response.TLS.VerifiedChains) == 0 {
|
|
t.Fatal("unverified TLS")
|
|
}
|
|
if transport == omni.TransportHTTP2 && response.ProtoMajor != 2 {
|
|
t.Fatal(response.Proto)
|
|
}
|
|
return response, string(data)
|
|
}
|
|
for n := 0; n < 2; n++ {
|
|
response, body := request("GET", "/api/catalog", "", "", "catalogRequest", "demo")
|
|
if response.StatusCode != 200 || !strings.Contains(body, "upstream catalog") {
|
|
t.Fatal(response.StatusCode, body)
|
|
}
|
|
}
|
|
if loads.Load() != 1 {
|
|
t.Fatal("cache missed", loads.Load())
|
|
}
|
|
first, original := request("POST", "/api/jobs", `{"name":"report"}`, "job1", "firstRequest", "demo")
|
|
second, replayed := request("POST", "/api/jobs", `{"name":"report"}`, "job1", "secondRequest", "demo")
|
|
if first.StatusCode != 202 || second.StatusCode != 202 || original != replayed || second.Header.Get("Idempotency-Replayed") != "true" || second.Header.Get("X-Request-ID") != "firstRequest" {
|
|
t.Fatal(first.StatusCode, second.StatusCode, original, replayed)
|
|
}
|
|
if response, _ := request("POST", "/api/jobs", `{"name":"other"}`, "job1", "conflictRequest", "demo"); response.StatusCode != 409 {
|
|
t.Fatal(response.StatusCode)
|
|
}
|
|
if response, _ := request("POST", "/api/jobs", `{"name":"report"}`, "job1", "unauthorized", "wrong"); response.StatusCode != 401 {
|
|
t.Fatal("replay bypassed authentication", response.StatusCode)
|
|
}
|
|
deadline := time.Now().Add(time.Second)
|
|
for {
|
|
response, body := request("GET", "/api/jobs", "", "", "statsRequest", "demo")
|
|
if response.StatusCode != 200 {
|
|
t.Fatal(response.StatusCode)
|
|
}
|
|
if strings.Contains(body, `"completed":1`) {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("job did not finish once", body)
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
})
|
|
}
|
|
}
|