add a get to trigger SSE

Author: adilsitos

e2e/proxy/proxy_test.go

@@ -90,7 +90,7 @@ func startProxy(t *testing.T, ctx context.Context, infisicalURL string, config P
 }
 
 // setupProxyTest sets up the common test
-func setupProxyTest(t *testing.T, ctx context.Context, proxyConfig ProxyTestConfig) (*helpers.InfisicalService, *proxyHelpers.ProxyTestHelper, *helpers.Command, string, string) {
+func setupProxyTest(t *testing.T, ctx context.Context, proxyConfig ProxyTestConfig) (*helpers.InfisicalService, *proxyHelpers.ProxyTestHelper, *helpers.Command, string, string, helpers.MachineIdentity) {
 	infisical := helpers.NewInfisicalService().Up(t, ctx)
 
 	// create machine identity with token auth (and universal auth if SSE is enabled)
@@ -145,14 +145,14 @@ func setupProxyTest(t *testing.T, ctx context.Context, proxyConfig ProxyTestConf
 	// create test helper with both proxy and direct API clients
 	helper := proxyHelpers.NewProxyTestHelper(t, proxyURL, infisical.ApiUrl(t), identityToken, projectID)
 
-	return infisical, helper, proxyCmd, identityToken, proxyURL
+	return infisical, helper, proxyCmd, identityToken, proxyURL, identity
 }
 
 func TestProxy_CacheHitMiss(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	_, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
+	_, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
 
 	// create a test secret
 	secret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
@@ -209,7 +209,7 @@ func TestProxy_MutationPurging(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	_, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
+	_, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
 
 	initialSecret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
 		Prefix: "MUTATION_TEST_",
@@ -281,7 +281,7 @@ func TestProxy_DeleteMutationPurging(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	_, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
+	_, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
 
 	// create a test secret
 	secret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
@@ -338,7 +338,7 @@ func TestProxy_TokenInvalidation(t *testing.T) {
 	config := DefaultProxyTestConfig()
 	config.AccessTokenCheckInterval = "2s"
 
-	_, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, config)
+	_, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, config)
 
 	// create and cache a secret
 	secret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
@@ -377,7 +377,7 @@ func TestProxy_HighAvailability(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	infisical, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
+	infisical, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
 
 	// create and cache a secret
 	secret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
@@ -451,7 +451,7 @@ func TestProxy_BackgroundRefresh(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	_, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
+	_, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
 
 	// create a test initialSecret
 	initialSecret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
@@ -520,7 +520,7 @@ func TestProxy_MultipleSecrets(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	_, helper, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
+	_, helper, _, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
 
 	var secrets []proxyHelpers.Secret
 	for i := 0; i < 3; i++ {
@@ -557,7 +557,7 @@ func TestProxy_SingleSecretEndpoint(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	_, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
+	_, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, DefaultProxyTestConfig())
 
 	// create a test secret
 	secret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
@@ -591,7 +591,7 @@ func TestProxy_SSECacheUpdate(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	_, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, SSEProxyTestConfig())
+	_, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, SSEProxyTestConfig())
 
 	// wait for SSE manager initialization
 	result := helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
@@ -687,7 +687,7 @@ func TestProxy_SSEConnectionRecovery(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	infisical, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, SSEProxyTestConfig())
+	infisical, helper, proxyCmd, _, _, _ := setupProxyTest(t, ctx, SSEProxyTestConfig())
 
 	// create and cache a secret to trigger SSE subscription
 	secret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
@@ -777,7 +777,7 @@ func TestProxy_SSEMultipleProjects(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	infisical, helper1, proxyCmd, identityToken, proxyURL := setupProxyTest(t, ctx, SSEProxyTestConfig())
+	infisical, helper1, proxyCmd, identityToken, proxyURL, _ := setupProxyTest(t, ctx, SSEProxyTestConfig())
 
 	// create a second project using the same identity
 	bearerAuth, err := securityprovider.NewSecurityProviderBearerToken(identityToken)
@@ -889,85 +889,76 @@ func TestProxy_SSEPollingFallbackRecovery(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
-	infisical, helper, proxyCmd, _, _ := setupProxyTest(t, ctx, SSEProxyTestConfig())
+	infisical, helper, proxyCmd, _, _, identity := setupProxyTest(t, ctx, SSEProxyTestConfig())
 
-	// create and cache a secret to trigger SSE subscription
-	secret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
-		Prefix: "SSE_POLLING_RECOVERY_",
-	})
-	helper.CreateSecretWithApi(ctx, secret)
-
-	slog.Info("Caching secret via proxy")
-	resp1 := helper.GetSecretsWithProxy(ctx)
-	require.Equal(t, http.StatusOK, resp1.StatusCode())
-	require.NotEmpty(t, resp1.JSON200.Secrets)
-
-	// wait for SSE connection established
 	result := helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
 		EnsureCmdRunning: proxyCmd,
-		ExpectedString:   "SSE connection established",
+		ExpectedString:   "SSE manager initialized",
 		Timeout:          30 * time.Second,
 	})
 	require.Equal(t, helpers.WaitSuccess, result)
 
-	// stop the Infisical backend to break SSE connection
-	slog.Info("Stopping Infisical backend to break SSE connection")
-	backendContainer, err := infisical.Compose().ServiceContainer(ctx, "backend")
-	require.NoError(t, err)
-	err = backendContainer.Stop(ctx, nil)
-	require.NoError(t, err)
+	// create secret via API (not proxy)
+	initialSecret := helper.GenerateSecret(proxyHelpers.GenerateSecretOptions{
+		Prefix: "SSE_POLLING_FALLBACK_RECOVERY_",
+	})
+	helper.CreateSecretWithApi(ctx, initialSecret)
 
-	require.Eventually(t, func() bool {
-		state, err := backendContainer.State(ctx)
-		if err != nil {
-			return false
+	// fetch via proxy -> cache miss, now cached
+	slog.Info("Fetching secret via proxy (expecting cache miss)")
+	resp1 := helper.GetSecretsWithProxy(ctx)
+	require.Equal(t, http.StatusOK, resp1.StatusCode())
+	require.NotNil(t, resp1.JSON200)
+
+	var foundInitial bool
+	for _, s := range resp1.JSON200.Secrets {
+		if s.SecretKey == initialSecret.SecretKey {
+			assert.Equal(t, initialSecret.SecretValue, s.SecretValue)
+			foundInitial = true
+			break
 		}
-		return !state.Running
-	}, 60*time.Second, 200*time.Millisecond, "Backend container should have stopped")
-	slog.Info("Backend stopped")
+	}
+	require.True(t, foundInitial, "Initial secret not found in response")
 
-	// wait for SSE connection loss
+	// wait for SSE connection (demand-driven, triggered by first fetch)
 	result = helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
 		EnsureCmdRunning: proxyCmd,
-		ExpectedString:   "SSE connection lost",
+		ExpectedString:   "SSE connection established",
 		Timeout:          30 * time.Second,
+		Interval:         200 * time.Millisecond,
 	})
-	require.Equal(t, helpers.WaitSuccess, result, "SSE connection loss should be detected")
+	require.Equal(t, helpers.WaitSuccess, result)
+
+	// change identity role to no-access to break SSE connection with auth errors
+	slog.Info("Changing identity role to no-access to break SSE connection")
+	infisical.UpdateMachineIdentityRole(t, ctx, identity.Id, "no-access")
 
-	// wait for SSE retries to exhaust and polling fallback to start (~62s worst case with backoff)
+	// wait for SSE retries to exhaust and polling fallback to start
 	slog.Info("Waiting for SSE retries to exhaust and polling fallback to start")
 	result = helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
 		EnsureCmdRunning: proxyCmd,
 		ExpectedString:   "transitioning to polling fallback",
-		Timeout:          180 * time.Second,
+		Timeout:          120 * time.Second,
+		Interval:         200 * time.Millisecond,
 	})
 	require.Equal(t, helpers.WaitSuccess, result, "Should transition to polling fallback after retries exhausted")
 
 	result = helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
 		EnsureCmdRunning: proxyCmd,
 		ExpectedString:   "Starting polling fallback for project",
 		Timeout:          10 * time.Second,
+		Interval:         200 * time.Millisecond,
 	})
 	require.Equal(t, helpers.WaitSuccess, result, "Polling fallback loop should start")
 
-	// restart the Infisical backend
-	slog.Info("Restarting Infisical backend")
-	err = backendContainer.Start(ctx)
-	require.NoError(t, err)
-
-	require.Eventually(t, func() bool {
-		state, err := backendContainer.State(ctx)
-		if err != nil {
-			return false
-		}
-		return state.Running
-	}, 60*time.Second, 200*time.Millisecond, "Backend container should have restarted")
-	slog.Info("Backend restarted")
+	// restore identity role to member to allow SSE reconnection
+	slog.Info("Restoring identity role to member")
+	infisical.UpdateMachineIdentityRole(t, ctx, identity.Id, "member")
 
 	// trigger EnsureSubscription by making a request — detects polling, attempts SSE reconnection
 	slog.Info("Fetching secrets to trigger SSE reconnection from polling mode")
-	respAfterRestart := helper.GetSecretsWithProxy(ctx)
-	require.Equal(t, http.StatusOK, respAfterRestart.StatusCode())
+	respAfterRestore := helper.GetSecretsWithProxy(ctx)
+	require.Equal(t, http.StatusOK, respAfterRestore.StatusCode())
 
 	// wait for SSE to re-establish (second occurrence after reconnection from polling)
 	slog.Info("Waiting for SSE reconnection from polling mode")
@@ -987,7 +978,7 @@ func TestProxy_SSEPollingFallbackRecovery(t *testing.T) {
 	// verify polling was cancelled after SSE recovery
 	result = helpers.WaitForStderr(t, helpers.WaitForStderrOptions{
 		EnsureCmdRunning: proxyCmd,
-		ExpectedString:   "cancelled polling fallback",
+		ExpectedString:   "Cancelled polling fallback",
 		Timeout:          30 * time.Second,
 	})
 	require.Equal(t, helpers.WaitSuccess, result, "Polling should be cancelled after SSE recovery")
@@ -997,9 +988,4 @@ func TestProxy_SSEPollingFallbackRecovery(t *testing.T) {
 	respFinal := helper.GetSecretsWithProxy(ctx)
 	require.Equal(t, http.StatusOK, respFinal.StatusCode())
 	require.True(t, proxyCmd.IsRunning(), "Proxy should still be running")
-
-	// tear down compose stack for clean state in subsequent tests
-	slog.Info("Tearing down compose stack for clean state")
-	err = infisical.DownWithForce(ctx)
-	require.NoError(t, err, "Failed to tear down compose stack")
 }

e2e/util/helpers.go

@@ -1,7 +1,9 @@
 package util
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -265,6 +267,29 @@ func WithUniversalAuth() MachineIdentityOption {
 	}
 }
 
+func (s *InfisicalService) UpdateMachineIdentityRole(t *testing.T, ctx context.Context, identityId string, role string) {
+	apiUrl := s.ApiUrl(t)
+	url := fmt.Sprintf("%s/api/v1/identities/%s", apiUrl, identityId)
+
+	body, err := json.Marshal(map[string]string{"role": role})
+	require.NoError(t, err)
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodPatch, url, bytes.NewReader(body))
+	require.NoError(t, err)
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Bearer "+s.provisionResult.Token)
+
+	resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
+	require.NoError(t, err)
+	defer resp.Body.Close()
+
+	respBody, _ := io.ReadAll(resp.Body)
+	require.True(t, resp.StatusCode >= 200 && resp.StatusCode < 300,
+		"Failed to update identity role to '%s', status %d: %s", role, resp.StatusCode, string(respBody))
+
+	slog.Info("Updated machine identity role", "identityId", identityId, "role", role)
+}
+
 type RunMethod string
 
 const (