spec/unit/plugins/mcollective/connector/activemq_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/activemq.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 Activemq 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 = Activemq.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 = Activemq.new
  62           c.instance_variable_get("@config").should == @config
  63         end
  64
  65         it "should set @subscriptions to an empty list" do
  66           c = Activemq.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 = {"activemq.pool.size" => "2",
  79                         "activemq.pool.1.host" => "host1",
  80                         "activemq.pool.1.port" => "6163",
  81                         "activemq.pool.1.user" => "user1",
  82                         "activemq.pool.1.password" => "password1",
  83                         "activemq.pool.1.ssl" => "false",
  84                         "activemq.pool.2.host" => "host2",
  85                         "activemq.pool.2.port" => "6164",
  86                         "activemq.pool.2.user" => "user2",
  87                         "activemq.pool.2.password" => "password2",
  88                         "activemq.pool.2.ssl" => "true",
  89                         "activemq.pool.2.ssl.fallback" => "true",
  90                         "activemq.initial_reconnect_delay" => "0.02",
  91                         "activemq.max_reconnect_delay" => "40",
  92                         "activemq.use_exponential_back_off" => "false",
  93                         "activemq.back_off_multiplier" => "3",
  94                         "activemq.max_reconnect_attempts" => "5",
  95                         "activemq.randomize" => "true",
  96                         "activemq.backup" => "true",
  97                         "activemq.timeout" => "1",
  98                         "activemq.connect_timeout" => "5"}
  99
 100
 101           ENV.delete("STOMP_USER")
 102           ENV.delete("STOMP_PASSWORD")
 103
 104           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 105
 106           Activemq::EventLogger.expects(:new).returns("logger")
 107
 108           connector = mock
 109           connector.expects(:new).with(:backup => true,
 110                                        :back_off_multiplier => 3,
 111                                        :max_reconnect_delay => 40.0,
 112                                        :timeout => 1,
 113                                        :connect_timeout => 5,
 114                                        :use_exponential_back_off => false,
 115                                        :max_reconnect_attempts => 5,
 116                                        :initial_reconnect_delay => 0.02,
 117                                        :randomize => true,
 118                                        :reliable => true,
 119                                        :logger => "logger",
 120                                        :hosts => [{:passcode => 'password1',
 121                                                    :host => 'host1',
 122                                                    :port => 6163,
 123                                                    :ssl => false,
 124                                                    :login => 'user1'},
 125                                                   {:passcode => 'password2',
 126                                                    :host => 'host2',
 127                                                    :port => 6164,
 128                                                    :ssl => true,
 129                                                    :login => 'user2'}
 130           ])
 131
 132           @c.expects(:ssl_parameters).with(2, true).returns(true)
 133
 134           @c.instance_variable_set("@connection", nil)
 135           @c.connect(connector)
 136         end
 137       end
 138
 139       describe "#ssl_paramaters" do
 140         it "should ensure all settings are provided" do
 141           pluginconf = {"activemq.pool.1.host" => "host1",
 142                         "activemq.pool.1.port" => "6164",
 143                         "activemq.pool.1.user" => "user1",
 144                         "activemq.pool.1.password" => "password1",
 145                         "activemq.pool.1.ssl" => "true",
 146                         "activemq.pool.1.ssl.cert" => "rspec"}
 147
 148           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 149
 150           expect { @c.ssl_parameters(1, false) }.to raise_error("cert, key and ca has to be supplied for verified SSL mode")
 151         end
 152
 153         it "should verify the ssl files exist" do
 154           pluginconf = {"activemq.pool.1.host" => "host1",
 155                         "activemq.pool.1.port" => "6164",
 156                         "activemq.pool.1.user" => "user1",
 157                         "activemq.pool.1.password" => "password1",
 158                         "activemq.pool.1.ssl" => "true",
 159                         "activemq.pool.1.ssl.cert" => "rspec.cert",
 160                         "activemq.pool.1.ssl.key" => "rspec.key",
 161                         "activemq.pool.1.ssl.ca" => "rspec1.ca,rspec2.ca"}
 162
 163           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 164
 165           File.expects(:exist?).with("rspec.cert").twice.returns(true)
 166           File.expects(:exist?).with("rspec.key").twice.returns(true)
 167           File.expects(:exist?).with("rspec1.ca").twice.returns(true)
 168           File.expects(:exist?).with("rspec2.ca").twice.returns(false)
 169
 170           expect { @c.ssl_parameters(1, false) }.to raise_error("Cannot find CA file rspec2.ca")
 171
 172           @c.ssl_parameters(1, true).should == true
 173         end
 174
 175         it "should support fallback mode when there are errors" do
 176           pluginconf = {"activemq.pool.1.host" => "host1",
 177                         "activemq.pool.1.port" => "6164",
 178                         "activemq.pool.1.user" => "user1",
 179                         "activemq.pool.1.password" => "password1",
 180                         "activemq.pool.1.ssl" => "true"}
 181
 182           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 183
 184           @c.ssl_parameters(1, true).should == true
 185         end
 186
 187         it "should fail if fallback isnt enabled" do
 188           pluginconf = {"activemq.pool.1.host" => "host1",
 189                         "activemq.pool.1.port" => "6164",
 190                         "activemq.pool.1.user" => "user1",
 191                         "activemq.pool.1.password" => "password1",
 192                         "activemq.pool.1.ssl" => "true"}
 193
 194           @config.expects(:pluginconf).returns(pluginconf).at_least_once
 195
 196           expect { @c.ssl_parameters(1, false) }.to raise_error
 197         end
 198       end
 199
 200       describe "#receive" do
 201         it "should receive from the middleware" do
 202           payload = mock
 203           payload.stubs(:body).returns("msg")
 204           payload.stubs(:headers).returns("headers")
 205
 206           @connection.expects(:receive).returns(payload)
 207
 208           Message.expects(:new).with("msg", payload, :base64 => true, :headers => "headers").returns("message")
 209           @c.instance_variable_set("@base64", true)
 210
 211           received = @c.receive
 212           received.should == "message"
 213         end
 214
 215         it "should sleep and retry if recieving while disconnected" do
 216           payload = mock
 217           payload.stubs(:body).returns("msg")
 218           payload.stubs(:headers).returns("headers")
 219
 220           Message.stubs(:new).returns("rspec")
 221           @connection.expects(:receive).raises(::Stomp::Error::NoCurrentConnection).returns(payload).twice
 222           @c.expects(:sleep).with(1)
 223
 224           @c.receive.should == "rspec"
 225         end
 226       end
 227
 228       describe "#publish" do
 229         before do
 230           @connection.stubs(:publish).with("test", "msg", {}).returns(true)
 231         end
 232
 233         it "should base64 encode a message if configured to do so" do
 234           @c.instance_variable_set("@base64", true)
 235           @c.expects(:headers_for).returns({})
 236           @c.expects(:target_for).returns({:name => "test", :headers => {}})
 237           @connection.expects(:publish).with("test", "msg", {})
 238           @msg.expects(:base64_encode!)
 239
 240           @c.publish(@msg)
 241         end
 242
 243         it "should not base64 encode if not configured to do so" do
 244           @c.instance_variable_set("@base64", false)
 245           @c.expects(:headers_for).returns({})
 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", {"test" => "test"}).once
 255           @c.expects(:headers_for).returns({"test" => "test"})
 256           @c.expects(:target_for).returns({:name => "test", :headers => {}})
 257
 258           @c.publish(@msg)
 259         end
 260
 261         it "should publish direct messages based on discovered_hosts" do
 262           msg = mock
 263           msg.stubs(:base64_encode!)
 264           msg.stubs(:payload).returns("msg")
 265           msg.stubs(:agent).returns("agent")
 266           msg.stubs(:collective).returns("mcollective")
 267           msg.stubs(:type).returns(:direct_request)
 268           msg.expects(:discovered_hosts).returns(["one", "two"])
 269
 270           @c.expects(:headers_for).with(msg, "one")
 271           @c.expects(:headers_for).with(msg, "two")
 272           @connection.expects(:publish).with('/queue/mcollective.nodes', 'msg', nil).twice
 273
 274           @c.publish(msg)
 275         end
 276       end
 277
 278       describe "#subscribe" do
 279         it "should handle duplicate subscription errors" do
 280           @connection.expects(:subscribe).raises(::Stomp::Error::DuplicateSubscription)
 281           Log.expects(:error).with(regexp_matches(/already had a matching subscription, ignoring/))
 282           @c.subscribe("test", :broadcast, "mcollective")
 283         end
 284
 285         it "should use the make_target correctly" do
 286           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}})
 287           @c.subscribe("test", :broadcast, "mcollective")
 288         end
 289
 290         it "should check for existing subscriptions" do
 291           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 292           @subscription.expects("include?").with("rspec").returns(false)
 293           @connection.expects(:subscribe).never
 294
 295           @c.subscribe("test", :broadcast, "mcollective")
 296         end
 297
 298         it "should subscribe to the middleware" do
 299           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 300           @connection.expects(:subscribe).with("test", {}, "rspec")
 301           @c.subscribe("test", :broadcast, "mcollective")
 302         end
 303
 304         it "should add to the list of subscriptions" do
 305           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 306           @subscription.expects("<<").with("rspec")
 307           @c.subscribe("test", :broadcast, "mcollective")
 308         end
 309       end
 310
 311       describe "#unsubscribe" do
 312         it "should use make_target correctly" do
 313           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}})
 314           @c.unsubscribe("test", :broadcast, "mcollective")
 315         end
 316
 317         it "should unsubscribe from the target" do
 318           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 319           @connection.expects(:unsubscribe).with("test", {}, "rspec").once
 320
 321           @c.unsubscribe("test", :broadcast, "mcollective")
 322         end
 323
 324         it "should delete the source from subscriptions" do
 325           @c.expects("make_target").with("test", :broadcast, "mcollective").returns({:name => "test", :headers => {}, :id => "rspec"})
 326           @subscription.expects(:delete).with("rspec").once
 327
 328           @c.unsubscribe("test", :broadcast, "mcollective")
 329         end
 330       end
 331
 332       describe "#target_for" do
 333         it "should create reply targets based on reply-to headers in requests" do
 334           message = mock
 335           message.expects(:type).returns(:reply)
 336
 337           request = mock
 338           request.expects(:headers).returns({"reply-to" => "foo"})
 339
 340           message.expects(:request).returns(request)
 341
 342           @c.target_for(message).should == {:name => "foo", :headers => {}}
 343         end
 344
 345         it "should create new request targets" do
 346           message = mock
 347           message.expects(:type).returns(:request).times(3)
 348           message.expects(:agent).returns("rspecagent")
 349           message.expects(:collective).returns("mcollective")
 350
 351           @c.expects(:make_target).with("rspecagent", :request, "mcollective")
 352           @c.target_for(message)
 353         end
 354
 355         it "should support direct requests" do
 356           message = mock
 357           message.expects(:type).returns(:direct_request).times(3)
 358           message.expects(:agent).returns("rspecagent")
 359           message.expects(:collective).returns("mcollective")
 360
 361           @c.expects(:make_target).with("rspecagent", :direct_request, "mcollective")
 362           @c.target_for(message)
 363         end
 364
 365         it "should fail for unknown message types" do
 366           message = mock
 367           message.stubs(:type).returns(:fail)
 368
 369           expect {
 370             @c.target_for(message)
 371           }.to raise_error("Don't now how to create a target for message type fail")
 372         end
 373       end
 374
 375       describe "#disconnect" do
 376         it "should disconnect from the stomp connection" do
 377           @connection.expects(:disconnect)
 378           @c.disconnect
 379           @c.connection.should == nil
 380         end
 381       end
 382
 383       describe "#headers_for" do
 384         it "should return empty headers if priority is 0" do
 385           message = mock
 386           message.expects(:type).returns(:foo)
 387
 388           @c.instance_variable_set("@msgpriority", 0)
 389           @c.headers_for(message).should == {}
 390         end
 391
 392         it "should return a priority if priority is non 0" do
 393           message = mock
 394           message.expects(:type).returns(:foo)
 395
 396           @c.instance_variable_set("@msgpriority", 1)
 397           @c.headers_for(message).should == {"priority" => 1}
 398         end
 399
 400         it "should set mc_identity for direct requests" do
 401           message = mock
 402           message.expects(:type).returns(:direct_request).twice
 403           message.expects(:agent).returns("rspecagent")
 404           message.expects(:collective).returns("mcollective")
 405           message.expects(:reply_to).returns(nil)
 406
 407           @c.instance_variable_set("@msgpriority", 0)
 408           @c.expects(:make_target).with("rspecagent", :reply, "mcollective").returns({:name => "test"})
 409           @c.headers_for(message, "some.node").should == {"mc_identity"=>"some.node", "reply-to"=>"test"}
 410         end
 411
 412         it "should set a reply-to header for :request type messages" do
 413           message = mock
 414           message.expects(:type).returns(:request).twice
 415           message.expects(:agent).returns("rspecagent")
 416           message.expects(:collective).returns("mcollective")
 417           message.expects(:reply_to).returns(nil)
 418
 419           @c.instance_variable_set("@msgpriority", 0)
 420           @c.expects(:make_target).with("rspecagent", :reply, "mcollective").returns({:name => "test"})
 421           @c.headers_for(message).should == {"reply-to" => "test"}
 422         end
 423
 424         it "should set reply-to correctly if the message defines it" do
 425           message = mock
 426           message.expects(:type).returns(:request).twice
 427           message.expects(:agent).returns("rspecagent")
 428           message.expects(:collective).returns("mcollective")
 429           message.expects(:reply_to).returns("rspec").twice
 430
 431           @c.headers_for(message).should == {"reply-to" => "rspec"}
 432
 433         end
 434       end
 435
 436       describe "#make_target" do
 437         it "should create correct targets" do
 438           @c.make_target("test", :reply, "mcollective").should == {:name => "/queue/mcollective.reply.rspec_#{$$}", :headers => {}, :id => "/queue/mcollective.reply.rspec_#{$$}"}
 439           @c.make_target("test", :broadcast, "mcollective").should == {:name => "/topic/mcollective.test.agent", :headers => {}, :id => "/topic/mcollective.test.agent"}
 440           @c.make_target("test", :request, "mcollective").should == {:name => "/topic/mcollective.test.agent", :headers => {}, :id => "/topic/mcollective.test.agent"}
 441           @c.make_target("test", :direct_request, "mcollective").should == {:headers=>{}, :name=>"/queue/mcollective.nodes", :id => "/queue/mcollective.nodes"}
 442           @c.make_target("test", :directed, "mcollective").should == {:name => "/queue/mcollective.nodes", :headers=>{"selector"=>"mc_identity = 'rspec'"}, :id => "mcollective_directed_to_identity"}
 443         end
 444
 445         it "should raise an error for unknown collectives" do
 446           expect {
 447             @c.make_target("test", :broadcast, "foo")
 448           }.to raise_error("Unknown collective 'foo' known collectives are 'mcollective'")
 449         end
 450
 451         it "should raise an error for unknown types" do
 452           expect {
 453             @c.make_target("test", :test, "mcollective")
 454           }.to raise_error("Unknown target type test")
 455         end
 456       end
 457
 458
 459       describe "#get_env_or_option" do
 460         it "should return the environment variable if set" do
 461           ENV["test"] = "rspec_env_test"
 462
 463           @c.get_env_or_option("test", nil, nil).should == "rspec_env_test"
 464
 465           ENV.delete("test")
 466         end
 467
 468         it "should return the config option if set" do
 469           @config.expects(:pluginconf).returns({"test" => "rspec_test"}).twice
 470           @c.get_env_or_option("test", "test", "test").should == "rspec_test"
 471         end
 472
 473         it "should return default if nothing else matched" do
 474           @config.expects(:pluginconf).returns({}).once
 475           @c.get_env_or_option("test", "test", "test").should == "test"
 476         end
 477
 478         it "should raise an error if no default is supplied" do
 479           @config.expects(:pluginconf).returns({}).once
 480
 481           expect {
 482             @c.get_env_or_option("test", "test")
 483           }.to raise_error("No test environment or plugin.test configuration option given")
 484         end
 485       end
 486
 487       describe "#get_option" do
 488         it "should return the config option if set" do
 489           @config.expects(:pluginconf).returns({"test" => "rspec_test"}).twice
 490           @c.get_option("test").should == "rspec_test"
 491         end
 492
 493         it "should return default option was not found" do
 494           @config.expects(:pluginconf).returns({}).once
 495           @c.get_option("test", "test").should == "test"
 496         end
 497
 498         it "should raise an error if no default is supplied" do
 499           @config.expects(:pluginconf).returns({}).once
 500
 501           expect {
 502             @c.get_option("test")
 503           }.to raise_error("No plugin.test configuration option given")
 504         end
 505       end
 506
 507       describe "#get_bool_option" do
 508         it "should return the default if option isnt set" do
 509           @config.expects(:pluginconf).returns({}).once
 510           @c.get_bool_option("test", "default").should == "default"
 511         end
 512
 513         ["1", "yes", "true"].each do |boolean|
 514           it "should map options to true correctly" do
 515             @config.expects(:pluginconf).returns({"test" => boolean}).twice
 516             @c.get_bool_option("test", "default").should == true
 517           end
 518         end
 519
 520         ["0", "no", "false"].each do |boolean|
 521           it "should map options to false correctly" do
 522             @config.expects(:pluginconf).returns({"test" => boolean}).twice
 523             @c.get_bool_option("test", "default").should == false
 524           end
 525         end
 526
 527         it "should return default for non boolean options" do
 528           @config.expects(:pluginconf).returns({"test" => "foo"}).twice
 529           @c.get_bool_option("test", "default").should == "default"
 530         end
 531       end
 532     end
 533   end
 534 end