补齐真实应用持久化与Nginx双层切换集成用例
This commit is contained in:
parent
53ed3d6021
commit
c7d5344329
246
backend/internal/app/management_integration_test.go
Normal file
246
backend/internal/app/management_integration_test.go
Normal file
@ -0,0 +1,246 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/ecdsa"
|
||||
"crypto/elliptic"
|
||||
"crypto/rand"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"crypto/x509/pkix"
|
||||
"encoding/pem"
|
||||
"errors"
|
||||
"math/big"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/certificate"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/nginxactivation"
|
||||
)
|
||||
|
||||
type applicationTLSFixture struct {
|
||||
root string
|
||||
config config.Config
|
||||
path string
|
||||
caPath string
|
||||
oldLeaf []byte
|
||||
newLeaf []byte
|
||||
candidate certificate.SavedResource
|
||||
upstream nginxactivation.Endpoint
|
||||
proxy nginxactivation.Endpoint
|
||||
}
|
||||
|
||||
func writeApplicationTLSFile(t *testing.T, path string, content []byte) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(path, content, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func applicationTestPort(t *testing.T) int {
|
||||
t.Helper()
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
port := listener.Addr().(*net.TCPAddr).Port
|
||||
if err := listener.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return port
|
||||
}
|
||||
|
||||
func newApplicationTLSFixture(t *testing.T) *applicationTLSFixture {
|
||||
t.Helper()
|
||||
cfg := readyApplicationConfig(t)
|
||||
for _, path := range []string{cfg.Paths.Config, cfg.Paths.Certificate, filepath.Join(cfg.Paths.Install, "empty-roots")} {
|
||||
if err := os.MkdirAll(path, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
now := time.Now()
|
||||
authority := &x509.Certificate{
|
||||
SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "Scert integration CA"},
|
||||
NotBefore: now.Add(-time.Hour), NotAfter: now.Add(7 * 24 * time.Hour),
|
||||
IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign,
|
||||
}
|
||||
der, err := x509.CreateCertificate(rand.Reader, authority, authority, &key.PublicKey, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
authority, err = x509.ParseCertificate(der)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
|
||||
writeApplicationTLSFile(t, filepath.Join(cfg.Paths.Install, "ca.pem"), caPEM)
|
||||
for index, name := range []string{"previous", "candidate"} {
|
||||
version := filepath.Join(cfg.Paths.Certificate, name)
|
||||
if err := os.Mkdir(version, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
leafKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
leaf := &x509.Certificate{
|
||||
SerialNumber: big.NewInt(int64(index + 2)), DNSNames: []string{cfg.Web.Hostname},
|
||||
NotBefore: now.Add(-time.Hour), NotAfter: now.Add(time.Duration(index+1) * 24 * time.Hour),
|
||||
KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
|
||||
}
|
||||
leafDER, err := x509.CreateCertificate(rand.Reader, leaf, authority, &leafKey.PublicKey, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
keyDER, err := x509.MarshalPKCS8PrivateKey(leafKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
chain := append(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: leafDER}), caPEM...)
|
||||
writeApplicationTLSFile(t, filepath.Join(version, "fullchain.pem"), chain)
|
||||
writeApplicationTLSFile(t, filepath.Join(version, "key.pem"), pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}))
|
||||
}
|
||||
cfg.Server.Host, cfg.Server.HTTPSMode = "127.0.0.1", "proxy"
|
||||
cfg.Server.Port, cfg.Server.ProxyHTTPSPort = applicationTestPort(t), applicationTestPort(t)
|
||||
for cfg.Server.ProxyHTTPSPort == cfg.Server.Port {
|
||||
cfg.Server.ProxyHTTPSPort = applicationTestPort(t)
|
||||
}
|
||||
cfg.Server.TLSCert = filepath.Join(cfg.Paths.Certificate, "previous", "fullchain.pem")
|
||||
cfg.Server.TLSKey = filepath.Join(cfg.Paths.Certificate, "previous", "key.pem")
|
||||
if err := cfg.Save(filepath.Join(cfg.Paths.Config, "scert.json")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return loadApplicationTLSFixture(t, cfg.Paths.Install)
|
||||
}
|
||||
|
||||
func loadApplicationTLSFixture(t *testing.T, root string) *applicationTLSFixture {
|
||||
t.Helper()
|
||||
path := filepath.Join(root, "conf", "scert.json")
|
||||
cfg, err := config.Load(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f := &applicationTLSFixture{
|
||||
root: root, config: cfg, path: path, caPath: filepath.Join(root, "ca.pem"),
|
||||
candidate: certificate.SavedResource{
|
||||
FullChainPath: filepath.Join(cfg.Paths.Certificate, "candidate", "fullchain.pem"),
|
||||
PrivateKeyPath: filepath.Join(cfg.Paths.Certificate, "candidate", "key.pem"),
|
||||
},
|
||||
}
|
||||
caPEM, err := os.ReadFile(f.caPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
roots := x509.NewCertPool()
|
||||
if !roots.AppendCertsFromPEM(caPEM) {
|
||||
t.Fatal("测试 CA 无法解析")
|
||||
}
|
||||
previous, err := tls.LoadX509KeyPair(cfg.Server.TLSCert, cfg.Server.TLSKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
candidate, err := tls.LoadX509KeyPair(f.candidate.FullChainPath, f.candidate.PrivateKeyPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.oldLeaf, f.newLeaf = previous.Certificate[0], candidate.Certificate[0]
|
||||
f.upstream = nginxactivation.Endpoint{Address: net.JoinHostPort("127.0.0.1", strconv.Itoa(cfg.Server.Port)), Hostname: cfg.Web.Hostname, Roots: roots}
|
||||
f.proxy = nginxactivation.Endpoint{Address: net.JoinHostPort("127.0.0.1", strconv.Itoa(cfg.Server.ProxyHTTPSPort)), Hostname: cfg.Web.Hostname, Roots: roots}
|
||||
return f
|
||||
}
|
||||
|
||||
func (f *applicationTLSFixture) start(t *testing.T) (*App, func()) {
|
||||
t.Helper()
|
||||
cfg, err := config.Load(f.path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
application, err := NewWithConfigPath(t.Context(), cfg, f.path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 隔离测试不连接固定生产 socket;恢复场景随后接入本轮独立 IPC。
|
||||
if application.managementStop != nil {
|
||||
application.managementStop()
|
||||
<-application.managementDone
|
||||
}
|
||||
done := make(chan error, 1)
|
||||
var once sync.Once
|
||||
stop := func() {
|
||||
once.Do(func() {
|
||||
if err := application.Close(); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, http.ErrServerClosed) {
|
||||
t.Errorf("应用 HTTPS 退出异常:%v", err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Error("应用 HTTPS 未按时退出")
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Cleanup(stop)
|
||||
go func() { done <- application.HTTP.ListenAndServe() }()
|
||||
select {
|
||||
case <-application.HTTP.Listening():
|
||||
case err := <-done:
|
||||
done <- err
|
||||
t.Fatalf("应用 HTTPS 启动失败:%v", err)
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("应用 HTTPS 未按时启动")
|
||||
}
|
||||
return application, stop
|
||||
}
|
||||
|
||||
func (f *applicationTLSFixture) seed(t *testing.T, application *App) {
|
||||
t.Helper()
|
||||
if err := application.SecretStore.Put(t.Context(), f.config.DNS.CredentialRef, map[string]string{
|
||||
"access_key_id": "integration-test-id", "access_key_secret": "integration-test-secret",
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := application.Database.DB.ExecContext(t.Context(), "CREATE TABLE activation_test (value TEXT NOT NULL); INSERT INTO activation_test VALUES ('preserved')"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (f *applicationTLSFixture) assertDatabase(t *testing.T, application *App) {
|
||||
t.Helper()
|
||||
var value string
|
||||
if err := application.Database.DB.QueryRowContext(t.Context(), "SELECT value FROM activation_test").Scan(&value); err != nil || value != "preserved" {
|
||||
t.Fatal("应用重建后数据库内容未保留", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplicationManagementListenerRestartsWithPersistentReadiness(t *testing.T) {
|
||||
f := newApplicationTLSFixture(t)
|
||||
first, stop := f.start(t)
|
||||
f.seed(t, first)
|
||||
if _, err := f.upstream.Check(t.Context(), f.oldLeaf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
stop()
|
||||
second, _ := f.start(t)
|
||||
if _, err := f.upstream.Check(t.Context(), f.oldLeaf); err != nil {
|
||||
t.Fatal("重建应用后 HTTPS、凭据与数据库就绪检查失败", err)
|
||||
}
|
||||
f.assertDatabase(t, second)
|
||||
ctx, cancel := context.WithTimeout(t.Context(), time.Second)
|
||||
defer cancel()
|
||||
if err := second.Database.Ping(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
141
backend/internal/app/management_nginx_fixture_linux_test.go
Normal file
141
backend/internal/app/management_nginx_fixture_linux_test.go
Normal file
@ -0,0 +1,141 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/nginxactivation"
|
||||
)
|
||||
|
||||
func startApplicationNginx(t *testing.T, f *applicationTLSFixture, binary string) nginxactivation.Rotation {
|
||||
t.Helper()
|
||||
prefix := filepath.Join(f.root, "nginx")
|
||||
directory := filepath.Join(prefix, "activation")
|
||||
version := filepath.Join(directory, "00000000000000000000000000000001")
|
||||
if err := os.MkdirAll(version, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for destination, source := range map[string]string{"fullchain.pem": f.config.Server.TLSCert, "key.pem": f.config.Server.TLSKey} {
|
||||
content, err := os.ReadFile(source)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writeApplicationTLSFile(t, filepath.Join(version, destination), content)
|
||||
}
|
||||
include := fmt.Sprintf("ssl_certificate %s/fullchain.pem;\nssl_certificate_key %s/key.pem;\n", version, version)
|
||||
writeApplicationTLSFile(t, filepath.Join(directory, "active.conf"), []byte(include))
|
||||
configuration := fmt.Sprintf(`worker_processes 1;
|
||||
pid %s/nginx.pid;
|
||||
error_log stderr warn;
|
||||
events { worker_connections 64; }
|
||||
http {
|
||||
access_log off;
|
||||
client_body_temp_path %s/client;
|
||||
proxy_temp_path %s/proxy;
|
||||
fastcgi_temp_path %s/fastcgi;
|
||||
uwsgi_temp_path %s/uwsgi;
|
||||
scgi_temp_path %s/scgi;
|
||||
server {
|
||||
listen %s ssl;
|
||||
server_name %s;
|
||||
include %s/active.conf;
|
||||
ssl_protocols TLSv1.2 TLSv1.3;
|
||||
location / {
|
||||
proxy_pass https://%s;
|
||||
proxy_set_header Host %s;
|
||||
proxy_ssl_server_name on;
|
||||
proxy_ssl_name %s;
|
||||
proxy_ssl_verify on;
|
||||
proxy_ssl_trusted_certificate %s;
|
||||
}
|
||||
}
|
||||
}
|
||||
`, prefix, prefix, prefix, prefix, prefix, prefix, f.proxy.Address, f.proxy.Hostname,
|
||||
directory, f.upstream.Address, f.config.Web.Hostname, f.config.Web.Hostname, f.caPath)
|
||||
configPath := filepath.Join(prefix, "nginx.conf")
|
||||
writeApplicationTLSFile(t, configPath, []byte(configuration))
|
||||
reloader := nginxactivation.CommandReloader{Binary: binary, Config: configPath, Prefix: prefix + "/"}
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
|
||||
defer cancel()
|
||||
if err := reloader.Check(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
logPath := filepath.Join(prefix, "process.log")
|
||||
log, err := os.OpenFile(logPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
command := exec.Command(binary, "-p", prefix+"/", "-c", configPath, "-g", "daemon off;")
|
||||
command.Stdout, command.Stderr = log, log
|
||||
command.Env = []string{"PATH=/usr/sbin:/usr/bin:/sbin:/bin", "LANG=C"}
|
||||
command.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGTERM}
|
||||
if err := command.Start(); err != nil {
|
||||
_ = log.Close()
|
||||
t.Fatal(err)
|
||||
}
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- command.Wait() }()
|
||||
t.Cleanup(func() {
|
||||
_ = command.Process.Signal(syscall.SIGQUIT)
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Errorf("隔离 Nginx 退出失败:%v", err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
_ = command.Process.Kill()
|
||||
<-done
|
||||
t.Error("隔离 Nginx 未按时退出")
|
||||
}
|
||||
_ = log.Close()
|
||||
if t.Failed() {
|
||||
content, _ := os.ReadFile(logPath)
|
||||
t.Logf("隔离 Nginx 日志:\n%s", content)
|
||||
}
|
||||
})
|
||||
if err := f.proxy.Wait(ctx, f.oldLeaf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return nginxactivation.Rotation{Directory: directory, Proxy: f.proxy, Upstream: f.upstream, Reloader: reloader}
|
||||
}
|
||||
|
||||
func startApplicationActivationIPC(t *testing.T, rotation nginxactivation.Rotation) nginxactivation.Client {
|
||||
t.Helper()
|
||||
directory := t.TempDir()
|
||||
if err := os.Chmod(directory, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
path := filepath.Join(directory, "activation.sock")
|
||||
listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: path, Net: "unix"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
listener.SetUnlinkOnClose(false)
|
||||
if err := os.Chmod(path, 0o600); err != nil {
|
||||
_ = listener.Close()
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
service := nginxactivation.LocalService{Rotation: rotation, AllowedUID: uint32(os.Geteuid())}
|
||||
go func() { done <- service.Serve(ctx, listener) }()
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Error("隔离激活服务未按时退出")
|
||||
}
|
||||
})
|
||||
return nginxactivation.Client{SocketPath: path, ServiceUID: uint32(os.Geteuid())}
|
||||
}
|
||||
231
backend/internal/app/management_nginx_linux_test.go
Normal file
231
backend/internal/app/management_nginx_linux_test.go
Normal file
@ -0,0 +1,231 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/config"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/nginxactivation"
|
||||
)
|
||||
|
||||
// 系统根证书缓存仅在子进程内加载测试 CA,不改生产依赖或主机信任库。
|
||||
func TestApplicationNginxIntegration(t *testing.T) {
|
||||
binary := os.Getenv("SCERT_TEST_NGINX_BINARY")
|
||||
if binary == "" {
|
||||
t.Skip("未指定隔离验收使用的 Nginx 可执行文件")
|
||||
}
|
||||
if !filepath.IsAbs(binary) {
|
||||
t.Fatal("测试 Nginx 路径必须是绝对路径")
|
||||
}
|
||||
if scenario := os.Getenv("SCERT_APP_NGINX_CHILD"); scenario != "" {
|
||||
runApplicationNginxScenario(t, binary, scenario)
|
||||
return
|
||||
}
|
||||
executable, err := os.Executable()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, scenario := range []string{"normal", "stage_response", "commit_response", "restart_staged", "restart_upstream"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
f := newApplicationTLSFixture(t)
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 120*time.Second)
|
||||
defer cancel()
|
||||
command := exec.CommandContext(ctx, executable, "-test.run=^TestApplicationNginxIntegration$", "-test.v", "-test.timeout=90s")
|
||||
command.Env = append(os.Environ(),
|
||||
"SCERT_APP_NGINX_CHILD="+scenario, "SCERT_APP_NGINX_ROOT="+f.root,
|
||||
"SSL_CERT_FILE="+f.caPath, "SSL_CERT_DIR="+filepath.Join(f.root, "empty-roots"),
|
||||
"GODEBUG=x509usefallbackroots=0",
|
||||
)
|
||||
if output, err := command.CombinedOutput(); err != nil {
|
||||
t.Fatalf("隔离应用与 Nginx 测试失败:%v\n%s", err, output)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type applicationFaultProxy struct {
|
||||
nginxactivation.ProxyClient
|
||||
scenario string
|
||||
pending bool
|
||||
}
|
||||
|
||||
func (p *applicationFaultProxy) StageRequest(ctx context.Context, id string, chain, key []byte) (string, error) {
|
||||
result, err := p.ProxyClient.StageRequest(ctx, id, chain, key)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
if p.scenario == "stage_response" || p.scenario == "restart_staged" {
|
||||
p.pending = p.scenario == "restart_staged"
|
||||
return "", errors.New("注入暂存应答丢失")
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (p *applicationFaultProxy) Commit(ctx context.Context, id string) error {
|
||||
if p.scenario == "restart_upstream" {
|
||||
p.pending = true
|
||||
return errors.New("注入上游持久化后辅助服务不可达")
|
||||
}
|
||||
if err := p.ProxyClient.Commit(ctx, id); err != nil {
|
||||
return err
|
||||
}
|
||||
if p.scenario == "commit_response" {
|
||||
return errors.New("注入提交应答丢失")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *applicationFaultProxy) ResolveRequest(ctx context.Context, id string) (nginxactivation.State, error) {
|
||||
if p.pending {
|
||||
return nginxactivation.State{}, errors.New("注入恢复查询不可达,保留未完成事务")
|
||||
}
|
||||
return p.ProxyClient.ResolveRequest(ctx, id)
|
||||
}
|
||||
|
||||
func runApplicationNginxScenario(t *testing.T, binary, scenario string) {
|
||||
f := loadApplicationTLSFixture(t, os.Getenv("SCERT_APP_NGINX_ROOT"))
|
||||
application, stop := f.start(t)
|
||||
f.seed(t, application)
|
||||
rotation := startApplicationNginx(t, f, binary)
|
||||
client := startApplicationActivationIPC(t, rotation)
|
||||
proxy := &applicationFaultProxy{ProxyClient: client, scenario: scenario}
|
||||
activator := coordinatedManagementActivator{
|
||||
server: application.HTTP, directory: filepath.Join(f.config.Paths.Data, "management-activation"), proxy: proxy,
|
||||
}
|
||||
state, err := application.HTTP.ManagementCertificate()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 60*time.Second)
|
||||
defer cancel()
|
||||
if err := activator.Ready(ctx, state); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = activator.Activate(ctx, state, f.candidate)
|
||||
want, leaf := "committed", f.newLeaf
|
||||
if scenario == "stage_response" || scenario == "restart_staged" {
|
||||
want, leaf = "rolled_back", f.oldLeaf
|
||||
}
|
||||
if scenario == "stage_response" || strings.HasPrefix(scenario, "restart_") {
|
||||
if err == nil {
|
||||
t.Fatal("注入故障后不应报告切换成功")
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if strings.HasPrefix(scenario, "restart_") {
|
||||
phase := "prepared"
|
||||
if scenario == "restart_upstream" {
|
||||
phase = "upstream"
|
||||
}
|
||||
assertApplicationCoordination(t, activator.directory, phase, false)
|
||||
pending, err := client.State(ctx)
|
||||
if err != nil || pending.Status != "staged" {
|
||||
t.Fatal("重建前必须保留真实代理未完成事务", pending, err)
|
||||
}
|
||||
} else {
|
||||
assertApplicationCoordination(t, activator.directory, want, true)
|
||||
}
|
||||
proxyLeaf := leaf
|
||||
if scenario == "restart_staged" {
|
||||
proxyLeaf = f.newLeaf
|
||||
}
|
||||
if _, err := f.proxy.Check(ctx, proxyLeaf); err != nil {
|
||||
t.Fatal("重建前代理证书与预期阶段不一致", err)
|
||||
}
|
||||
if _, err := f.upstream.Check(ctx, leaf); err != nil {
|
||||
t.Fatal("重建前应用监听证书与预期阶段不一致", err)
|
||||
}
|
||||
assertApplicationTLSConfig(t, f, want)
|
||||
stop()
|
||||
restarted, _ := f.start(t)
|
||||
recovered := coordinatedManagementActivator{server: restarted.HTTP, directory: activator.directory, proxy: client}
|
||||
// 使用实际应用恢复工作循环,首次完成后停止,避免与后续断言抢占独占会话。
|
||||
completed := make(chan error, 1)
|
||||
restarted.startManagementRecovery(func(ctx context.Context) error {
|
||||
err := recovered.coordinator(nil).Recover(ctx)
|
||||
completed <- err
|
||||
return err
|
||||
})
|
||||
select {
|
||||
case err := <-completed:
|
||||
if err != nil {
|
||||
t.Fatal("应用恢复工作循环失败", err)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatal("应用恢复工作循环超时", ctx.Err())
|
||||
}
|
||||
restarted.managementStop()
|
||||
<-restarted.managementDone
|
||||
assertApplicationCoordination(t, recovered.directory, want, true)
|
||||
assertApplicationTLSConfig(t, f, want)
|
||||
for _, endpoint := range []nginxactivation.Endpoint{f.proxy, f.upstream} {
|
||||
if err := endpoint.Wait(ctx, leaf); err != nil {
|
||||
t.Fatal("重建应用后真实双层证书未收敛", err)
|
||||
}
|
||||
}
|
||||
actual, err := client.State(ctx)
|
||||
if err != nil || actual.Status != want {
|
||||
t.Fatal("代理最终事务状态错误", actual, err)
|
||||
}
|
||||
var snapshot nginxactivation.ManagedCertificate
|
||||
if err := (managementUpstream{server: restarted.HTTP}).WithCertificate(ctx, func(session nginxactivation.ManagedSession) error {
|
||||
var err error
|
||||
snapshot, err = session.Snapshot()
|
||||
return err
|
||||
}); err != nil || snapshot.LeafHash != fmt.Sprintf("%x", sha256.Sum256(leaf)) {
|
||||
t.Fatal("重建后应用管理快照错误", err)
|
||||
}
|
||||
current, err := restarted.HTTP.ManagementCertificate()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := recovered.Ready(ctx, current); err != nil {
|
||||
t.Fatal("恢复后未允许后续续期", err)
|
||||
}
|
||||
f.assertDatabase(t, restarted)
|
||||
}
|
||||
|
||||
func assertApplicationTLSConfig(t *testing.T, f *applicationTLSFixture, phase string) {
|
||||
t.Helper()
|
||||
actual, err := config.Load(f.path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := f.config
|
||||
if phase == "committed" {
|
||||
want.Server.TLSCert, want.Server.TLSKey = f.candidate.FullChainPath, f.candidate.PrivateKeyPath
|
||||
}
|
||||
if !reflect.DeepEqual(actual, want) {
|
||||
t.Fatal("双层协调改变了证书路径以外的配置,或证书路径未按终态持久化")
|
||||
}
|
||||
}
|
||||
|
||||
func assertApplicationCoordination(t *testing.T, directory, phase string, terminal bool) {
|
||||
t.Helper()
|
||||
content, err := os.ReadFile(filepath.Join(directory, "coordination.json"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var record struct {
|
||||
Phase string `json:"phase"`
|
||||
RequestID string `json:"request_id"`
|
||||
ProxyID string `json:"proxy_id"`
|
||||
}
|
||||
if err := json.Unmarshal(content, &record); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if record.Phase != phase || record.RequestID == "" || (terminal && record.RequestID != record.ProxyID) {
|
||||
t.Fatalf("应用协调记录错误:%+v,预期阶段 %s", record, phase)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user