Merge branch '404560-gq-apis-update' into 'master'

Draft: Adding update instance headers API

See merge request !122846



Merged-by: Hitesh Raghuvanshi <hraghuvanshi@gitlab.com>

app/assets/javascripts/graphql_shared/possible_types.json

@@ -3,6 +3,10 @@
     "AlertManagementHttpIntegration",
     "AlertManagementPrometheusIntegration"
   ],
+  "BaseHeaderInterface": [
+    "AuditEventStreamingHeader",
+    "AuditEventsStreamingInstanceHeader"
+  ],
   "CiVariable": [
     "CiGroupVariable",
     "CiInstanceVariable",

doc/api/graphql/reference/index.md

@@ -1256,6 +1256,48 @@ Input type: `AuditEventsStreamingHeadersUpdateInput`
 | <a id="mutationauditeventsstreamingheadersupdateerrors"></a>`errors` | [`[String!]!`](#string) | Errors encountered during execution of the mutation. |
 | <a id="mutationauditeventsstreamingheadersupdateheader"></a>`header` | [`AuditEventStreamingHeader`](#auditeventstreamingheader) | Updates header. |
+### `Mutation.auditEventsStreamingInstanceHeadersCreate`
+
+Input type: `AuditEventsStreamingInstanceHeadersCreateInput`
+
+#### Arguments
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="mutationauditeventsstreaminginstanceheaderscreateclientmutationid"></a>`clientMutationId` | [`String`](#string) | A unique identifier for the client performing the mutation. |
+| <a id="mutationauditeventsstreaminginstanceheaderscreatedestinationid"></a>`destinationId` | [`AuditEventsInstanceExternalAuditEventDestinationID!`](#auditeventsinstanceexternalauditeventdestinationid) | Instance level external destination to associate header with. |
+| <a id="mutationauditeventsstreaminginstanceheaderscreatekey"></a>`key` | [`String!`](#string) | Header key. |
+| <a id="mutationauditeventsstreaminginstanceheaderscreatevalue"></a>`value` | [`String!`](#string) | Header value. |
+
+#### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="mutationauditeventsstreaminginstanceheaderscreateclientmutationid"></a>`clientMutationId` | [`String`](#string) | A unique identifier for the client performing the mutation. |
+| <a id="mutationauditeventsstreaminginstanceheaderscreateerrors"></a>`errors` | [`[String!]!`](#string) | Errors encountered during execution of the mutation. |
+| <a id="mutationauditeventsstreaminginstanceheaderscreateheader"></a>`header` | [`AuditEventsStreamingInstanceHeader`](#auditeventsstreaminginstanceheader) | Created header. |
+
+### `Mutation.auditEventsStreamingInstanceHeadersUpdate`
+
+Input type: `AuditEventsStreamingInstanceHeadersUpdateInput`
+
+#### Arguments
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="mutationauditeventsstreaminginstanceheadersupdateclientmutationid"></a>`clientMutationId` | [`String`](#string) | A unique identifier for the client performing the mutation. |
+| <a id="mutationauditeventsstreaminginstanceheadersupdateheaderid"></a>`headerId` | [`AuditEventsStreamingInstanceHeaderID!`](#auditeventsstreaminginstanceheaderid) | Header to update. |
+| <a id="mutationauditeventsstreaminginstanceheadersupdatekey"></a>`key` | [`String!`](#string) | Header key. |
+| <a id="mutationauditeventsstreaminginstanceheadersupdatevalue"></a>`value` | [`String!`](#string) | Header value. |
+
+#### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="mutationauditeventsstreaminginstanceheadersupdateclientmutationid"></a>`clientMutationId` | [`String`](#string) | A unique identifier for the client performing the mutation. |
+| <a id="mutationauditeventsstreaminginstanceheadersupdateerrors"></a>`errors` | [`[String!]!`](#string) | Errors encountered during execution of the mutation. |
+| <a id="mutationauditeventsstreaminginstanceheadersupdateheader"></a>`header` | [`AuditEventsStreamingInstanceHeader`](#auditeventsstreaminginstanceheader) | Updates header. |
+
 ### `Mutation.awardEmojiAdd`
 Input type: `AwardEmojiAddInput`
@@ -7467,6 +7509,29 @@ The edge type for [`AuditEventStreamingHeader`](#auditeventstreamingheader).
 | <a id="auditeventstreamingheaderedgecursor"></a>`cursor` | [`String!`](#string) | A cursor for use in pagination. |
 | <a id="auditeventstreamingheaderedgenode"></a>`node` | [`AuditEventStreamingHeader`](#auditeventstreamingheader) | The item at the end of the edge. |
+#### `AuditEventsStreamingInstanceHeaderConnection`
+
+The connection type for [`AuditEventsStreamingInstanceHeader`](#auditeventsstreaminginstanceheader).
+
+##### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="auditeventsstreaminginstanceheaderconnectionedges"></a>`edges` | [`[AuditEventsStreamingInstanceHeaderEdge]`](#auditeventsstreaminginstanceheaderedge) | A list of edges. |
+| <a id="auditeventsstreaminginstanceheaderconnectionnodes"></a>`nodes` | [`[AuditEventsStreamingInstanceHeader]`](#auditeventsstreaminginstanceheader) | A list of nodes. |
+| <a id="auditeventsstreaminginstanceheaderconnectionpageinfo"></a>`pageInfo` | [`PageInfo!`](#pageinfo) | Information to aid in pagination. |
+
+#### `AuditEventsStreamingInstanceHeaderEdge`
+
+The edge type for [`AuditEventsStreamingInstanceHeader`](#auditeventsstreaminginstanceheader).
+
+##### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="auditeventsstreaminginstanceheaderedgecursor"></a>`cursor` | [`String!`](#string) | A cursor for use in pagination. |
+| <a id="auditeventsstreaminginstanceheaderedgenode"></a>`node` | [`AuditEventsStreamingInstanceHeader`](#auditeventsstreaminginstanceheader) | The item at the end of the edge. |
+
 #### `AwardEmojiConnection`
 The connection type for [`AwardEmoji`](#awardemoji).
@@ -12135,6 +12200,18 @@ Represents a HTTP header key/value that belongs to an audit streaming destinatio
 | <a id="auditeventstreamingheaderkey"></a>`key` | [`String!`](#string) | Key of the header. |
 | <a id="auditeventstreamingheadervalue"></a>`value` | [`String!`](#string) | Value of the header. |
+### `AuditEventsStreamingInstanceHeader`
+
+Represents a HTTP header key/value that belongs to an audit streaming destination.
+
+#### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="auditeventsstreaminginstanceheaderid"></a>`id` | [`ID!`](#id) | ID of the header. |
+| <a id="auditeventsstreaminginstanceheaderkey"></a>`key` | [`String!`](#string) | Key of the header. |
+| <a id="auditeventsstreaminginstanceheadervalue"></a>`value` | [`String!`](#string) | Value of the header. |
+
 ### `AwardEmoji`
 An emoji awarded by a user.
@@ -16744,6 +16821,7 @@ Represents an external resource to send instance audit events to.
 | Name | Type | Description |
 | ---- | ---- | ----------- |
 | <a id="instanceexternalauditeventdestinationdestinationurl"></a>`destinationUrl` | [`String!`](#string) | External destination to send audit events to. |
+| <a id="instanceexternalauditeventdestinationheaders"></a>`headers` | [`AuditEventsStreamingInstanceHeaderConnection!`](#auditeventsstreaminginstanceheaderconnection) | List of additional HTTP headers sent with each event. (see [Connections](#connections)) |
 | <a id="instanceexternalauditeventdestinationid"></a>`id` | [`ID!`](#id) | ID of the destination. |
 | <a id="instanceexternalauditeventdestinationverificationtoken"></a>`verificationToken` | [`String!`](#string) | Verification token to validate source of event. |
@@ -26668,6 +26746,12 @@ A `AuditEventsStreamingHeaderID` is a global ID. It is encoded as a string.
 An example `AuditEventsStreamingHeaderID` is: `"gid://gitlab/AuditEvents::Streaming::Header/1"`.
+### `AuditEventsStreamingInstanceHeaderID`
+
+A `AuditEventsStreamingInstanceHeaderID` is a global ID. It is encoded as a string.
+
+An example `AuditEventsStreamingInstanceHeaderID` is: `"gid://gitlab/AuditEvents::Streaming::InstanceHeader/1"`.
+
 ### `AwardableID`
 A `AwardableID` is a global ID. It is encoded as a string.
@@ -27454,6 +27538,21 @@ Implementations:
 | <a id="alertmanagementintegrationtype"></a>`type` | [`AlertManagementIntegrationType!`](#alertmanagementintegrationtype) | Type of integration. |
 | <a id="alertmanagementintegrationurl"></a>`url` | [`String`](#string) | Endpoint which accepts alert notifications. |
+#### `BaseHeaderInterface`
+
+Implementations:
+
+- [`AuditEventStreamingHeader`](#auditeventstreamingheader)
+- [`AuditEventsStreamingInstanceHeader`](#auditeventsstreaminginstanceheader)
+
+##### Fields
+
+| Name | Type | Description |
+| ---- | ---- | ----------- |
+| <a id="baseheaderinterfaceid"></a>`id` | [`ID!`](#id) | ID of the header. |
+| <a id="baseheaderinterfacekey"></a>`key` | [`String!`](#string) | Key of the header. |
+| <a id="baseheaderinterfacevalue"></a>`value` | [`String!`](#string) | Value of the header. |
+
 #### `CiVariable`
 Implementations:

ee/app/graphql/ee/types/mutation_type.rb

@@ -119,6 +119,8 @@ module MutationType
         mount_mutation ::Mutations::AuditEvents::GoogleCloudLoggingConfigurations::Destroy
         mount_mutation ::Mutations::AuditEvents::GoogleCloudLoggingConfigurations::Update
         mount_mutation ::Mutations::Forecasting::BuildForecast, alpha: { milestone: '16.0' }
+        mount_mutation ::Mutations::AuditEvents::Streaming::InstanceHeaders::Create
+        mount_mutation ::Mutations::AuditEvents::Streaming::InstanceHeaders::Update
 
         prepend(Types::DeprecatedMutations)
       end

ee/app/graphql/mutations/audit_events/streaming/instance_headers/base.rb

+# frozen_string_literal: true
+
+module Mutations
+  module AuditEvents
+    module Streaming
+      module InstanceHeaders
+        class Base < BaseMutation
+          ERROR_MESSAGE = 'You do not have access to this mutation.'
+
+          authorize :admin_instance_external_audit_events
+
+          def ready?(**args)
+            unless current_user&.can?(:admin_instance_external_audit_events)
+              raise_resource_not_available_error! ERROR_MESSAGE
+            end
+
+            super
+          end
+
+          private
+
+          def find_destination(destination_id)
+            GitlabSchema.object_from_id(
+              destination_id, expected_type: ::AuditEvents::InstanceExternalAuditEventDestination
+            ).sync
+          end
+
+          def find_header(header_id)
+            GitlabSchema.object_from_id(
+              header_id, expected_type: ::AuditEvents::Streaming::InstanceHeader
+            ).sync
+          end
+        end
+      end
+    end
+  end
+end

ee/app/graphql/mutations/audit_events/streaming/instance_headers/create.rb

+# frozen_string_literal: true
+
+module Mutations
+  module AuditEvents
+    module Streaming
+      module InstanceHeaders
+        class Create < Base
+          graphql_name 'AuditEventsStreamingInstanceHeadersCreate'
+
+          argument :key, GraphQL::Types::String,
+            required: true,
+            description: 'Header key.'
+
+          argument :value, GraphQL::Types::String,
+            required: true,
+            description: 'Header value.'
+
+          argument :destination_id, ::Types::GlobalIDType[::AuditEvents::InstanceExternalAuditEventDestination],
+            required: true,
+            description: 'Instance level external destination to associate header with.'
+
+          field :header, ::Types::AuditEvents::Streaming::InstanceHeaderType,
+            null: true,
+            description: 'Created header.'
+
+          def resolve(destination_id:, key:, value:)
+            response = ::AuditEvents::Streaming::InstanceHeaders::CreateService.new(
+              params: { key: key, value: value, destination: find_destination(destination_id) }
+            ).execute
+
+            if response.success?
+              { header: response.payload[:header], errors: [] }
+            else
+              { header: nil, errors: response.errors }
+            end
+          end
+        end
+      end
+    end
+  end
+end

ee/app/graphql/mutations/audit_events/streaming/instance_headers/update.rb

+# frozen_string_literal: true
+
+module Mutations
+  module AuditEvents
+    module Streaming
+      module InstanceHeaders
+        class Update < Base
+          graphql_name 'AuditEventsStreamingInstanceHeadersUpdate'
+
+          argument :header_id, ::Types::GlobalIDType[::AuditEvents::Streaming::InstanceHeader],
+            required: true,
+            description: 'Header to update.'
+
+          argument :key, GraphQL::Types::String,
+            required: true,
+            description: 'Header key.'
+
+          argument :value, GraphQL::Types::String,
+            required: true,
+            description: 'Header value.'
+
+          field :header, ::Types::AuditEvents::Streaming::InstanceHeaderType,
+            null: true,
+            description: 'Updates header.'
+
+          def resolve(header_id:, key:, value:)
+            header = find_header(header_id)
+
+            response = ::AuditEvents::Streaming::InstanceHeaders::UpdateService.new(
+              params: { header: header, key: key, value: value }
+            ).execute
+
+            if response.success?
+              { header: response.payload[:header], errors: [] }
+            elsif header.present?
+              { header: header.reset, errors: response.errors }
+            else
+              { errors: response.errors }
+            end
+          end
+        end
+      end
+    end
+  end
+end

ee/app/graphql/types/audit_events/instance_external_audit_event_destination_type.rb

@@ -9,6 +9,10 @@ class InstanceExternalAuditEventDestinationType < ::Types::BaseObject
       authorize :admin_instance_external_audit_events
 
       implements(ExternalAuditEventDestinationInterface)
+
+      field :headers, ::Types::AuditEvents::Streaming::InstanceHeaderType.connection_type,
+        null: false,
+        description: 'List of additional HTTP headers sent with each event.'
     end
   end
 end

ee/app/graphql/types/audit_events/streaming/base_header_interface.rb

+# frozen_string_literal: true
+
+module Types
+  module AuditEvents
+    module Streaming
+      module BaseHeaderInterface
+        include Types::BaseInterface
+
+        field :id, GraphQL::Types::ID,
+          null: false,
+          description: 'ID of the header.'
+
+        field :key, GraphQL::Types::String,
+          null: false,
+          description: 'Key of the header.'
+
+        field :value, GraphQL::Types::String,
+          null: false,
+          description: 'Value of the header.'
+      end
+    end
+  end
+end

ee/app/graphql/types/audit_events/streaming/header_type.rb

@@ -7,21 +7,12 @@ module AuditEvents
     module Streaming
       class HeaderType < ::Types::BaseObject
         graphql_name 'AuditEventStreamingHeader'
+
         description 'Represents a HTTP header key/value that belongs to an audit streaming destination.'
 
         authorize :admin_external_audit_events
 
-        field :id, GraphQL::Types::ID,
-              null: false,
-              description: 'ID of the header.'
-
-        field :key, GraphQL::Types::String,
-              null: false,
-              description: 'Key of the header.'
-
-        field :value, GraphQL::Types::String,
-              null: false,
-              description: 'Value of the header.'
+        implements(BaseHeaderInterface)
       end
     end
   end

ee/app/graphql/types/audit_events/streaming/instance_header_type.rb

+# frozen_string_literal: true
+
+module Types
+  module AuditEvents
+    module Streaming
+      class InstanceHeaderType < ::Types::BaseObject
+        graphql_name 'AuditEventsStreamingInstanceHeader'
+
+        description 'Represents a HTTP header key/value that belongs to an audit streaming destination.'
+
+        authorize :admin_instance_external_audit_events
+
+        implements(BaseHeaderInterface)
+      end
+    end
+  end
+end

ee/app/policies/audit_events/streaming/instance_header_policy.rb

+# frozen_string_literal: true
+
+module AuditEvents
+  module Streaming
+    class InstanceHeaderPolicy < ::BasePolicy
+      delegate { @subject.instance_external_audit_event_destination }
+    end
+  end
+end

ee/app/services/audit_events/streaming/instance_headers/base_service.rb

+# frozen_string_literal: true
+
+module AuditEvents
+  module Streaming
+    module InstanceHeaders
+      class BaseService
+        attr_reader :params
+
+        DESTINATION_ERROR_MESSAGE = "Please provide a valid destinationId."
+        HEADER_ERROR_MESSAGE = "Please provide a valid headerId."
+
+        def initialize(params: {})
+          @params = params
+        end
+
+        private
+
+        def destination_error
+          ServiceResponse.error(message: DESTINATION_ERROR_MESSAGE)
+        end
+
+        def header_error
+          ServiceResponse.error(message: HEADER_ERROR_MESSAGE)
+        end
+      end
+    end
+  end
+end

ee/app/services/audit_events/streaming/instance_headers/create_service.rb

+# frozen_string_literal: true
+
+module AuditEvents
+  module Streaming
+    module InstanceHeaders
+      class CreateService < BaseService
+        def execute
+          destination = params[:destination]
+          return destination_error if destination.blank?
+
+          header = destination.headers.new(key: params[:key], value: params[:value])
+
+          if header.save
+            ServiceResponse.success(payload: { header: header, errors: [] })
+          else
+            ServiceResponse.error(message: Array(header.errors))
+          end
+        end
+      end
+    end
+  end
+end

ee/app/services/audit_events/streaming/instance_headers/update_service.rb

+# frozen_string_literal: true
+
+module AuditEvents
+  module Streaming
+    module InstanceHeaders
+      class UpdateService < BaseService
+        def execute
+          header = params[:header]
+          return header_error if header.blank?
+
+          if header.update(key: params[:key], value: params[:value])
+            ServiceResponse.success(payload: { header: header, errors: [] })
+          else
+            ServiceResponse.error(message: Array(header.errors))
+          end
+        end
+      end
+    end
+  end
+end

ee/spec/requests/api/graphql/audit_events/streaming/instance_headers/create_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe 'Create an instance external audit event destination header', feature_category: :audit_events do
+  include GraphqlHelpers
+
+  let_it_be(:destination) { create(:instance_external_audit_event_destination) }
+  let_it_be(:admin) { create(:admin) }
+  let_it_be(:user) { create(:user) }
+
+  let_it_be(:current_user) { admin }
+
+  let(:mutation) { graphql_mutation(:audit_events_streaming_instance_headers_create, input) }
+  let(:mutation_response) { graphql_mutation_response(:audit_events_streaming_instance_headers_create) }
+
+  let(:input) do
+    {
+      destinationId: destination.to_gid,
+      key: 'foo',
+      value: 'bar'
+    }
+  end
+
+  subject { post_graphql_mutation(mutation, current_user: current_user) }
+
+  shared_examples 'a mutation that does not create a header' do
+    it 'does not create a header' do
+      expect { post_graphql_mutation(mutation, current_user: current_user) }
+        .not_to change { destination.headers.count }
+    end
+  end
+
+  context 'when feature is licensed' do
+    before do
+      stub_licensed_features(external_audit_events: true)
+    end
+
+    context 'when feature flag `ff_external_audit_events` is enabled' do
+      context 'when current user is instance admin' do
+        it 'creates the header with the correct attributes', :aggregate_failures do
+          expect { subject }
+            .to change { destination.headers.count }.by(1)
+
+          header = AuditEvents::Streaming::InstanceHeader.last
+
+          expect(header.key).to eq('foo')
+          expect(header.value).to eq('bar')
+          expect(mutation_response['errors']).to be_empty
+        end
+
+        context 'when the header attributes are invalid' do
+          let_it_be(:invalid_headers_input) do
+            {
+              destinationId: destination.to_gid,
+              key: '',
+              value: 'bar'
+            }
+          end
+
+          let(:mutation) { graphql_mutation(:audit_events_streaming_instance_headers_create, invalid_headers_input) }
+
+          it 'returns correct errors' do
+            subject
+
+            expect(mutation_response['header']).to be_nil
+            expect(mutation_response['errors']).to contain_exactly("Key can't be blank")
+          end
+
+          it_behaves_like 'a mutation that does not create a header' do
+            let_it_be(:current_user) { admin }
+          end
+        end
+
+        context 'when the destination id is wrong' do
+          let_it_be(:invalid_destination_input) do
+            {
+              destinationId: "gid://gitlab/AuditEvents::InstanceExternalAuditEventDestination/14566",
+              key: 'foo',
+              value: 'bar'
+            }
+          end
+
+          let(:mutation) do
+            graphql_mutation(:audit_events_streaming_instance_headers_create, invalid_destination_input)
+          end
+
+          it 'returns correct errors' do
+            subject
+
+            expect(mutation_response['header']).to be_nil
+            expect(mutation_response['errors']).to contain_exactly(
+              AuditEvents::Streaming::InstanceHeaders::BaseService::DESTINATION_ERROR_MESSAGE)
+          end
+
+          it 'does not create any header' do
+            expect { post_graphql_mutation(mutation, current_user: current_user) }
+              .not_to change { AuditEvents::Streaming::InstanceHeader.count }
+          end
+        end
+      end
+
+      context 'when current user is not instance admin' do
+        it_behaves_like 'a mutation that does not create a header' do
+          let_it_be(:current_user) { user }
+        end
+      end
+    end
+
+    context 'when feature flag `ff_external_audit_events` is disabled' do
+      before do
+        stub_feature_flags(ff_external_audit_events: false)
+      end
+
+      context 'when current user is instance admin' do
+        it_behaves_like 'a mutation that does not create a header' do
+          let_it_be(:current_user) { admin }
+        end
+      end
+
+      context 'when current user is not instance admin' do
+        it_behaves_like 'a mutation that does not create a header' do
+          let_it_be(:current_user) { user }
+        end
+      end
+    end
+  end
+
+  context 'when feature is unlicensed' do
+    before do
+      stub_licensed_features(external_audit_events: false)
+    end
+
+    it_behaves_like 'a mutation that returns top-level errors',
+      errors: [Mutations::AuditEvents::Streaming::InstanceHeaders::Base::ERROR_MESSAGE]
+
+    it_behaves_like 'a mutation that does not create a header'
+  end
+end

ee/spec/requests/api/graphql/audit_events/streaming/instance_headers/update_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe 'Update an external audit event destination header', feature_category: :audit_events do
+  include GraphqlHelpers
+
+  let_it_be(:destination) { create(:instance_external_audit_event_destination) }
+  let_it_be(:header) do
+    create(:instance_audit_events_streaming_header,
+      key: 'key-1',
+      instance_external_audit_event_destination: destination
+    )
+  end
+
+  let_it_be(:admin) { create(:admin) }
+  let_it_be(:user) { create(:user) }
+
+  let(:current_user) { admin }
+  let(:mutation) { graphql_mutation(:audit_events_streaming_instance_headers_update, input) }
+  let(:mutation_response) { graphql_mutation_response(:audit_events_streaming_instance_headers_update) }
+
+  let(:input) do
+    {
+      headerId: header.to_gid,
+      key: 'new-key',
+      value: 'new-value'
+    }
+  end
+
+  subject { post_graphql_mutation(mutation, current_user: current_user) }
+
+  shared_examples 'a mutation that does not update a header' do
+    it 'does not update a header key' do
+      expect { post_graphql_mutation(mutation, current_user: actioner) }.not_to change { header.key }
+    end
+
+    it 'does not update a header value' do
+      expect { post_graphql_mutation(mutation, current_user: actioner) }.not_to change { header.value }
+    end
+  end
+
+  context 'when feature is licensed' do
+    before do
+      stub_licensed_features(external_audit_events: true)
+    end
+
+    context 'when feature flag `ff_external_audit_events` is enabled' do
+      context 'when current user is instance admin' do
+        it 'updates the header with the correct attributes', :aggregate_failures do
+          expect { subject }.to change { header.reload.key }.from('key-1').to('new-key')
+                                                            .and change { header.reload.value }.from('bar')
+                                                                                               .to('new-value')
+        end
+
+        context 'when the header attributes are invalid' do
+          let(:invalid_key_input) do
+            {
+              headerId: header.to_gid,
+              key: '',
+              value: 'bar'
+            }
+          end
+
+          let(:mutation) { graphql_mutation(:audit_events_streaming_instance_headers_update, invalid_key_input) }
+
+          it 'returns correct errors' do
+            subject
+
+            expect(mutation_response['errors']).to contain_exactly("Key can't be blank")
+          end
+
+          it 'returns the unmutated attribute values', :aggregate_failures do
+            subject
+
+            expect(mutation_response.dig('header', 'key')).to eq('key-1')
+            expect(mutation_response.dig('header', 'value')).to eq('bar')
+          end
+
+          it_behaves_like 'a mutation that does not update a header' do
+            let_it_be(:actioner) { admin }
+          end
+        end
+
+        context 'when the header id is wrong' do
+          let_it_be(:invalid_header_input) do
+            {
+              headerId: "gid://gitlab/AuditEvents::Streaming::InstanceHeader/-1",
+              key: 'foo',
+              value: 'bar'
+            }
+          end
+
+          let(:mutation) { graphql_mutation(:audit_events_streaming_instance_headers_update, invalid_header_input) }
+
+          it 'returns correct errors' do
+            subject
+
+            expect(mutation_response['errors'])
+              .to contain_exactly(AuditEvents::Streaming::InstanceHeaders::BaseService::HEADER_ERROR_MESSAGE)
+          end
+
+          it_behaves_like 'a mutation that does not update a header' do
+            let_it_be(:actioner) { admin }
+          end
+        end
+      end
+
+      context 'when current user is not instance admin' do
+        it_behaves_like 'a mutation that does not update a header' do
+          let_it_be(:actioner) { user }
+        end
+      end
+    end
+
+    context 'when feature flag `ff_external_audit_events` is disabled' do
+      before do
+        stub_feature_flags(ff_external_audit_events: false)
+      end
+
+      context 'when current user is instance admin' do
+        it_behaves_like 'a mutation that does not update a header' do
+          let_it_be(:actioner) { admin }
+        end
+      end
+
+      context 'when current user is not instance admin' do
+        it_behaves_like 'a mutation that does not update a header' do
+          let_it_be(:actioner) { user }
+        end
+      end
+    end
+  end
+
+  context 'when feature is unlicensed' do
+    before do
+      stub_licensed_features(external_audit_events: false)
+    end
+
+    it_behaves_like 'a mutation that returns top-level errors',
+      errors: [Mutations::AuditEvents::Streaming::InstanceHeaders::Base::ERROR_MESSAGE]
+
+    it_behaves_like 'a mutation that does not update a header' do
+      let_it_be(:actioner) { admin }
+    end
+  end
+end

ee/spec/services/audit_events/streaming/instance_headers/create_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe AuditEvents::Streaming::InstanceHeaders::CreateService, feature_category: :audit_events do
+  let_it_be(:destination) { create(:instance_external_audit_event_destination) }
+  let_it_be(:event_type) { "audit_events_streaming_instance_headers_create" }
+
+  let(:params) { { destination: destination } }
+
+  subject(:service) do
+    described_class.new(
+      params: params
+    )
+  end
+
+  describe '#execute' do
+    subject(:response) { service.execute }
+
+    context 'when destination is nil' do
+      let(:params) { { destination: nil, key: 'a_key', value: 'a_value' } }
+
+      it 'has error in the response' do
+        expect(response).to be_error
+        expect(response.message).to eql(AuditEvents::Streaming::InstanceHeaders::BaseService::DESTINATION_ERROR_MESSAGE)
+      end
+    end
+
+    context 'when there are validation issues' do
+      let(:expected_errors) { ["Key can't be blank", "Value can't be blank"] }
+
+      it 'has an array of errors in the response' do
+        expect(response).to be_error
+        expect(response.errors).to match_array expected_errors
+      end
+    end
+
+    context 'when the header is created successfully' do
+      let(:params) { super().merge(key: 'a_key', value: 'a_value') }
+
+      it 'has the header in the response payload' do
+        expect(response).to be_success
+        expect(response.payload[:header].key).to eq 'a_key'
+        expect(response.payload[:header].value).to eq 'a_value'
+      end
+
+      it 'creates header for destination' do
+        expect { response }
+          .to change { AuditEvents::Streaming::InstanceHeader.count }.by(1)
+
+        header = AuditEvents::Streaming::InstanceHeader.last
+
+        expect(header.key).to eq('a_key')
+        expect(header.value).to eq('a_value')
+      end
+    end
+  end
+end

ee/spec/services/audit_events/streaming/instance_headers/update_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe AuditEvents::Streaming::InstanceHeaders::UpdateService, feature_category: :audit_events do
+  let(:header) { create(:instance_audit_events_streaming_header, key: 'old', value: 'old') }
+
+  let(:params) do
+    {
+      header: header,
+      key: 'new',
+      value: 'new'
+    }
+  end
+
+  subject(:service) do
+    described_class.new(
+      params: params
+    )
+  end
+
+  describe '#execute' do
+    shared_examples 'does not update header' do
+      it 'does not update the header' do
+        expect { subject }.not_to change { header.reload.key }
+        expect(header.value).to eq 'old'
+      end
+
+      it 'has an error response' do
+        expect(response).to be_error
+        expect(response.errors)
+          .to match_array [error_message]
+      end
+    end
+
+    subject(:response) { service.execute }
+
+    context 'when no header is provided' do
+      let(:params) { super().merge(header: nil) }
+
+      it_behaves_like 'does not update header' do
+        let_it_be(:error_message) { AuditEvents::Streaming::InstanceHeaders::BaseService::HEADER_ERROR_MESSAGE }
+      end
+    end
+
+    context 'when the header is updated successfully' do
+      it 'updates the header' do
+        expect(response).to be_success
+        expect(header.reload.key).to eq 'new'
+        expect(header.value).to eq 'new'
+      end
+    end
+
+    context 'when header update is unsuccessful' do
+      let(:params) { super().merge(key: '') }
+
+      it_behaves_like 'does not update header' do
+        let_it_be(:error_message) { "Key can't be blank" }
+      end
+    end
+  end
+end