mirror of
https://gh-proxy.org/https://github.com/klinsmaya/CLIProxyAPI.git
synced 2026-11-08 17:36:47 +08:00
Merge pull request #3872 from router-for-me/codex/pluginhost-async-reload
Fix async plugin reload races
This commit is contained in:
@@ -24,7 +24,6 @@ func sumRecentRequestBuckets(buckets []coreauth.RecentRequestBucket) (int64, int
|
||||
|
||||
func TestGetAPIKeyUsage_GroupsByProviderAndAPIKey(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
manager := coreauth.NewManager(nil, nil, nil)
|
||||
if _, err := manager.Register(context.Background(), &coreauth.Auth{
|
||||
|
||||
@@ -28,7 +28,6 @@ import (
|
||||
geminiAuth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/gemini"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/auth/kimi"
|
||||
xaiauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/xai"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/config"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/interfaces"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/misc"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/registry"
|
||||
@@ -1267,13 +1266,11 @@ func (h *Handler) PatchAuthFileStatus(c *gin.Context) {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "config api key entry not found"})
|
||||
return
|
||||
}
|
||||
if errSave := config.SaveConfigPreserveComments(h.configFilePath, h.cfg); errSave != nil {
|
||||
h.mu.Unlock()
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to save config: %v", errSave)})
|
||||
cfgSnapshot, okSnapshot := h.saveConfigAndSnapshotLocked(c)
|
||||
h.mu.Unlock()
|
||||
if !okSnapshot {
|
||||
return
|
||||
}
|
||||
cfgSnapshot := h.cfg
|
||||
h.mu.Unlock()
|
||||
h.reloadConfigAfterManagementSave(ctx, cfgSnapshot)
|
||||
if h.tokenStore != nil {
|
||||
_ = h.tokenStore.Delete(ctx, targetAuth.ID)
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
|
||||
func TestUploadAuthFile_BatchMultipart(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
manager := coreauth.NewManager(nil, nil, nil)
|
||||
@@ -86,7 +85,6 @@ func TestUploadAuthFile_BatchMultipart(t *testing.T) {
|
||||
|
||||
func TestUploadAuthFile_BatchMultipart_InvalidJSONDoesNotOverwriteExistingFile(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
manager := coreauth.NewManager(nil, nil, nil)
|
||||
@@ -152,7 +150,6 @@ func TestUploadAuthFile_BatchMultipart_InvalidJSONDoesNotOverwriteExistingFile(t
|
||||
|
||||
func TestDeleteAuthFile_BatchQuery(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
files := []string{"alpha.json", "beta.json"}
|
||||
|
||||
@@ -17,7 +17,6 @@ import (
|
||||
|
||||
func TestDeleteAuthFile_UsesAuthPathFromManager(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
tempDir := t.TempDir()
|
||||
authDir := filepath.Join(tempDir, "auth")
|
||||
@@ -101,7 +100,6 @@ func TestDeleteAuthFile_UsesAuthPathFromManager(t *testing.T) {
|
||||
|
||||
func TestDeleteAuthFile_FallbackToAuthDirPath(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
fileName := "fallback-user.json"
|
||||
@@ -130,7 +128,6 @@ func TestDeleteAuthFile_FallbackToAuthDirPath(t *testing.T) {
|
||||
|
||||
func TestDeleteAuthFile_RemovesRuntimeAuth(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
fileName := "runtime-remove-user.json"
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
|
||||
func TestDownloadAuthFile_ReturnsFile(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
fileName := "download-user.json"
|
||||
@@ -40,7 +39,6 @@ func TestDownloadAuthFile_ReturnsFile(t *testing.T) {
|
||||
|
||||
func TestDownloadAuthFile_RejectsPathSeparators(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := NewHandlerWithoutConfigFilePath(&config.Config{AuthDir: t.TempDir()}, nil)
|
||||
|
||||
|
||||
@@ -16,7 +16,6 @@ import (
|
||||
|
||||
func TestDownloadAuthFile_PreventsWindowsSlashTraversal(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
tempDir := t.TempDir()
|
||||
authDir := filepath.Join(tempDir, "auth")
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
|
||||
func TestPatchAuthFileFields_MergeHeadersAndDeleteEmptyValues(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
store := &memoryAuthStore{}
|
||||
manager := coreauth.NewManager(store, nil, nil)
|
||||
@@ -113,7 +112,6 @@ func TestPatchAuthFileFields_MergeHeadersAndDeleteEmptyValues(t *testing.T) {
|
||||
|
||||
func TestPatchAuthFileFields_HeadersEmptyMapIsNoop(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
store := &memoryAuthStore{}
|
||||
manager := coreauth.NewManager(store, nil, nil)
|
||||
@@ -168,7 +166,6 @@ func TestPatchAuthFileFields_HeadersEmptyMapIsNoop(t *testing.T) {
|
||||
|
||||
func TestPatchAuthFileFields_WebsocketsFalseIsUpdate(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
store := &memoryAuthStore{}
|
||||
manager := coreauth.NewManager(store, nil, nil)
|
||||
@@ -217,7 +214,6 @@ func TestPatchAuthFileFields_WebsocketsFalseIsUpdate(t *testing.T) {
|
||||
|
||||
func TestPatchAuthFileFields_ArbitraryFieldsPersistToFile(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
fileName := "generic.json"
|
||||
|
||||
@@ -16,7 +16,6 @@ import (
|
||||
|
||||
func TestListAuthFiles_IncludesProjectIDFromManager(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
fileName := "[email protected]"
|
||||
@@ -55,7 +54,6 @@ func TestListAuthFiles_IncludesProjectIDFromManager(t *testing.T) {
|
||||
|
||||
func TestListAuthFilesFromDisk_IncludesProjectID(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
filePath := filepath.Join(authDir, "[email protected]")
|
||||
@@ -73,7 +71,6 @@ func TestListAuthFilesFromDisk_IncludesProjectID(t *testing.T) {
|
||||
|
||||
func TestListAuthFiles_IncludesWebsocketsFromManager(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
fileName := "[email protected]"
|
||||
@@ -111,7 +108,6 @@ func TestListAuthFiles_IncludesWebsocketsFromManager(t *testing.T) {
|
||||
|
||||
func TestListAuthFilesFromDisk_IncludesWebsockets(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := t.TempDir()
|
||||
filePath := filepath.Join(authDir, "[email protected]")
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
|
||||
func TestListAuthFiles_IncludesRecentRequestsBuckets(t *testing.T) {
|
||||
t.Setenv("MANAGEMENT_PASSWORD", "")
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
manager := coreauth.NewManager(nil, nil, nil)
|
||||
record := &coreauth.Auth{
|
||||
|
||||
@@ -24,7 +24,6 @@ func writeTestConfigFile(t *testing.T) string {
|
||||
|
||||
func TestDeleteGeminiKey_RequiresBaseURLWhenAPIKeyDuplicated(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -52,7 +51,6 @@ func TestDeleteGeminiKey_RequiresBaseURLWhenAPIKeyDuplicated(t *testing.T) {
|
||||
|
||||
func TestDeleteGeminiKey_DeletesOnlyMatchingBaseURL(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -83,7 +81,6 @@ func TestDeleteGeminiKey_DeletesOnlyMatchingBaseURL(t *testing.T) {
|
||||
|
||||
func TestDeleteClaudeKey_DeletesEmptyBaseURLWhenExplicitlyProvided(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -114,7 +111,6 @@ func TestDeleteClaudeKey_DeletesEmptyBaseURLWhenExplicitlyProvided(t *testing.T)
|
||||
|
||||
func TestDeleteVertexCompatKey_DeletesOnlyMatchingBaseURL(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -145,7 +141,6 @@ func TestDeleteVertexCompatKey_DeletesOnlyMatchingBaseURL(t *testing.T) {
|
||||
|
||||
func TestDeleteCodexKey_RequiresBaseURLWhenAPIKeyDuplicated(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
|
||||
@@ -38,25 +38,33 @@ const attemptMaxIdleTime = 2 * time.Hour
|
||||
|
||||
// Handler aggregates config reference, persistence path and helpers.
|
||||
type Handler struct {
|
||||
cfg *config.Config
|
||||
configFilePath string
|
||||
mu sync.Mutex
|
||||
attemptsMu sync.Mutex
|
||||
failedAttempts map[string]*attemptInfo // keyed by client IP
|
||||
authManager *coreauth.Manager
|
||||
tokenStore coreauth.Store
|
||||
localPassword string
|
||||
allowRemoteOverride bool
|
||||
envSecret string
|
||||
logDir string
|
||||
postAuthHook coreauth.PostAuthHook
|
||||
postAuthPersistHook coreauth.PostAuthHook
|
||||
pluginHost *pluginhost.Host
|
||||
configReloadHook func(context.Context, *config.Config)
|
||||
pluginStoreRegistryURL string
|
||||
pluginStoreHTTPClient pluginstore.HTTPDoer
|
||||
pluginReleaseCacheMu sync.Mutex
|
||||
pluginReleaseCache map[string]pluginReleaseCacheEntry
|
||||
cfg *config.Config
|
||||
configFilePath string
|
||||
mu sync.Mutex
|
||||
reloadMu sync.Mutex
|
||||
reloadGeneration uint64
|
||||
appliedReloadGeneration uint64
|
||||
attemptsMu sync.Mutex
|
||||
failedAttempts map[string]*attemptInfo // keyed by client IP
|
||||
authManager *coreauth.Manager
|
||||
tokenStore coreauth.Store
|
||||
localPassword string
|
||||
allowRemoteOverride bool
|
||||
envSecret string
|
||||
logDir string
|
||||
postAuthHook coreauth.PostAuthHook
|
||||
postAuthPersistHook coreauth.PostAuthHook
|
||||
pluginHost *pluginhost.Host
|
||||
configReloadHook func(context.Context, *config.Config)
|
||||
pluginStoreRegistryURL string
|
||||
pluginStoreHTTPClient pluginstore.HTTPDoer
|
||||
pluginReleaseCacheMu sync.Mutex
|
||||
pluginReleaseCache map[string]pluginReleaseCacheEntry
|
||||
}
|
||||
|
||||
type configReloadSnapshot struct {
|
||||
cfg *config.Config
|
||||
generation uint64
|
||||
}
|
||||
|
||||
// NewHandler creates a new management handler instance.
|
||||
@@ -152,25 +160,63 @@ func (h *Handler) SetConfigReloadHook(hook func(context.Context, *config.Config)
|
||||
h.mu.Unlock()
|
||||
}
|
||||
|
||||
func (h *Handler) reloadConfigAfterManagementSave(ctx context.Context, cfg *config.Config) {
|
||||
if h == nil || cfg == nil {
|
||||
// reloadSnapshotConfigLocked clones the runtime config and assigns a reload generation.
|
||||
// Callers must hold h.mu.
|
||||
func (h *Handler) reloadSnapshotConfigLocked() configReloadSnapshot {
|
||||
if h == nil || h.cfg == nil {
|
||||
return configReloadSnapshot{}
|
||||
}
|
||||
h.reloadGeneration++
|
||||
return configReloadSnapshot{
|
||||
cfg: h.cfg.CloneForRuntime(),
|
||||
generation: h.reloadGeneration,
|
||||
}
|
||||
}
|
||||
|
||||
// saveConfigAndSnapshotLocked saves h.cfg and returns a full runtime config snapshot.
|
||||
// Callers must hold h.mu.
|
||||
func (h *Handler) saveConfigAndSnapshotLocked(c *gin.Context) (configReloadSnapshot, bool) {
|
||||
if errSave := config.SaveConfigPreserveComments(h.configFilePath, h.cfg); errSave != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("failed to save config: %v", errSave)})
|
||||
return configReloadSnapshot{}, false
|
||||
}
|
||||
return h.reloadSnapshotConfigLocked(), true
|
||||
}
|
||||
|
||||
// reloadConfigAfterManagementSave reloads from an independent config snapshot.
|
||||
// Callers must pass a full Config clone captured immediately after a successful save.
|
||||
func (h *Handler) reloadConfigAfterManagementSave(ctx context.Context, snapshot configReloadSnapshot) {
|
||||
if h == nil || snapshot.cfg == nil || snapshot.generation == 0 {
|
||||
return
|
||||
}
|
||||
h.reloadMu.Lock()
|
||||
defer h.reloadMu.Unlock()
|
||||
|
||||
h.mu.Lock()
|
||||
if snapshot.generation < h.appliedReloadGeneration {
|
||||
h.mu.Unlock()
|
||||
return
|
||||
}
|
||||
hook := h.configReloadHook
|
||||
host := h.pluginHost
|
||||
h.mu.Unlock()
|
||||
if hook != nil {
|
||||
hook(ctx, cfg)
|
||||
return
|
||||
hook(ctx, snapshot.cfg)
|
||||
} else if host != nil {
|
||||
host.ApplyConfig(ctx, snapshot.cfg)
|
||||
}
|
||||
if host != nil {
|
||||
host.ApplyConfig(ctx, cfg)
|
||||
|
||||
h.mu.Lock()
|
||||
if snapshot.generation > h.appliedReloadGeneration {
|
||||
h.appliedReloadGeneration = snapshot.generation
|
||||
}
|
||||
h.mu.Unlock()
|
||||
}
|
||||
|
||||
func (h *Handler) reloadConfigAfterManagementSaveAsync(ctx context.Context, cfg *config.Config) {
|
||||
if h == nil || cfg == nil {
|
||||
// reloadConfigAfterManagementSaveAsync reloads from an independent config snapshot.
|
||||
// Callers must pass a full Config clone captured immediately after a successful save.
|
||||
func (h *Handler) reloadConfigAfterManagementSaveAsync(ctx context.Context, snapshot configReloadSnapshot) {
|
||||
if h == nil || snapshot.cfg == nil || snapshot.generation == 0 {
|
||||
return
|
||||
}
|
||||
reloadCtx := context.Background()
|
||||
@@ -183,7 +229,7 @@ func (h *Handler) reloadConfigAfterManagementSaveAsync(ctx context.Context, cfg
|
||||
log.WithField("panic", recovered).Error("management: async config reload panicked")
|
||||
}
|
||||
}()
|
||||
h.reloadConfigAfterManagementSave(reloadCtx, cfg)
|
||||
h.reloadConfigAfterManagementSave(reloadCtx, snapshot)
|
||||
}()
|
||||
}
|
||||
|
||||
|
||||
@@ -41,7 +41,6 @@ func TestAuthenticateManagementKey_LocalhostIPBan_BlocksCorrectKeyDuringBan(t *t
|
||||
}
|
||||
|
||||
func TestMiddlewareSetsSupportPluginHeader(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{},
|
||||
|
||||
@@ -706,7 +706,6 @@ func performGetLogs(t *testing.T, h *Handler, target string) logsAPIResponse {
|
||||
|
||||
func performGetLogsRaw(t *testing.T, h *Handler, target string) (int, string) {
|
||||
t.Helper()
|
||||
gin.SetMode(gin.TestMode)
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
c.Request = httptest.NewRequest(http.MethodGet, target, nil)
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
)
|
||||
|
||||
func TestPostOAuthCallbackCreatesMissingAuthDir(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
authDir := filepath.Join(t.TempDir(), "missing-auth")
|
||||
state := "test-antigravity-state"
|
||||
|
||||
@@ -198,15 +198,15 @@ func (h *Handler) installPluginFromStore(c *gin.Context, goos, goarch string) {
|
||||
return
|
||||
}
|
||||
|
||||
pluginIsLoaded := func() bool { return pluginLoaded(host, id) }
|
||||
pluginIsBusy := func() bool { return pluginBusy(host, id) }
|
||||
unloadedBeforeWrite := false
|
||||
result, errInstall := client.Install(installCtx, plugin, pluginstore.InstallOptions{
|
||||
PluginsDir: pluginsDir,
|
||||
GOOS: goos,
|
||||
GOARCH: goarch,
|
||||
PluginLoaded: pluginIsLoaded,
|
||||
PluginLoaded: pluginIsBusy,
|
||||
BeforeWrite: func() error {
|
||||
if !pluginIsLoaded() {
|
||||
if !pluginIsBusy() {
|
||||
return nil
|
||||
}
|
||||
if host == nil {
|
||||
@@ -215,8 +215,8 @@ func (h *Handler) installPluginFromStore(c *gin.Context, goos, goarch string) {
|
||||
log.WithFields(log.Fields{
|
||||
"plugin_id": id,
|
||||
"version": plugin.Version,
|
||||
}).Info("pluginstore: unloading loaded plugin before install")
|
||||
if !host.UnloadPlugin(id) && pluginIsLoaded() {
|
||||
}).Info("pluginstore: unloading busy plugin before install")
|
||||
if !host.UnloadPlugin(id) && pluginIsBusy() {
|
||||
return pluginstore.ErrLoadedPluginLocked
|
||||
}
|
||||
unloadedBeforeWrite = true
|
||||
@@ -226,9 +226,9 @@ func (h *Handler) installPluginFromStore(c *gin.Context, goos, goarch string) {
|
||||
if errInstall != nil {
|
||||
if unloadedBeforeWrite {
|
||||
h.mu.Lock()
|
||||
reloadCfg := h.cfg
|
||||
cfgSnapshot := h.reloadSnapshotConfigLocked()
|
||||
h.mu.Unlock()
|
||||
h.reloadConfigAfterManagementSave(c.Request.Context(), reloadCfg)
|
||||
h.reloadConfigAfterManagementSave(c.Request.Context(), cfgSnapshot)
|
||||
}
|
||||
if errors.Is(errInstall, pluginstore.ErrLoadedPluginLocked) {
|
||||
c.JSON(http.StatusConflict, gin.H{
|
||||
@@ -271,10 +271,10 @@ func (h *Handler) installPluginFromStore(c *gin.Context, goos, goarch string) {
|
||||
})
|
||||
return
|
||||
}
|
||||
reloadCfg := h.cfg
|
||||
cfgSnapshot := h.reloadSnapshotConfigLocked()
|
||||
h.mu.Unlock()
|
||||
|
||||
h.reloadConfigAfterManagementSave(c.Request.Context(), reloadCfg)
|
||||
h.reloadConfigAfterManagementSaveAsync(c.Request.Context(), cfgSnapshot)
|
||||
log.WithFields(log.Fields{
|
||||
"plugin_id": result.ID,
|
||||
"source_id": source.ID,
|
||||
@@ -560,9 +560,9 @@ func pluginLocalStatuses(pluginsEnabled bool, pluginsDir string, configs map[str
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func pluginLoaded(host *pluginhost.Host, id string) bool {
|
||||
func pluginBusy(host *pluginhost.Host, id string) bool {
|
||||
if host == nil {
|
||||
return false
|
||||
}
|
||||
return host.PluginLoaded(id)
|
||||
return host.PluginBusy(id)
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package management
|
||||
import (
|
||||
"archive/zip"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
@@ -25,7 +24,6 @@ import (
|
||||
|
||||
func TestListPluginStoreMergesInstalledStatus(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
pluginsDir := writeManagementPluginFile(t, "sample-provider")
|
||||
h := &Handler{
|
||||
@@ -84,7 +82,6 @@ func TestListPluginStoreMergesInstalledStatus(t *testing.T) {
|
||||
|
||||
func TestListPluginStoreEscapesRegistryStrings(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -150,7 +147,6 @@ func TestListPluginStoreEscapesRegistryStrings(t *testing.T) {
|
||||
|
||||
func TestListPluginStoreShowsLatestReleaseVersionAndCaches(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
httpClient := &countingPluginStoreHTTPClient{responses: fakePluginStoreHTTPClient{
|
||||
"https://registry.example/registry.json": registryJSON(t),
|
||||
@@ -203,7 +199,6 @@ func TestListPluginStoreShowsLatestReleaseVersionAndCaches(t *testing.T) {
|
||||
|
||||
func TestListPluginStoreFallsBackToRegistryVersion(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -242,7 +237,6 @@ func TestListPluginStoreFallsBackToRegistryVersion(t *testing.T) {
|
||||
|
||||
func TestListPluginStoreIncludesThirdPartySources(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -304,7 +298,6 @@ func TestListPluginStoreIncludesThirdPartySources(t *testing.T) {
|
||||
|
||||
func TestInstallPluginFromStoreWritesFileAndEnablesConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
pluginsDir := t.TempDir()
|
||||
archiveData := makeManagementPluginStoreZip(t, "sample-provider"+managementPluginExtension(runtime.GOOS), "library-data")
|
||||
@@ -335,6 +328,7 @@ func TestInstallPluginFromStoreWritesFileAndEnablesConfig(t *testing.T) {
|
||||
"https://downloads.example/checksums.txt": []byte(hex.EncodeToString(checksum[:]) + " " + archiveName + "\n"),
|
||||
},
|
||||
}
|
||||
reloads, reloadDone := captureConfigReload(h)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
@@ -346,6 +340,11 @@ func TestInstallPluginFromStoreWritesFileAndEnablesConfig(t *testing.T) {
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
|
||||
}
|
||||
cfgSnapshot := waitForAsyncReload(t, reloads)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
if cfgSnapshot == h.cfg {
|
||||
t.Fatalf("reload config = handler config %p, want independent snapshot", h.cfg)
|
||||
}
|
||||
var body pluginInstallResponse
|
||||
if errDecode := json.Unmarshal(rec.Body.Bytes(), &body); errDecode != nil {
|
||||
t.Fatalf("Unmarshal() error = %v; body=%s", errDecode, rec.Body.String())
|
||||
@@ -371,18 +370,27 @@ func TestInstallPluginFromStoreWritesFileAndEnablesConfig(t *testing.T) {
|
||||
if item.Enabled == nil || !*item.Enabled {
|
||||
t.Fatalf("plugin enabled = %#v, want true", item.Enabled)
|
||||
}
|
||||
snapshotItem := cfgSnapshot.Plugins.Configs["sample-provider"]
|
||||
if snapshotItem.Enabled == nil || !*snapshotItem.Enabled {
|
||||
t.Fatalf("snapshot plugin enabled = %#v, want true", snapshotItem.Enabled)
|
||||
}
|
||||
if h.cfg.Plugins.Enabled {
|
||||
t.Fatal("global plugins.enabled changed to true")
|
||||
}
|
||||
if cfgSnapshot.Plugins.Enabled {
|
||||
t.Fatal("snapshot global plugins.enabled changed to true")
|
||||
}
|
||||
raw := marshalPluginRaw(t, item)
|
||||
if !strings.Contains(raw, "mode: fast") {
|
||||
t.Fatalf("plugin raw config lost custom field:\n%s", raw)
|
||||
}
|
||||
if raw := marshalPluginRaw(t, snapshotItem); !strings.Contains(raw, "mode: fast") {
|
||||
t.Fatalf("snapshot plugin raw config lost custom field:\n%s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInstallPluginFromStoreUsesRequestedThirdPartySource(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
pluginsDir := t.TempDir()
|
||||
archiveData := makeManagementPluginStoreZip(t, "sample-provider"+managementPluginExtension(runtime.GOOS), "third-party-library-data")
|
||||
@@ -411,6 +419,7 @@ func TestInstallPluginFromStoreUsesRequestedThirdPartySource(t *testing.T) {
|
||||
"https://downloads.example/checksums.txt": []byte(hex.EncodeToString(checksum[:]) + " " + archiveName + "\n"),
|
||||
},
|
||||
}
|
||||
reloads, reloadDone := captureConfigReload(h)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
@@ -423,6 +432,11 @@ func TestInstallPluginFromStoreUsesRequestedThirdPartySource(t *testing.T) {
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
|
||||
}
|
||||
cfgSnapshot := waitForAsyncReload(t, reloads)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
if cfgSnapshot == h.cfg {
|
||||
t.Fatalf("reload config = handler config %p, want independent snapshot", h.cfg)
|
||||
}
|
||||
var body pluginInstallResponse
|
||||
if errDecode := json.Unmarshal(rec.Body.Bytes(), &body); errDecode != nil {
|
||||
t.Fatalf("Unmarshal() error = %v; body=%s", errDecode, rec.Body.String())
|
||||
@@ -438,11 +452,14 @@ func TestInstallPluginFromStoreUsesRequestedThirdPartySource(t *testing.T) {
|
||||
if string(data) != "third-party-library-data" {
|
||||
t.Fatalf("installed file = %q, want third-party-library-data", data)
|
||||
}
|
||||
snapshotItem := cfgSnapshot.Plugins.Configs["sample-provider"]
|
||||
if snapshotItem.Enabled == nil || !*snapshotItem.Enabled {
|
||||
t.Fatalf("snapshot plugin enabled = %#v, want true", snapshotItem.Enabled)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInstallPluginFromStoreRequiresSourceForDuplicateIDs(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -476,7 +493,6 @@ func TestInstallPluginFromStoreRequiresSourceForDuplicateIDs(t *testing.T) {
|
||||
|
||||
func TestInstallPluginFromStoreOverwritesFilePreservesConfigAndReloads(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
pluginsDir := t.TempDir()
|
||||
existingPath := filepath.Join(pluginsDir, "sample-provider"+managementPluginExtension(runtime.GOOS))
|
||||
@@ -511,13 +527,7 @@ func TestInstallPluginFromStoreOverwritesFilePreservesConfigAndReloads(t *testin
|
||||
"https://downloads.example/checksums.txt": []byte(hex.EncodeToString(checksum[:]) + " " + archiveName + "\n"),
|
||||
},
|
||||
}
|
||||
reloads := 0
|
||||
h.SetConfigReloadHook(func(_ context.Context, cfg *config.Config) {
|
||||
reloads++
|
||||
if cfg != h.cfg {
|
||||
t.Fatalf("reload config = %p, want handler config %p", cfg, h.cfg)
|
||||
}
|
||||
})
|
||||
reloads, reloadDone := captureConfigReload(h)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
@@ -529,8 +539,10 @@ func TestInstallPluginFromStoreOverwritesFilePreservesConfigAndReloads(t *testin
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
|
||||
}
|
||||
if reloads != 1 {
|
||||
t.Fatalf("reloads = %d, want 1", reloads)
|
||||
cfgSnapshot := waitForAsyncReload(t, reloads)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
if cfgSnapshot == h.cfg {
|
||||
t.Fatalf("reload config = handler config %p, want independent snapshot", h.cfg)
|
||||
}
|
||||
data, errRead := os.ReadFile(existingPath)
|
||||
if errRead != nil {
|
||||
@@ -543,13 +555,23 @@ func TestInstallPluginFromStoreOverwritesFilePreservesConfigAndReloads(t *testin
|
||||
if item.Enabled == nil || !*item.Enabled {
|
||||
t.Fatalf("plugin enabled = %#v, want true", item.Enabled)
|
||||
}
|
||||
snapshotItem := cfgSnapshot.Plugins.Configs["sample-provider"]
|
||||
if snapshotItem.Enabled == nil || !*snapshotItem.Enabled {
|
||||
t.Fatalf("snapshot plugin enabled = %#v, want true", snapshotItem.Enabled)
|
||||
}
|
||||
if item.Priority != 5 {
|
||||
t.Fatalf("plugin priority = %d, want 5", item.Priority)
|
||||
}
|
||||
if snapshotItem.Priority != 5 {
|
||||
t.Fatalf("snapshot plugin priority = %d, want 5", snapshotItem.Priority)
|
||||
}
|
||||
raw := marshalPluginRaw(t, item)
|
||||
if !strings.Contains(raw, "mode: fast") || !strings.Contains(raw, "extra: keep") {
|
||||
t.Fatalf("plugin raw config lost custom fields:\n%s", raw)
|
||||
}
|
||||
if raw := marshalPluginRaw(t, snapshotItem); !strings.Contains(raw, "mode: fast") || !strings.Contains(raw, "extra: keep") {
|
||||
t.Fatalf("snapshot plugin raw config lost custom fields:\n%s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnablePluginConfigLockedPreservesExistingFields(t *testing.T) {
|
||||
|
||||
@@ -215,18 +215,25 @@ func (h *Handler) PatchPluginEnabled(c *gin.Context) {
|
||||
}
|
||||
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
ensurePluginConfigMap(h.cfg)
|
||||
item := h.cfg.Plugins.Configs[id]
|
||||
node := pluginConfigNode(item)
|
||||
setYAMLMappingValue(node, "enabled", boolYAMLNode(*body.Enabled))
|
||||
updated, errConfig := pluginInstanceConfigFromNode(node)
|
||||
if errConfig != nil {
|
||||
h.mu.Unlock()
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid_config", "message": errConfig.Error()})
|
||||
return
|
||||
}
|
||||
h.cfg.Plugins.Configs[id] = updated
|
||||
h.persistLocked(c)
|
||||
cfgSnapshot, okSnapshot := h.saveConfigAndSnapshotLocked(c)
|
||||
h.mu.Unlock()
|
||||
if !okSnapshot {
|
||||
return
|
||||
}
|
||||
|
||||
h.reloadConfigAfterManagementSaveAsync(c.Request.Context(), cfgSnapshot)
|
||||
c.JSON(http.StatusOK, gin.H{"status": "ok"})
|
||||
}
|
||||
|
||||
// PutPluginConfig replaces plugins.configs.<id> with the request object.
|
||||
@@ -331,7 +338,7 @@ func (h *Handler) DeletePlugin(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
if pluginLoaded(host, id) && (host == nil || !host.UnloadPlugin(id)) && pluginLoaded(host, id) {
|
||||
if pluginBusy(host, id) && (host == nil || !host.UnloadPlugin(id)) && pluginBusy(host, id) {
|
||||
c.JSON(http.StatusConflict, gin.H{
|
||||
"error": "plugin_delete_requires_restart",
|
||||
"message": "loaded plugin cannot be deleted while the server is running",
|
||||
@@ -366,10 +373,10 @@ func (h *Handler) DeletePlugin(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
}
|
||||
reloadCfg := h.cfg
|
||||
cfgSnapshot := h.reloadSnapshotConfigLocked()
|
||||
h.mu.Unlock()
|
||||
|
||||
h.reloadConfigAfterManagementSaveAsync(c.Request.Context(), reloadCfg)
|
||||
h.reloadConfigAfterManagementSaveAsync(c.Request.Context(), cfgSnapshot)
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "deleted",
|
||||
"id": htmlsanitize.String(id),
|
||||
|
||||
@@ -32,9 +32,60 @@ func waitForAsyncReload(t *testing.T, reloads <-chan *config.Config) *config.Con
|
||||
}
|
||||
}
|
||||
|
||||
func waitForReloadDone(t *testing.T, done <-chan struct{}) {
|
||||
t.Helper()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for config reload hook to finish")
|
||||
}
|
||||
}
|
||||
|
||||
func captureConfigReload(h *Handler) (<-chan *config.Config, <-chan struct{}) {
|
||||
reloads := make(chan *config.Config, 1)
|
||||
done := make(chan struct{})
|
||||
h.SetConfigReloadHook(func(_ context.Context, cfg *config.Config) {
|
||||
defer close(done)
|
||||
reloads <- cfg
|
||||
})
|
||||
return reloads, done
|
||||
}
|
||||
|
||||
func TestConfigReloadGenerationSkipsOlderSnapshot(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
Plugins: config.PluginsConfig{
|
||||
Configs: map[string]config.PluginInstanceConfig{
|
||||
"sample": pluginConfigFromYAML(t, "enabled: true\nmode: old\n"),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
reloadedModes := make([]string, 0, 1)
|
||||
h.SetConfigReloadHook(func(_ context.Context, cfg *config.Config) {
|
||||
reloadedModes = append(reloadedModes, pluginRawScalarValue(t, cfg.Plugins.Configs["sample"], "mode"))
|
||||
})
|
||||
|
||||
h.mu.Lock()
|
||||
older := h.reloadSnapshotConfigLocked()
|
||||
item := h.cfg.Plugins.Configs["sample"]
|
||||
setPluginRawScalarValue(t, &item.Raw, "mode", "new")
|
||||
h.cfg.Plugins.Configs["sample"] = item
|
||||
newer := h.reloadSnapshotConfigLocked()
|
||||
h.mu.Unlock()
|
||||
|
||||
h.reloadConfigAfterManagementSave(context.Background(), newer)
|
||||
h.reloadConfigAfterManagementSave(context.Background(), older)
|
||||
|
||||
if len(reloadedModes) != 1 || reloadedModes[0] != "new" {
|
||||
t.Fatalf("reloaded modes = %#v, want only new snapshot", reloadedModes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListPluginsIncludesScannedAndConfiguredPlugins(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
pluginsDir := writeManagementPluginFile(t, "scanned")
|
||||
disabled := false
|
||||
@@ -117,7 +168,6 @@ func TestListPluginsIncludesScannedAndConfiguredPlugins(t *testing.T) {
|
||||
|
||||
func TestGetPluginConfigReturnsPreservedRawConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -174,7 +224,6 @@ options:
|
||||
|
||||
func TestGetPluginConfigReturnsEmptyObjectForKnownUnconfiguredPlugin(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
pluginsDir := writeManagementPluginFile(t, "scanned")
|
||||
h := &Handler{
|
||||
@@ -207,7 +256,6 @@ func TestGetPluginConfigReturnsEmptyObjectForKnownUnconfiguredPlugin(t *testing.
|
||||
|
||||
func TestGetPluginConfigReturnsNotFoundForUnknownPlugin(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{},
|
||||
@@ -228,7 +276,6 @@ func TestGetPluginConfigReturnsNotFoundForUnknownPlugin(t *testing.T) {
|
||||
|
||||
func TestPatchPluginEnabledUpdatesOnlyPluginConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -241,6 +288,7 @@ func TestPatchPluginEnabledUpdatesOnlyPluginConfig(t *testing.T) {
|
||||
},
|
||||
configFilePath: writeTestConfigFile(t),
|
||||
}
|
||||
reloads, reloadDone := captureConfigReload(h)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
@@ -253,6 +301,21 @@ func TestPatchPluginEnabledUpdatesOnlyPluginConfig(t *testing.T) {
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
|
||||
}
|
||||
cfgSnapshot := waitForAsyncReload(t, reloads)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
if cfgSnapshot == h.cfg {
|
||||
t.Fatalf("reload config = handler config %p, want independent snapshot", h.cfg)
|
||||
}
|
||||
if cfgSnapshot.Plugins.Enabled {
|
||||
t.Fatal("snapshot global Plugins.Enabled changed to true")
|
||||
}
|
||||
snapshotItem := cfgSnapshot.Plugins.Configs["sample"]
|
||||
if snapshotItem.Enabled == nil || !*snapshotItem.Enabled {
|
||||
t.Fatalf("snapshot sample enabled = %#v, want true", snapshotItem.Enabled)
|
||||
}
|
||||
if raw := marshalPluginRaw(t, snapshotItem); !strings.Contains(raw, "mode: safe") {
|
||||
t.Fatalf("snapshot raw config lost custom field:\n%s", raw)
|
||||
}
|
||||
if h.cfg.Plugins.Enabled {
|
||||
t.Fatal("global Plugins.Enabled changed to true")
|
||||
}
|
||||
@@ -266,9 +329,71 @@ func TestPatchPluginEnabledUpdatesOnlyPluginConfig(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPatchPluginEnabledReloadSnapshotRawImmutability(t *testing.T) {
|
||||
t.Parallel()
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
Plugins: config.PluginsConfig{
|
||||
Configs: map[string]config.PluginInstanceConfig{
|
||||
"sample": pluginConfigFromYAML(t, "enabled: false\nmode: first\n"),
|
||||
},
|
||||
},
|
||||
},
|
||||
configFilePath: writeTestConfigFile(t),
|
||||
}
|
||||
reloads := make(chan *config.Config, 1)
|
||||
releaseReload := make(chan struct{})
|
||||
reloadDone := make(chan struct{})
|
||||
h.SetConfigReloadHook(func(_ context.Context, cfg *config.Config) {
|
||||
defer close(reloadDone)
|
||||
reloads <- cfg
|
||||
<-releaseReload
|
||||
})
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(rec)
|
||||
c.Params = gin.Params{{Key: "id", Value: "sample"}}
|
||||
c.Request = httptest.NewRequest(http.MethodPatch, "/v0/management/plugins/sample/enabled", strings.NewReader(`{"enabled":true}`))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
|
||||
h.PatchPluginEnabled(c)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
close(releaseReload)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
t.Fatalf("status = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
|
||||
}
|
||||
cfgSnapshot := waitForAsyncReload(t, reloads)
|
||||
|
||||
h.mu.Lock()
|
||||
item := h.cfg.Plugins.Configs["sample"]
|
||||
setPluginRawScalarValue(t, &item.Raw, "mode", "second")
|
||||
h.cfg.Plugins.Configs["sample"] = item
|
||||
h.mu.Unlock()
|
||||
|
||||
if cfgSnapshot == h.cfg {
|
||||
t.Fatalf("reload config = handler config %p, want independent snapshot", h.cfg)
|
||||
}
|
||||
snapshotItem := cfgSnapshot.Plugins.Configs["sample"]
|
||||
if snapshotItem.Enabled == nil || !*snapshotItem.Enabled {
|
||||
t.Fatalf("snapshot sample enabled = %#v, want true", snapshotItem.Enabled)
|
||||
}
|
||||
if got := pluginRawScalarValue(t, snapshotItem, "mode"); got != "first" {
|
||||
t.Fatalf("snapshot raw mode = %q, want first", got)
|
||||
}
|
||||
h.mu.Lock()
|
||||
handlerItem := h.cfg.Plugins.Configs["sample"]
|
||||
h.mu.Unlock()
|
||||
if got := pluginRawScalarValue(t, handlerItem, "mode"); got != "second" {
|
||||
t.Fatalf("handler raw mode = %q, want second", got)
|
||||
}
|
||||
|
||||
close(releaseReload)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
}
|
||||
|
||||
func TestPutPluginConfigReplacesPluginConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -304,7 +429,6 @@ func TestPutPluginConfigReplacesPluginConfig(t *testing.T) {
|
||||
|
||||
func TestPatchPluginConfigMergesAndDeletesFields(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{
|
||||
@@ -340,7 +464,6 @@ func TestPatchPluginConfigMergesAndDeletesFields(t *testing.T) {
|
||||
|
||||
func TestDeletePluginRemovesDiscoveredFileAndConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
pluginsDir := writeManagementPluginFile(t, "sample")
|
||||
h := &Handler{
|
||||
@@ -356,8 +479,9 @@ func TestDeletePluginRemovesDiscoveredFileAndConfig(t *testing.T) {
|
||||
}
|
||||
reloads := make(chan *config.Config, 1)
|
||||
releaseReload := make(chan struct{})
|
||||
defer close(releaseReload)
|
||||
reloadDone := make(chan struct{})
|
||||
h.SetConfigReloadHook(func(_ context.Context, cfg *config.Config) {
|
||||
defer close(reloadDone)
|
||||
reloads <- cfg
|
||||
<-releaseReload
|
||||
})
|
||||
@@ -396,14 +520,23 @@ func TestDeletePluginRemovesDiscoveredFileAndConfig(t *testing.T) {
|
||||
if _, errStat := os.Stat(path); !os.IsNotExist(errStat) {
|
||||
t.Fatalf("plugin file stat error = %v, want not exist", errStat)
|
||||
}
|
||||
if cfg := waitForAsyncReload(t, reloads); cfg != h.cfg {
|
||||
t.Fatalf("reload config = %p, want handler config %p", cfg, h.cfg)
|
||||
cfgSnapshot := waitForAsyncReload(t, reloads)
|
||||
if cfgSnapshot == h.cfg {
|
||||
close(releaseReload)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
t.Fatalf("reload config = handler config %p, want independent snapshot", h.cfg)
|
||||
}
|
||||
if _, ok := cfgSnapshot.Plugins.Configs["sample"]; ok {
|
||||
close(releaseReload)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
t.Fatal("snapshot plugin config still exists after delete")
|
||||
}
|
||||
close(releaseReload)
|
||||
waitForReloadDone(t, reloadDone)
|
||||
}
|
||||
|
||||
func TestDeletePluginReturnsNotFoundForUnknownPlugin(t *testing.T) {
|
||||
t.Parallel()
|
||||
gin.SetMode(gin.TestMode)
|
||||
|
||||
h := &Handler{
|
||||
cfg: &config.Config{},
|
||||
@@ -516,3 +649,25 @@ func marshalPluginRaw(t *testing.T, item config.PluginInstanceConfig) string {
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
|
||||
func pluginRawScalarValue(t *testing.T, item config.PluginInstanceConfig, key string) string {
|
||||
t.Helper()
|
||||
for i := 0; i+1 < len(item.Raw.Content); i += 2 {
|
||||
if item.Raw.Content[i] != nil && item.Raw.Content[i].Value == key && item.Raw.Content[i+1] != nil {
|
||||
return item.Raw.Content[i+1].Value
|
||||
}
|
||||
}
|
||||
t.Fatalf("plugin raw missing scalar key %q", key)
|
||||
return ""
|
||||
}
|
||||
|
||||
func setPluginRawScalarValue(t *testing.T, node *yaml.Node, key, value string) {
|
||||
t.Helper()
|
||||
for i := 0; i+1 < len(node.Content); i += 2 {
|
||||
if node.Content[i] != nil && node.Content[i].Value == key && node.Content[i+1] != nil {
|
||||
node.Content[i+1].Value = value
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("plugin raw missing scalar key %q", key)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
package management
|
||||
|
||||
import (
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
os.Exit(m.Run())
|
||||
}
|
||||
@@ -11,7 +11,6 @@ import (
|
||||
)
|
||||
|
||||
func TestGetUsageQueuePopsRequestedRecords(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
withManagementUsageQueue(t, func() {
|
||||
redisqueue.Enqueue([]byte(`{"id":1}`))
|
||||
redisqueue.Enqueue([]byte(`{"id":2}`))
|
||||
@@ -46,7 +45,6 @@ func TestGetUsageQueuePopsRequestedRecords(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGetUsageQueueInvalidCountDoesNotPop(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
withManagementUsageQueue(t, func() {
|
||||
redisqueue.Enqueue([]byte(`{"id":1}`))
|
||||
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
var yamlNodeType = reflect.TypeOf(yaml.Node{})
|
||||
|
||||
// CloneForRuntime returns an independent in-memory snapshot of the full config.
|
||||
func (cfg *Config) CloneForRuntime() *Config {
|
||||
if cfg == nil {
|
||||
return nil
|
||||
}
|
||||
cloned := cloneRuntimeValue(reflect.ValueOf(cfg))
|
||||
return cloned.Interface().(*Config)
|
||||
}
|
||||
|
||||
func cloneRuntimeValue(v reflect.Value) reflect.Value {
|
||||
if !v.IsValid() {
|
||||
return v
|
||||
}
|
||||
|
||||
if v.Type() == yamlNodeType {
|
||||
node := v.Interface().(yaml.Node)
|
||||
return reflect.ValueOf(*deepCopyNode(&node))
|
||||
}
|
||||
|
||||
switch v.Kind() {
|
||||
case reflect.Pointer:
|
||||
if v.IsNil() {
|
||||
return reflect.Zero(v.Type())
|
||||
}
|
||||
out := reflect.New(v.Type().Elem())
|
||||
out.Elem().Set(cloneRuntimeValue(v.Elem()))
|
||||
return out
|
||||
case reflect.Interface:
|
||||
if v.IsNil() {
|
||||
return reflect.Zero(v.Type())
|
||||
}
|
||||
return cloneRuntimeValue(v.Elem())
|
||||
case reflect.Struct:
|
||||
out := reflect.New(v.Type()).Elem()
|
||||
for i := 0; i < v.NumField(); i++ {
|
||||
dst := out.Field(i)
|
||||
if !dst.CanSet() {
|
||||
return v
|
||||
}
|
||||
dst.Set(cloneRuntimeValue(v.Field(i)))
|
||||
}
|
||||
return out
|
||||
case reflect.Slice:
|
||||
if v.IsNil() {
|
||||
return reflect.Zero(v.Type())
|
||||
}
|
||||
out := reflect.MakeSlice(v.Type(), v.Len(), v.Len())
|
||||
for i := 0; i < v.Len(); i++ {
|
||||
out.Index(i).Set(cloneRuntimeValue(v.Index(i)))
|
||||
}
|
||||
return out
|
||||
case reflect.Array:
|
||||
out := reflect.New(v.Type()).Elem()
|
||||
for i := 0; i < v.Len(); i++ {
|
||||
out.Index(i).Set(cloneRuntimeValue(v.Index(i)))
|
||||
}
|
||||
return out
|
||||
case reflect.Map:
|
||||
if v.IsNil() {
|
||||
return reflect.Zero(v.Type())
|
||||
}
|
||||
out := reflect.MakeMapWithSize(v.Type(), v.Len())
|
||||
iter := v.MapRange()
|
||||
for iter.Next() {
|
||||
out.SetMapIndex(cloneRuntimeValue(iter.Key()), cloneRuntimeValue(iter.Value()))
|
||||
}
|
||||
return out
|
||||
default:
|
||||
return v
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,309 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/registry"
|
||||
"gopkg.in/yaml.v3"
|
||||
)
|
||||
|
||||
func TestCloneForRuntimeNil(t *testing.T) {
|
||||
var cfg *Config
|
||||
if got := cfg.CloneForRuntime(); got != nil {
|
||||
t.Fatalf("CloneForRuntime() = %#v, want nil", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCloneForRuntimeDeepCopiesConfig(t *testing.T) {
|
||||
cfg := sampleCloneRuntimeConfig()
|
||||
|
||||
clone := cfg.CloneForRuntime()
|
||||
if clone == nil {
|
||||
t.Fatal("CloneForRuntime() = nil")
|
||||
}
|
||||
if clone == cfg {
|
||||
t.Fatal("CloneForRuntime() returned original pointer")
|
||||
}
|
||||
|
||||
mutateOriginalConfig(cfg)
|
||||
|
||||
if clone.Home.Host != "home.local" {
|
||||
t.Fatalf("clone.Home.Host = %q, want home.local", clone.Home.Host)
|
||||
}
|
||||
if clone.APIKeys[0] != "client-key" {
|
||||
t.Fatalf("clone.APIKeys[0] = %q, want client-key", clone.APIKeys[0])
|
||||
}
|
||||
if clone.OAuthExcludedModels["codex"][0] != "hidden-model" {
|
||||
t.Fatalf("clone.OAuthExcludedModels[codex][0] = %q, want hidden-model", clone.OAuthExcludedModels["codex"][0])
|
||||
}
|
||||
if clone.OAuthModelAlias["codex"][0].Alias != "client-model" {
|
||||
t.Fatalf("clone.OAuthModelAlias[codex][0].Alias = %q, want client-model", clone.OAuthModelAlias["codex"][0].Alias)
|
||||
}
|
||||
if got := pluginRawScalar(t, clone.Plugins.Configs["sample"].Raw, "mode"); got != "first" {
|
||||
t.Fatalf("clone plugin raw mode = %q, want first", got)
|
||||
}
|
||||
if clone.OpenAICompatibility[0].Models[0].Thinking.Levels[0] != "low" {
|
||||
t.Fatalf("clone thinking level = %q, want low", clone.OpenAICompatibility[0].Models[0].Thinking.Levels[0])
|
||||
}
|
||||
if got := clone.Payload.Default[0].Params["object"].(map[string]any)["key"]; got != "value" {
|
||||
t.Fatalf("clone payload object key = %#v, want value", got)
|
||||
}
|
||||
|
||||
clone.APIKeys[0] = "clone-client-key"
|
||||
clone.OAuthExcludedModels["codex"][0] = "clone-hidden-model"
|
||||
clone.OAuthModelAlias["codex"][0].Alias = "clone-client-model"
|
||||
clone.OpenAICompatibility[0].Models[0].Thinking.Levels[0] = "clone-low"
|
||||
clone.Payload.Default[0].Params["object"].(map[string]any)["key"] = "clone-value"
|
||||
plugin := clone.Plugins.Configs["sample"]
|
||||
setPluginRawScalar(t, &plugin.Raw, "mode", "third")
|
||||
clone.Plugins.Configs["sample"] = plugin
|
||||
|
||||
if cfg.APIKeys[0] != "mutated-client-key" {
|
||||
t.Fatalf("cfg.APIKeys[0] = %q, want mutated-client-key", cfg.APIKeys[0])
|
||||
}
|
||||
if cfg.OAuthExcludedModels["codex"][0] != "mutated-hidden-model" {
|
||||
t.Fatalf("cfg.OAuthExcludedModels[codex][0] = %q, want mutated-hidden-model", cfg.OAuthExcludedModels["codex"][0])
|
||||
}
|
||||
if cfg.OAuthModelAlias["codex"][0].Alias != "mutated-client-model" {
|
||||
t.Fatalf("cfg.OAuthModelAlias[codex][0].Alias = %q, want mutated-client-model", cfg.OAuthModelAlias["codex"][0].Alias)
|
||||
}
|
||||
if got := pluginRawScalar(t, cfg.Plugins.Configs["sample"].Raw, "mode"); got != "second" {
|
||||
t.Fatalf("cfg plugin raw mode = %q, want second", got)
|
||||
}
|
||||
if cfg.OpenAICompatibility[0].Models[0].Thinking.Levels[0] != "mutated-low" {
|
||||
t.Fatalf("cfg thinking level = %q, want mutated-low", cfg.OpenAICompatibility[0].Models[0].Thinking.Levels[0])
|
||||
}
|
||||
if got := cfg.Payload.Default[0].Params["object"].(map[string]any)["key"]; got != "mutated-value" {
|
||||
t.Fatalf("cfg payload object key = %#v, want mutated-value", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCloneForRuntimeDoesNotShareReferenceFields(t *testing.T) {
|
||||
cfg := sampleCloneRuntimeConfig()
|
||||
clone := cfg.CloneForRuntime()
|
||||
|
||||
assertNoSharedRuntimeReferences(t, reflect.ValueOf(cfg), reflect.ValueOf(clone), "Config")
|
||||
}
|
||||
|
||||
func sampleCloneRuntimeConfig() *Config {
|
||||
cacheStrict := true
|
||||
bypassStrict := false
|
||||
pluginEnabled := false
|
||||
cacheUserID := true
|
||||
|
||||
return &Config{
|
||||
SDKConfig: SDKConfig{
|
||||
APIKeys: []string{"client-key"},
|
||||
Streaming: StreamingConfig{
|
||||
KeepAliveSeconds: 3,
|
||||
BootstrapRetries: 2,
|
||||
},
|
||||
},
|
||||
Home: HomeConfig{
|
||||
Enabled: true,
|
||||
Host: "home.local",
|
||||
Port: 8081,
|
||||
TLS: HomeTLSConfig{
|
||||
Enable: true,
|
||||
ServerName: "home.local",
|
||||
CACert: "ca",
|
||||
ClientCert: "cert",
|
||||
ClientKey: "key",
|
||||
UseTargetServerName: true,
|
||||
},
|
||||
},
|
||||
Plugins: PluginsConfig{
|
||||
Enabled: true,
|
||||
Dir: "plugins",
|
||||
StoreSources: []string{"https://plugins.example/store.json"},
|
||||
Configs: map[string]PluginInstanceConfig{
|
||||
"sample": {
|
||||
Enabled: &pluginEnabled,
|
||||
Priority: 10,
|
||||
Raw: samplePluginRawNode("first"),
|
||||
},
|
||||
},
|
||||
},
|
||||
AntigravitySignatureCacheEnabled: &cacheStrict,
|
||||
AntigravitySignatureBypassStrict: &bypassStrict,
|
||||
GeminiKey: []GeminiKey{{
|
||||
APIKey: "gemini-key",
|
||||
Models: []GeminiModel{{Name: "gemini-upstream", Alias: "gemini-client"}},
|
||||
Headers: map[string]string{"X-Gemini": "one"},
|
||||
ExcludedModels: []string{"gemini-hidden"},
|
||||
}},
|
||||
CodexKey: []CodexKey{{
|
||||
APIKey: "codex-key",
|
||||
Models: []CodexModel{{Name: "codex-upstream", Alias: "codex-client"}},
|
||||
Headers: map[string]string{"X-Codex": "one"},
|
||||
ExcludedModels: []string{"codex-hidden-key"},
|
||||
}},
|
||||
ClaudeKey: []ClaudeKey{{
|
||||
APIKey: "claude-key",
|
||||
Models: []ClaudeModel{{Name: "claude-upstream", Alias: "claude-client"}},
|
||||
Headers: map[string]string{"X-Claude": "one"},
|
||||
ExcludedModels: []string{"claude-hidden"},
|
||||
Cloak: &CloakConfig{
|
||||
SensitiveWords: []string{"secret"},
|
||||
CacheUserID: &cacheUserID,
|
||||
},
|
||||
}},
|
||||
OpenAICompatibility: []OpenAICompatibility{{
|
||||
Name: "compat",
|
||||
APIKeyEntries: []OpenAICompatibilityAPIKey{{APIKey: "compat-key", ProxyURL: "http://proxy.local"}},
|
||||
Models: []OpenAICompatibilityModel{{
|
||||
Name: "compat-upstream",
|
||||
Alias: "compat-client",
|
||||
Thinking: ®istry.ThinkingSupport{Levels: []string{"low", "high"}},
|
||||
}},
|
||||
Headers: map[string]string{"X-Compat": "one"},
|
||||
}},
|
||||
VertexCompatAPIKey: []VertexCompatKey{{
|
||||
APIKey: "vertex-key",
|
||||
Headers: map[string]string{"X-Vertex": "one"},
|
||||
Models: []VertexCompatModel{{Name: "vertex-upstream", Alias: "vertex-client"}},
|
||||
ExcludedModels: []string{"vertex-hidden"},
|
||||
}},
|
||||
OAuthExcludedModels: map[string][]string{
|
||||
"codex": {"hidden-model"},
|
||||
},
|
||||
OAuthModelAlias: map[string][]OAuthModelAlias{
|
||||
"codex": {{Name: "upstream-model", Alias: "client-model", Fork: true}},
|
||||
},
|
||||
Payload: PayloadConfig{
|
||||
Default: []PayloadRule{{
|
||||
Models: []PayloadModelRule{{
|
||||
Name: "model-*",
|
||||
Headers: map[string]string{"X-Tier": "gold"},
|
||||
Match: []map[string]any{{"tier": "gold"}},
|
||||
Exist: []string{"$.messages"},
|
||||
}},
|
||||
Params: map[string]any{
|
||||
"object": map[string]any{"key": "value"},
|
||||
"array": []any{"first", map[string]any{"nested": "value"}},
|
||||
},
|
||||
}},
|
||||
Filter: []PayloadFilterRule{{
|
||||
Models: []PayloadModelRule{{Name: "model-*"}},
|
||||
Params: []string{"$.secret"},
|
||||
}},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func mutateOriginalConfig(cfg *Config) {
|
||||
cfg.Home.Host = "mutated-home.local"
|
||||
cfg.APIKeys[0] = "mutated-client-key"
|
||||
cfg.OAuthExcludedModels["codex"][0] = "mutated-hidden-model"
|
||||
cfg.OAuthModelAlias["codex"][0].Alias = "mutated-client-model"
|
||||
cfg.OpenAICompatibility[0].Models[0].Thinking.Levels[0] = "mutated-low"
|
||||
cfg.Payload.Default[0].Params["object"].(map[string]any)["key"] = "mutated-value"
|
||||
plugin := cfg.Plugins.Configs["sample"]
|
||||
setPluginRawScalar(nil, &plugin.Raw, "mode", "second")
|
||||
cfg.Plugins.Configs["sample"] = plugin
|
||||
}
|
||||
|
||||
func samplePluginRawNode(mode string) yaml.Node {
|
||||
modeValue := &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: mode, Anchor: "modeAnchor"}
|
||||
return yaml.Node{
|
||||
Kind: yaml.MappingNode,
|
||||
Tag: "!!map",
|
||||
Content: []*yaml.Node{
|
||||
{Kind: yaml.ScalarNode, Tag: "!!str", Value: "enabled"},
|
||||
{Kind: yaml.ScalarNode, Tag: "!!bool", Value: "false"},
|
||||
{Kind: yaml.ScalarNode, Tag: "!!str", Value: "mode"},
|
||||
modeValue,
|
||||
{Kind: yaml.ScalarNode, Tag: "!!str", Value: "mode-alias"},
|
||||
{Kind: yaml.AliasNode, Alias: modeValue},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func pluginRawScalar(t *testing.T, node yaml.Node, key string) string {
|
||||
t.Helper()
|
||||
for i := 0; i+1 < len(node.Content); i += 2 {
|
||||
if node.Content[i] != nil && node.Content[i].Value == key && node.Content[i+1] != nil {
|
||||
return node.Content[i+1].Value
|
||||
}
|
||||
}
|
||||
t.Fatalf("raw plugin node missing key %q", key)
|
||||
return ""
|
||||
}
|
||||
|
||||
func setPluginRawScalar(t *testing.T, node *yaml.Node, key, value string) {
|
||||
if t != nil {
|
||||
t.Helper()
|
||||
}
|
||||
for i := 0; i+1 < len(node.Content); i += 2 {
|
||||
if node.Content[i] != nil && node.Content[i].Value == key && node.Content[i+1] != nil {
|
||||
node.Content[i+1].Value = value
|
||||
return
|
||||
}
|
||||
}
|
||||
if t != nil {
|
||||
t.Fatalf("raw plugin node missing key %q", key)
|
||||
}
|
||||
}
|
||||
|
||||
func assertNoSharedRuntimeReferences(t *testing.T, original, clone reflect.Value, path string) {
|
||||
t.Helper()
|
||||
if !original.IsValid() || !clone.IsValid() {
|
||||
return
|
||||
}
|
||||
if original.Kind() == reflect.Interface {
|
||||
if original.IsNil() || clone.IsNil() {
|
||||
return
|
||||
}
|
||||
assertNoSharedRuntimeReferences(t, original.Elem(), clone.Elem(), path)
|
||||
return
|
||||
}
|
||||
if original.Kind() != clone.Kind() {
|
||||
t.Fatalf("%s kind mismatch: %s != %s", path, original.Kind(), clone.Kind())
|
||||
}
|
||||
|
||||
switch original.Kind() {
|
||||
case reflect.Pointer:
|
||||
if original.IsNil() || clone.IsNil() {
|
||||
return
|
||||
}
|
||||
if original.Pointer() == clone.Pointer() {
|
||||
t.Fatalf("%s shares pointer %x", path, original.Pointer())
|
||||
}
|
||||
assertNoSharedRuntimeReferences(t, original.Elem(), clone.Elem(), path+"->"+original.Type().Elem().String())
|
||||
case reflect.Map:
|
||||
if original.IsNil() || clone.IsNil() {
|
||||
return
|
||||
}
|
||||
if original.Pointer() == clone.Pointer() {
|
||||
t.Fatalf("%s shares map pointer %x", path, original.Pointer())
|
||||
}
|
||||
iter := original.MapRange()
|
||||
for iter.Next() {
|
||||
key := iter.Key()
|
||||
assertNoSharedRuntimeReferences(t, iter.Value(), clone.MapIndex(key), path+"["+keyForPath(key)+"]")
|
||||
}
|
||||
case reflect.Slice:
|
||||
if original.IsNil() || clone.IsNil() {
|
||||
return
|
||||
}
|
||||
if original.Pointer() == clone.Pointer() {
|
||||
t.Fatalf("%s shares slice pointer %x", path, original.Pointer())
|
||||
}
|
||||
for i := 0; i < original.Len(); i++ {
|
||||
assertNoSharedRuntimeReferences(t, original.Index(i), clone.Index(i), path+"[]")
|
||||
}
|
||||
case reflect.Struct:
|
||||
for i := 0; i < original.NumField(); i++ {
|
||||
field := original.Type().Field(i)
|
||||
assertNoSharedRuntimeReferences(t, original.Field(i), clone.Field(i), path+"."+field.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func keyForPath(key reflect.Value) string {
|
||||
if key.Kind() == reflect.String {
|
||||
return key.String()
|
||||
}
|
||||
return key.Type().String()
|
||||
}
|
||||
@@ -1525,14 +1525,25 @@ func isZeroValueNode(node *yaml.Node) bool {
|
||||
|
||||
// deepCopyNode creates a deep copy of a yaml.Node graph.
|
||||
func deepCopyNode(n *yaml.Node) *yaml.Node {
|
||||
return deepCopyNodeSeen(n, map[*yaml.Node]*yaml.Node{})
|
||||
}
|
||||
|
||||
func deepCopyNodeSeen(n *yaml.Node, seen map[*yaml.Node]*yaml.Node) *yaml.Node {
|
||||
if n == nil {
|
||||
return nil
|
||||
}
|
||||
if cp, ok := seen[n]; ok {
|
||||
return cp
|
||||
}
|
||||
cp := *n
|
||||
seen[n] = &cp
|
||||
if n.Alias != nil {
|
||||
cp.Alias = deepCopyNodeSeen(n.Alias, seen)
|
||||
}
|
||||
if len(n.Content) > 0 {
|
||||
cp.Content = make([]*yaml.Node, len(n.Content))
|
||||
for i := range n.Content {
|
||||
cp.Content[i] = deepCopyNode(n.Content[i])
|
||||
cp.Content[i] = deepCopyNodeSeen(n.Content[i], seen)
|
||||
}
|
||||
}
|
||||
return &cp
|
||||
|
||||
+62
-12
@@ -3,7 +3,6 @@ package pluginhost
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@@ -36,9 +35,11 @@ type pluginUnloadTarget struct {
|
||||
}
|
||||
|
||||
type Host struct {
|
||||
applyMu sync.Mutex
|
||||
mu sync.Mutex
|
||||
loader pluginLoader
|
||||
loaded map[string]*loadedPlugin
|
||||
loading map[string]struct{}
|
||||
fused map[string]string
|
||||
runtimeConfig *config.Config
|
||||
authManager *coreauth.Manager
|
||||
@@ -65,6 +66,7 @@ func New() *Host {
|
||||
h := &Host{
|
||||
loader: defaultPluginLoader(),
|
||||
loaded: make(map[string]*loadedPlugin),
|
||||
loading: make(map[string]struct{}),
|
||||
fused: make(map[string]string),
|
||||
modelClientIDs: make(map[string]struct{}),
|
||||
executorModelClientIDs: make(map[string]struct{}),
|
||||
@@ -137,16 +139,38 @@ func (h *Host) PluginLoaded(id string) bool {
|
||||
return ok
|
||||
}
|
||||
|
||||
// PluginBusy reports whether a plugin dynamic library is loaded or being loaded.
|
||||
func (h *Host) PluginBusy(id string) bool {
|
||||
if h == nil {
|
||||
return false
|
||||
}
|
||||
id = strings.TrimSpace(id)
|
||||
if id == "" {
|
||||
return false
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if _, ok := h.loaded[id]; ok {
|
||||
return true
|
||||
}
|
||||
_, ok := h.loading[id]
|
||||
return ok
|
||||
}
|
||||
|
||||
func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
|
||||
if h == nil {
|
||||
return
|
||||
}
|
||||
h.applyMu.Lock()
|
||||
defer h.applyMu.Unlock()
|
||||
|
||||
rc := runtimeConfigFromConfig(cfg)
|
||||
h.mu.Lock()
|
||||
h.runtimeConfig = cfg
|
||||
h.mu.Unlock()
|
||||
|
||||
if !rc.Enabled {
|
||||
h.mu.Lock()
|
||||
h.managementRoutes = make(map[string]managementRouteRecord)
|
||||
h.resourceRoutes = make(map[string]resourceRouteRecord)
|
||||
h.snapshot.Store(emptySnapshot())
|
||||
@@ -158,6 +182,7 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
|
||||
files, errSelect := selectPluginFiles(rc.Dir)
|
||||
if errSelect != nil {
|
||||
log.Warnf("pluginhost: failed to select plugin files: %v", errSelect)
|
||||
h.mu.Lock()
|
||||
h.managementRoutes = make(map[string]managementRouteRecord)
|
||||
h.resourceRoutes = make(map[string]resourceRouteRecord)
|
||||
h.snapshot.Store(emptySnapshot())
|
||||
@@ -175,26 +200,39 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
|
||||
if !item.Enabled {
|
||||
continue
|
||||
}
|
||||
if _, disabled := h.fused[file.ID]; disabled {
|
||||
h.mu.Lock()
|
||||
lp := h.loaded[file.ID]
|
||||
_, disabled := h.fused[file.ID]
|
||||
h.mu.Unlock()
|
||||
if disabled {
|
||||
continue
|
||||
}
|
||||
|
||||
lp := h.loaded[file.ID]
|
||||
if lp == nil {
|
||||
loaded, errLoad := h.loadLocked(file)
|
||||
h.mu.Lock()
|
||||
h.loading[file.ID] = struct{}{}
|
||||
h.mu.Unlock()
|
||||
|
||||
loaded, errLoad := h.load(file)
|
||||
h.mu.Lock()
|
||||
delete(h.loading, file.ID)
|
||||
if errLoad != nil {
|
||||
h.mu.Unlock()
|
||||
log.Warnf("pluginhost: failed to load plugin %s from %s: %v", file.ID, file.Path, errLoad)
|
||||
continue
|
||||
}
|
||||
// ApplyConfig, UnloadPlugin, and ShutdownAll are serialized by applyMu,
|
||||
// so a nil read cannot race into a duplicate load.
|
||||
lp = loaded
|
||||
h.loaded[file.ID] = lp
|
||||
h.mu.Unlock()
|
||||
log.WithFields(log.Fields{
|
||||
"plugin_id": file.ID,
|
||||
"path": file.Path,
|
||||
}).Info("pluginhost: plugin loaded")
|
||||
}
|
||||
|
||||
plugin, okCall := h.callRegisterLocked(ctx, lp, item)
|
||||
plugin, okCall := h.callRegister(ctx, lp, item)
|
||||
if !okCall {
|
||||
continue
|
||||
}
|
||||
@@ -208,12 +246,13 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
|
||||
}
|
||||
|
||||
sortRecords(records)
|
||||
h.mu.Lock()
|
||||
h.snapshot.Store(&Snapshot{enabled: true, records: records})
|
||||
h.mu.Unlock()
|
||||
h.refreshThinkingProviders(records)
|
||||
}
|
||||
|
||||
func (h *Host) loadLocked(file pluginFile) (*loadedPlugin, error) {
|
||||
func (h *Host) load(file pluginFile) (*loadedPlugin, error) {
|
||||
client, errOpen := h.loader.Open(file, h)
|
||||
if errOpen != nil {
|
||||
return nil, errOpen
|
||||
@@ -236,6 +275,9 @@ func (h *Host) UnloadPlugin(id string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
h.applyMu.Lock()
|
||||
defer h.applyMu.Unlock()
|
||||
|
||||
var target pluginUnloadTarget
|
||||
h.mu.Lock()
|
||||
lp := h.loaded[id]
|
||||
@@ -269,6 +311,9 @@ func (h *Host) ShutdownAll() {
|
||||
return
|
||||
}
|
||||
|
||||
h.applyMu.Lock()
|
||||
defer h.applyMu.Unlock()
|
||||
|
||||
targets := make([]pluginUnloadTarget, 0)
|
||||
h.mu.Lock()
|
||||
for _, lp := range h.loaded {
|
||||
@@ -282,6 +327,7 @@ func (h *Host) ShutdownAll() {
|
||||
})
|
||||
}
|
||||
h.loaded = make(map[string]*loadedPlugin)
|
||||
h.loading = make(map[string]struct{})
|
||||
h.modelClientIDs = make(map[string]struct{})
|
||||
h.executorModelClientIDs = make(map[string]struct{})
|
||||
h.modelProviders = make(map[string]string)
|
||||
@@ -346,17 +392,20 @@ func (h *Host) removePluginRuntimeStateLocked(id string) {
|
||||
delete(h.modelRegistrations, id)
|
||||
}
|
||||
|
||||
func (h *Host) callRegisterLocked(ctx context.Context, lp *loadedPlugin, item runtimeItemConfig) (pluginapi.Plugin, bool) {
|
||||
func (h *Host) callRegister(ctx context.Context, lp *loadedPlugin, item runtimeItemConfig) (pluginapi.Plugin, bool) {
|
||||
if lp == nil {
|
||||
return pluginapi.Plugin{}, false
|
||||
}
|
||||
|
||||
method := pluginabi.MethodPluginRegister
|
||||
if lp.registered {
|
||||
h.mu.Lock()
|
||||
registered := lp.registered
|
||||
h.mu.Unlock()
|
||||
if registered {
|
||||
method = pluginabi.MethodPluginReconfigure
|
||||
}
|
||||
|
||||
plugin, okCall := h.safePluginCallLocked(ctx, lp.id, method, func() pluginapi.Plugin {
|
||||
plugin, okCall := h.safePluginCall(ctx, lp.id, method, func() pluginapi.Plugin {
|
||||
plugin, errRegister := registerRPCPlugin(ctx, h, lp.id, lp.client, method, item.ConfigYAML)
|
||||
if errRegister != nil {
|
||||
log.Warnf("pluginhost: plugin %s %s failed: %v", lp.id, method, errRegister)
|
||||
@@ -367,7 +416,9 @@ func (h *Host) callRegisterLocked(ctx context.Context, lp *loadedPlugin, item ru
|
||||
if !okCall {
|
||||
return pluginapi.Plugin{}, false
|
||||
}
|
||||
h.mu.Lock()
|
||||
lp.registered = true
|
||||
h.mu.Unlock()
|
||||
if !validPlugin(plugin) {
|
||||
log.Warnf("pluginhost: plugin %s returned invalid metadata or no capabilities", lp.id)
|
||||
return pluginapi.Plugin{}, false
|
||||
@@ -375,11 +426,10 @@ func (h *Host) callRegisterLocked(ctx context.Context, lp *loadedPlugin, item ru
|
||||
return plugin, true
|
||||
}
|
||||
|
||||
func (h *Host) safePluginCallLocked(ctx context.Context, id, method string, fn func() pluginapi.Plugin) (out pluginapi.Plugin, ok bool) {
|
||||
func (h *Host) safePluginCall(ctx context.Context, id, method string, fn func() pluginapi.Plugin) (out pluginapi.Plugin, ok bool) {
|
||||
defer func() {
|
||||
if recovered := recover(); recovered != nil {
|
||||
h.fused[id] = fmt.Sprintf("%s panic: %v", method, recovered)
|
||||
log.WithField("plugin_id", id).WithField("method", method).Errorf("pluginhost: plugin panic recovered: %v\n%s", recovered, debug.Stack())
|
||||
h.fusePlugin(id, method, recovered)
|
||||
out = pluginapi.Plugin{}
|
||||
ok = false
|
||||
}
|
||||
|
||||
@@ -4,7 +4,10 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/config"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/thinking"
|
||||
@@ -605,6 +608,225 @@ func TestHostApplyConfig_PanicFusesPluginForProcessLifetime(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostApplyConfigDoesNotHoldHostMuDuringRegister(t *testing.T) {
|
||||
h, cfg, registerStarted, releaseRegister := newBlockingRegisterHost(t)
|
||||
applyDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(applyDone)
|
||||
}()
|
||||
|
||||
waitForHostTestSignal(t, registerStarted, "register start")
|
||||
probeDone := make(chan struct{})
|
||||
go func() {
|
||||
_ = h.currentModelExecutor()
|
||||
close(probeDone)
|
||||
}()
|
||||
waitForHostTestSignal(t, probeDone, "Host.mu probe")
|
||||
|
||||
releaseRegister()
|
||||
waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
|
||||
|
||||
snap := h.Snapshot()
|
||||
if !snap.enabled || len(snap.records) != 1 || snap.records[0].id != "alpha" {
|
||||
t.Fatalf("Snapshot() = %+v, want alpha registered", snap)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostApplyConfigSerializesLifecycleCalls(t *testing.T) {
|
||||
loader := newTestSymbolLoader()
|
||||
started := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
secondEntered := make(chan struct{})
|
||||
var releaseOnce sync.Once
|
||||
releaseFirst := func() { releaseOnce.Do(func() { close(release) }) }
|
||||
t.Cleanup(releaseFirst)
|
||||
|
||||
var startOnce sync.Once
|
||||
var secondOnce sync.Once
|
||||
var lifecycleCalls int32
|
||||
var activeLifecycleCalls int32
|
||||
var concurrentLifecycleCalls int32
|
||||
lifecycle := func([]byte) pluginapi.Plugin {
|
||||
if active := atomic.AddInt32(&activeLifecycleCalls, 1); active > 1 {
|
||||
atomic.StoreInt32(&concurrentLifecycleCalls, 1)
|
||||
}
|
||||
call := atomic.AddInt32(&lifecycleCalls, 1)
|
||||
if call == 1 {
|
||||
startOnce.Do(func() { close(started) })
|
||||
<-release
|
||||
} else {
|
||||
secondOnce.Do(func() { close(secondEntered) })
|
||||
}
|
||||
atomic.AddInt32(&activeLifecycleCalls, -1)
|
||||
return validTestPlugin("alpha")
|
||||
}
|
||||
plugin := &testPlugin{
|
||||
registerResult: validTestPlugin("alpha"),
|
||||
reconfigureResult: validTestPlugin("alpha"),
|
||||
}
|
||||
lookup := newTestSymbolLookup(plugin)
|
||||
lookup.registerOverride = lifecycle
|
||||
lookup.reconfigureOverride = lifecycle
|
||||
loader.lookups["alpha"] = lookup
|
||||
h := NewForTest(loader)
|
||||
cfg := &config.Config{
|
||||
Plugins: config.PluginsConfig{
|
||||
Enabled: true,
|
||||
Dir: makePluginDir(t, "alpha"),
|
||||
},
|
||||
}
|
||||
|
||||
firstDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(firstDone)
|
||||
}()
|
||||
waitForHostTestSignal(t, started, "first register start")
|
||||
|
||||
secondDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(secondDone)
|
||||
}()
|
||||
select {
|
||||
case <-secondEntered:
|
||||
t.Fatal("second ApplyConfig entered plugin lifecycle before first ApplyConfig finished")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
|
||||
releaseFirst()
|
||||
waitForHostTestSignal(t, firstDone, "first ApplyConfig completion")
|
||||
waitForHostTestSignal(t, secondDone, "second ApplyConfig completion")
|
||||
|
||||
if got := atomic.LoadInt32(&lifecycleCalls); got != 2 {
|
||||
t.Fatalf("lifecycle calls = %d, want 2", got)
|
||||
}
|
||||
if atomic.LoadInt32(&concurrentLifecycleCalls) != 0 {
|
||||
t.Fatal("plugin lifecycle calls ran concurrently")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostPluginBusyReportsLoadingPlugin(t *testing.T) {
|
||||
h, cfg, openStarted, releaseOpen := newBlockingOpenHost(t)
|
||||
t.Cleanup(h.ShutdownAll)
|
||||
|
||||
applyDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(applyDone)
|
||||
}()
|
||||
|
||||
waitForHostTestSignal(t, openStarted, "plugin open start")
|
||||
if h.PluginLoaded("alpha") {
|
||||
t.Fatal("PluginLoaded(alpha) = true, want false while plugin is still loading")
|
||||
}
|
||||
if !h.PluginBusy("alpha") {
|
||||
t.Fatal("PluginBusy(alpha) = false, want true while plugin is loading")
|
||||
}
|
||||
|
||||
releaseOpen()
|
||||
waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
|
||||
if !h.PluginLoaded("alpha") {
|
||||
t.Fatal("PluginLoaded(alpha) = false, want true after load")
|
||||
}
|
||||
if !h.PluginBusy("alpha") {
|
||||
t.Fatal("PluginBusy(alpha) = false, want true after load")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostUnloadWaitsForBlockingLoad(t *testing.T) {
|
||||
h, cfg, openStarted, releaseOpen := newBlockingOpenHost(t)
|
||||
applyDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(applyDone)
|
||||
}()
|
||||
waitForHostTestSignal(t, openStarted, "plugin open start")
|
||||
|
||||
unloadDone := make(chan bool)
|
||||
go func() {
|
||||
unloadDone <- h.UnloadPlugin("alpha")
|
||||
}()
|
||||
select {
|
||||
case <-unloadDone:
|
||||
t.Fatal("UnloadPlugin completed while ApplyConfig was still loading")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
|
||||
releaseOpen()
|
||||
waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
|
||||
if ok := waitForHostTestBool(t, unloadDone, "UnloadPlugin completion"); !ok {
|
||||
t.Fatal("UnloadPlugin returned false, want true after loading completes")
|
||||
}
|
||||
if h.PluginBusy("alpha") {
|
||||
t.Fatal("PluginBusy(alpha) = true, want false after unload")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostUnloadAndShutdownWaitForBlockingRegister(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
action func(*Host) bool
|
||||
assertDone func(*testing.T, *Host)
|
||||
}{
|
||||
{
|
||||
name: "unload",
|
||||
action: func(h *Host) bool {
|
||||
return h.UnloadPlugin("alpha")
|
||||
},
|
||||
assertDone: func(t *testing.T, h *Host) {
|
||||
t.Helper()
|
||||
if h.PluginLoaded("alpha") {
|
||||
t.Fatal("PluginLoaded(alpha) = true, want false after unload")
|
||||
}
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "shutdown",
|
||||
action: func(h *Host) bool {
|
||||
h.ShutdownAll()
|
||||
return true
|
||||
},
|
||||
assertDone: func(t *testing.T, h *Host) {
|
||||
t.Helper()
|
||||
if h.PluginLoaded("alpha") {
|
||||
t.Fatal("PluginLoaded(alpha) = true, want false after shutdown")
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
h, cfg, registerStarted, releaseRegister := newBlockingRegisterHost(t)
|
||||
applyDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(applyDone)
|
||||
}()
|
||||
waitForHostTestSignal(t, registerStarted, "register start")
|
||||
|
||||
actionDone := make(chan bool)
|
||||
go func() {
|
||||
actionDone <- tt.action(h)
|
||||
}()
|
||||
select {
|
||||
case <-actionDone:
|
||||
t.Fatalf("%s completed while ApplyConfig was still registering", tt.name)
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
|
||||
releaseRegister()
|
||||
waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
|
||||
if ok := waitForHostTestBool(t, actionDone, tt.name+" completion"); !ok {
|
||||
t.Fatalf("%s returned false, want true", tt.name)
|
||||
}
|
||||
tt.assertDone(t, h)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSortRecordsPriorityDescendingAndIDTieBreak(t *testing.T) {
|
||||
records := []capabilityRecord{
|
||||
{id: "charlie", priority: 1},
|
||||
@@ -635,3 +857,98 @@ func (c *capturePluginClient) Call(ctx context.Context, method string, request [
|
||||
}
|
||||
|
||||
func (c *capturePluginClient) Shutdown() {}
|
||||
|
||||
type blockingOpenLoader struct {
|
||||
inner *testSymbolLoader
|
||||
started chan struct{}
|
||||
release <-chan struct{}
|
||||
startOnce sync.Once
|
||||
}
|
||||
|
||||
func (l *blockingOpenLoader) Open(file pluginFile, host *Host) (pluginClient, error) {
|
||||
l.startOnce.Do(func() { close(l.started) })
|
||||
<-l.release
|
||||
return l.inner.Open(file, host)
|
||||
}
|
||||
|
||||
func newBlockingOpenHost(t *testing.T) (*Host, *config.Config, <-chan struct{}, func()) {
|
||||
t.Helper()
|
||||
|
||||
inner := newTestSymbolLoader()
|
||||
plugin := &testPlugin{
|
||||
registerResult: validTestPlugin("alpha"),
|
||||
reconfigureResult: validTestPlugin("alpha"),
|
||||
}
|
||||
inner.lookups["alpha"] = newTestSymbolLookup(plugin)
|
||||
|
||||
openStarted := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
var releaseOnce sync.Once
|
||||
releaseOpen := func() { releaseOnce.Do(func() { close(release) }) }
|
||||
t.Cleanup(releaseOpen)
|
||||
|
||||
h := NewForTest(&blockingOpenLoader{
|
||||
inner: inner,
|
||||
started: openStarted,
|
||||
release: release,
|
||||
})
|
||||
cfg := &config.Config{
|
||||
Plugins: config.PluginsConfig{
|
||||
Enabled: true,
|
||||
Dir: makePluginDir(t, "alpha"),
|
||||
},
|
||||
}
|
||||
return h, cfg, openStarted, releaseOpen
|
||||
}
|
||||
|
||||
func newBlockingRegisterHost(t *testing.T) (*Host, *config.Config, <-chan struct{}, func()) {
|
||||
t.Helper()
|
||||
|
||||
loader := newTestSymbolLoader()
|
||||
registerStarted := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
var startOnce sync.Once
|
||||
var releaseOnce sync.Once
|
||||
releaseRegister := func() { releaseOnce.Do(func() { close(release) }) }
|
||||
t.Cleanup(releaseRegister)
|
||||
|
||||
plugin := &testPlugin{
|
||||
registerResult: validTestPlugin("alpha"),
|
||||
reconfigureResult: validTestPlugin("alpha"),
|
||||
}
|
||||
lookup := newTestSymbolLookup(plugin)
|
||||
lookup.registerOverride = func([]byte) pluginapi.Plugin {
|
||||
startOnce.Do(func() { close(registerStarted) })
|
||||
<-release
|
||||
return validTestPlugin("alpha")
|
||||
}
|
||||
loader.lookups["alpha"] = lookup
|
||||
h := NewForTest(loader)
|
||||
cfg := &config.Config{
|
||||
Plugins: config.PluginsConfig{
|
||||
Enabled: true,
|
||||
Dir: makePluginDir(t, "alpha"),
|
||||
},
|
||||
}
|
||||
return h, cfg, registerStarted, releaseRegister
|
||||
}
|
||||
|
||||
func waitForHostTestSignal(t *testing.T, ch <-chan struct{}, name string) {
|
||||
t.Helper()
|
||||
select {
|
||||
case <-ch:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatalf("timed out waiting for %s", name)
|
||||
}
|
||||
}
|
||||
|
||||
func waitForHostTestBool(t *testing.T, ch <-chan bool, name string) bool {
|
||||
t.Helper()
|
||||
select {
|
||||
case ok := <-ch:
|
||||
return ok
|
||||
case <-time.After(time.Second):
|
||||
t.Fatalf("timed out waiting for %s", name)
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user