diff --git a/.rspec b/.rspec
index 8c18f1a..7a2cc1a 100644
--- a/.rspec
+++ b/.rspec
@@ -1,2 +1,3 @@
+--require spec_helper
 --format documentation
 --color
diff --git a/.rubocop.yml b/.rubocop.yml
index 5679916..88d6ac0 100644
--- a/.rubocop.yml
+++ b/.rubocop.yml
@@ -6,4 +6,3 @@ AllCops:
   TargetRubyVersion: 3.1
   Exclude:
     - "vendor/**/*"
-    - "spec/**/*"
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index c01c1fa..c959d28 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -40,7 +40,7 @@ bundle exec rake style   # Cookstyle / RuboCop
 To run a single spec file:
 
 ```sh
-bundle exec rspec spec/kitchen/driver/vro_spec.rb
+bundle exec rspec spec/vro_spec.rb
 ```
 
 Many style offenses can be corrected automatically:
diff --git a/Gemfile b/Gemfile
index 8a25339..793b300 100644
--- a/Gemfile
+++ b/Gemfile
@@ -5,7 +5,6 @@ gemspec
 group :test do
   gem "rake"
   gem "rspec", "~> 3.2"
-  gem "webmock",   "~> 3.0"
 end
 
 group :docs do
diff --git a/lib/kitchen/driver/vro.rb b/lib/kitchen/driver/vro.rb
index a0f80de..32e207b 100644
--- a/lib/kitchen/driver/vro.rb
+++ b/lib/kitchen/driver/vro.rb
@@ -16,6 +16,8 @@
 # limitations under the License.
 #
 
+require "timeout" unless defined?(Timeout)
+
 require "kitchen"
 require "vcoworkflows"
 require_relative "version"
@@ -87,14 +89,27 @@ def create(state)
 
       # Runs the destroy workflow for the server named in state.
       #
-      # @param state [Hash] instance state naming the server
+      # Clears +server_id+ and +hostname+ afterwards. Test Kitchen persists
+      # the state hash even when an action fails, so leaving a destroyed
+      # server named in it means the next +kitchen destroy+ runs the destroy
+      # workflow again against a machine that is already gone.
+      #
+      # @param state [Hash] mutable instance state naming the server; loses
+      #   +server_id+ and +hostname+
       # @return [void]
       def destroy(state)
         return if state[:server_id].nil?
 
-        info("Executing the destroy-server workflow for #{state[:hostname]} (#{state[:server_id]})...")
+        server_id = state[:server_id]
+        hostname  = state[:hostname]
+
+        info("Executing the destroy-server workflow for #{hostname} (#{server_id})...")
         execute_destroy_workflow(state)
-        info("Server #{state[:hostname]} (#{state[:server_id]}) destroyed.")
+
+        state.delete(:server_id)
+        state.delete(:hostname)
+
+        info("Server #{hostname} (#{server_id}) destroyed.")
       end
 
       # Connection settings for the vRO API.
@@ -131,16 +146,20 @@ def verify_ssl?
 
       # Points the driver at a different workflow.
       #
-      # Clears the memoized client, which is what makes it safe to run the
-      # destroy workflow after the create workflow in the same process.
+      # Clears the memoized client and the memoized output parameters, which
+      # is what makes it safe to run the destroy workflow after the create
+      # workflow in the same process. Without the second reset, anything
+      # reading {#output_parameters} after the switch would still be looking
+      # at the previous workflow's output.
       #
       # @param name [String] workflow name
       # @param id [String, nil] workflow id, when the name is ambiguous
       # @return [void]
       def set_workflow_vars(name, id)
-        @vro_client    = nil
-        @workflow_name = name
-        @workflow_id   = id
+        @vro_client        = nil
+        @output_parameters = nil
+        @workflow_name     = name
+        @workflow_id       = id
       end
 
       # Runs the create workflow and records what it produced.
@@ -221,17 +240,26 @@ def wait_for_workflow
       # @return [void]
       def wait_for_server(state)
         instance.transport.connection(state).wait_until_ready
-      rescue
+      rescue => e
         error("Server #{state[:hostname]} (#{state[:server_id]}) not reachable. Destroying server...")
-        destroy(state)
-        raise
+
+        begin
+          destroy(state)
+        rescue => destroy_error
+          # Report the leaked machine, but keep the connection failure as the
+          # error that surfaces -- it is the one that explains the run.
+          error("The destroy-server workflow also failed: #{destroy_error.message}. " \
+                "The server may still exist; check the vRO UI.")
+        end
+
+        raise e
       end
 
       # Sets input parameters on the current workflow.
       #
       # @param data [Hash] parameter names to values; names are stringified
       # @return [void]
-      def set_workflow_parameters(data) # rubocop:disable Style/AccessorMethodName
+      def set_workflow_parameters(data) # rubocop:disable Naming/AccessorMethodName
         data.each do |key, value|
           vro_client.parameter(key.to_s, value)
         end
@@ -245,15 +273,19 @@ def output_parameters
       end
 
       # @param key [String] output parameter name
-      # @return [String] the parameter's value, stringified
+      # @return [String] the parameter's value, stringified; an empty string
+      #   when the workflow did not return that parameter at all
       def output_parameter_value(key)
-        output_parameters[key].value.to_s
+        parameter = output_parameters[key]
+        return "" if parameter.nil?
+
+        parameter.value.to_s
       end
 
       # @param key [String] output parameter name
       # @return [Boolean] true when the parameter is missing or empty
       def output_parameter_empty?(key)
-        output_parameter_value(key).nil? || output_parameter_value(key).empty?
+        output_parameter_value(key).empty?
       end
 
       # Checks that the create workflow returned what the driver needs.
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index 30e9e58..cbddcf6 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -15,3 +15,55 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+
+require "rspec"
+
+require "kitchen"
+require "kitchen/driver/vro"
+require "kitchen/provisioner/dummy"
+require "kitchen/transport/dummy"
+require "kitchen/verifier/dummy"
+
+Dir[File.join(__dir__, "support", "**", "*.rb")].sort.each { |f| require f }
+
+# These specs never talk to a vRO appliance. Everything this driver does is
+# turn configuration into vcoworkflows calls and turn workflow output back
+# into instance state, so all of it can be exercised against doubles of the
+# vcoworkflows API. Anything that genuinely needs an appliance belongs in a
+# manual run against a real vRO -- see CONTRIBUTING.md.
+RSpec.configure do |config|
+  config.expect_with :rspec do |expectations|
+    # `should` has been deprecated since RSpec 3. Only `expect` is available.
+    expectations.syntax = :expect
+    expectations.include_chain_clauses_in_custom_matcher_descriptions = true
+  end
+
+  config.mock_with :rspec do |mocks|
+    mocks.syntax = :expect
+
+    # Fail if an example stubs a method the real object does not have. Without
+    # this, a rename in lib/ leaves the specs stubbing a method that no longer
+    # exists and passing while the driver is broken.
+    mocks.verify_partial_doubles = true
+  end
+
+  # No `describe` at the top level and no `should` on every object: the specs
+  # use `RSpec.describe` and `expect` explicitly.
+  config.disable_monkey_patching!
+
+  # Surface deprecations as failures rather than warnings that scroll past.
+  config.raise_errors_for_deprecations!
+
+  config.shared_context_metadata_behavior = :apply_to_host_groups
+
+  config.include VroDoubles
+
+  config.filter_run_when_matching :focus
+
+  # Run specs in random order to surface order dependencies. If you find an
+  # order dependency and want to debug it, you can fix the order by providing
+  # the seed, which is printed after each run.
+  #     --seed 1234
+  config.order = "random"
+  Kernel.srand config.seed
+end
diff --git a/spec/support/vro_doubles.rb b/spec/support/vro_doubles.rb
new file mode 100644
index 0000000..7010754
--- /dev/null
+++ b/spec/support/vro_doubles.rb
@@ -0,0 +1,40 @@
+#
+# License:: Apache License, Version 2.0
+#
+# Verifying doubles for the small slice of the vcoworkflows API that this
+# driver uses. Building them here keeps the specs honest -- if vcoworkflows
+# renames `token` or drops `output_parameters`, every example that leans on
+# these helpers fails loudly instead of passing against a stale stub.
+#
+module VroDoubles
+  # @param value [Object] the value vRO returned for the parameter
+  # @return [VcoWorkflows::WorkflowParameter] a verifying double
+  def workflow_parameter(value)
+    instance_double(VcoWorkflows::WorkflowParameter, value: value)
+  end
+
+  # @param output [Hash{String => Object}] output parameter names to raw values
+  # @return [Hash{String => VcoWorkflows::WorkflowParameter}]
+  def workflow_output(output)
+    output.transform_values { |value| workflow_parameter(value) }
+  end
+
+  # @param state [String] the workflow run state vRO reports
+  # @param alive [Boolean, Array<Boolean>] whether the run is still going; an
+  #   array is returned one element per call, so a poll loop can be driven
+  # @param output [Hash{String => Object}] output parameter names to raw values
+  # @return [VcoWorkflows::WorkflowToken] a verifying double
+  def workflow_token(state: "completed", alive: false, output: {})
+    token = instance_double(VcoWorkflows::WorkflowToken,
+      state: state,
+      output_parameters: workflow_output(output))
+    allow(token).to receive(:alive?).and_return(*Array(alive))
+    token
+  end
+
+  # @param token [VcoWorkflows::WorkflowToken] the token the client hands back
+  # @return [VcoWorkflows::Workflow] a verifying double
+  def workflow_client(token: workflow_token)
+    instance_double(VcoWorkflows::Workflow, execute: nil, parameter: nil, token: token)
+  end
+end
diff --git a/spec/vro_spec.rb b/spec/vro_spec.rb
index ba3546b..4388c08 100644
--- a/spec/vro_spec.rb
+++ b/spec/vro_spec.rb
@@ -1,5 +1,3 @@
-# Encoding: UTF-8
-
 #
 # Authors:: Chef Partner Engineering (<partnereng@chef.io>)
 # Copyright:: Copyright (c) 2015 Chef Software, Inc.
@@ -17,77 +15,76 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-require 'spec_helper'
-require 'kitchen/driver/vro'
-require 'kitchen/provisioner/dummy'
-require 'kitchen/transport/dummy'
-require 'kitchen/verifier/dummy'
-
-shared_examples 'output parameters that are missing a required key' do
-  it 'raises an exception' do
+RSpec.shared_examples "output parameters that are missing a required key" do
+  it "raises an exception" do
     allow(driver).to receive(:output_parameters).and_return(output_parameters)
+
     expect { driver.validate_create_output_parameters! }.to raise_error(
       RuntimeError,
-      'The workflow output did not contain a server_id and ip_address parameter.'
+      "The workflow output did not contain a server_id and ip_address parameter."
     )
   end
 end
 
-describe Kitchen::Driver::Vro do
+RSpec.describe Kitchen::Driver::Vro do
   let(:logged_output) { StringIO.new }
   let(:logger)        { Logger.new(logged_output) }
-  let(:platform)      { Kitchen::Platform.new(name: 'fake_platform') }
+  let(:platform)      { Kitchen::Platform.new(name: "fake_platform") }
   let(:transport)     { Kitchen::Transport::Dummy.new }
-  let(:driver)        { Kitchen::Driver::Vro.new(config) }
+  let(:driver)        { described_class.new(config) }
 
   let(:config) do
     {
-      vro_username:          'myuser',
-      vro_password:          'mypassword',
-      vro_base_url:          'https://vra.corp.local:8281',
-      create_workflow_name:  'Create Workflow',
-      create_workflow_id:    'workflow-1',
-      destroy_workflow_name: 'Destroy Workflow',
-      destroy_workflow_id:   'workflow-2'
+      vro_username: "myuser",
+      vro_password: "mypassword",
+      vro_base_url: "https://vra.corp.local:8281",
+      create_workflow_name: "Create Workflow",
+      create_workflow_id: "workflow-1",
+      destroy_workflow_name: "Destroy Workflow",
+      destroy_workflow_id: "workflow-2",
     }
   end
 
   let(:instance) do
     instance_double(Kitchen::Instance,
-                    logger:    logger,
-                    transport: transport,
-                    platform:  platform,
-                    to_str:    'instance_str')
+      logger: logger,
+      transport: transport,
+      platform: platform,
+      to_str: "instance_str")
   end
 
   before do
     allow(driver).to receive(:instance).and_return(instance)
   end
 
-  describe 'plugin metadata' do
-    it 'is a Test Kitchen driver' do
+  describe "plugin metadata" do
+    it "is a Test Kitchen driver" do
       expect(driver).to be_a(Kitchen::Driver::Base)
     end
 
-    it 'shows up as vRO in kitchen list' do
-      expect(driver.name).to eq('vRO')
+    it "shows up as vRO in kitchen list" do
+      expect(driver.name).to eq("vRO")
     end
 
-    it 'reports its own gem version to kitchen diagnose' do
+    it "reports its own gem version to kitchen diagnose" do
       expect(driver.diagnose_plugin[:version]).to eq(Kitchen::Driver::VRO_VERSION)
     end
+
+    it "speaks version 2 of the driver API" do
+      expect(driver.diagnose_plugin[:api_version]).to eq(2)
+    end
   end
 
-  describe 'configuration' do
-    it 'verifies TLS by default' do
-      expect(driver[:vro_disable_ssl_verify]).to eq(false)
+  describe "configuration" do
+    it "verifies TLS by default" do
+      expect(driver[:vro_disable_ssl_verify]).to be false
     end
 
-    it 'defaults to a 300 second workflow timeout' do
+    it "defaults to a 300 second workflow timeout" do
       expect(driver[:request_timeout]).to eq(300)
     end
 
-    it 'defaults both workflow ids to nil so the name is used alone' do
+    it "defaults both workflow ids to nil so the name is used alone" do
       config.delete(:create_workflow_id)
       config.delete(:destroy_workflow_id)
 
@@ -95,13 +92,14 @@
       expect(driver[:destroy_workflow_id]).to be_nil
     end
 
-    it 'defaults both workflow parameter sets to empty' do
+    it "defaults both workflow parameter sets to empty" do
       expect(driver[:create_workflow_parameters]).to eq({})
       expect(driver[:destroy_workflow_parameters]).to eq({})
     end
 
-    %i[vro_username vro_password vro_base_url
-       create_workflow_name destroy_workflow_name].each do |key|
+    required = %i{vro_username vro_password vro_base_url create_workflow_name destroy_workflow_name}
+
+    required.each do |key|
       it "refuses to run without #{key}" do
         config.delete(key)
 
@@ -109,147 +107,253 @@
           .to raise_error(Kitchen::UserError, /#{key}/)
       end
     end
+
+    it "accepts a configuration that supplies every required key" do
+      expect { driver.finalize_config!(instance) }.not_to raise_error
+    end
   end
 
-  describe '#set_workflow_vars' do
-    it 'points the driver at the named workflow' do
-      driver.set_workflow_vars('Some Workflow', 'workflow-9')
+  describe "#vro_config" do
+    it "passes the configured endpoint and credentials through to vcoworkflows" do
+      vro_config = driver.vro_config
+
+      expect(vro_config).to be_a(VcoWorkflows::Config)
+      expect(vro_config.username).to eq("myuser")
+      expect(vro_config.password).to eq("mypassword")
+      expect(vro_config.url).to start_with("https://vra.corp.local:8281")
+    end
+
+    it "asks vcoworkflows to verify TLS by default" do
+      expect(driver.vro_config.verify_ssl).to be true
+    end
+
+    it "asks vcoworkflows to skip TLS verification when it is disabled" do
+      config[:vro_disable_ssl_verify] = true
 
-      expect(driver.workflow_name).to eq('Some Workflow')
-      expect(driver.workflow_id).to eq('workflow-9')
+      expect(driver.vro_config.verify_ssl).to be false
     end
 
-    it 'discards the memoized client so the next call builds a new one' do
-      first = double('first client')
-      allow(VcoWorkflows::Workflow).to receive(:new).and_return(first)
-      allow(driver).to receive(:vro_config).and_return(double('config'))
-      driver.set_workflow_vars('Create Workflow', 'workflow-1')
-      expect(driver.vro_client).to eq(first)
+    it "builds the config only once" do
+      expect(driver.vro_config).to equal(driver.vro_config)
+    end
+  end
 
-      second = double('second client')
-      allow(VcoWorkflows::Workflow).to receive(:new).and_return(second)
-      driver.set_workflow_vars('Destroy Workflow', 'workflow-2')
+  describe "#verify_ssl?" do
+    it "is false when vro_disable_ssl_verify is true" do
+      config[:vro_disable_ssl_verify] = true
 
-      expect(driver.vro_client).to eq(second)
+      expect(driver.verify_ssl?).to be false
+    end
+
+    it "is true when vro_disable_ssl_verify is false" do
+      config[:vro_disable_ssl_verify] = false
+
+      expect(driver.verify_ssl?).to be true
     end
   end
 
-  describe '#workflow_successful?' do
-    let(:vro_client) { double('vro_client') }
+  describe "#vro_client" do
+    let(:vro_config) { instance_double(VcoWorkflows::Config) }
 
     before do
-      allow(driver).to receive(:vro_client).and_return(vro_client)
+      allow(driver).to receive(:vro_config).and_return(vro_config)
     end
 
-    it 'is true when vRO reports the run completed' do
-      allow(vro_client).to receive(:token).and_return(double(state: 'completed'))
+    it "creates a VcoWorkflows::Workflow for the current workflow" do
+      driver.set_workflow_vars("workflow name", "workflow-12345")
 
-      expect(driver.workflow_successful?).to eq(true)
+      expect(VcoWorkflows::Workflow).to receive(:new)
+        .with("workflow name", id: "workflow-12345", config: vro_config)
+
+      driver.vro_client
     end
 
-    it 'is false for any other state' do
-      allow(vro_client).to receive(:token).and_return(double(state: 'failed'))
+    it "builds the client only once" do
+      allow(VcoWorkflows::Workflow).to receive(:new).and_return(workflow_client)
+
+      driver.vro_client
+      driver.vro_client
 
-      expect(driver.workflow_successful?).to eq(false)
+      expect(VcoWorkflows::Workflow).to have_received(:new).once
     end
   end
 
-  describe '#create' do
-    context 'when a server already exists' do
-      let(:state) { { server_id: 'server-12345' } }
+  describe "#set_workflow_vars" do
+    it "points the driver at the named workflow" do
+      driver.set_workflow_vars("Some Workflow", "workflow-9")
+
+      expect(driver.workflow_name).to eq("Some Workflow")
+      expect(driver.workflow_id).to eq("workflow-9")
+    end
 
-      it 'does not create the server' do
-        expect(driver).not_to receive(:execute_create_workflow)
+    it "accepts a nil id for a workflow addressed by name alone" do
+      driver.set_workflow_vars("Some Workflow", nil)
 
-        driver.create(state)
-      end
+      expect(driver.workflow_id).to be_nil
+    end
+
+    it "discards the memoized output parameters so they cannot leak between workflows" do
+      create_token = workflow_token(output: { "server_id" => "server-12345" })
+      allow(driver).to receive(:vro_client).and_return(workflow_client(token: create_token))
+      expect(driver.output_parameters.keys).to eq(["server_id"])
+
+      driver.set_workflow_vars("Destroy Workflow", "workflow-2")
+      allow(driver).to receive(:vro_client).and_return(workflow_client(token: workflow_token))
+
+      expect(driver.output_parameters).to be_empty
     end
 
+    it "discards the memoized client so the next call builds a new one" do
+      first  = workflow_client
+      second = workflow_client
+      allow(driver).to receive(:vro_config).and_return(instance_double(VcoWorkflows::Config))
+      allow(VcoWorkflows::Workflow).to receive(:new).and_return(first, second)
+
+      driver.set_workflow_vars("Create Workflow", "workflow-1")
+      expect(driver.vro_client).to equal(first)
+
+      driver.set_workflow_vars("Destroy Workflow", "workflow-2")
+      expect(driver.vro_client).to equal(second)
+    end
+  end
+
+  describe "#create" do
     let(:state) { {} }
 
-    it 'calls the expected methods' do
-      expect(driver).to receive(:execute_create_workflow).with(state)
-      expect(driver).to receive(:wait_for_server).with(state)
+    it "runs the create workflow and then waits for the server" do
+      expect(driver).to receive(:execute_create_workflow).with(state).ordered
+      expect(driver).to receive(:wait_for_server).with(state).ordered
 
       driver.create(state)
     end
-  end
 
-  describe '#destroy' do
-    context 'when a server does not exist' do
-      let(:state) { {} }
+    it "does nothing when state already names a server" do
+      state[:server_id] = "server-12345"
 
-      it 'does not destroy the server' do
-        expect(driver).not_to receive(:execute_destroy_workflow)
+      expect(driver).not_to receive(:execute_create_workflow)
+      expect(driver).not_to receive(:wait_for_server)
 
-        driver.destroy(state)
+      driver.create(state)
+    end
+
+    context "with the vRO API stubbed end to end" do
+      let(:token) do
+        workflow_token(output: { "server_id" => "server-12345", "ip_address" => "1.2.3.4" })
+      end
+      let(:client)     { workflow_client(token: token) }
+      let(:connection) { transport.connection(state) }
+
+      before do
+        allow(VcoWorkflows::Workflow).to receive(:new).and_return(client)
+        allow(transport).to receive(:connection).and_return(connection)
+        allow(connection).to receive(:wait_until_ready)
+      end
+
+      it "records the server the workflow reported" do
+        driver.create(state)
+
+        expect(state[:server_id]).to eq("server-12345")
+        expect(state[:hostname]).to eq("1.2.3.4")
+      end
+
+      it "waits for the transport to accept a connection" do
+        driver.create(state)
+
+        expect(connection).to have_received(:wait_until_ready)
+      end
+
+      it "asks vRO for the configured create workflow" do
+        driver.create(state)
+
+        expect(VcoWorkflows::Workflow).to have_received(:new)
+          .with("Create Workflow", hash_including(id: "workflow-1"))
+      end
+
+      it "sends the configured create parameters to the workflow" do
+        config[:create_workflow_parameters] = { os_name: "centos", cpus: 4 }
+
+        driver.create(state)
+
+        expect(client).to have_received(:parameter).with("os_name", "centos")
+        expect(client).to have_received(:parameter).with("cpus", 4)
       end
     end
+  end
 
-    let(:state) { { server_id: 'server-12345' } }
+  describe "#destroy" do
+    let(:state) { { server_id: "server-12345", hostname: "1.2.3.4" } }
 
-    it 'calls the expected methods' do
+    it "runs the destroy workflow" do
       expect(driver).to receive(:execute_destroy_workflow).with(state)
+
       driver.destroy(state)
     end
-  end
 
-  describe '#vro_config' do
-    it 'creates a VcoWorkflows::Config object' do
-      expect(VcoWorkflows::Config).to receive(:new).with(url: 'https://vra.corp.local:8281',
-                                                         username: 'myuser',
-                                                         password: 'mypassword',
-                                                         verify_ssl: true)
-      driver.vro_config
+    it "does nothing when state does not name a server" do
+      expect(driver).not_to receive(:execute_destroy_workflow)
+
+      driver.destroy({})
     end
-  end
 
-  describe '#vro_client' do
-    let(:vro_config) { double('vro_config') }
-    it 'creates a VcoWorkflows::Workflow object' do
-      allow(driver).to receive(:workflow_name).and_return('workflow name')
-      allow(driver).to receive(:workflow_id).and_return('workflow-12345')
-      allow(driver).to receive(:vro_config).and_return(vro_config)
+    it "clears the destroyed server out of state" do
+      allow(driver).to receive(:execute_destroy_workflow)
 
-      expect(VcoWorkflows::Workflow).to receive(:new).with('workflow name',
-                                                           id: 'workflow-12345',
-                                                           config: vro_config)
+      driver.destroy(state)
 
-      driver.vro_client
+      expect(state).not_to have_key(:server_id)
+      expect(state).not_to have_key(:hostname)
     end
-  end
 
-  describe '#verify_ssl?' do
-    context 'when vro_disable_ssl_verify is true' do
+    it "leaves state alone when the destroy workflow fails" do
+      allow(driver).to receive(:execute_destroy_workflow).and_raise(RuntimeError)
+
+      expect { driver.destroy(state) }.to raise_error(RuntimeError)
+      expect(state[:server_id]).to eq("server-12345")
+    end
+
+    it "does not run the destroy workflow twice for the same server" do
+      allow(driver).to receive(:execute_destroy_workflow)
+
+      driver.destroy(state)
+      driver.destroy(state)
+
+      expect(driver).to have_received(:execute_destroy_workflow).once
+    end
+
+    context "with the vRO API stubbed end to end" do
+      let(:client) { workflow_client }
+
       before do
-        config[:vro_disable_ssl_verify] = true
+        allow(VcoWorkflows::Workflow).to receive(:new).and_return(client)
       end
 
-      it 'returns false' do
-        expect(driver.verify_ssl?).to eq(false)
+      it "asks vRO for the configured destroy workflow" do
+        driver.destroy(state)
+
+        expect(VcoWorkflows::Workflow).to have_received(:new)
+          .with("Destroy Workflow", hash_including(id: "workflow-2"))
       end
-    end
 
-    context 'when vro_disable_ssl_verify is false' do
-      before do
-        config[:vro_disable_ssl_verify] = false
+      it "hands the stored server id to the destroy workflow" do
+        driver.destroy(state)
+
+        expect(client).to have_received(:parameter).with("server_id", "server-12345")
       end
 
-      it 'returns true' do
-        expect(driver.verify_ssl?).to eq(true)
+      it "sends the configured destroy parameters to the workflow" do
+        config[:destroy_workflow_parameters] = { force: true }
+
+        driver.destroy(state)
+
+        expect(client).to have_received(:parameter).with("force", true)
       end
     end
   end
 
-  describe '#execute_create_workflow' do
-    let(:state)      { {} }
-    let(:server_id)  { double('server_id', value: 'server-12345') }
-    let(:ip_address) { double('ip_address', value: '1.2.3.4') }
+  describe "#execute_create_workflow" do
+    let(:state) { {} }
     let(:output_parameters) do
-      {
-        'server_id'  => server_id,
-        'ip_address' => ip_address
-      }
+      workflow_output("server_id" => "server-12345", "ip_address" => "1.2.3.4")
     end
 
     before do
@@ -262,8 +366,8 @@
       allow(driver).to receive(:output_parameters).and_return(output_parameters)
     end
 
-    it 'calls the expected methods' do
-      expect(driver).to receive(:set_workflow_vars).with('Create Workflow', 'workflow-1')
+    it "selects the create workflow and submits it" do
+      expect(driver).to receive(:set_workflow_vars).with("Create Workflow", "workflow-1")
       expect(driver).to receive(:set_workflow_parameters).with({})
       expect(driver).to receive(:execute_workflow)
       expect(driver).to receive(:wait_for_workflow)
@@ -272,26 +376,42 @@
       driver.execute_create_workflow(state)
     end
 
-    it 'raises an error if the workflow did not complete successfully' do
+    it "raises an error if the workflow did not complete successfully" do
+      allow(driver).to receive(:workflow_successful?).and_return(false)
+
+      expect { driver.execute_create_workflow(state) }.to raise_error(RuntimeError)
+    end
+
+    it "does not validate the output of a workflow that failed" do
       allow(driver).to receive(:workflow_successful?).and_return(false)
+
+      expect(driver).not_to receive(:validate_create_output_parameters!)
+
       expect { driver.execute_create_workflow(state) }.to raise_error(RuntimeError)
     end
 
-    it 'sets the state hash with the proper info' do
+    it "sets the state hash with the proper info" do
       driver.execute_create_workflow(state)
 
-      expect(state[:server_id]).to eq('server-12345')
-      expect(state[:hostname]).to eq('1.2.3.4')
+      expect(state[:server_id]).to eq("server-12345")
+      expect(state[:hostname]).to eq("1.2.3.4")
+    end
+
+    it "leaves state untouched when the output does not validate" do
+      allow(driver).to receive(:validate_create_output_parameters!)
+        .and_raise(RuntimeError, "The server_id parameter was empty.")
+
+      expect { driver.execute_create_workflow(state) }.to raise_error(RuntimeError)
+      expect(state).to be_empty
     end
   end
 
-  describe '#execute_destroy_workflow' do
-    let(:state)      { { server_id: 'server-12345' } }
-    let(:vro_client) { double('vro_client') }
+  describe "#execute_destroy_workflow" do
+    let(:state)      { { server_id: "server-12345" } }
+    let(:vro_client) { workflow_client }
 
     before do
       allow(driver).to receive(:vro_client).and_return(vro_client)
-      allow(vro_client).to receive(:parameter)
       allow(driver).to receive(:set_workflow_vars)
       allow(driver).to receive(:set_workflow_parameters)
       allow(driver).to receive(:execute_workflow)
@@ -299,10 +419,10 @@
       allow(driver).to receive(:workflow_successful?).and_return(true)
     end
 
-    it 'calls the expected methods' do
-      expect(driver).to receive(:set_workflow_vars).with('Destroy Workflow', 'workflow-2')
+    it "selects the destroy workflow and submits it with the server id" do
+      expect(driver).to receive(:set_workflow_vars).with("Destroy Workflow", "workflow-2")
       expect(driver).to receive(:set_workflow_parameters).with({})
-      expect(vro_client).to receive(:parameter).with('server_id', 'server-12345')
+      expect(vro_client).to receive(:parameter).with("server_id", "server-12345")
       expect(driver).to receive(:execute_workflow)
       expect(driver).to receive(:wait_for_workflow)
       expect(driver).to receive(:workflow_successful?)
@@ -310,244 +430,350 @@
       driver.execute_destroy_workflow(state)
     end
 
-    it 'raises an error if the workflow did not complete successfully' do
+    it "raises an error if the workflow did not complete successfully" do
       allow(driver).to receive(:workflow_successful?).and_return(false)
+
       expect { driver.execute_destroy_workflow(state) }.to raise_error(RuntimeError)
     end
   end
 
-  describe '#execute_workflow' do
-    let(:vro_client) { double('vro_client') }
+  describe "#execute_workflow" do
+    let(:vro_client) { workflow_client }
+
     before do
       allow(driver).to receive(:vro_client).and_return(vro_client)
     end
 
-    it 'executes the workflow' do
+    it "executes the workflow" do
       expect(vro_client).to receive(:execute)
+
       driver.execute_workflow
     end
 
-    context 'when execute fails with a RestClient::BadRequest' do
-      it 'prints an error with the HTTP response' do
-        HTTPResponse = Struct.new(:code, :to_s)
-        response = HTTPResponse.new(400, 'an HTTP error occurred')
-        exception = RestClient::BadRequest.new
-        exception.response = response
+    context "when execute fails with a RestClient::BadRequest" do
+      let(:exception) do
+        RestClient::BadRequest.new.tap do |e|
+          e.response = "an HTTP error occurred"
+        end
+      end
+
+      it "logs the HTTP response body, which is where vRO puts the reason" do
         allow(vro_client).to receive(:execute).and_raise(exception)
-        expect(driver).to receive(:error).with('The workflow execution request failed: an HTTP error occurred')
+
+        expect(driver).to receive(:error)
+          .with("The workflow execution request failed: an HTTP error occurred")
+
         expect { driver.execute_workflow }.to raise_error(RestClient::BadRequest)
       end
     end
 
-    context 'when execute fails with any other exception' do
-      it 'prints an error with the exception message' do
-        allow(vro_client).to receive(:execute).and_raise(RuntimeError, 'a non-HTTP error occurred')
-        expect(driver).to receive(:error).with('The workflow execution request failed: a non-HTTP error occurred')
-        expect { driver.execute_workflow }.to raise_error(RuntimeError)
+    context "when execute fails with any other exception" do
+      it "logs the exception message and re-raises" do
+        allow(vro_client).to receive(:execute)
+          .and_raise(RuntimeError, "a non-HTTP error occurred")
+
+        expect(driver).to receive(:error)
+          .with("The workflow execution request failed: a non-HTTP error occurred")
+
+        expect { driver.execute_workflow }.to raise_error(RuntimeError, "a non-HTTP error occurred")
       end
     end
   end
 
-  describe '#wait_for_workflow' do
-    let(:vro_client) { double('vro_client') }
-    let(:token)      { double('token') }
+  describe "#wait_for_workflow" do
+    let(:token)      { workflow_token(alive: [true, true, false]) }
+    let(:vro_client) { workflow_client(token: token) }
 
     before do
       allow(driver).to receive(:vro_client).and_return(vro_client)
-      allow(vro_client).to receive(:token).and_return(token)
 
-      # don't actually sleep
+      # Don't actually sleep between polls.
       allow(driver).to receive(:sleep)
     end
 
-    context 'when the requests completes normally, 3 loops' do
-      it 'only fetches the token 3 times' do
-        allow(token).to receive(:alive?).exactly(3).times.and_return(true, true, false)
-        expect(vro_client).to receive(:token).exactly(3).times
+    it "polls until the token is no longer alive" do
+      driver.wait_for_workflow
 
-        driver.wait_for_workflow
-      end
+      expect(token).to have_received(:alive?).exactly(3).times
     end
 
-    context 'when the request is completed on the first loop' do
-      it 'only refreshes the request 1 time' do
-        expect(token).to receive(:alive?).once.and_return(false)
-        driver.wait_for_workflow
-      end
+    it "re-reads the token on every poll rather than caching it" do
+      driver.wait_for_workflow
+
+      expect(vro_client).to have_received(:token).exactly(3).times
     end
 
-    context 'when the timeout is exceeded' do
-      it 'raises a Timeout exception' do
+    it "waits two seconds between polls" do
+      driver.wait_for_workflow
+
+      expect(driver).to have_received(:sleep).with(2).twice
+    end
+
+    it "does not sleep when the workflow is already finished on the first poll" do
+      allow(token).to receive(:alive?).and_return(false)
+
+      driver.wait_for_workflow
+
+      expect(driver).not_to have_received(:sleep)
+    end
+
+    context "when the timeout is exceeded" do
+      it "raises a Timeout error naming the configured timeout" do
         allow(Timeout).to receive(:timeout).and_raise(Timeout::Error)
+
         expect { driver.wait_for_workflow }.to raise_error(
-          Timeout::Error, 'Workflow did not complete in 300 seconds. ' \
-          'Please check the vRO UI for more information.'
+          Timeout::Error, "Workflow did not complete in 300 seconds. " \
+          "Please check the vRO UI for more information."
         )
       end
-    end
 
-    context 'when a non-timeout exception is raised' do
-      it 'raises the original exception' do
-        allow(vro_client).to receive(:token).and_raise(RuntimeError, 'an error occurred')
-        expect { driver.wait_for_workflow }.to raise_error(RuntimeError, 'an error occurred')
+      it "reports a non-default request_timeout" do
+        config[:request_timeout] = 30
+        allow(Timeout).to receive(:timeout).and_raise(Timeout::Error)
+
+        expect { driver.wait_for_workflow }
+          .to raise_error(Timeout::Error, /did not complete in 30 seconds/)
+      end
+
+      it "actually gives up once the configured timeout elapses" do
+        config[:request_timeout] = 0.05
+        allow(token).to receive(:alive?).and_return(true)
+        allow(driver).to receive(:sleep) { Kernel.sleep(0.01) }
+
+        expect { driver.wait_for_workflow }.to raise_error(Timeout::Error)
       end
     end
+
+    it "lets a non-timeout exception through untouched" do
+      allow(vro_client).to receive(:token).and_raise(RuntimeError, "an error occurred")
+
+      expect { driver.wait_for_workflow }.to raise_error(RuntimeError, "an error occurred")
+    end
   end
 
-  describe '#wait_for_server' do
-    let(:connection) { instance.transport.connection(state) }
-    let(:state)      { { hostname: 'host1', server_id: 'server-12345' } }
+  describe "#wait_for_server" do
+    let(:state)      { { hostname: "host1", server_id: "server-12345" } }
+    let(:connection) { transport.connection(state) }
 
     before do
       allow(transport).to receive(:connection).and_return(connection)
     end
 
-    it 'calls wait_until_ready on the transport connection' do
+    it "calls wait_until_ready on the transport connection" do
       expect(connection).to receive(:wait_until_ready)
+
       driver.wait_for_server(state)
     end
 
-    it 'destroys the server if the server failed to become ready' do
+    it "destroys the server if the server failed to become ready" do
       allow(connection).to receive(:wait_until_ready).and_raise(RuntimeError)
+
       expect(driver).to receive(:destroy).with(state)
+
       expect { driver.wait_for_server(state) }.to raise_error(RuntimeError)
     end
+
+    it "re-raises the original connection failure" do
+      allow(connection).to receive(:wait_until_ready)
+        .and_raise(Kitchen::Transport::TransportFailed, "connection refused")
+      allow(driver).to receive(:destroy)
+
+      expect { driver.wait_for_server(state) }
+        .to raise_error(Kitchen::Transport::TransportFailed, "connection refused")
+    end
+
+    context "when the cleanup destroy also fails" do
+      before do
+        allow(connection).to receive(:wait_until_ready)
+          .and_raise(Kitchen::Transport::TransportFailed, "connection refused")
+        allow(driver).to receive(:destroy).and_raise(RuntimeError, "vRO rejected the request")
+      end
+
+      it "still surfaces the connection failure, not the cleanup failure" do
+        allow(driver).to receive(:error)
+
+        expect { driver.wait_for_server(state) }
+          .to raise_error(Kitchen::Transport::TransportFailed, "connection refused")
+      end
+
+      it "warns that the server may have been left behind" do
+        expect(driver).to receive(:error).with(/not reachable/)
+        expect(driver).to receive(:error).with(/destroy-server workflow also failed/)
+
+        expect { driver.wait_for_server(state) }
+          .to raise_error(Kitchen::Transport::TransportFailed)
+      end
+    end
   end
 
-  describe '#set_workflow_parameters' do
-    let(:vro_client) { double('vro_client') }
-    let(:params)     { { key1: 'value1', key2: 'value2' } }
+  describe "#set_workflow_parameters" do
+    let(:vro_client) { workflow_client }
 
-    it 'sets parameters on the client' do
+    before do
       allow(driver).to receive(:vro_client).and_return(vro_client)
-      expect(vro_client).to receive(:parameter).with('key1', 'value1')
-      expect(vro_client).to receive(:parameter).with('key2', 'value2')
+    end
+
+    it "sets parameters on the client" do
+      expect(vro_client).to receive(:parameter).with("key1", "value1")
+      expect(vro_client).to receive(:parameter).with("key2", "value2")
+
+      driver.set_workflow_parameters(key1: "value1", key2: "value2")
+    end
+
+    it "stringifies symbol keys, which is how they arrive from kitchen.yml" do
+      expect(vro_client).to receive(:parameter).with("os_version", "9")
 
-      driver.set_workflow_parameters(params)
+      driver.set_workflow_parameters(os_version: "9")
+    end
+
+    it "passes non-string values through untouched" do
+      expect(vro_client).to receive(:parameter).with("cpus", 4)
+      expect(vro_client).to receive(:parameter).with("ha", true)
+
+      driver.set_workflow_parameters("cpus" => 4, "ha" => true)
+    end
+
+    it "does nothing when there are no parameters" do
+      expect(vro_client).not_to receive(:parameter)
+
+      driver.set_workflow_parameters({})
     end
   end
 
-  describe '#output_parameters' do
-    let(:vro_client) { double('vro_client') }
-    let(:token)      { double('token') }
-    let(:output)     { double('output_parameters') }
+  describe "#output_parameters" do
+    let(:token)      { workflow_token(output: { "server_id" => "server-12345" }) }
+    let(:vro_client) { workflow_client(token: token) }
 
-    it 'returns the output parameters from the workflow token' do
+    before do
       allow(driver).to receive(:vro_client).and_return(vro_client)
-      allow(vro_client).to receive(:token).and_return(token)
-      expect(token).to receive(:output_parameters).and_return(output)
-      expect(driver.output_parameters).to eq(output)
     end
-  end
 
-  describe '#output_parameter_value' do
-    let(:test_key) { double('test_key', value: 'test_value') }
-    let(:output_parameters) { { 'test_key' => test_key } }
-    it 'returns the correct value' do
-      allow(driver).to receive(:output_parameters).and_return(output_parameters)
-      expect(driver.output_parameter_value('test_key')).to eq('test_value')
+    it "returns the output parameters from the workflow token" do
+      expect(driver.output_parameters.keys).to eq(["server_id"])
+    end
+
+    it "reads them from the token only once" do
+      driver.output_parameters
+      driver.output_parameters
+
+      expect(token).to have_received(:output_parameters).once
     end
   end
 
-  describe '#output_parameter_empty?' do
-    context 'when the value is not nil or empty' do
-      it 'returns false' do
-        allow(driver).to receive(:output_parameter_value).with('test_key').and_return('test_value')
-        expect(driver.output_parameter_empty?('test_key')).to eq(false)
-      end
+  describe "#output_parameter_value" do
+    it "returns the value vRO reported" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("test_key" => "test_value"))
+
+      expect(driver.output_parameter_value("test_key")).to eq("test_value")
     end
 
-    context 'when the value is nil' do
-      it 'returns true' do
-        allow(driver).to receive(:output_parameter_value).with('test_key').and_return(nil)
-        expect(driver.output_parameter_empty?('test_key')).to eq(true)
-      end
+    it "stringifies a value that vRO returned as a number" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("server_id" => 12_345))
+
+      expect(driver.output_parameter_value("server_id")).to eq("12345")
     end
 
-    context 'when the value is empty' do
-      it 'returns true' do
-        allow(driver).to receive(:output_parameter_value).with('test_key').and_return('')
-        expect(driver.output_parameter_empty?('test_key')).to eq(true)
-      end
+    it "returns an empty string for a parameter vRO left unset" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("ip_address" => nil))
+
+      expect(driver.output_parameter_value("ip_address")).to eq("")
+    end
+
+    it "returns an empty string for a parameter the workflow never returned" do
+      allow(driver).to receive(:output_parameters).and_return({})
+
+      expect(driver.output_parameter_value("ip_address")).to eq("")
     end
   end
 
-  describe '#validate_create_output_parameters!' do
-    let(:server_id)   { double('server_id', value: 'server-12345') }
-    let(:ip_address)  { double('ip_address', value: '1.2.3.4') }
+  describe "#output_parameter_empty?" do
+    it "is false when the value is neither nil nor empty" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("test_key" => "test_value"))
+
+      expect(driver.output_parameter_empty?("test_key")).to be false
+    end
+
+    it "is true when vRO returned the parameter unset" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("test_key" => nil))
+
+      expect(driver.output_parameter_empty?("test_key")).to be true
+    end
 
-    context 'when the output parameters do not include server_id and ip_address' do
+    it "is true when the value is empty" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("test_key" => ""))
+
+      expect(driver.output_parameter_empty?("test_key")).to be true
+    end
+
+    it "is true when the workflow did not return the parameter at all" do
+      allow(driver).to receive(:output_parameters).and_return({})
+
+      expect(driver.output_parameter_empty?("test_key")).to be true
+    end
+  end
+
+  describe "#validate_create_output_parameters!" do
+    context "when the output parameters do not include server_id and ip_address" do
       let(:output_parameters) { {} }
-      it_behaves_like 'output parameters that are missing a required key'
+
+      it_behaves_like "output parameters that are missing a required key"
     end
 
-    context 'when the output parameters do not include server_id' do
-      let(:output_parameters) { { 'ip_address' => ip_address } }
-      it_behaves_like 'output parameters that are missing a required key'
+    context "when the output parameters do not include server_id" do
+      let(:output_parameters) { workflow_output("ip_address" => "1.2.3.4") }
+
+      it_behaves_like "output parameters that are missing a required key"
     end
 
-    context 'when the output parameters do not include ip_address' do
-      let(:output_parameters) { { 'server_id' => server_id } }
-      it_behaves_like 'output parameters that are missing a required key'
+    context "when the output parameters do not include ip_address" do
+      let(:output_parameters) { workflow_output("server_id" => "server-12345") }
+
+      it_behaves_like "output parameters that are missing a required key"
     end
 
-    context 'when server_id is empty' do
-      let(:server_id) { double('server_id', value: '') }
-      let(:output_parameters) do
-        {
-          'server_id'  => server_id,
-          'ip_address' => ip_address
-        }
-      end
+    it "raises when server_id is empty" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("server_id" => "", "ip_address" => "1.2.3.4"))
 
-      it 'raises an exception' do
-        allow(driver).to receive(:output_parameters).and_return(output_parameters)
-        expect { driver.validate_create_output_parameters! }.to raise_error(
-          RuntimeError,
-          'The server_id parameter was empty.'
-        )
-      end
+      expect { driver.validate_create_output_parameters! }
+        .to raise_error(RuntimeError, "The server_id parameter was empty.")
     end
 
-    context 'when ip_address is empty' do
-      let(:ip_address) { double('ip_address', value: '') }
-      let(:output_parameters) do
-        {
-          'server_id'  => server_id,
-          'ip_address' => ip_address
-        }
-      end
+    it "raises when ip_address is empty" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("server_id" => "server-12345", "ip_address" => ""))
 
-      it 'raises an exception' do
-        allow(driver).to receive(:output_parameters).and_return(output_parameters)
-        expect { driver.validate_create_output_parameters! }.to raise_error(
-          RuntimeError,
-          'The ip_address parameter was empty.'
-        )
-      end
+      expect { driver.validate_create_output_parameters! }
+        .to raise_error(RuntimeError, "The ip_address parameter was empty.")
     end
-  end
 
-  describe '#workflow_successful?' do
-    let(:vro_client) { double('vro_client') }
-    let(:token)      { double('token') }
-    before do
-      allow(driver).to receive(:vro_client).and_return(vro_client)
-      allow(vro_client).to receive(:token).and_return(token)
+    it "accepts output that names both a server and an address" do
+      allow(driver).to receive(:output_parameters)
+        .and_return(workflow_output("server_id" => "server-12345", "ip_address" => "1.2.3.4"))
+
+      expect { driver.validate_create_output_parameters! }.not_to raise_error
     end
+  end
 
-    context 'when the state is completed' do
-      it 'returns true' do
-        allow(token).to receive(:state).and_return('completed')
-        expect(driver.workflow_successful?).to eq(true)
-      end
+  describe "#workflow_successful?" do
+    it "is true when vRO reports the run completed" do
+      allow(driver).to receive(:vro_client)
+        .and_return(workflow_client(token: workflow_token(state: "completed")))
+
+      expect(driver.workflow_successful?).to be true
     end
 
-    context 'when the state is failed' do
-      it 'returns true' do
-        allow(token).to receive(:state).and_return('failed')
-        expect(driver.workflow_successful?).to eq(false)
-      end
+    it "is false for any other state" do
+      allow(driver).to receive(:vro_client)
+        .and_return(workflow_client(token: workflow_token(state: "failed")))
+
+      expect(driver.workflow_successful?).to be false
     end
   end
 end