File: test_client.rb

package info (click to toggle)
ruby-sidekiq 3.2.6~dfsg-1
  • links: PTS, VCS
  • area: main
  • in suites: jessie, jessie-kfreebsd
  • size: 1,128 kB
  • ctags: 1,222
  • sloc: ruby: 5,848; makefile: 37; sh: 4
file content (269 lines) | stat: -rw-r--r-- 8,474 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
require 'helper'
require 'sidekiq/client'
require 'sidekiq/worker'

class TestClient < Sidekiq::Test
  describe 'with mock redis' do
    before do
      @redis = Minitest::Mock.new
      def @redis.multi; [yield] * 2 if block_given?; end
      def @redis.set(*); true; end
      def @redis.sadd(*); true; end
      def @redis.srem(*); true; end
      def @redis.get(*); nil; end
      def @redis.del(*); nil; end
      def @redis.incrby(*); nil; end
      def @redis.setex(*); true; end
      def @redis.expire(*); true; end
      def @redis.watch(*); true; end
      def @redis.with_connection; yield self; end
      def @redis.with; yield self; end
      def @redis.exec; true; end
      Sidekiq.instance_variable_set(:@redis, @redis)
      Sidekiq::Client.instance_variable_set(:@default, nil)
    end

    after do
      Sidekiq.redis = REDIS
      Sidekiq::Client.instance_variable_set(:@default, nil)
    end

    it 'raises ArgumentError with invalid params' do
      assert_raises ArgumentError do
        Sidekiq::Client.push('foo', 1)
      end

      assert_raises ArgumentError do
        Sidekiq::Client.push('foo', :class => 'Foo', :noargs => [1, 2])
      end

      assert_raises ArgumentError do
        Sidekiq::Client.push('queue' => 'foo', 'class' => MyWorker, 'noargs' => [1, 2])
      end

      assert_raises ArgumentError do
        Sidekiq::Client.push('queue' => 'foo', 'class' => 42, 'args' => [1, 2])
      end

      assert_raises ArgumentError do
        Sidekiq::Client.push('queue' => 'foo', 'class' => MyWorker, 'args' => 1)
      end
    end

    describe 'as instance' do
      it 'can push' do
        @redis.expect :lpush, 1, ['queue:default', Array]
        client = Sidekiq::Client.new
        jid = client.push('class' => 'Blah', 'args' => [1,2,3])
        assert_equal 24, jid.size
      end

      it 'allows local middleware modification' do
        @redis.expect :lpush, 1, ['queue:default', Array]
        $called = false
        mware = Class.new { def call(worker_klass,msg,q,r); $called = true; msg;end }
        client = Sidekiq::Client.new
        client.middleware do |chain|
          chain.add mware
        end
        client.push('class' => 'Blah', 'args' => [1,2,3])

        assert $called
        assert client.middleware.exists?(mware)
        refute Sidekiq.client_middleware.exists?(mware)
      end
    end

    it 'pushes messages to redis' do
      @redis.expect :lpush, 1, ['queue:foo', Array]
      pushed = Sidekiq::Client.push('queue' => 'foo', 'class' => MyWorker, 'args' => [1, 2])
      assert pushed
      assert_equal 24, pushed.size
      @redis.verify
    end

    it 'pushes messages to redis using a String class' do
      @redis.expect :lpush, 1, ['queue:foo', Array]
      pushed = Sidekiq::Client.push('queue' => 'foo', 'class' => 'MyWorker', 'args' => [1, 2])
      assert pushed
      assert_equal 24, pushed.size
      @redis.verify
    end

    class MyWorker
      include Sidekiq::Worker
    end

    it 'has default options' do
      assert_equal Sidekiq.default_worker_options, MyWorker.get_sidekiq_options
    end

    it 'handles perform_async' do
      @redis.expect :lpush, 1, ['queue:default', Array]
      pushed = MyWorker.perform_async(1, 2)
      assert pushed
      @redis.verify
    end

    it 'enqueues messages to redis' do
      @redis.expect :lpush, 1, ['queue:default', Array]
      pushed = Sidekiq::Client.enqueue(MyWorker, 1, 2)
      assert pushed
      @redis.verify
    end

    it 'enqueues messages to redis' do
      @redis.expect :lpush, 1, ['queue:custom_queue', Array]
      pushed = Sidekiq::Client.enqueue_to(:custom_queue, MyWorker, 1, 2)
      assert pushed
      @redis.verify
    end

    it 'enqueues messages to redis (delayed, custom queue)' do
      @redis.expect :zadd, 1, ['schedule', Array]
      pushed = Sidekiq::Client.enqueue_to_in(:custom_queue, 3.minutes, MyWorker, 1, 2)
      assert pushed
      @redis.verify
    end

    it 'enqueues messages to redis (delayed into past, custom queue)' do
      @redis.expect :lpush, 1, ['queue:custom_queue', Array]
      pushed = Sidekiq::Client.enqueue_to_in(:custom_queue, -3.minutes, MyWorker, 1, 2)
      assert pushed
      @redis.verify
    end

    it 'enqueues messages to redis (delayed)' do
      @redis.expect :zadd, 1, ['schedule', Array]
      pushed = Sidekiq::Client.enqueue_in(3.minutes, MyWorker, 1, 2)
      assert pushed
      @redis.verify
    end

    class QueuedWorker
      include Sidekiq::Worker
      sidekiq_options :queue => :flimflam
    end

    it 'enqueues to the named queue' do
      @redis.expect :lpush, 1, ['queue:flimflam', Array]
      pushed = QueuedWorker.perform_async(1, 2)
      assert pushed
      @redis.verify
    end

    it 'retrieves queues' do
      @redis.expect :smembers, ['bob'], ['queues']
      assert_equal ['bob'], Sidekiq::Queue.all.map(&:name)
    end
  end

  describe 'bulk' do
    after do
      Sidekiq::Queue.new.clear
    end
    it 'can push a large set of jobs at once' do
      jids = Sidekiq::Client.push_bulk('class' => QueuedWorker, 'args' => (1..1_000).to_a.map { |x| Array(x) })
      assert_equal 1_000, jids.size
    end
    it 'can push a large set of jobs at once using a String class' do
      jids = Sidekiq::Client.push_bulk('class' => 'QueuedWorker', 'args' => (1..1_000).to_a.map { |x| Array(x) })
      assert_equal 1_000, jids.size
    end
    it 'returns the jids for the jobs' do
      Sidekiq::Client.push_bulk('class' => 'QueuedWorker', 'args' => (1..2).to_a.map { |x| Array(x) }).each do |jid|
        assert_match(/[0-9a-f]{12}/, jid)
      end
    end
  end

  class BaseWorker
    include Sidekiq::Worker
    sidekiq_options 'retry' => 'base'
  end
  class AWorker < BaseWorker
  end
  class BWorker < BaseWorker
    sidekiq_options 'retry' => 'b'
  end
  class CWorker < BaseWorker
    sidekiq_options 'retry' => 2
  end

  describe 'client middleware' do

    class Stopper
      def call(worker_class, message, queue, r)
        raise ArgumentError unless r
        yield if message['args'].first.odd?
      end
    end

    it 'can stop some of the jobs from pushing' do
      Sidekiq.client_middleware.add Stopper
      begin
        assert_equal nil, Sidekiq::Client.push('class' => MyWorker, 'args' => [0])
        assert_match(/[0-9a-f]{12}/, Sidekiq::Client.push('class' => MyWorker, 'args' => [1]))
        Sidekiq::Client.push_bulk('class' => MyWorker, 'args' => [[0], [1]]).each do |jid|
          assert_match(/[0-9a-f]{12}/, jid)
        end
      ensure
        Sidekiq.client_middleware.remove Stopper
      end
    end
  end

  describe 'inheritance' do
    it 'inherits sidekiq options' do
      assert_equal 'base', AWorker.get_sidekiq_options['retry']
      assert_equal 'b', BWorker.get_sidekiq_options['retry']
    end
  end

  describe 'item normalization' do
    it 'defaults retry to true' do
      assert_equal true, Sidekiq::Client.new.__send__(:normalize_item, 'class' => QueuedWorker, 'args' => [])['retry']
    end

    it "does not normalize numeric retry's" do
      assert_equal 2, Sidekiq::Client.new.__send__(:normalize_item, 'class' => CWorker, 'args' => [])['retry']
    end
  end

  describe 'sharding' do
    class DWorker < BaseWorker
    end
    it 'allows sidekiq_options to point to different Redi' do
      conn = MiniTest::Mock.new
      conn.expect(:multi, [0, 1])
      DWorker.sidekiq_options('pool' => ConnectionPool.new(size: 1) { conn })
      DWorker.perform_async(1,2,3)
      conn.verify
    end
    it 'allows #via to point to different Redi' do
      conn = MiniTest::Mock.new
      conn.expect(:multi, [0, 1])
      default = Sidekiq::Client.new.redis_pool
      sharded_pool = ConnectionPool.new(size: 1) { conn }
      Sidekiq::Client.via(sharded_pool) do
        CWorker.perform_async(1,2,3)
        assert_equal sharded_pool, Sidekiq::Client.new.redis_pool
        assert_raises RuntimeError do
          Sidekiq::Client.via(default) do
            # nothing
          end
        end
      end
      assert_equal default, Sidekiq::Client.new.redis_pool
      conn.verify
    end
    it 'allows Resque helpers to point to different Redi' do
      conn = MiniTest::Mock.new
      conn.expect(:multi, []) { |*args, &block| block.call }
      conn.expect(:zadd, 1, [String, Array])
      DWorker.sidekiq_options('pool' => ConnectionPool.new(size: 1) { conn })
      Sidekiq::Client.enqueue_in(10, DWorker, 3)
      conn.verify
    end
  end
end