yms-daemon/integration/native_backend_executor_linux_test.go

//go:build linux

// Package integration_test 包含 yms-daemon 的集成测试，在真实文件系统与进程环境下
// 验证原生后端执行器与事务底座的协同行为。相关测试仅在 Linux 平台编译运行。
package integration_test

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"sync"
	"testing"

	"yms-daemon/internal/filestore"
	"yms-daemon/internal/nativebackendexecutor"
	"yms-daemon/internal/systemd"
	"yms-daemon/internal/transaction"
)

// TestNativeBackendExecutorIntegration 在临时目录中搭建完整的原生后端发布流程，
// 验证执行器能把传入的 JAR 制品安装到发布存储、更新槽位软链接并触发 systemd 单元启动，
// 且事务状态与事件能正确持久化到数据库。
func TestNativeBackendExecutorIntegration(t *testing.T) {
	ctx := context.Background()
	root := t.TempDir()
	databasePath := filepath.Join(root, "transactions.db")
	store, err := transaction.OpenStore(ctx, databasePath)
	if err != nil {
		t.Fatalf("open transaction store: %v", err)
	}
	coordinator, err := transaction.NewCoordinator(store, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create transaction coordinator: %v", err)
	}
	releaseStore, err := filestore.New(filepath.Join(root, "releases"))
	if err != nil {
		t.Fatalf("create native backend release store: %v", err)
	}

	artifactContent := []byte("integrated native backend jar")
	artifactPath := filepath.Join(root, "incoming", "glory-soft-yms.jar")
	if err := os.MkdirAll(filepath.Dir(artifactPath), 0o750); err != nil {
		t.Fatalf("create incoming directory: %v", err)
	}
	if err := os.WriteFile(artifactPath, artifactContent, 0o640); err != nil {
		t.Fatalf("write incoming backend JAR: %v", err)
	}
	previousTarget := filepath.Join(root, "previous", "glory-soft-yms.jar")
	if err := os.MkdirAll(filepath.Dir(previousTarget), 0o750); err != nil {
		t.Fatalf("create previous release directory: %v", err)
	}
	if err := os.WriteFile(previousTarget, []byte("previous native backend jar"), 0o640); err != nil {
		t.Fatalf("write previous backend JAR: %v", err)
	}
	slotJarPath := filepath.Join(root, "slots", "backend-green.jar")
	if err := os.MkdirAll(filepath.Dir(slotJarPath), 0o750); err != nil {
		t.Fatalf("create native backend slot directory: %v", err)
	}
	if err := os.Symlink(previousTarget, slotJarPath); err != nil {
		t.Fatalf("create native backend slot link: %v", err)
	}

	request := nativebackendexecutor.Request{
		ArtifactPath:       artifactPath,
		ArtifactIdentity:   integrationIdentity(artifactContent),
		ReleasePath:        "glory-soft-yms-20260815.jar",
		SlotJarPath:        slotJarPath,
		PreviousSlotTarget: previousTarget,
		UnitName:           "yms-green.service",
		Port:               8081,
		HealthEndpoint:     "http://127.0.0.1:8081/yms/actuator/health",
	}
	units := &integrationUnitManager{unit: systemd.Unit{
		Name:        request.UnitName,
		LoadState:   "loaded",
		ActiveState: "inactive",
		SubState:    "dead",
	}}
	healthBody := `{"status":"UP","components":{"db":{"status":"UP","components":{"dorisDataSource":{"status":"UP"},"postgresqlDataSource":{"status":"UP"}}},"diskSpace":{"status":"UP"},"ping":{"status":"UP"},"redis":{"status":"UP"}}}`
	httpClient := &http.Client{Transport: integrationRoundTripFunc(func(*http.Request) (*http.Response, error) {
		return &http.Response{
			StatusCode: http.StatusOK,
			Header:     make(http.Header),
			Body:       io.NopCloser(bytes.NewBufferString(healthBody)),
		}, nil
	})}
	executor, err := nativebackendexecutor.New(store, coordinator, releaseStore, units, httpClient)
	if err != nil {
		t.Fatalf("create native backend executor: %v", err)
	}
	record, created, err := store.CreateTransaction(ctx, transaction.CreateRequest{
		ID:             "native-backend-integration",
		IdempotencyKey: "native-backend-integration-request",
		Source:         "integration-test",
		Service:        "backend",
	})
	if err != nil || !created {
		t.Fatalf("create native backend transaction: record=%+v created=%t err=%v", record, created, err)
	}

	if err := executor.Run(ctx, record.ID, request); err != nil {
		t.Fatalf("run native backend executor: %v", err)
	}
	installed, found, err := releaseStore.Inspect(request.ReleasePath, request.ArtifactIdentity)
	if err != nil || !found {
		t.Fatalf("inspect installed backend JAR: file=%+v found=%t err=%v", installed, found, err)
	}
	actualContent, err := os.ReadFile(installed.Path)
	if err != nil || !bytes.Equal(actualContent, artifactContent) {
		t.Fatalf("verify installed backend JAR: content=%q err=%v", actualContent, err)
	}
	actualTarget, err := os.Readlink(slotJarPath)
	if err != nil || actualTarget != installed.Path {
		t.Fatalf("verify native backend slot link: target=%q installed=%q err=%v", actualTarget, installed.Path, err)
	}
	if err := store.Close(); err != nil {
		t.Fatalf("close transaction store: %v", err)
	}

	reopened, err := transaction.OpenStore(ctx, databasePath)
	if err != nil {
		t.Fatalf("reopen transaction store: %v", err)
	}
	t.Cleanup(func() { _ = reopened.Close() })
	persisted, err := reopened.Transaction(ctx, record.ID)
	if err != nil || persisted.State != transaction.StateSwitching {
		t.Fatalf("unexpected persisted native transaction: record=%+v err=%v", persisted, err)
	}
	pending, err := reopened.PendingSteps(ctx, record.ID)
	if err != nil || len(pending) != 0 {
		t.Fatalf("unexpected persisted pending steps: steps=%+v err=%v", pending, err)
	}
	units.mu.Lock()
	startCalls := units.startCalls
	startedName := units.startedName
	units.mu.Unlock()
	if startCalls != 1 || startedName != request.UnitName {
		t.Fatalf("unexpected integrated systemd calls: start=%d name=%s", startCalls, startedName)
	}
}

// integrationIdentity 根据内容计算文件的 SHA256 摘要与大小，返回用于发布存储校验的文件身份。
func integrationIdentity(content []byte) filestore.Identity {
	digest := sha256.Sum256(content)
	return filestore.Identity{Size: int64(len(content)), SHA256: hex.EncodeToString(digest[:])}
}

// integrationUnitManager systemd.Manager 接口的内存实现，用于在集成测试中模拟 systemd 单元，
// 记录启动与停止调用并跟踪单元状态变化。
type integrationUnitManager struct {
	// mu 保护以下所有字段的并发访问。
	mu sync.Mutex
	// unit 当前模拟的 systemd 单元状态。
	unit systemd.Unit
	// startCalls 统计 Start 被调用的次数。
	startCalls int
	// stopCalls 统计 Stop 被调用的次数。
	stopCalls int
	// startedName 记录最近一次 Start 的单元名。
	startedName string
}

// Inspect 返回指定名称的模拟单元状态；名称不匹配时返回 ErrUnitNotFound。
func (m *integrationUnitManager) Inspect(_ context.Context, name string) (systemd.Unit, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if name != m.unit.Name {
		return systemd.Unit{}, systemd.ErrUnitNotFound
	}
	return m.unit, nil
}

// Start 模拟启动指定单元：累计调用次数、记录单元名并把状态更新为 active/running；
// 名称不匹配时返回 ErrUnitNotFound。
func (m *integrationUnitManager) Start(_ context.Context, name string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if name != m.unit.Name {
		return systemd.ErrUnitNotFound
	}
	m.startCalls++
	m.startedName = name
	m.unit.ActiveState = "active"
	m.unit.SubState = "running"
	return nil
}

// Stop 模拟停止指定单元：累计调用次数并把状态更新为 inactive/dead；
// 名称不匹配时返回 ErrUnitNotFound。
func (m *integrationUnitManager) Stop(_ context.Context, name string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if name != m.unit.Name {
		return systemd.ErrUnitNotFound
	}
	m.stopCalls++
	m.unit.ActiveState = "inactive"
	m.unit.SubState = "dead"
	return nil
}

// integrationRoundTripFunc 一个可替换的 HTTP 传输实现，用于拦截健康检查请求并返回固定响应。
type integrationRoundTripFunc func(*http.Request) (*http.Response, error)

// RoundTrip 校验请求地址是否为预期的健康检查端点，然后委托给底层函数生成响应。
func (f integrationRoundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
	if request.URL.String() != "http://127.0.0.1:8081/yms/actuator/health" {
		return nil, errors.New("unexpected native backend health endpoint: " + request.URL.String())
	}
	return f(request)
}

// 编译期断言 integrationUnitManager 实现了 systemd.Manager 接口。
var _ systemd.Manager = (*integrationUnitManager)(nil)