yms-daemon/internal/daemonserver/server_test.go

package daemonserver

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"syscall"
	"testing"
	"time"

	"yms-daemon/internal/backendupdate"
	"yms-daemon/internal/daemonapi"
	"yms-daemon/internal/daemonclient"
	"yms-daemon/internal/transaction"
)

// TestServerAcceptsBackendUpdateThroughUnixSocket 验证服务端能通过 Unix Socket 接收后端 ZIP 更新请求，
// 正确分派到更新器，并向客户端返回事务终态以及流式的更新进度事件。
func TestServerAcceptsBackendUpdateThroughUnixSocket(t *testing.T) {
	socketPath := shortSocketPath(t)
	updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-01", State: transaction.StateCommitted}}
	server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create daemon server: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	serveResult := make(chan error, 1)
	go func() { serveResult <- server.Serve(ctx) }()
	waitForSocket(t, socketPath, serveResult)

	packagePath := filepath.Join(t.TempDir(), "package.zip")
	if err := os.WriteFile(packagePath, []byte("zip"), 0o600); err != nil {
		t.Fatalf("write update package: %v", err)
	}
	var progress []daemonapi.Response
	response, err := daemonclient.Update(context.Background(), socketPath, "backend", daemonapi.InputTypeRepackZIP, packagePath, func(event daemonapi.Response) {
		progress = append(progress, event)
	})
	if err != nil {
		t.Fatalf("submit backend update: %v", err)
	}
	if response.TransactionID != updater.record.ID || response.State != string(transaction.StateCommitted) || updater.file != packagePath || updater.inputType != daemonapi.InputTypeRepackZIP {
		t.Fatalf("unexpected update response or dispatch: response=%+v file=%s", response, updater.file)
	}
	if len(progress) != 1 || progress[0].Message != "test update progress" {
		t.Fatalf("unexpected streamed update progress: %+v", progress)
	}

	cancel()
	select {
	case err := <-serveResult:
		if err != nil {
			t.Fatalf("stop daemon server: %v", err)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("daemon server did not stop")
	}
}

// TestServerAcceptsDirectNativeJARThroughUnixSocket 验证服务端能接收原生后端 JAR 直接更新请求，
// 并将制品路径与输入类型正确分派到更新器。
func TestServerAcceptsDirectNativeJARThroughUnixSocket(t *testing.T) {
	socketPath := shortSocketPath(t)
	updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-direct-01", State: transaction.StateCommitted}}
	server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create daemon server: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	serveResult := make(chan error, 1)
	go func() { serveResult <- server.Serve(ctx) }()
	waitForSocket(t, socketPath, serveResult)

	jarPath := filepath.Join(t.TempDir(), "glory-soft-yms.jar")
	if err := os.WriteFile(jarPath, []byte("jar"), 0o600); err != nil {
		t.Fatalf("write direct native backend JAR: %v", err)
	}
	response, err := daemonclient.Update(context.Background(), socketPath, "backend", daemonapi.InputTypeNativeJAR, jarPath, nil)
	if err != nil {
		t.Fatalf("submit direct native backend update: %v", err)
	}
	if response.TransactionID != updater.record.ID || updater.file != jarPath || updater.inputType != daemonapi.InputTypeNativeJAR {
		t.Fatalf("unexpected direct update response or dispatch: response=%+v file=%s inputType=%s", response, updater.file, updater.inputType)
	}
}

// TestServerAcceptsContainerImageThroughUnixSocket 验证服务端能接收容器镜像更新请求，
// 并把镜像引用作为制品路径分派到更新器。
func TestServerAcceptsContainerImageThroughUnixSocket(t *testing.T) {
	socketPath := shortSocketPath(t)
	updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-container-01", State: transaction.StateCommitted}}
	server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create daemon server: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	serveResult := make(chan error, 1)
	go func() { serveResult <- server.Serve(ctx) }()
	waitForSocket(t, socketPath, serveResult)

	imageReference := "harbor.ymswell.asia/ymswell/glory-ymswell:20260813-184902-a37bf50d-v1.1.8.1"
	response, err := daemonclient.UpdateContainerImage(context.Background(), socketPath, "backend", imageReference, true, nil)
	if err != nil {
		t.Fatalf("submit container backend image: %v", err)
	}
	if response.TransactionID != updater.record.ID || updater.file != imageReference || updater.inputType != daemonapi.InputTypeContainerImage {
		t.Fatalf("unexpected container update dispatch: response=%+v image=%s inputType=%s", response, updater.file, updater.inputType)
	}
}

// TestServerReturnsTransactionFailure 验证更新器返回错误时，服务端会在最终结果中携带事务标识、
// 回滚终态以及错误信息，客户端据此返回错误。
func TestServerReturnsTransactionFailure(t *testing.T) {
	socketPath := shortSocketPath(t)
	updater := &fakeUpdater{
		record: transaction.Transaction{ID: "transaction-02", State: transaction.StateRolledBack},
		err:    errors.New("health check failed"),
	}
	server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create daemon server: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	serveResult := make(chan error, 1)
	go func() { serveResult <- server.Serve(ctx) }()
	waitForSocket(t, socketPath, serveResult)

	packagePath := filepath.Join(t.TempDir(), "package.zip")
	response, err := daemonclient.Update(context.Background(), socketPath, "backend", daemonapi.InputTypeRepackZIP, packagePath, nil)
	if err == nil || response.TransactionID != updater.record.ID || response.State != string(transaction.StateRolledBack) {
		t.Fatalf("unexpected failed update response: response=%+v err=%v", response, err)
	}
}

// TestServerAcceptsBackendRestartThroughUnixSocket 验证服务端能接收后端重启请求，
// 并返回事务终态与流式的重启进度事件。
func TestServerAcceptsBackendRestartThroughUnixSocket(t *testing.T) {
	socketPath := shortSocketPath(t)
	updater := &fakeUpdater{record: transaction.Transaction{ID: "transaction-restart-01", State: transaction.StateCommitted}}
	server, err := New(socketPath, updater, &fakeDiagnoser{}, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create daemon server: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	serveResult := make(chan error, 1)
	go func() { serveResult <- server.Serve(ctx) }()
	waitForSocket(t, socketPath, serveResult)

	var progress []daemonapi.Response
	response, err := daemonclient.Restart(context.Background(), socketPath, "backend", func(event daemonapi.Response) {
		progress = append(progress, event)
	})
	if err != nil {
		t.Fatalf("submit backend restart: %v", err)
	}
	if response.TransactionID != updater.record.ID || updater.operation != daemonapi.OperationRestart {
		t.Fatalf("unexpected restart response or dispatch: response=%+v operation=%s", response, updater.operation)
	}
	if len(progress) != 1 || progress[0].Message != "test restart progress" {
		t.Fatalf("unexpected streamed restart progress: %+v", progress)
	}
}

// TestServerReturnsStatusThroughUnixSocket 验证服务端能通过 Unix Socket 处理 status 请求，
// 并把诊断结果返回给客户端。
func TestServerReturnsStatusThroughUnixSocket(t *testing.T) {
	socketPath := shortSocketPath(t)
	updater := &fakeUpdater{}
	diagnosis := daemonapi.Diagnosis{Service: "backend", Type: "container", Healthy: true, Items: []daemonapi.DiagnosisItem{{Level: daemonapi.DiagnosisLevelOK, Code: "ok", Message: "ok"}}}
	diagnoser := &fakeDiagnoser{diagnosis: diagnosis}
	server, err := New(socketPath, updater, diagnoser, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create daemon server: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	serveResult := make(chan error, 1)
	go func() { serveResult <- server.Serve(ctx) }()
	waitForSocket(t, socketPath, serveResult)

	result, err := daemonclient.Status(context.Background(), socketPath, "backend")
	if err != nil {
		t.Fatalf("submit status: %v", err)
	}
	if result.Service != "backend" || !result.Healthy {
		t.Fatalf("unexpected status result: %+v", result)
	}
}

// TestServerReconcileForwardsApply 验证 reconcile --apply 会把 apply 标志正确传递给诊断器。
func TestServerReconcileForwardsApply(t *testing.T) {
	socketPath := shortSocketPath(t)
	updater := &fakeUpdater{}
	diagnoser := &fakeDiagnoser{diagnosis: daemonapi.Diagnosis{Service: "backend", Type: "container", Healthy: true}}
	server, err := New(socketPath, updater, diagnoser, slog.New(slog.NewTextHandler(io.Discard, nil)))
	if err != nil {
		t.Fatalf("create daemon server: %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	serveResult := make(chan error, 1)
	go func() { serveResult <- server.Serve(ctx) }()
	waitForSocket(t, socketPath, serveResult)

	if _, err := daemonclient.Reconcile(context.Background(), socketPath, "backend", true); err != nil {
		t.Fatalf("submit reconcile apply: %v", err)
	}
	if !diagnoser.applied {
		t.Fatal("reconcile --apply was not forwarded to diagnoser")
	}
}

// shortSocketPath 创建一个短目录并返回其下的 daemon.sock 路径，用于规避 Unix Socket 路径长度上限。
// 目录会在测试结束时自动清理。
func shortSocketPath(t *testing.T) string {
	t.Helper()
	directory, err := os.MkdirTemp("", "yd-")
	if err != nil {
		t.Fatalf("create short Unix Socket directory: %v", err)
	}
	t.Cleanup(func() { _ = os.RemoveAll(directory) })
	return filepath.Join(directory, "daemon.sock")
}

// waitForSocket 轮询等待 Unix Socket 文件出现，用于在测试中同步服务端就绪状态。
// 若服务端在 Socket 创建前就退出且原因为权限不足则跳过测试，否则视为失败；
// 超过 3 秒仍未见 Socket 也视为失败。
func waitForSocket(t *testing.T, socketPath string, serveResult <-chan error) {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		select {
		case err := <-serveResult:
			if errors.Is(err, syscall.EPERM) {
				t.Skip("Unix Socket creation is not permitted by the test sandbox")
			}
			t.Fatalf("daemon server stopped before creating Unix Socket: %v", err)
		default:
		}
		info, err := os.Lstat(socketPath)
		if err == nil && info.Mode()&os.ModeSocket != 0 {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("daemon Unix Socket was not created: %s", socketPath)
}

// fakeUpdater backendUpdater 接口的测试替身，记录被分派的参数并返回预设结果。
type fakeUpdater struct {
	// record 各更新方法返回的预设事务。
	record transaction.Transaction
	// err 各更新方法返回的预设错误。
	err error
	// file 记录最近一次分派收到的制品路径或镜像引用。
	file string
	// inputType 记录最近一次分派收到的输入类型。
	inputType string
	// operation 记录最近一次分派收到的操作类型。
	operation string
}

// UpdateRepack 记录分派参数并上报一条更新进度，返回预设事务与错误。
func (u *fakeUpdater) UpdateRepack(_ context.Context, file string, report backendupdate.ProgressReporter) (transaction.Transaction, error) {
	u.file = file
	u.inputType = daemonapi.InputTypeRepackZIP
	u.operation = daemonapi.OperationUpdate
	if report != nil {
		report(backendupdate.Progress{TransactionID: u.record.ID, State: transaction.StateStarting, Message: "test update progress"})
	}
	return u.record, u.err
}

// UpdateNativeJAR 记录分派参数并上报一条更新进度，返回预设事务与错误。
func (u *fakeUpdater) UpdateNativeJAR(_ context.Context, file string, report backendupdate.ProgressReporter) (transaction.Transaction, error) {
	u.file = file
	u.inputType = daemonapi.InputTypeNativeJAR
	u.operation = daemonapi.OperationUpdate
	if report != nil {
		report(backendupdate.Progress{TransactionID: u.record.ID, State: transaction.StateStarting, Message: "test update progress"})
	}
	return u.record, u.err
}

// UpdateContainerImage 记录分派参数并上报一条更新进度，返回预设事务与错误。
func (u *fakeUpdater) UpdateContainerImage(_ context.Context, imageReference string, _ bool, report backendupdate.ProgressReporter) (transaction.Transaction, error) {
	u.file = imageReference
	u.inputType = daemonapi.InputTypeContainerImage
	u.operation = daemonapi.OperationUpdate
	if report != nil {
		report(backendupdate.Progress{TransactionID: u.record.ID, State: transaction.StateStarting, Message: "test update progress"})
	}
	return u.record, u.err
}

// Restart 记录重启操作并上报一条重启进度，返回预设事务与错误。
func (u *fakeUpdater) Restart(_ context.Context, report backendupdate.ProgressReporter) (transaction.Transaction, error) {
	u.operation = daemonapi.OperationRestart
	if report != nil {
		report(backendupdate.Progress{TransactionID: u.record.ID, State: transaction.StateStarting, Message: "test restart progress"})
	}
	return u.record, u.err
}

// fakeDiagnoser backendDiagnoser 接口的测试替身，返回预设的空诊断结果。
type fakeDiagnoser struct {
	diagnosis daemonapi.Diagnosis
	err       error
	applied   bool
}

// Diagnose 返回预设的诊断结果。
func (d *fakeDiagnoser) Diagnose(context.Context) (daemonapi.Diagnosis, error) {
	return d.diagnosis, d.err
}

// Reconcile 返回预设的诊断结果，并记录 apply 标志。
func (d *fakeDiagnoser) Reconcile(_ context.Context, apply bool) (daemonapi.Diagnosis, error) {
	d.applied = apply
	return d.diagnosis, d.err
}