spec/unit/plugins/mcollective/connector/rabbitmq_spec.rb

   1 #!/usr/bin/env rspec
   2
   3 require 'spec_helper'
   4
   5 MCollective::PluginManager.clear
   6
   7 require File.dirname(__FILE__) + '/../../../../../plugins/mcollective/connector/rabbitmq.rb'
   8
   9 # create the stomp error class here as it does not always exist
  10 # all versions of the stomp gem and we do not want to tie tests
  11 # to the stomp gem
  12 module Stomp
  13   module Error
  14     class DuplicateSubscription < RuntimeError; end
  15   end
  16 end
  17
  18 module MCollective
  19   module Connector
  20     describe Rabbitmq do
  21       before do
  22         unless ::Stomp::Error.constants.map{|c| c.to_s}.include?("NoCurrentConnection")
  23           class ::Stomp::Error::NoCurrentConnection < RuntimeError ; end
  24         end
  25
  26         @config = mock
  27         @config.stubs(:configured).returns(true)
  28         @config.stubs(:identity).returns("rspec")
  29         @config.stubs(:collectives).returns(["mcollective"])
  30
  31         logger = mock
  32         logger.stubs(:log)
  33         logger.stubs(:start)
  34         Log.configure(logger)
  35
  36         Config.stubs(:instance).returns(@config)
  37
  38         @msg = mock
  39         @msg.stubs(:base64_encode!)
  40         @msg.stubs(:payload).returns("msg")
  41         @msg.stubs(:agent).returns("agent")
  42         @msg.stubs(:type).returns(:reply)
  43         @msg.stubs(:collective).returns("mcollective")
  44
  45         @subscription = mock
  46         @subscription.stubs("<<").returns(true)
  47         @subscription.stubs("include?").returns(false)
  48         @subscription.stubs("delete").returns(false)
  49
  50         @connection = mock
  51         @connection.stubs(:subscribe).returns(true)
  52         @connection.stubs(:unsubscribe).returns(true)
  53
  54         @c = Rabbitmq.new
  55         @c.instance_variable_set("@subscriptions", @subscription)
  56         @c.instance_variable_set("@connection", @connection)
  57       end
  58
  59       describe "#initialize" do
  60         it "should set the @config variable" do
  61           c = Rabbitmq.new
  62           c.instance_variable_get("@config").should == @config
  63         end
  64
  65         it "should set @subscriptions to an empty list" do
  66           c = Rabbitmq.new
  67           c.instance_variable_get("@subscriptions").should == []
  68         end
  69       end
  70
  71       describe "#connect" do
  72         it "should not try to reconnect if already connected" do
  73           Log.expects(:debug).with("Already connection, not re-initializing connection").once
  74           @c.connect
  75         end
  76
  77         it "should support new style config" do
  78           pluginconf = {"rabbitmq.pool.size" => "2",
  79                         "rabbitmq.pool.1.host" => "host1",
  80                         "rabbitmq.pool.1.port" => "6163",
  81                         "rabbitmq.pool.1.user" => "user1",
  82                         "rabbitmq.pool.1.password" => "password1",
  83                         "rabbitmq.pool.1.ssl" => "false",
  84                         "rabbitmq.pool.2.host" => "host2",
  85                         "rabbitmq.pool.2.port" => "6164",
  86                         "rabbitmq.pool.2.user" => "user2",
  87                         "rabbitmq.pool.2.password" => "password2",
  88                         "rabbitmq.pool.2.ssl" => "true",
  89                         "rabbitmq.pool.2.ssl.fallback" => "true",
  90                         "rabbitmq.initial_reconnect_delay" => "0.02",
  91                         "rabbitmq.max_reconnect_delay" => "40",
  92                         "rabbitmq.use_exponential_back_off" => "false",
  93                         "rabbitmq.back_off_multiplier" => "3",
  94                         "rabbitmq.max_reconnect_attempts" => "5",
  95                         "rabbitmq.randomize" => "true",
  96                         "rabbitmq.backup" => "true",
  97                         "rabbitmq.timeout" => "1",
  98                         "rabbitmq.vhost" => "mcollective",
  99                         "rabbitmq.connect_timeout" => "5"}
 100
 101
 102           ENV.delete("STOMP_USER")
 103           ENV.delete("STOMP_PASSWORD")
 104
 105           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 106
 107           Rabbitmq::EventLogger.expects(:new).returns("logger")
 108
 109           connector = mock
 110           connector.expects(:new).with(:backup => true,
 111                                        :back_off_multiplier => 3,
 112                                        :max_reconnect_delay => 40.0,
 113                                        :timeout => 1,
 114                                        :connect_timeout => 5,
 115                                        :use_exponential_back_off => false,
 116                                        :max_reconnect_attempts => 5,
 117                                        :initial_reconnect_delay => 0.02,
 118                                        :randomize => true,
 119                                        :reliable => true,
 120                                        :logger => "logger",
 121                                        :connect_headers => {'accept-version' => '1.0', 'host' => 'mcollective'},
 122                                        :hosts => [{:passcode => 'password1',
 123                                                    :host => 'host1',
 124                                                    :port => 6163,
 125                                                    :ssl => false,
 126                                                    :login => 'user1'},
 127                                                   {:passcode => 'password2',
 128                                                    :host => 'host2',
 129                                                    :port => 6164,
 130                                                    :ssl => true,
 131                                                    :login => 'user2'}
 132           ])
 133
 134           @c.expects(:ssl_parameters).with(2, true).returns(true)
 135
 136           @c.instance_variable_set("@connection", nil)
 137           @c.connect(connector)
 138         end
 139       end
 140
 141       describe "#ssl_paramaters" do
 142         it "should ensure all settings are provided" do
 143           pluginconf = {"rabbitmq.pool.1.host" => "host1",
 144                         "rabbitmq.pool.1.port" => "6164",
 145                         "rabbitmq.pool.1.user" => "user1",
 146                         "rabbitmq.pool.1.password" => "password1",
 147                         "rabbitmq.pool.1.ssl" => "true",
 148                         "rabbitmq.pool.1.ssl.cert" => "rspec"}
 149
 150           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 151
 152           expect { @c.ssl_parameters(1, false) }.to raise_error("cert, key and ca has to be supplied for verified SSL mode")
 153         end
 154
 155         it "should verify the ssl files exist" do
 156           pluginconf = {"rabbitmq.pool.1.host" => "host1",
 157                         "rabbitmq.pool.1.port" => "6164",
 158                         "rabbitmq.pool.1.user" => "user1",
 159                         "rabbitmq.pool.1.password" => "password1",
 160                         "rabbitmq.pool.1.ssl" => "true",
 161                         "rabbitmq.pool.1.ssl.cert" => "rspec.cert",
 162                         "rabbitmq.pool.1.ssl.key" => "rspec.key",
 163                         "rabbitmq.pool.1.ssl.ca" => "rspec1.ca,rspec2.ca"}
 164
 165           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 166
 167           File.expects(:exist?).with("rspec.cert").twice.returns(true)
 168           File.expects(:exist?).with("rspec.key").twice.returns(true)
 169           File.expects(:exist?).with("rspec1.ca").twice.returns(true)
 170           File.expects(:exist?).with("rspec2.ca").twice.returns(false)
 171
 172           expect { @c.ssl_parameters(1, false) }.to raise_error("Cannot find CA file rspec2.ca")
 173
 174           @c.ssl_parameters(1, true).should == true
 175         end
 176
 177         it "should support fallback mode when there are errors" do
 178           pluginconf = {"rabbitmq.pool.1.host" => "host1",
 179                         "rabbitmq.pool.1.port" => "6164",
 180                         "rabbitmq.pool.1.user" => "user1",
 181                         "rabbitmq.pool.1.password" => "password1",
 182                         "rabbitmq.pool.1.ssl" => "true"}
 183
 184           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 185
 186           @c.ssl_parameters(1, true).should == true
 187         end
 188
 189         it "should fail if fallback isnt enabled" do
 190           pluginconf = {"rabbitmq.pool.1.host" => "host1",
 191                         "rabbitmq.pool.1.port" => "6164",
 192                         "rabbitmq.pool.1.user" => "user1",
 193                         "rabbitmq.pool.1.password" => "password1",
 194                         "rabbitmq.pool.1.ssl" => "true"}
 195
 196           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 197
 198           expect { @c.ssl_parameters(1, false) }.to raise_error
 199         end
 200       end
 201
 202       describe "#receive" do
 203         it "should receive from the middleware" do
 204           payload = mock
 205           payload.stubs(:body).returns("msg")
 206           payload.stubs(:headers).returns("headers")
 207
 208           @connection.expects(:receive).returns(payload)
 209
 210           Message.expects(:new).with("msg", payload, :base64 => true, :headers => "headers").returns("message")
 211           @c.instance_variable_set("@base64", true)
 212
 213           received = @c.receive
 214           received.should == "message"
 215         end
 216
 217         it "should sleep and retry if recieving while disconnected" do
 218           payload = mock
 219           payload.stubs(:body).returns("msg")
 220           payload.stubs(:headers).returns("headers")
 221
 222           Message.stubs(:new).returns("rspec")
 223           @connection.expects(:receive).raises(::Stomp::Error::NoCurrentConnection).returns(payload).twice
 224           @c.expects(:sleep).with(1)
 225
 226           @c.receive.should == "rspec"
 227         end
 228       end
 229
 230       describe "#publish" do
 231         before do
 232           @connection.stubs(:publish).with("test", "msg", {}).returns(true)
 233         end
 234
 235         it "should base64 encode a message if configured to do so" do
 236           @c.instance_variable_set("@base64", true)
 237           @c.expects(:target_for).returns({:name => "test", :headers => {}})
 238           @connection.expects(:publish).with("test", "msg", {})
 239           @msg.expects(:base64_encode!)
 240
 241           @c.publish(@msg)
 242         end
 243
 244         it "should not base64 encode if not configured to do so" do
 245           @c.instance_variable_set("@base64", false)
 246           @c.expects(:target_for).returns({:name => "test", :headers => {}})
 247           @connection.expects(:publish).with("test", "msg", {})
 248           @msg.expects(:base64_encode!).never
 249
 250           @c.publish(@msg)
 251         end
 252
 253         it "should publish the correct message to the correct target with msgheaders" do
 254           @connection.expects(:publish).with("test", "msg", {}).once
 255           @c.expects(:target_for).returns({:name => "test", :headers => {}})
 256
 257           @c.publish(@msg)
 258         end
 259
 260         it "should publish direct messages based on discovered_hosts" do
 261           msg = mock
 262           msg.stubs(:base64_encode!)
 263           msg.stubs(:payload).returns("msg")
 264           msg.stubs(:agent).returns("agent")
 265           msg.stubs(:collective).returns("mcollective")
 266           msg.stubs(:type).returns(:direct_request)
 267           msg.expects(:discovered_hosts).returns(["one", "two"])
 268
 269           @connection.expects(:publish).with('/exchange/mcollective_directed/one', 'msg', {'reply-to' => '/temp-queue/mcollective_reply_agent'})
 270           @connection.expects(:publish).with('/exchange/mcollective_directed/two', 'msg', {'reply-to' => '/temp-queue/mcollective_reply_agent'})
 271
 272           @c.publish(msg)
 273         end
 274       end
 275
 276       describe "#subscribe" do
 277         it "should handle duplicate subscription errors" do
 278           @connection.expects(:subscribe).raises(::Stomp::Error::DuplicateSubscription)
 279           Log.expects(:error).with(regexp_matches(/already had a matching subscription, ignoring/))
 280           @c.subscribe("test", :broadcast, "mcollective")
 281         end
 282
 283         it "should use the make_target correctly" do
 284           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}})
 285           @c.subscribe("test", :broadcast, "mcollective")
 286         end
 287
 288         it "should check for existing subscriptions" do
 289           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 290           @subscription.expects("include?").with("rspec").returns(false)
 291           @connection.expects(:subscribe).never
 292
 293           @c.subscribe("test", :broadcast, "mcollective")
 294         end
 295
 296         it "should subscribe to the middleware" do
 297           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 298           @connection.expects(:subscribe).with("test", {}, "rspec")
 299           @c.subscribe("test", :broadcast, "mcollective")
 300         end
 301
 302         it "should add to the list of subscriptions" do
 303           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 304           @subscription.expects("<<").with("rspec")
 305           @c.subscribe("test", :broadcast, "mcollective")
 306         end
 307       end
 308
 309       describe "#unsubscribe" do
 310         it "should use make_target correctly" do
 311           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}})
 312           @c.unsubscribe("test", :broadcast, "mcollective")
 313         end
 314
 315         it "should unsubscribe from the target" do
 316           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 317           @connection.expects(:unsubscribe).with("test", {}, "rspec").once
 318
 319           @c.unsubscribe("test", :broadcast, "mcollective")
 320         end
 321
 322         it "should delete the source from subscriptions" do
 323           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 324           @subscription.expects(:delete).with("rspec").once
 325
 326           @c.unsubscribe("test", :broadcast, "mcollective")
 327         end
 328       end
 329
 330       describe "#target_for" do
 331         it "should create reply targets based on reply-to headers in requests" do
 332           message = mock
 333           message.expects(:type).returns(:reply)
 334
 335           request = mock
 336           request.expects(:headers).returns({"reply-to" => "foo"})
 337
 338           message.expects(:request).returns(request)
 339
 340           @c.target_for(message).should == {:name => "foo", :headers => {}, :id => ""}
 341         end
 342
 343         it "should create new request targets" do
 344           message = mock
 345           message.expects(:type).returns(:request).times(3)
 346           message.expects(:agent).returns("rspecagent")
 347           message.expects(:collective).returns("mcollective")
 348
 349           @c.expects(:make_target).with("rspecagent", :request, "mcollective", nil)
 350           @c.target_for(message)
 351         end
 352
 353         it "should support direct requests" do
 354           message = mock
 355           message.expects(:type).returns(:direct_request).times(3)
 356           message.expects(:agent).returns("rspecagent")
 357           message.expects(:collective).returns("mcollective")
 358
 359           @c.expects(:make_target).with("rspecagent", :direct_request, "mcollective", nil)
 360           @c.target_for(message)
 361         end
 362
 363         it "should fail for unknown message types" do
 364           message = mock
 365           message.stubs(:type).returns(:fail)
 366
 367           expect {
 368             @c.target_for(message)
 369           }.to raise_error("Don't now how to create a target for message type fail")
 370         end
 371       end
 372
 373       describe "#disconnect" do
 374         it "should disconnect from the stomp connection" do
 375           @connection.expects(:disconnect)
 376           @c.disconnect
 377           @c.connection.should == nil
 378         end
 379       end
 380
 381       describe "#make_target" do
 382         it "should create correct targets" do
 383           @c.make_target("test", :reply, "mcollective").should == {:name => "/temp-queue/mcollective_reply_test", :headers => {}, :id => "mcollective_test_replies"}
 384           @c.make_target("test", :broadcast, "mcollective").should == {:name => "/exchange/mcollective_broadcast/test", :headers => {"reply-to"=>"/temp-queue/mcollective_reply_test"}, :id => "mcollective_broadcast_test"}
 385           @c.make_target("test", :request, "mcollective").should == {:name => "/exchange/mcollective_broadcast/test", :headers => {"reply-to"=>"/temp-queue/mcollective_reply_test"}, :id => "mcollective_broadcast_test"}
 386           @c.make_target("test", :direct_request, "mcollective", "rspec").should == {:headers=>{"reply-to"=>"/temp-queue/mcollective_reply_test"}, :name=>"/exchange/mcollective_directed/rspec", :id => nil}
 387           @c.make_target("test", :directed, "mcollective").should == {:name => "/exchange/mcollective_directed/rspec", :headers=>{}, :id => "rspec_directed_to_identity"}
 388         end
 389
 390         it "should raise an error for unknown collectives" do
 391           expect {
 392             @c.make_target("test", :broadcast, "foo")
 393           }.to raise_error("Unknown collective 'foo' known collectives are 'mcollective'")
 394         end
 395
 396         it "should raise an error for unknown types" do
 397           expect {
 398             @c.make_target("test", :test, "mcollective")
 399           }.to raise_error("Unknown target type test")
 400         end
 401       end
 402
 403
 404       describe "#get_env_or_option" do
 405         it "should return the environment variable if set" do
 406           ENV["test"] = "rspec_env_test"
 407
 408           @c.get_env_or_option("test", nil, nil).should == "rspec_env_test"
 409
 410           ENV.delete("test")
 411         end
 412
 413         it "should return the config option if set" do
 414           @config.expects(:pluginconf).returns({"test" => "rspec_test"}).twice
 415           @c.get_env_or_option("test", "test", "test").should == "rspec_test"
 416         end
 417
 418         it "should return default if nothing else matched" do
 419           @config.expects(:pluginconf).returns({}).once
 420           @c.get_env_or_option("test", "test", "test").should == "test"
 421         end
 422
 423         it "should raise an error if no default is supplied" do
 424           @config.expects(:pluginconf).returns({}).once
 425
 426           expect {
 427             @c.get_env_or_option("test", "test")
 428           }.to raise_error("No test environment or plugin.test configuration option given")
 429         end
 430       end
 431
 432       describe "#get_option" do
 433         it "should return the config option if set" do
 434           @config.expects(:pluginconf).returns({"test" => "rspec_test"}).twice
 435           @c.get_option("test").should == "rspec_test"
 436         end
 437
 438         it "should return default option was not found" do
 439           @config.expects(:pluginconf).returns({}).once
 440           @c.get_option("test", "test").should == "test"
 441         end
 442
 443         it "should raise an error if no default is supplied" do
 444           @config.expects(:pluginconf).returns({}).once
 445
 446           expect {
 447             @c.get_option("test")
 448           }.to raise_error("No plugin.test configuration option given")
 449         end
 450       end
 451
 452       describe "#get_bool_option" do
 453         it "should return the default if option isnt set" do
 454           @config.expects(:pluginconf).returns({}).once
 455           @c.get_bool_option("test", "default").should == "default"
 456         end
 457
 458         ["1", "yes", "true"].each do |boolean|
 459           it "should map options to true correctly" do
 460             @config.expects(:pluginconf).returns({"test" => boolean}).twice
 461             @c.get_bool_option("test", "default").should == true
 462           end
 463         end
 464
 465         ["0", "no", "false"].each do |boolean|
 466           it "should map options to false correctly" do
 467             @config.expects(:pluginconf).returns({"test" => boolean}).twice
 468             @c.get_bool_option("test", "default").should == false
 469           end
 470         end
 471
 472         it "should return default for non boolean options" do
 473           @config.expects(:pluginconf).returns({"test" => "foo"}).twice
 474           @c.get_bool_option("test", "default").should == "default"
 475         end
 476       end
 477     end
 478   end
 479 end