test: add sink functionality for idp intents

cmd/start/start.go

@@ -142,10 +142,6 @@ type Server struct {
 func startZitadel(ctx context.Context, config *Config, masterKey string, server chan<- *Server) error {
 	showBasicInformation(config)
 
-	// sink Server is stubbed out in production builds, see function's godoc.
-	closeSink := sink.StartServer()
-	defer closeSink()
-
 	i18n.MustLoadSupportedLanguagesFromDir()
 
 	queryDBClient, err := database.Connect(config.Database, false, dialect.DBPurposeQuery)
@@ -261,6 +257,10 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server
 	}
 	defer commands.Close(ctx) // wait for background jobs
 
+	// sink Server is stubbed out in production builds, see function's godoc.
+	closeSink := sink.StartServer(commands)
+	defer closeSink()
+
 	clock := clockpkg.New()
 	actionsExecutionStdoutEmitter, err := logstore.NewEmitter[*record.ExecutionLog](ctx, clock, &logstore.EmitterConfig{Enabled: config.LogStore.Execution.Stdout.Enabled}, stdout.NewStdoutEmitter[*record.ExecutionLog]())
 	if err != nil {

internal/api/grpc/session/v2/integration_test/session_test.go

@@ -23,6 +23,7 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/zitadel/zitadel/internal/integration"
+	"github.com/zitadel/zitadel/internal/integration/sink"
 	mgmt "github.com/zitadel/zitadel/pkg/grpc/management"
 	"github.com/zitadel/zitadel/pkg/grpc/object/v2"
 	"github.com/zitadel/zitadel/pkg/grpc/session/v2"
@@ -32,6 +33,7 @@ import (
 var (
 	CTX             context.Context
 	IAMOwnerCTX     context.Context
+	LoginCTX        context.Context
 	Instance        *integration.Instance
 	Client          session.SessionServiceClient
 	User            *user.AddHumanUserResponse
@@ -49,6 +51,7 @@ func TestMain(m *testing.M) {
 
 		CTX = Instance.WithAuthorization(ctx, integration.UserTypeOrgOwner)
 		IAMOwnerCTX = Instance.WithAuthorization(ctx, integration.UserTypeIAMOwner)
+		LoginCTX = Instance.WithAuthorization(ctx, integration.UserTypeLogin)
 		User = createFullUser(CTX)
 		DeactivatedUser = createDeactivatedUser(CTX)
 		LockedUser = createLockedUser(CTX)
@@ -424,10 +427,9 @@ func TestServer_CreateSession_webauthn(t *testing.T) {
 	verifyCurrentSession(t, createResp.GetSessionId(), updateResp.GetSessionToken(), updateResp.GetDetails().GetSequence(), time.Minute, nil, nil, 0, User.GetUserId(), wantUserFactor, wantWebAuthNFactorUserVerified)
 }
 
-/*
 func TestServer_CreateSession_successfulIntent(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
-	createResp, err := Client.CreateSession(CTX, &session.CreateSessionRequest{
+	createResp, err := Client.CreateSession(LoginCTX, &session.CreateSessionRequest{
 		Checks: &session.Checks{
 			User: &session.CheckUser{
 				Search: &session.CheckUser_UserId{
@@ -439,8 +441,9 @@ func TestServer_CreateSession_successfulIntent(t *testing.T) {
 	require.NoError(t, err)
 	verifyCurrentSession(t, createResp.GetSessionId(), createResp.GetSessionToken(), createResp.GetDetails().GetSequence(), time.Minute, nil, nil, 0, User.GetUserId())
 
-	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, User.GetUserId(), "id")
+	intentID, token, _, _, err := sink.SuccessfulOAuthIntent(Instance.ID(), idpID, "id", User.GetUserId())
-	updateResp, err := Client.SetSession(CTX, &session.SetSessionRequest{
+	require.NoError(t, err)
+	updateResp, err := Client.SetSession(LoginCTX, &session.SetSessionRequest{
 		SessionId: createResp.GetSessionId(),
 		Checks: &session.Checks{
 			IdpIntent: &session.CheckIDPIntent{
@@ -454,9 +457,10 @@ func TestServer_CreateSession_successfulIntent(t *testing.T) {
 }
 
 func TestServer_CreateSession_successfulIntent_instant(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
 
-	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, User.GetUserId(), "id")
+	intentID, token, _, _, err := sink.SuccessfulOAuthIntent(Instance.ID(), idpID, "id", User.GetUserId())
+	require.NoError(t, err)
 	createResp, err := Client.CreateSession(CTX, &session.CreateSessionRequest{
 		Checks: &session.Checks{
 			User: &session.CheckUser{
@@ -475,11 +479,11 @@ func TestServer_CreateSession_successfulIntent_instant(t *testing.T) {
 }
 
 func TestServer_CreateSession_successfulIntentUnknownUserID(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
 
 	// successful intent without known / linked user
 	idpUserID := "id"
-	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, "", idpUserID)
+	intentID, token, _, _, err := sink.SuccessfulOAuthIntent(Instance.ID(), idpID, idpUserID, "")
 
 	// link the user (with info from intent)
 	Instance.CreateUserIDPlink(CTX, User.GetUserId(), idpUserID, idpID, User.GetUserId())
@@ -503,7 +507,7 @@ func TestServer_CreateSession_successfulIntentUnknownUserID(t *testing.T) {
 }
 
 func TestServer_CreateSession_startedIntentFalseToken(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
 
 	createResp, err := Client.CreateSession(CTX, &session.CreateSessionRequest{
 		Checks: &session.Checks{
@@ -517,19 +521,18 @@ func TestServer_CreateSession_startedIntentFalseToken(t *testing.T) {
 	require.NoError(t, err)
 	verifyCurrentSession(t, createResp.GetSessionId(), createResp.GetSessionToken(), createResp.GetDetails().GetSequence(), time.Minute, nil, nil, 0, User.GetUserId())
 
-	intentID := Instance.CreateIntent(t, CTX, idpID)
+	intent := Instance.CreateIntent(CTX, idpID)
 	_, err = Client.SetSession(CTX, &session.SetSessionRequest{
 		SessionId: createResp.GetSessionId(),
 		Checks: &session.Checks{
 			IdpIntent: &session.CheckIDPIntent{
-				IdpIntentId:    intentID,
+				IdpIntentId:    intent.GetIdpIntent().GetIdpIntentId(),
 				IdpIntentToken: "false",
 			},
 		},
 	})
 	require.Error(t, err)
 }
-*/
 
 func registerTOTP(ctx context.Context, t *testing.T, userID string) (secret string) {
 	resp, err := Instance.Client.UserV2.RegisterTOTP(ctx, &user.RegisterTOTPRequest{

internal/api/grpc/session/v2beta/integration_test/session_test.go

@@ -424,9 +424,8 @@ func TestServer_CreateSession_webauthn(t *testing.T) {
 	verifyCurrentSession(t, createResp.GetSessionId(), updateResp.GetSessionToken(), updateResp.GetDetails().GetSequence(), time.Minute, nil, nil, 0, User.GetUserId(), wantUserFactor, wantWebAuthNFactorUserVerified)
 }
 
-/*
 func TestServer_CreateSession_successfulIntent(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
 	createResp, err := Client.CreateSession(CTX, &session.CreateSessionRequest{
 		Checks: &session.Checks{
 			User: &session.CheckUser{
@@ -439,7 +438,7 @@ func TestServer_CreateSession_successfulIntent(t *testing.T) {
 	require.NoError(t, err)
 	verifyCurrentSession(t, createResp.GetSessionId(), createResp.GetSessionToken(), createResp.GetDetails().GetSequence(), time.Minute, nil, nil, 0, User.GetUserId())
 
-	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, User.GetUserId(), "id")
+	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, idpID, User.GetUserId(), "id")
 	updateResp, err := Client.SetSession(CTX, &session.SetSessionRequest{
 		SessionId: createResp.GetSessionId(),
 		Checks: &session.Checks{
@@ -454,9 +453,9 @@ func TestServer_CreateSession_successfulIntent(t *testing.T) {
 }
 
 func TestServer_CreateSession_successfulIntent_instant(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
 
-	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, User.GetUserId(), "id")
+	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, idpID, User.GetUserId(), "id")
 	createResp, err := Client.CreateSession(CTX, &session.CreateSessionRequest{
 		Checks: &session.Checks{
 			User: &session.CheckUser{
@@ -475,11 +474,11 @@ func TestServer_CreateSession_successfulIntent_instant(t *testing.T) {
 }
 
 func TestServer_CreateSession_successfulIntentUnknownUserID(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
 
 	// successful intent without known / linked user
 	idpUserID := "id"
-	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, "", idpUserID)
+	intentID, token, _, _ := Instance.CreateSuccessfulOAuthIntent(t, idpID, "", idpUserID)
 
 	// link the user (with info from intent)
 	Instance.CreateUserIDPlink(CTX, User.GetUserId(), idpUserID, idpID, User.GetUserId())
@@ -503,7 +502,7 @@ func TestServer_CreateSession_successfulIntentUnknownUserID(t *testing.T) {
 }
 
 func TestServer_CreateSession_startedIntentFalseToken(t *testing.T) {
-	idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IAMOwnerCTX, gofakeit.AppName()).GetId()
 
 	createResp, err := Client.CreateSession(CTX, &session.CreateSessionRequest{
 		Checks: &session.Checks{
@@ -517,19 +516,18 @@ func TestServer_CreateSession_startedIntentFalseToken(t *testing.T) {
 	require.NoError(t, err)
 	verifyCurrentSession(t, createResp.GetSessionId(), createResp.GetSessionToken(), createResp.GetDetails().GetSequence(), time.Minute, nil, nil, 0, User.GetUserId())
 
-	intentID := Instance.CreateIntent(t, CTX, idpID)
+	intent := Instance.CreateIntent(CTX, idpID)
 	_, err = Client.SetSession(CTX, &session.SetSessionRequest{
 		SessionId: createResp.GetSessionId(),
 		Checks: &session.Checks{
 			IdpIntent: &session.CheckIDPIntent{
-				IdpIntentId:    intentID,
+				IdpIntentId:    intent.GetIdpIntent().GetIdpIntentId(),
 				IdpIntentToken: "false",
 			},
 		},
 	})
 	require.Error(t, err)
 }
-*/
 
 func registerTOTP(ctx context.Context, t *testing.T, userID string) (secret string) {
 	resp, err := Instance.Client.UserV2.RegisterTOTP(ctx, &user.RegisterTOTPRequest{

internal/api/grpc/user/v2/integration_test/user_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/zitadel/logging"
+	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/brianvoe/gofakeit/v6"
 	"github.com/muhlemmer/gu"
@@ -20,6 +21,7 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/zitadel/zitadel/internal/api/grpc"
 	"github.com/zitadel/zitadel/internal/integration"
 	"github.com/zitadel/zitadel/pkg/grpc/auth"
 	"github.com/zitadel/zitadel/pkg/grpc/idp"
@@ -2111,15 +2113,14 @@ func TestServer_StartIdentityProviderIntent(t *testing.T) {
 	}
 }
 
-/*
 func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
-		idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IamCTX, gofakeit.AppName()).GetId()
-	intentID := Instance.CreateIntent(t, CTX, idpID)
+	intentID := Instance.CreateIntent(CTX, idpID).GetIdpIntent().GetIdpIntentId()
-	successfulID, token, changeDate, sequence := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, "", "id")
+	successfulID, token, changeDate, sequence := Instance.CreateSuccessfulOAuthIntent(t, idpID, "", "id")
-	successfulWithUserID, withUsertoken, withUserchangeDate, withUsersequence := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID, "user", "id")
+	successfulWithUserID, withUsertoken, withUserchangeDate, withUsersequence := Instance.CreateSuccessfulOAuthIntent(t, idpID, "user", "id")
-	ldapSuccessfulID, ldapToken, ldapChangeDate, ldapSequence := Instance.CreateSuccessfulLDAPIntent(t, CTX, idpID, "", "id")
+	ldapSuccessfulID, ldapToken, ldapChangeDate, ldapSequence := Instance.CreateSuccessfulLDAPIntent(t, idpID, "", "id")
-	ldapSuccessfulWithUserID, ldapWithUserToken, ldapWithUserChangeDate, ldapWithUserSequence := Instance.CreateSuccessfulLDAPIntent(t, CTX, idpID, "user", "id")
+	ldapSuccessfulWithUserID, ldapWithUserToken, ldapWithUserChangeDate, ldapWithUserSequence := Instance.CreateSuccessfulLDAPIntent(t, idpID, "user", "id")
-	samlSuccessfulID, samlToken, samlChangeDate, samlSequence := Instance.CreateSuccessfulSAMLIntent(t, CTX, idpID, "", "id")
+	samlSuccessfulID, samlToken, samlChangeDate, samlSequence := Instance.CreateSuccessfulSAMLIntent(t, idpID, "", "id")
 	type args struct {
 		ctx context.Context
 		req *user.RetrieveIdentityProviderIntentRequest
@@ -2370,7 +2371,6 @@ func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
 		})
 	}
 }
-*/
 
 func ctxFromNewUserWithRegisteredPasswordlessLegacy(t *testing.T) (context.Context, string, *auth.AddMyPasswordlessResponse) {
 	userID := Instance.CreateHumanUser(CTX).GetUserId()

internal/api/grpc/user/v2beta/integration_test/user_test.go

@@ -16,8 +16,10 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/structpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/zitadel/zitadel/internal/api/grpc"
 	"github.com/zitadel/zitadel/internal/integration"
 	"github.com/zitadel/zitadel/pkg/grpc/idp"
 	mgmt "github.com/zitadel/zitadel/pkg/grpc/management"
@@ -2142,15 +2144,14 @@ func TestServer_StartIdentityProviderIntent(t *testing.T) {
 	}
 }
 
-/*
 func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
-		idpID := Instance.AddGenericOAuthProvider(t, CTX)
+	idpID := Instance.AddGenericOAuthProvider(IamCTX, gofakeit.AppName()).GetId()
-	intentID := Instance.CreateIntent(t, CTX, idpID)
+	intentID := Instance.CreateIntent(CTX, idpID).GetIdpIntent().GetIdpIntentId()
-	successfulID, token, changeDate, sequence := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID.Id, "", "id")
+	successfulID, token, changeDate, sequence := Instance.CreateSuccessfulOAuthIntent(t, idpID, "", "id")
-	successfulWithUserID, withUsertoken, withUserchangeDate, withUsersequence := Instance.CreateSuccessfulOAuthIntent(t, CTX, idpID.Id, "user", "id")
+	successfulWithUserID, withUsertoken, withUserchangeDate, withUsersequence := Instance.CreateSuccessfulOAuthIntent(t, idpID, "user", "id")
-	ldapSuccessfulID, ldapToken, ldapChangeDate, ldapSequence := Instance.CreateSuccessfulLDAPIntent(t, CTX, idpID.Id, "", "id")
+	ldapSuccessfulID, ldapToken, ldapChangeDate, ldapSequence := Instance.CreateSuccessfulLDAPIntent(t, idpID, "", "id")
-	ldapSuccessfulWithUserID, ldapWithUserToken, ldapWithUserChangeDate, ldapWithUserSequence := Instance.CreateSuccessfulLDAPIntent(t, CTX, idpID.Id, "user", "id")
+	ldapSuccessfulWithUserID, ldapWithUserToken, ldapWithUserChangeDate, ldapWithUserSequence := Instance.CreateSuccessfulLDAPIntent(t, idpID, "user", "id")
-	samlSuccessfulID, samlToken, samlChangeDate, samlSequence := Instance.CreateSuccessfulSAMLIntent(t, CTX, idpID.Id, "", "id")
+	samlSuccessfulID, samlToken, samlChangeDate, samlSequence := Instance.CreateSuccessfulSAMLIntent(t, idpID, "", "id")
 	type args struct {
 		ctx context.Context
 		req *user.RetrieveIdentityProviderIntentRequest
@@ -2205,7 +2206,7 @@ func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
 							IdToken:     gu.Ptr("idToken"),
 						},
 					},
-					IdpId:    idpID.Id,
+					IdpId:    idpID,
 					UserId:   "id",
 					UserName: "username",
 					RawInformation: func() *structpb.Struct {
@@ -2243,7 +2244,7 @@ func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
 							IdToken:     gu.Ptr("idToken"),
 						},
 					},
-					IdpId:    idpID.Id,
+					IdpId:    idpID,
 					UserId:   "id",
 					UserName: "username",
 					RawInformation: func() *structpb.Struct {
@@ -2287,7 +2288,7 @@ func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
 							}(),
 						},
 					},
-					IdpId:    idpID.Id,
+					IdpId:    idpID,
 					UserId:   "id",
 					UserName: "username",
 					RawInformation: func() *structpb.Struct {
@@ -2333,7 +2334,7 @@ func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
 							}(),
 						},
 					},
-					IdpId:    idpID.Id,
+					IdpId:    idpID,
 					UserId:   "id",
 					UserName: "username",
 					RawInformation: func() *structpb.Struct {
@@ -2370,7 +2371,7 @@ func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
 							Assertion: []byte("<Assertion xmlns=\"urn:oasis:names:tc:SAML:2.0:assertion\" ID=\"id\" IssueInstant=\"0001-01-01T00:00:00Z\" Version=\"\"><Issuer xmlns=\"urn:oasis:names:tc:SAML:2.0:assertion\" NameQualifier=\"\" SPNameQualifier=\"\" Format=\"\" SPProvidedID=\"\"></Issuer></Assertion>"),
 						},
 					},
-					IdpId:    idpID.Id,
+					IdpId:    idpID,
 					UserId:   "id",
 					UserName: "",
 					RawInformation: func() *structpb.Struct {
@@ -2401,7 +2402,6 @@ func TestServer_RetrieveIdentityProviderIntent(t *testing.T) {
 		})
 	}
 }
-*/
 
 func TestServer_ListAuthenticationMethodTypes(t *testing.T) {
 	userIDWithoutAuth := Instance.CreateHumanUser(CTX).GetUserId()

internal/integration/client.go

@@ -17,6 +17,7 @@ import (
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/zitadel/zitadel/internal/domain"
+	"github.com/zitadel/zitadel/internal/integration/sink"
 	"github.com/zitadel/zitadel/pkg/grpc/admin"
 	"github.com/zitadel/zitadel/pkg/grpc/auth"
 	"github.com/zitadel/zitadel/pkg/grpc/feature/v2"
@@ -523,101 +524,38 @@ func (i *Instance) AddSAMLPostProvider(ctx context.Context) string {
 	return resp.GetId()
 }
 
-/*
+func (i *Instance) CreateIntent(ctx context.Context, idpID string) *user_v2.StartIdentityProviderIntentResponse {
-func (s *Instance) CreateIntent(t *testing.T, ctx context.Context, idpID string) string {
+	resp, err := i.Client.UserV2.StartIdentityProviderIntent(ctx, &user_v2.StartIdentityProviderIntentRequest{
-	resp, err := i.Client.UserV2.StartIdentityProviderIntent(ctx, &user.StartIdentityProviderIntentRequest{
 		IdpId: idpID,
-		Content: &user.StartIdentityProviderIntentRequest_Urls{
+		Content: &user_v2.StartIdentityProviderIntentRequest_Urls{
-			Urls: &user.RedirectURLs{
+			Urls: &user_v2.RedirectURLs{
 				SuccessUrl: "https://example.com/success",
 				FailureUrl: "https://example.com/failure",
 			},
-			AutoLinking:       idp.AutoLinkingOption_AUTO_LINKING_OPTION_USERNAME,
 		},
 	})
 	logging.OnError(err).Fatal("create generic OAuth idp")
 	return resp
 }
 
-func (i *Instance) CreateIntent(t *testing.T, ctx context.Context, idpID string) string {
+func (i *Instance) CreateSuccessfulOAuthIntent(t *testing.T, idpID, userID, idpUserID string) (string, string, time.Time, uint64) {
-	ctx = authz.WithInstance(context.WithoutCancel(ctx), s.Instance)
+	intentID, token, changeDate, sequence, err := sink.SuccessfulOAuthIntent(i.ID(), idpID, idpUserID, userID)
-	writeModel, _, err := s.Commands.CreateIntent(ctx, idpID, "https://example.com/success", "https://example.com/failure", s.Instance.InstanceID())
 	require.NoError(t, err)
-	return writeModel.AggregateID
+	return intentID, token, changeDate, sequence
 }
 
-func (i *Instance) CreateSuccessfulOAuthIntent(t *testing.T, ctx context.Context, idpID, userID, idpUserID string) (string, string, time.Time, uint64) {
+func (i *Instance) CreateSuccessfulLDAPIntent(t *testing.T, idpID, userID, idpUserID string) (string, string, time.Time, uint64) {
-	ctx = authz.WithInstance(context.WithoutCancel(ctx), s.Instance)
+	intentID, token, changeDate, sequence, err := sink.SuccessfulLDAPIntent(i.ID(), idpID, idpUserID, userID)
-	intentID := s.CreateIntent(t, ctx, idpID)
-	writeModel, err := s.Commands.GetIntentWriteModel(ctx, intentID, s.Instance.InstanceID())
 	require.NoError(t, err)
-	idpUser := openid.NewUser(
+	return intentID, token, changeDate, sequence
-		&oidc.UserInfo{
-			Subject: idpUserID,
-			UserInfoProfile: oidc.UserInfoProfile{
-				PreferredUsername: "username",
-			},
-		},
-	)
-	idpSession := &openid.Session{
-		Tokens: &oidc.Tokens[*oidc.IDTokenClaims]{
-			Token: &oauth2.Token{
-				AccessToken: "accessToken",
-			},
-			IDToken: "idToken",
-		},
-	}
-	token, err := s.Commands.SucceedIDPIntent(ctx, writeModel, idpUser, idpSession, userID)
-	require.NoError(t, err)
-	return intentID, token, writeModel.ChangeDate, writeModel.ProcessedSequence
 }
 
-func (s *Instance) CreateSuccessfulLDAPIntent(t *testing.T, ctx context.Context, idpID, userID, idpUserID string) (string, string, time.Time, uint64) {
+func (i *Instance) CreateSuccessfulSAMLIntent(t *testing.T, idpID, userID, idpUserID string) (string, string, time.Time, uint64) {
-	ctx = authz.WithInstance(context.WithoutCancel(ctx), s.Instance)
+	intentID, token, changeDate, sequence, err := sink.SuccessfulSAMLIntent(i.ID(), idpID, idpUserID, userID)
-	intentID := s.CreateIntent(t, ctx, idpID)
-	writeModel, err := s.Commands.GetIntentWriteModel(ctx, intentID, s.Instance.InstanceID())
 	require.NoError(t, err)
-	username := "username"
+	return intentID, token, changeDate, sequence
-	lang := language.Make("en")
-	idpUser := ldap.NewUser(
-		idpUserID,
-		"",
-		"",
-		"",
-		"",
-		username,
-		"",
-		false,
-		"",
-		false,
-		lang,
-		"",
-		"",
-	)
-	attributes := map[string][]string{"id": {idpUserID}, "username": {username}, "language": {lang.String()}}
-	token, err := s.Commands.SucceedLDAPIDPIntent(ctx, writeModel, idpUser, userID, attributes)
-	require.NoError(t, err)
-	return intentID, token, writeModel.ChangeDate, writeModel.ProcessedSequence
 }
 
-func (s *Instance) CreateSuccessfulSAMLIntent(t *testing.T, ctx context.Context, idpID, userID, idpUserID string) (string, string, time.Time, uint64) {
-	ctx = authz.WithInstance(context.WithoutCancel(ctx), s.Instance)
-	intentID := s.CreateIntent(t, ctx, idpID)
-	writeModel, err := s.Server.Commands.GetIntentWriteModel(ctx, intentID, s.Instance.InstanceID())
-	require.NoError(t, err)
-
-	idpUser := &saml.UserMapper{
-		ID:         idpUserID,
-		Attributes: map[string][]string{"attribute1": {"value1"}},
-	}
-	assertion := &crewjam_saml.Assertion{ID: "id"}
-
-	token, err := s.Server.Commands.SucceedSAMLIDPIntent(ctx, writeModel, idpUser, userID, assertion)
-	require.NoError(t, err)
-	return intentID, token, writeModel.ChangeDate, writeModel.ProcessedSequence
-}
-*/
-
 func (i *Instance) CreateVerifiedWebAuthNSession(t *testing.T, ctx context.Context, userID string) (id, token string, start, change time.Time) {
 	return i.CreateVerifiedWebAuthNSessionWithLifetime(t, ctx, userID, 0)
 }

internal/integration/sink/server.go

@@ -3,6 +3,9 @@
 package sink
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
 	"errors"
 	"io"
 	"net/http"
@@ -10,11 +13,23 @@ import (
 	"path"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/gorilla/websocket"
 	"github.com/sirupsen/logrus"
 	"github.com/zitadel/logging"
+	"github.com/zitadel/oidc/v3/pkg/oidc"
+	"golang.org/x/oauth2"
+	"golang.org/x/text/language"
+
+	crewjam_saml "github.com/crewjam/saml"
+
+	"github.com/zitadel/zitadel/internal/api/authz"
+	"github.com/zitadel/zitadel/internal/command"
+	"github.com/zitadel/zitadel/internal/idp/providers/ldap"
+	openid "github.com/zitadel/zitadel/internal/idp/providers/oidc"
+	"github.com/zitadel/zitadel/internal/idp/providers/saml"
 )
 
 const (
@@ -33,6 +48,60 @@ func CallURL(ch Channel) string {
 	return u.String()
 }
 
+func SuccessfulOAuthIntent(instanceID, idpID, idpUserID, userID string) (string, string, time.Time, uint64, error) {
+	u := url.URL{
+		Scheme: "http",
+		Host:   host,
+		Path:   successfulIntentOAuthPath(),
+	}
+	resp, err := callIntent(u.String(), &SuccessfulIntentRequest{
+		InstanceID: instanceID,
+		IDPID:      idpID,
+		IDPUserID:  idpUserID,
+		UserID:     userID,
+	})
+	if err != nil {
+		return "", "", time.Time{}, uint64(0), err
+	}
+	return resp.IntentID, resp.Token, resp.ChangeDate, resp.Sequence, nil
+}
+
+func SuccessfulSAMLIntent(instanceID, idpID, idpUserID, userID string) (string, string, time.Time, uint64, error) {
+	u := url.URL{
+		Scheme: "http",
+		Host:   host,
+		Path:   successfulIntentSAMLPath(),
+	}
+	resp, err := callIntent(u.String(), &SuccessfulIntentRequest{
+		InstanceID: instanceID,
+		IDPID:      idpID,
+		IDPUserID:  idpUserID,
+		UserID:     userID,
+	})
+	if err != nil {
+		return "", "", time.Time{}, uint64(0), err
+	}
+	return resp.IntentID, resp.Token, resp.ChangeDate, resp.Sequence, nil
+}
+
+func SuccessfulLDAPIntent(instanceID, idpID, idpUserID, userID string) (string, string, time.Time, uint64, error) {
+	u := url.URL{
+		Scheme: "http",
+		Host:   host,
+		Path:   successfulIntentLDAPPath(),
+	}
+	resp, err := callIntent(u.String(), &SuccessfulIntentRequest{
+		InstanceID: instanceID,
+		IDPID:      idpID,
+		IDPUserID:  idpUserID,
+		UserID:     userID,
+	})
+	if err != nil {
+		return "", "", time.Time{}, uint64(0), err
+	}
+	return resp.IntentID, resp.Token, resp.ChangeDate, resp.Sequence, nil
+}
+
 // StartServer starts a simple HTTP server on localhost:8081
 // ZITADEL can use the server to send HTTP requests which can be
 // used to validate tests through [Subscribe]rs.
@@ -41,7 +110,7 @@ func CallURL(ch Channel) string {
 // [CallURL] can be used to obtain the full URL for a given Channel.
 //
 // This function is only active when the `integration` build tag is enabled
-func StartServer() (close func()) {
+func StartServer(commands *command.Commands) (close func()) {
 	router := chi.NewRouter()
 	for _, ch := range ChannelValues() {
 		fwd := &forwarder{
@@ -50,6 +119,9 @@ func StartServer() (close func()) {
 		}
 		router.HandleFunc(rootPath(ch), fwd.receiveHandler)
 		router.HandleFunc(subscribePath(ch), fwd.subscriptionHandler)
+		router.HandleFunc(successfulIntentOAuthPath(), successfulIntentHandler(commands, createSuccessfulOAuthIntent))
+		router.HandleFunc(successfulIntentSAMLPath(), successfulIntentHandler(commands, createSuccessfulSAMLIntent))
+		router.HandleFunc(successfulIntentLDAPPath(), successfulIntentHandler(commands, createSuccessfulLDAPIntent))
 	}
 	s := &http.Server{
 		Addr:    listenAddr,
@@ -76,6 +148,26 @@ func subscribePath(c Channel) string {
 	return path.Join("/", c.String(), "subscribe")
 }
 
+func intentPath() string {
+	return path.Join("/", "intent")
+}
+
+func successfulIntentPath() string {
+	return path.Join(intentPath(), "/", "successful")
+}
+
+func successfulIntentOAuthPath() string {
+	return path.Join(successfulIntentPath(), "/", "oauth")
+}
+
+func successfulIntentSAMLPath() string {
+	return path.Join(successfulIntentPath(), "/", "saml")
+}
+
+func successfulIntentLDAPPath() string {
+	return path.Join(successfulIntentPath(), "/", "ldap")
+}
+
 // forwarder handles incoming HTTP requests from ZITADEL and
 // forwards them to all subscribed web sockets.
 type forwarder struct {
@@ -165,3 +257,165 @@ func readLoop(ws *websocket.Conn) (done chan error) {
 
 	return done
 }
+
+type SuccessfulIntentRequest struct {
+	InstanceID string `json:"instance_id"`
+	IDPID      string `json:"idp_id"`
+	IDPUserID  string `json:"idp_user_id"`
+	UserID     string `json:"user_id"`
+}
+type SuccessfulIntentResponse struct {
+	IntentID   string    `json:"intent_id"`
+	Token      string    `json:"token"`
+	ChangeDate time.Time `json:"change_date"`
+	Sequence   uint64    `json:"sequence"`
+}
+
+func callIntent(url string, req *SuccessfulIntentRequest) (*SuccessfulIntentResponse, error) {
+	data, err := json.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.Post(url, "application/json", io.NopCloser(bytes.NewReader(data)))
+	if err != nil {
+		return nil, err
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, errors.New(string(body))
+	}
+	result := new(SuccessfulIntentResponse)
+	if err := json.Unmarshal(body, result); err != nil {
+		return nil, err
+	}
+	return result, nil
+}
+
+func successfulIntentHandler(cmd *command.Commands, createIntent func(ctx context.Context, cmd *command.Commands, req *SuccessfulIntentRequest) (*SuccessfulIntentResponse, error)) func(w http.ResponseWriter, r *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		req := &SuccessfulIntentRequest{}
+		if err := json.Unmarshal(body, req); err != nil {
+		}
+
+		ctx := authz.WithInstanceID(r.Context(), req.InstanceID)
+		resp, err := createIntent(ctx, cmd, req)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		data, err := json.Marshal(resp)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		w.Write(data)
+		return
+	}
+}
+
+func createIntent(ctx context.Context, cmd *command.Commands, instanceID, idpID string) (string, error) {
+	writeModel, _, err := cmd.CreateIntent(ctx, idpID, "https://example.com/success", "https://example.com/failure", instanceID)
+	if err != nil {
+		return "", err
+	}
+	return writeModel.AggregateID, nil
+}
+
+func createSuccessfulOAuthIntent(ctx context.Context, cmd *command.Commands, req *SuccessfulIntentRequest) (*SuccessfulIntentResponse, error) {
+	intentID, err := createIntent(ctx, cmd, req.InstanceID, req.IDPID)
+	writeModel, err := cmd.GetIntentWriteModel(ctx, intentID, req.InstanceID)
+	idpUser := openid.NewUser(
+		&oidc.UserInfo{
+			Subject: req.IDPUserID,
+			UserInfoProfile: oidc.UserInfoProfile{
+				PreferredUsername: "username",
+			},
+		},
+	)
+	idpSession := &openid.Session{
+		Tokens: &oidc.Tokens[*oidc.IDTokenClaims]{
+			Token: &oauth2.Token{
+				AccessToken: "accessToken",
+			},
+			IDToken: "idToken",
+		},
+	}
+	token, err := cmd.SucceedIDPIntent(ctx, writeModel, idpUser, idpSession, req.UserID)
+	if err != nil {
+		return nil, err
+	}
+	return &SuccessfulIntentResponse{
+		intentID,
+		token,
+		writeModel.ChangeDate,
+		writeModel.ProcessedSequence,
+	}, nil
+}
+
+func createSuccessfulSAMLIntent(ctx context.Context, cmd *command.Commands, req *SuccessfulIntentRequest) (*SuccessfulIntentResponse, error) {
+	intentID, err := createIntent(ctx, cmd, req.InstanceID, req.IDPID)
+	writeModel, err := cmd.GetIntentWriteModel(ctx, intentID, req.InstanceID)
+
+	idpUser := &saml.UserMapper{
+		ID:         req.IDPUserID,
+		Attributes: map[string][]string{"attribute1": {"value1"}},
+	}
+	assertion := &crewjam_saml.Assertion{ID: "id"}
+
+	token, err := cmd.SucceedSAMLIDPIntent(ctx, writeModel, idpUser, req.UserID, assertion)
+	if err != nil {
+		return nil, err
+	}
+	return &SuccessfulIntentResponse{
+		intentID,
+		token,
+		writeModel.ChangeDate,
+		writeModel.ProcessedSequence,
+	}, nil
+}
+
+func createSuccessfulLDAPIntent(ctx context.Context, cmd *command.Commands, req *SuccessfulIntentRequest) (*SuccessfulIntentResponse, error) {
+	intentID, err := createIntent(ctx, cmd, req.InstanceID, req.IDPID)
+	writeModel, err := cmd.GetIntentWriteModel(ctx, intentID, req.InstanceID)
+	username := "username"
+	lang := language.Make("en")
+	idpUser := ldap.NewUser(
+		req.IDPUserID,
+		"",
+		"",
+		"",
+		"",
+		username,
+		"",
+		false,
+		"",
+		false,
+		lang,
+		"",
+		"",
+	)
+	attributes := map[string][]string{"id": {req.IDPUserID}, "username": {username}, "language": {lang.String()}}
+	token, err := cmd.SucceedLDAPIDPIntent(ctx, writeModel, idpUser, req.UserID, attributes)
+	if err != nil {
+		return nil, err
+	}
+	return &SuccessfulIntentResponse{
+		intentID,
+		token,
+		writeModel.ChangeDate,
+		writeModel.ProcessedSequence,
+	}, nil
+}