From 585f8383810b38cde7f5435be578ff71dc230cc0 Mon Sep 17 00:00:00 2001
From: Oleksii Fedorov <ofedorov@pivotal.io>
Date: Thu, 22 Aug 2019 18:42:29 +0200
Subject: [PATCH] As an API user, I can delete service brokers that have no
 instances via v3

[finishes #166631760]
[PR]

Co-authored-by: Derik Evangelista <devangelista@pivotal.io>
Co-authored-by: Felisia Martini <fmartini@pivotal.io>
Co-authored-by: Joao Lebre <jlebre@pivotal.io>
---
 .../v3/service_brokers_controller.rb          |  19 +-
 bin/rspec                                     |   2 +-
 config/routes.rb                              |   2 +-
 spec/request/service_brokers_spec.rb          | 418 +++++++++++-------
 spec/support/controller_helpers.rb            |  19 +-
 .../support/matchers/be_reported_as_events.rb |  11 +
 spec/support/matchers/find_broker.rb          |  15 +
 spec/support/matchers/match_broker.rb         |  36 ++
 spec/support/service_broker_helpers.rb        |  13 +
 .../request/service_brokers.rb                |  62 +++
 spec/support/user_helpers.rb                  |   8 +
 .../v3/service_bindings_controller_spec.rb    |   2 +-
 .../v3/service_brokers_controller_spec.rb     | 133 +++++-
 13 files changed, 553 insertions(+), 187 deletions(-)
 create mode 100644 spec/support/matchers/be_reported_as_events.rb
 create mode 100644 spec/support/matchers/find_broker.rb
 create mode 100644 spec/support/matchers/match_broker.rb
 create mode 100644 spec/support/shared_examples/request/service_brokers.rb

diff --git a/app/controllers/v3/service_brokers_controller.rb b/app/controllers/v3/service_brokers_controller.rb
index e05bc3ec97d..8f029b1b411 100644
--- a/app/controllers/v3/service_brokers_controller.rb
+++ b/app/controllers/v3/service_brokers_controller.rb
@@ -59,9 +59,22 @@ def create
     unprocessable!(e.message)
   end
 
-  def delete
-    raise "real,ly unique mesage with stypos"
-    #render status: :ok, json: {}
+  def destroy
+    service_broker = VCAP::CloudController::ServiceBroker.find(guid: hashed_params[:guid])
+    broker_not_found! unless service_broker
+
+    if service_broker.space.nil?
+      broker_not_found! unless permission_queryer.can_read_globally?
+      unauthorized! unless permission_queryer.can_write_global_service_broker?
+    else
+      broker_not_found! unless permission_queryer.can_read_from_space?(service_broker.space.guid, service_broker.space.organization.guid)
+      unauthorized! unless permission_queryer.can_write_space_scoped_service_broker?(service_broker.space.guid)
+    end
+
+    service_event_repository = VCAP::CloudController::Repositories::ServiceEventRepository.new(user_audit_info)
+    service_broker_remover = VCAP::Services::ServiceBrokers::ServiceBrokerRemover.new(service_event_repository)
+
+    service_broker_remover.remove(service_broker)
   end
 
   private
diff --git a/bin/rspec b/bin/rspec
index 6e6709219af..d7be13f3f43 100755
--- a/bin/rspec
+++ b/bin/rspec
@@ -1,6 +1,6 @@
 #!/usr/bin/env ruby
 begin
-  load File.expand_path('../spring', __FILE__)
+  load File.expand_path('spring', __dir__)
 rescue LoadError => e
   raise unless e.message.include?('spring')
 end
diff --git a/config/routes.rb b/config/routes.rb
index 7147dab1a2c..bd9657380e9 100644
--- a/config/routes.rb
+++ b/config/routes.rb
@@ -164,7 +164,7 @@
   get '/service_brokers', to: 'service_brokers#index'
   get '/service_brokers/:guid', to: 'service_brokers#show'
   post '/service_brokers', to: 'service_brokers#create'
-  delete '/service_brokers/:guid', to: 'service_brokers#delete'
+  delete '/service_brokers/:guid', to: 'service_brokers#destroy'
 
   # space_manifests
   post '/spaces/:guid/actions/apply_manifest', to: 'space_manifests#apply_manifest'
diff --git a/spec/request/service_brokers_spec.rb b/spec/request/service_brokers_spec.rb
index 5bbf8fc9bbc..aa15dd247a4 100644
--- a/spec/request/service_brokers_spec.rb
+++ b/spec/request/service_brokers_spec.rb
@@ -1,39 +1,53 @@
 require 'spec_helper'
+require 'cloud_controller'
+require 'services'
 
 RSpec.describe 'V3 service brokers' do
-  def catalog
+  let(:global_broker_id) { 'global-service-id' }
+  let(:space_broker_id) { 'space-service-id' }
+
+  def catalog(id=global_broker_id)
     {
-        'services' => [
+      'services' => [
+        {
+          'id' => "#{id}-1",
+          'name' => 'service_name-1',
+          'description' => 'some description 1',
+          'bindable' => true,
+          'plans' => [
             {
-                'id' => 'service_id-1',
-                'name' => 'service_name-1',
-                'description' => 'some description 1',
-                'bindable' => true,
-                'plans' => [
-                    {
-                        'id' => 'fake_plan_id-1',
-                        'name' => 'plan_name-1',
-                        'description' => 'fake_plan_description 1',
-                        'schemas' => nil
-                    }
-                ]
-            },
+                'id' => 'fake_plan_id-1',
+                'name' => 'plan_name-1',
+                'description' => 'fake_plan_description 1',
+                'schemas' => nil
+            }
+          ],
+          'dashboard_client' => dashboard_client(id)
+        },
+        {
+          'id' => "#{id}-2",
+          'name' => 'route_volume_service_name-2',
+          'requires' => ['volume_mount', 'route_forwarding'],
+          'description' => 'some description 2',
+          'bindable' => true,
+          'plans' => [
             {
-                'id' => 'service_id-2',
-                'name' => 'route_volume_service_name-2',
-                'requires' => ['volume_mount', 'route_forwarding'],
-                'description' => 'some description 2',
-                'bindable' => true,
-                'plans' => [
-                    {
-                        'id' => 'fake_plan_id-2',
-                        'name' => 'plan_name-2',
-                        'description' => 'fake_plan_description 2',
-                        'schemas' => nil
-                    }
-                ]
-            },
-        ]
+                'id' => 'fake_plan_id-2',
+                'name' => 'plan_name-2',
+                'description' => 'fake_plan_description 2',
+                'schemas' => nil
+            }
+          ]
+        },
+      ]
+    }
+  end
+
+  def dashboard_client(id=global_broker_id)
+    {
+      'id' => "#{id}-uaa-id",
+      'secret' => 'my-dashboard-secret',
+      'redirect_uri' => 'http://example.org'
     }
   end
 
@@ -70,10 +84,59 @@ def catalog
     headers_for(user)
   }
 
+  let(:global_broker_request_body) do
+    {
+      name: 'broker name',
+      url: 'http://example.org/broker-url',
+      credentials: {
+        type: 'basic',
+        data: {
+          username: 'admin',
+          password: 'welcome',
+        }
+      }
+    }
+  end
+
+  let(:space_scoped_broker_request_body) do
+    {
+      name: 'space-scoped broker name',
+      url: 'http://example.org/space-broker-url',
+      credentials: {
+        type: 'basic',
+        data: {
+          username: 'admin',
+          password: 'welcome',
+        },
+      },
+      relationships: {
+        space: {
+          data: {
+            guid: space.guid
+          },
+        },
+      },
+    }
+  end
+
   let(:parsed_body) {
     JSON.parse(last_response.body)
   }
 
+  before do
+    stub_request(:get, 'http://example.org/broker-url/v2/catalog').
+      to_return(status: 200, body: catalog.to_json, headers: {})
+    stub_request(:get, 'http://example.org/space-broker-url/v2/catalog').
+      to_return(status: 200, body: catalog(space_broker_id).to_json, headers: {})
+
+    token = { token_type: 'Bearer', access_token: 'my-favourite-access-token' }
+    stub_request(:post, 'https://uaa.service.cf.internal/oauth/token').
+      to_return(status: 200, body: token.to_json, headers: { 'Content-Type' => 'application/json' })
+
+    stub_uaa_for(global_broker_id)
+    stub_uaa_for(space_broker_id)
+  end
+
   describe 'GET /v3/service_brokers' do
     context 'when there are no service brokers' do
       it 'returns 200 OK and an empty list for admin' do
@@ -221,7 +284,7 @@ def expect_a_list_of_brokers(user_headers, list)
       list.each do |expected_broker|
         actual_broker = brokers.find { |b| b.fetch('name') == expected_broker.name }
         expect(actual_broker).to_not be_nil, "Could not find broker with name '#{expected_broker.name}'"
-        expect_broker_to_match(expected: expected_broker, actual: actual_broker)
+        expect(actual_broker).to match_broker(expected_broker)
       end
     end
   end
@@ -229,7 +292,7 @@ def expect_a_list_of_brokers(user_headers, list)
   describe 'GET /v3/service_brokers/:guid' do
     context 'when the service broker does not exist' do
       it 'return with 404 Not Found' do
-        expect_broker_not_found('does-not-exist', with: admin_headers)
+        is_expected.to_not find_broker(broker_guid: 'does-not-exist', with: admin_headers)
       end
     end
 
@@ -246,11 +309,11 @@ def expect_a_list_of_brokers(user_headers, list)
       end
 
       it 'returns 404 Not Found for space developer' do
-        expect_broker_not_found(global_service_broker_1.guid, with: space_developer_headers)
+        is_expected.to_not find_broker(broker_guid: global_service_broker_1.guid, with: space_developer_headers)
       end
 
       it 'returns 404 Not Found for org/space auditor/manager' do
-        expect_broker_not_found(global_service_broker_2.guid, with: org_space_manager_headers)
+        is_expected.to_not find_broker(broker_guid: global_service_broker_2.guid, with: org_space_manager_headers)
       end
     end
 
@@ -270,78 +333,28 @@ def expect_a_list_of_brokers(user_headers, list)
       end
 
       it 'returns 404 Not Found for space developer in another space' do
-        expect_broker_not_found(space_scoped_service_broker.guid, with: space_developer_alternate_space_headers)
+        is_expected.to_not find_broker(broker_guid: space_scoped_service_broker.guid, with: space_developer_alternate_space_headers)
       end
 
       it 'returns 404 Not Found for org/space auditor/manager' do
-        expect_broker_not_found(space_scoped_service_broker.guid, with: org_space_manager_headers)
+        is_expected.to_not find_broker(broker_guid: space_scoped_service_broker.guid, with: org_space_manager_headers)
       end
     end
 
-    def expect_broker_not_found(guid, with:)
-      get("/v3/service_brokers/#{guid}", {}, with)
-      expect(last_response.status).to eq(404)
-    end
-
     def expect_broker(expected_broker, with:)
       get("/v3/service_brokers/#{expected_broker.guid}", {}, with)
       expect(last_response.status).to eq(200)
-      expect_broker_to_match(expected: expected_broker, actual: parsed_body)
+      expect(parsed_body).to match_broker(expected_broker)
     end
   end
 
   describe 'POST /v3/service_brokers' do
-    let(:global_broker_request_body) do
-      {
-          name: 'broker name',
-          url: 'http://example.org/broker-url',
-          credentials: {
-              type: 'basic',
-              data: {
-                  username: 'admin',
-                  password: 'welcome',
-              }
-          }
-      }
-    end
-
-    let(:space_scoped_broker_request_body) do
-      {
-          name: 'space-scoped broker name',
-          url: 'http://example.org/broker-url',
-          credentials: {
-              type: 'basic',
-              data: {
-                  username: 'admin',
-                  password: 'welcome',
-              },
-          },
-          relationships: {
-              space: {
-                  data: {
-                      guid: space.guid
-                  },
-              },
-          },
-      }
-    end
-
-    before do
-      stub_request(:get, 'http://example.org/broker-url/v2/catalog').
-          to_return(status: 200, body: catalog.to_json, headers: {})
-    end
-
-    def create_broker(broker_body, with:)
-      @count_before_creation = VCAP::CloudController::ServiceBroker.count
-      post('/v3/service_brokers', broker_body.to_json, with)
-    end
-
     context 'as admin' do
       context 'when route and volume mount services are enabled' do
         before do
           TestConfig.config[:route_services_enabled] = true
           TestConfig.config[:volume_services_enabled] = true
-          create_broker(global_broker_request_body, with: admin_headers)
+          create_broker_successfully(global_broker_request_body, with: admin_headers)
         end
 
         it 'returns 201 Created' do
@@ -355,12 +368,13 @@ def create_broker(broker_body, with:)
 
         it 'reports service events' do
           # FIXME: there is an event missing for registering/creating the broker itself by the respective user
-          expect_events([
-                            {type: 'audit.service.create', actor: 'broker name'},
-                            {type: 'audit.service.create', actor: 'broker name'},
-                            {type: 'audit.service_plan.create', actor: 'broker name'},
-                            {type: 'audit.service_plan.create', actor: 'broker name'},
-                        ])
+          expect([
+            { type: 'audit.service.create', actor: 'broker name' },
+            { type: 'audit.service.create', actor: 'broker name' },
+            { type: 'audit.service_dashboard_client.create', actor: 'broker name' },
+            { type: 'audit.service_plan.create', actor: 'broker name' },
+            { type: 'audit.service_plan.create', actor: 'broker name' },
+          ]).to be_reported_as_events
         end
       end
 
@@ -368,7 +382,7 @@ def create_broker(broker_body, with:)
         before do
           TestConfig.config[:route_services_enabled] = false
           TestConfig.config[:volume_services_enabled] = false
-          create_broker(global_broker_request_body, with: admin_headers)
+          create_broker_successfully(global_broker_request_body, with: admin_headers)
         end
 
         it 'returns 201 Created' do
@@ -382,19 +396,25 @@ def create_broker(broker_body, with:)
 
         it 'returns warning in the header' do
           expect(last_response_warnings).
-              to eq([
-                        'Service route_volume_service_name-2 is declared to be a route service but support for route services is disabled.' \
+            to eq([
+              'Service route_volume_service_name-2 is declared to be a route service but support for route services is disabled.' \
 ' Users will be prevented from binding instances of this service with routes.',
-                        'Service route_volume_service_name-2 is declared to be a volume mount service but support for volume mount services is disabled.' \
+              'Service route_volume_service_name-2 is declared to be a volume mount service but support for volume mount services is disabled.' \
 ' Users will be prevented from binding instances of this service with apps.'
-                    ])
+            ])
+        end
+
+        let(:uaa_uri) { VCAP::CloudController::Config.config.get(:uaa, :internal_url) }
+        let(:tx_url) { uaa_uri + '/oauth/clients/tx/modify' }
+        it 'creates some UAA stuff' do
+          expect(a_request(:post, tx_url)).to have_been_made
         end
       end
 
       context 'when user provides a malformed request' do
         let(:malformed_body) do
           {
-              whatever: 'oopsie'
+            whatever: 'oopsie'
           }
         end
 
@@ -422,7 +442,7 @@ def create_broker(broker_body, with:)
       context 'when fetching broker catalog fails' do
         before do
           stub_request(:get, 'http://example.org/broker-url/v2/catalog').
-              to_return(status: 418, body: {}.to_json)
+            to_return(status: 418, body: {}.to_json)
           create_broker(global_broker_request_body, with: admin_headers)
         end
 
@@ -438,7 +458,7 @@ def create_broker(broker_body, with:)
         create_broker(global_broker_request_body, with: space_developer_headers)
 
         expect_no_broker_created
-        expect_error(status: 403, error: 'CF-NotAuthorized', description: 'You are not authorized to perform the requested action')
+        expect_unauthorized
       end
 
       describe 'registering a space scoped service broker' do
@@ -457,12 +477,13 @@ def create_broker(broker_body, with:)
 
         it 'reports service events' do
           # FIXME: there is an event missing for registering/creating the broker itself by the respective user
-          expect_events([
-                            {type: 'audit.service.create', actor: 'space-scoped broker name'},
-                            {type: 'audit.service.create', actor: 'space-scoped broker name'},
-                            {type: 'audit.service_plan.create', actor: 'space-scoped broker name'},
-                            {type: 'audit.service_plan.create', actor: 'space-scoped broker name'},
-                        ])
+          expect([
+            { type: 'audit.service.create', actor: 'space-scoped broker name' },
+            { type: 'audit.service.create', actor: 'space-scoped broker name' },
+            { type: 'audit.service_dashboard_client.create', actor: 'space-scoped broker name' },
+            { type: 'audit.service_plan.create', actor: 'space-scoped broker name' },
+            { type: 'audit.service_plan.create', actor: 'space-scoped broker name' },
+          ]).to be_reported_as_events
         end
       end
     end
@@ -472,14 +493,14 @@ def create_broker(broker_body, with:)
         create_broker(global_broker_request_body, with: org_space_manager_headers)
 
         expect_no_broker_created
-        expect_error(status: 403, error: 'CF-NotAuthorized', description: 'You are not authorized to perform the requested action')
+        expect_unauthorized
       end
 
       it 'returns 403 when registering a space-scoped broker' do
         create_broker(space_scoped_broker_request_body, with: org_space_manager_headers)
 
         expect_no_broker_created
-        expect_error(status: 403, error: 'CF-NotAuthorized', description: 'You are not authorized to perform the requested action')
+        expect_unauthorized
       end
     end
 
@@ -489,11 +510,11 @@ def expect_created_broker(expected_broker)
       service_broker = VCAP::CloudController::ServiceBroker.last
 
       expect(service_broker).to include(
-                                    'name' => expected_broker[:name],
-                                    'broker_url' => expected_broker[:url],
-                                    'auth_username' => expected_broker.dig(:credentials, :data, :username),
-                                    'space_guid' => expected_broker.dig(:relationships, :space, :data, :guid),
-                                )
+        'name' => expected_broker[:name],
+        'broker_url' => expected_broker[:url],
+        'auth_username' => expected_broker.dig(:credentials, :data, :username),
+        'space_guid' => expected_broker.dig(:relationships, :space, :data, :guid),
+      )
       # password not exported in to_hash
       expect(service_broker.auth_password).to eq(expected_broker[:credentials][:data][:password])
     end
@@ -510,82 +531,159 @@ def expect_catalog_synchronized(catalog)
       end
     end
 
-    def expect_events(expected_events)
-      events = VCAP::CloudController::Event.all
-      expect(events.map { |e| {type: e.type, actor: e.actor_name} }).to eq(expected_events)
-    end
-
     def expect_no_broker_created
       expect(VCAP::CloudController::ServiceBroker.count).to eq(@count_before_creation)
     end
-
-    def expect_error(status:, error: '', description: '')
-      expect(last_response).to have_status_code(status)
-      expect(last_response.body).to include(error)
-      expect(last_response.body).to include(description)
-    end
   end
 
   describe 'DELETE /v3/service_brokers/:guid' do
-    let!(:global_service_broker) { VCAP::CloudController::ServiceBroker.make }
-    let!(:space_scoped_service_broker) { VCAP::CloudController::ServiceBroker.make(space: space) }
+    let!(:global_broker) {
+      create_broker_successfully(global_broker_request_body, with: admin_headers)
+    }
+    let!(:global_broker_services) { VCAP::CloudController::Service.where(service_broker_id: global_broker.id) }
+    let!(:global_broker_plans) { VCAP::CloudController::ServicePlan.where(service_id: global_broker_services.map(&:id)) }
+
+    let!(:space_scoped_service_broker) {
+      create_broker_successfully(space_scoped_broker_request_body, with: admin_headers)
+    }
+    let!(:space_broker_services) { VCAP::CloudController::Service.where(service_broker_id: space_scoped_service_broker.id) }
+    let!(:space_broker_plans) { VCAP::CloudController::ServicePlan.where(service_id: space_broker_services.map(&:id)) }
 
     context 'as an admin user' do
-      # what about space scoped?
-      # what about when there are SIs?
       context 'when the broker does not exist' do
-        xit 'responds with 404 Not Found' do
-          delete '/v3/service_brokers/guid-that-does-not-exist', {}, admin_headers
-          expect(last_response).to have_status_code(404)
+        it 'responds with 404 Not Found' do
+          delete_broker('guid-that-does-not-exist', with: admin_headers)
+          expect_error(status: 404, error: 'CF-ResourceNotFound', description: 'Service broker not found')
         end
       end
 
-      context 'there are no service instances' do
-        xit 'returns 204 No Content' do
-        end
+      context 'when there are no service instances' do
+        let(:broker) { global_broker }
+        let(:broker_services) { global_broker_services }
+        let(:broker_plans) { global_broker_plans }
+        let(:actor) { 'broker name' }
+        let(:user_headers) { admin_headers }
+        let(:broker_id) { global_broker_id }
+
+        it_behaves_like 'a successful broker delete'
       end
     end
 
-    context 'as an admin-read-only user' do
+    context 'as an admin-read-only/global auditor user' do
       it 'fails authorization' do
-        delete("/v3/service_brokers/#{global_service_broker.guid}", {}, admin_read_only_headers)
+        delete_broker(global_broker.guid, with: admin_read_only_headers)
+        expect_unauthorized
 
-        expect(last_response).to have_status_code(403)
+        delete_broker(global_broker.guid, with: global_auditor_headers)
+        expect_unauthorized
+      end
+    end
+
+    context 'as a space developer' do
+      context 'with access to the broker' do
+        context 'when the broker has no service instances' do
+          let(:broker) { space_scoped_service_broker }
+          let(:broker_services) { space_broker_services }
+          let(:broker_plans) { space_broker_plans }
+          let(:actor) { 'space-scoped broker name' }
+          let(:user_headers) { space_developer_headers }
+          let(:broker_id) { space_broker_id }
+
+          it_behaves_like 'a successful broker delete'
+        end
+      end
+
+      context 'without access to the broker' do
+        it 'fails authorization' do
+          delete_broker(space_scoped_service_broker.guid, with: space_developer_alternate_space_headers)
+          expect(last_response.status).to eq(404)
+        end
       end
     end
 
     context 'as an org/space auditor/manager/billing manager user' do
-      xit 'responds with 404 Not Found' do
-        delete "/v3/service_brokers/#{global_service_broker.guid}", {}, org_space_manager_headers
+      it 'responds with 404 Not Found for global brokers' do
+        delete "/v3/service_brokers/#{global_broker.guid}", {}, org_space_manager_headers
         expect(last_response).to have_status_code(404)
+      end
 
+      it 'responds with 403 Not Authorized for space-scoped broker' do
         delete "/v3/service_brokers/#{space_scoped_service_broker.guid}", {}, org_space_manager_headers
-        expect(last_response).to have_status_code(404)
+        expect(last_response).to have_status_code(403)
       end
     end
-  end
 
-  def expect_broker_to_match(actual:, expected:)
-    expect(actual['url']).to eq(expected.broker_url)
-    expect(actual['created_at']).to eq(expected.created_at.iso8601)
-    expect(actual['updated_at']).to eq(expected.updated_at.iso8601)
-    expect(actual).to have_key('links')
-    expect(actual['links']).to have_key('self')
-    expect(actual['links']['self']['href']).to include("/v3/service_brokers/#{expected.guid}")
-
-    if expected.space.nil?
-      expect(actual['relationships'].length).to eq(0)
-      expect(actual['links']).not_to have_key('space')
-    else
-      expect(actual['relationships'].length).to eq(1)
-      expect(actual['relationships']['space']).to have_key('data')
-      expect(actual['relationships']['space']['data']['guid']).to eq(expected.space.guid)
-      expect(actual['links']).to have_key('space')
-      expect(actual['links']['space']['href']).to include("/v3/spaces/#{expected.space.guid}")
+    def delete_broker(guid, with:)
+      delete "/v3/service_brokers/#{guid}", {}, with
     end
   end
 
+  def create_broker(broker_body, with:)
+    @count_before_creation = VCAP::CloudController::ServiceBroker.count
+    post('/v3/service_brokers', broker_body.to_json, with)
+  end
+
+  def create_broker_successfully(broker_body, with:)
+    create_broker(broker_body, with: with)
+    expect(last_response).to have_status_code(201)
+    VCAP::CloudController::ServiceBroker.last
+  end
+
   def last_response_warnings
     last_response.headers['X-Cf-Warnings'].split(',').map { |w| CGI.unescape(w) }
   end
+
+  def expect_error(status:, error: '', description: '')
+    expect(last_response).to have_status_code(status)
+    expect(last_response.body).to include(error)
+    expect(last_response.body).to include(description)
+  end
+
+  def expect_unauthorized
+    expect_error(
+      status: 403,
+      error: 'CF-NotAuthorized',
+      description: 'You are not authorized to perform the requested action'
+    )
+  end
+
+  def stub_uaa_for(broker_id)
+    stub_request(:get, "https://uaa.service.cf.internal/oauth/clients/#{broker_id}-uaa-id").
+      to_return(
+        { status: 404, body: {}.to_json, headers: { 'Content-Type' => 'application/json' } },
+            { status: 200, body: { client_id: dashboard_client(broker_id)['id'] }.to_json, headers: { 'Content-Type' => 'application/json' } }
+        )
+
+    stub_request(:post, 'https://uaa.service.cf.internal/oauth/clients/tx/modify').
+      with(
+        body: [
+          {
+                "client_id": "#{broker_id}-uaa-id",
+                "client_secret": 'my-dashboard-secret',
+                "redirect_uri": 'http://example.org',
+                "scope": %w(openid cloud_controller_service_permissions.read),
+                "authorities": ['uaa.resource'],
+                "authorized_grant_types": ['authorization_code'],
+                "action": 'add'
+            }
+        ].to_json
+        ).
+      to_return(status: 201, body: {}.to_json, headers: { 'Content-Type' => 'application/json' })
+
+    stub_request(:post, 'https://uaa.service.cf.internal/oauth/clients/tx/modify').
+      with(
+        body: [
+          {
+                "client_id": "#{broker_id}-uaa-id",
+                "client_secret": nil,
+                "redirect_uri": nil,
+                "scope": %w(openid cloud_controller_service_permissions.read),
+                "authorities": ['uaa.resource'],
+                "authorized_grant_types": ['authorization_code'],
+                "action": 'delete'
+            }
+        ].to_json
+        ).
+      to_return(status: 200, body: {}.to_json, headers: { 'Content-Type' => 'application/json' })
+  end
 end
diff --git a/spec/support/controller_helpers.rb b/spec/support/controller_helpers.rb
index f0ba289a427..d4b37dcdc87 100644
--- a/spec/support/controller_helpers.rb
+++ b/spec/support/controller_helpers.rb
@@ -47,7 +47,8 @@ def admin_headers_for(user, opts={})
   end
 
   def headers_for(user, opts={})
-    opts = { email: Sham.email,
+    generated_email = Sham.email
+    opts = { email: generated_email,
              https: false }.merge(opts)
 
     headers = {}
@@ -55,7 +56,13 @@ def headers_for(user, opts={})
     headers['HTTP_AUTHORIZATION'] = "bearer #{token}"
 
     headers['HTTP_X_FORWARDED_PROTO'] = 'https' if opts[:https]
-    json_headers(headers)
+    result = json_headers(headers)
+
+    result.define_singleton_method('_generated_email') do
+      generated_email
+    end
+
+    result
   end
 
   def json_headers(headers)
@@ -106,4 +113,12 @@ def admin_read_only_headers
     @admin_read_only_headers
   end
 
+  def global_auditor_headers
+    if !@global_auditor_headers
+      user = User.make
+      @global_auditor_headers = headers_for(user, scopes: %w(cloud_controller.global_auditor))
+      user.destroy
+    end
+    @global_auditor_headers
+  end
 end
diff --git a/spec/support/matchers/be_reported_as_events.rb b/spec/support/matchers/be_reported_as_events.rb
new file mode 100644
index 00000000000..2a3b80caf42
--- /dev/null
+++ b/spec/support/matchers/be_reported_as_events.rb
@@ -0,0 +1,11 @@
+RSpec::Matchers.define :be_reported_as_events do
+  events = []
+  match do |expected_events|
+    events = VCAP::CloudController::Event.all.sort_by(&:type).map { |e| { type: e.type, actor: e.actor_name } }
+    events == expected_events
+  end
+
+  failure_message do
+    "Expected events were not reported. Events reported: #{events}"
+  end
+end
diff --git a/spec/support/matchers/find_broker.rb b/spec/support/matchers/find_broker.rb
new file mode 100644
index 00000000000..7c5cd8d2778
--- /dev/null
+++ b/spec/support/matchers/find_broker.rb
@@ -0,0 +1,15 @@
+RSpec::Matchers.define :find_broker do |expected|
+  match do
+    get("/v3/service_brokers/#{expected[:broker_guid]}", {}, expected[:with] || nil)
+
+    last_response.status == 200
+  end
+
+  failure_message do
+    "expected broker with GUID '#{expected[:broker_guid]}' to be found"
+  end
+
+  failure_message_when_negated do
+    "expected broker with GUID '#{expected[:broker_guid]}' to not be found"
+  end
+end
diff --git a/spec/support/matchers/match_broker.rb b/spec/support/matchers/match_broker.rb
new file mode 100644
index 00000000000..47c77b5555f
--- /dev/null
+++ b/spec/support/matchers/match_broker.rb
@@ -0,0 +1,36 @@
+RSpec::Matchers.define :match_broker do |expected|
+  problems = []
+
+  match do |actual|
+    problems << "Expected #{actual['url']} to be equal to #{expected.broker_url}" unless actual['url'] == expected.broker_url
+    problems << "Expected #{actual['created_at']} to be equal to #{expected.created_at.iso8601}" unless actual['created_at'] == expected.created_at.iso8601
+    problems << "Expected #{actual['updated_at']} to be equal to #{expected.updated_at.iso8601}" unless actual['updated_at'] == expected.updated_at.iso8601
+    problems << "Expected broker object to have key 'links'" if actual['links'].nil?
+    problems << "Expected broker.links to have key 'self'" if actual['links']['self'].nil?
+    unless actual['links']['self']['href'].include?("/v3/service_brokers/#{expected.guid}")
+      problems << "Expected #{actual['links']['self']['href']} to include '/v3/service_brokers/#{expected.guid}'"
+    end
+
+    if expected.space.nil?
+      problems << 'Expected broker relationships to be empty' unless actual['relationships'].empty?
+      # problems << "Expected #{actual['links']}).not_to have_key('space')
+      problems << "Expected broker.links to not have key 'space'" unless actual['links']['space'].nil?
+    else
+      problems << 'Expected broker relationships to not be empty' if actual['relationships'].empty?
+      problems << "Expected broker.links to have key 'space'" if actual['links']['space'].nil?
+      problems << "Expected broker.relationships.space to have key 'data'" if actual['relationships']['space']['data'].nil?
+      unless actual['relationships']['space']['data']['guid'] == expected.space.guid
+        problems << "Expected #{actual['relationships']['space']['data']['guid']} to be equal to #{expected.space.guid}"
+      end
+      unless actual['links']['space']['href'].include?("/v3/spaces/#{expected.space.guid}")
+        problems << "Expected #{actual['links']['space']['href']} to include '/v3/spaces/#{expected.space.guid}'"
+      end
+    end
+
+    problems.empty?
+  end
+
+  failure_message do |actual_event|
+    "Expect brokers to match, but it did not. Problems were:\n" + problems.join("\n")
+  end
+end
diff --git a/spec/support/service_broker_helpers.rb b/spec/support/service_broker_helpers.rb
index 8a6f791c985..becbed52c25 100644
--- a/spec/support/service_broker_helpers.rb
+++ b/spec/support/service_broker_helpers.rb
@@ -85,6 +85,15 @@ def stub_unbind_for_instance(service_instance, opts={})
       to_return(status: status, body: body)
   end
 
+  def stub_delete(broker, opts={})
+    status = opts[:status] || 204
+    body = opts[:body] || '{}'
+
+    stub_request(:delete, delete_broker_url(broker)).
+      with(basic_auth: basic_auth(service_broker: broker)).
+      to_return(status: status, body: body)
+  end
+
   def provision_url_for_broker(broker, accepts_incomplete: nil)
     path = "/v2/service_instances/#{guid_pattern}"
     async_query = "accepts_incomplete=#{accepts_incomplete}" if !accepts_incomplete.nil?
@@ -150,6 +159,10 @@ def service_binding_url(service_binding, query=nil)
     build_broker_url(service_instance.service_broker, path, query)
   end
 
+  def delete_broker_url(broker)
+    build_broker_url(broker)
+  end
+
   def remove_basic_auth(url)
     uri = URI(url)
     uri.user = nil
diff --git a/spec/support/shared_examples/request/service_brokers.rb b/spec/support/shared_examples/request/service_brokers.rb
new file mode 100644
index 00000000000..eefb57a4bd8
--- /dev/null
+++ b/spec/support/shared_examples/request/service_brokers.rb
@@ -0,0 +1,62 @@
+RSpec.shared_examples 'a successful broker delete' do
+  before do
+    clear_events
+    delete "/v3/service_brokers/#{broker.guid}", {}, user_headers
+  end
+
+  it 'returns 204 No Content' do
+    expect(last_response).to have_status_code(204)
+  end
+
+  it 'deletes the broker' do
+    is_expected.to_not find_broker(broker_guid: broker.guid, with: user_headers)
+  end
+
+  it 'deletes the service offerings and plans' do
+    services = VCAP::CloudController::Service.where(id: broker_services.map(&:id))
+    expect(services).to have(0).items
+
+    plans = VCAP::CloudController::ServicePlan.where(id: broker_plans.map(&:id))
+    expect(plans).to have(0).items
+  end
+
+  it 'emits service and plan deletion events, and broker deletion event' do
+    expect(broker_delete_events(actor, user_headers._generated_email)).to be_reported_as_events
+  end
+
+  it 'deletes the UAA clients related to this broker' do
+    # see service_broker_remover.rb
+    uaa_client_id = "#{broker_id}-uaa-id"
+    expect(VCAP::CloudController::ServiceDashboardClient.find_client_by_uaa_id(uaa_client_id)).to be_nil
+
+    expect(a_request(:post, 'https://uaa.service.cf.internal/oauth/clients/tx/modify').
+        with(
+          body: [
+            {
+                  "client_id": uaa_client_id,
+                  "client_secret": nil,
+                  "redirect_uri": nil,
+                  "scope": %w(openid cloud_controller_service_permissions.read),
+                  "authorities": ['uaa.resource'],
+                  "authorized_grant_types": ['authorization_code'],
+                  "action": 'delete'
+              }
+          ].to_json
+        )).to have_been_made
+  end
+
+  def clear_events
+    VCAP::CloudController::Event.dataset.destroy
+  end
+
+  def broker_delete_events(actor, email)
+    [
+      { type: 'audit.service.delete', actor: actor },
+      { type: 'audit.service.delete', actor: actor },
+      { type: 'audit.service_broker.delete', actor: email },
+      { type: 'audit.service_dashboard_client.delete', actor: actor },
+      { type: 'audit.service_plan.delete', actor: actor },
+      { type: 'audit.service_plan.delete', actor: actor },
+    ]
+  end
+end
diff --git a/spec/support/user_helpers.rb b/spec/support/user_helpers.rb
index 8d104cd7f41..908fdc2a85d 100644
--- a/spec/support/user_helpers.rb
+++ b/spec/support/user_helpers.rb
@@ -223,6 +223,10 @@ def allow_user_global_read_access(user)
     allow(permissions_double(user)).to receive(:can_read_globally?).and_return(true)
   end
 
+  def allow_user_global_write_access(user)
+    allow(permissions_double(user)).to receive(:can_write_globally?).and_return(true)
+  end
+
   def allow_user_read_access_for_isolation_segment(user)
     allow(permissions_double(user)).to receive(:can_read_from_isolation_segment?).and_return(true)
   end
@@ -235,6 +239,10 @@ def disallow_user_global_read_access(user)
     allow(permissions_double(user)).to receive(:can_read_globally?).and_return(false)
   end
 
+  def disallow_user_global_write_access(user)
+    allow(permissions_double(user)).to receive(:can_write_globally?).and_return(false)
+  end
+
   def disallow_user_read_access(user, space:)
     allow(permissions_double(user)).to receive(:can_read_from_space?).with(space.guid, space.organization_guid).and_return(false)
   end
diff --git a/spec/unit/controllers/v3/service_bindings_controller_spec.rb b/spec/unit/controllers/v3/service_bindings_controller_spec.rb
index 507b524c1dc..0cfec17f0e4 100644
--- a/spec/unit/controllers/v3/service_bindings_controller_spec.rb
+++ b/spec/unit/controllers/v3/service_bindings_controller_spec.rb
@@ -514,7 +514,7 @@
         end
       end
 
-      context 'when the user has read, but not write persimmons on the space' do
+      context 'when the user has read, but not write permissions on the space' do
         before do
           allow_user_read_access_for(user, spaces: [space])
           disallow_user_write_access(user, space: space)
diff --git a/spec/unit/controllers/v3/service_brokers_controller_spec.rb b/spec/unit/controllers/v3/service_brokers_controller_spec.rb
index 5578544d0cc..47b2a115336 100644
--- a/spec/unit/controllers/v3/service_brokers_controller_spec.rb
+++ b/spec/unit/controllers/v3/service_brokers_controller_spec.rb
@@ -2,29 +2,29 @@
 require 'permissions_spec_helper'
 
 RSpec.describe ServiceBrokersController, type: :controller do
-  describe '#create' do
-    let(:user) { set_current_user(VCAP::CloudController::User.make) }
-    let(:space) { VCAP::CloudController::Space.make }
+  let(:user) { set_current_user(VCAP::CloudController::User.make) }
+  let(:space) { VCAP::CloudController::Space.make }
+
+  before do
+    allow_user_read_access_for(user, spaces: [space])
+    allow_user_write_access(user, space: space)
+  end
 
-    let(:request_body) do
+  describe '#create' do
+    let(:request_body) {
       {
-        name: 'some-name',
-        relationships: { space: { data: { guid: space_guid } } },
-        url: 'https://fake.url',
-        credentials: {
-          type: 'basic',
-          data: {
-            username: 'fake username',
-            password: 'fake password',
+          name: 'some-name',
+          relationships: { space: { data: { guid: space_guid } } },
+          url: 'https://fake.url',
+          credentials: {
+              type: 'basic',
+              data: {
+                  username: 'fake username',
+                  password: 'fake password',
+              },
           },
-        },
       }
-    end
-
-    before do
-      allow_user_read_access_for(user, spaces: [space])
-      allow_user_write_access(user, space: space)
-    end
+    }
 
     context 'when a non-existent space is provided' do
       let(:space_guid) { 'space-that-does-not-exist' }
@@ -49,4 +49,99 @@
       end
     end
   end
+
+  describe '#destroy' do
+    let(:service_broker) { VCAP::CloudController::ServiceBroker.make }
+
+    before do
+      allow_user_global_read_access(user)
+      allow_user_global_write_access(user)
+      stub_delete(service_broker)
+    end
+
+    it 'returns a 204' do
+      delete :destroy, params: { guid: service_broker.guid }
+      expect(response.status).to eq 204
+      expect(service_broker.exists?).to be_falsey
+    end
+
+    context 'permissions' do
+      context 'when the service broker does not exist' do
+        it 'returns a 404' do
+          delete :destroy, params: { guid: 'a-guid-that-doesnt-exist' }
+          expect(response).to have_status_code(404)
+          expect(response.body).to include 'Service broker not found'
+        end
+      end
+
+      context 'global brokers' do
+        context 'when the user has read, but not write permissions' do
+          before do
+            allow_user_global_read_access(user)
+            disallow_user_global_write_access(user)
+          end
+
+          it 'returns a 403 Not Authorized and does NOT delete the broker' do
+            delete :destroy, params: { guid: service_broker.guid }
+
+            expect(response.status).to eq 403
+            expect(response.body).to include 'NotAuthorized'
+            expect(service_broker.exists?).to be_truthy
+          end
+        end
+
+        context 'when the user does not have read permissions' do
+          before do
+            disallow_user_global_read_access(user)
+          end
+
+          it 'returns a 404 and does NOT delete the broker' do
+            delete :destroy, params: { guid: service_broker.guid }
+
+            expect(response.status).to eq 404
+            expect(response.body).to include 'ResourceNotFound'
+            expect(response.body).to include 'Service broker not found'
+            expect(service_broker.exists?).to be_truthy
+          end
+        end
+      end
+
+      context 'space scoped brokers' do
+        let(:service_broker) { VCAP::CloudController::ServiceBroker.make(space: space) }
+
+        before do
+          stub_delete(service_broker)
+        end
+
+        context 'when the user has read, but not write permissions on the space' do
+          before do
+            disallow_user_write_access(user, space: space)
+          end
+
+          it 'returns a 403 Not Authorized and does NOT delete the broker' do
+            delete :destroy, params: { guid: service_broker.guid }
+
+            expect(response.status).to eq 403
+            expect(response.body).to include 'NotAuthorized'
+            expect(service_broker.exists?).to be_truthy
+          end
+        end
+
+        context 'when the user does not have read permissions on the space' do
+          before do
+            disallow_user_read_access(user, space: space)
+          end
+
+          it 'returns a 404 and does NOT delete the broker' do
+            delete :destroy, params: { guid: service_broker.guid }
+
+            expect(response.status).to eq 404
+            expect(response.body).to include 'ResourceNotFound'
+            expect(response.body).to include 'Service broker not found'
+            expect(service_broker.exists?).to be_truthy
+          end
+        end
+      end
+    end
+  end
 end