Class: Zizq::Test::Client

Inherits:
Client
  • Object
show all
Defined in:
lib/zizq/test/client.rb,
sig/generated/zizq/test/client.rbs

Overview

A Zizq::Client stand-in for use in test suites.

Buffers enqueue / enqueue_bulk calls in memory and returns synthetic Resources::Job instances (with generated ids) so that callers depending on the regular client's return contract don't need to special-case test mode.

Read operations (get_queues, list_jobs, count_jobs, …) are explicitly not supported in test mode and raise NotSupported. Tests that need those should either run against a real server or stub at a higher level.

Activated indirectly via Zizq.configuration.test_mode = true — Zizq.client then lazily builds a Test::Client instead of a real Client.

Defined Under Namespace

Classes: Entry, NotSupported

Constant Summary collapse

ID_LENGTH =

Length of a real scru128 id in its base-32 representation. Synthetic test ids are sized to match (test prefix + zero padded counter) so they fit anywhere a real id would.

Returns:

  • (::Integer) 
25
ID_PREFIX =

Returns:

  • (::String) 
"test"
STATUS_SCHEDULED =

The canonical Zizq lifecycle states. We mirror these so the status on a buffered job reflects what the real server would report. Test mode never retries — in_flight only ever transitions to completed or dead.

Returns:

  • (::String) 
"scheduled"
STATUS_READY =

Returns:

  • (::String) 
"ready"
STATUS_IN_FLIGHT =

Returns:

  • (::String) 
"in_flight"
STATUS_COMPLETED =

Returns:

  • (::String) 
"completed"
STATUS_DEAD =

Returns:

  • (::String) 
"dead"
PASS_ALL_FILTER =

Default filter: lambda — passes every job. Named so the filter pipeline is always callable without nil-checking.

Returns:

->(_job) { true }

Constants inherited from Client

Client::CONTENT_TYPES, Client::STREAM_ACCEPT

Instance Attribute Summary

Attributes inherited from Client

#format, #url

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Client

#add_cron_group_entry, #build_cron_entry, #build_cron_job, #build_path, #build_set_body, #build_where_params, #cleanup_internal_clients, #consume_response, #count_jobs, #decode_body, #delete, #delete_all_crons, #delete_all_jobs, #delete_cron_group, #delete_cron_group_entry, #delete_job, #enc, #encode_body, #encode_range, #ensure_io_thread, #erase_all_data, #get, #get_cron_group, #get_cron_group_entry, #get_error, #get_job, #get_path, #get_queues, #handle_response!, #health, #http, #io_thread_run, #list_cron_groups, #list_errors, #list_jobs, make_finalizer, parse_msgpack_stream, parse_ndjson, #patch, #post, #put, #raw_post, #replace_cron_group, #replace_cron_group_entry, #report_failure, #report_success, #report_success_bulk, #request, #server_version, #stream_http, #sync_call, #take_jobs, #thread_local_http, #update_all_jobs, #update_cron_group, #update_cron_group_entry, #update_job, #validate_and_build_set, #validate_where

Constructor Details

#initialize ⇒ Client

Returns a new instance of Client.

Signature:

  • () -> void



59
60
61
62
63
64
65
66
 
# File 'lib/zizq/test/client.rb', line 59

def initialize #: () -> void
  # Skip the parent's HTTP setup — we don't open connections in
  # test mode. The parent's @http and friends stay nil; methods
  # that would touch them are overridden below.

  @entries = [] #: Array[Entry]
  @mutex = Mutex.new
end

Class Method Details

.normalize_payload(payload) ⇒ Object

Round-trip the payload through JSON so the in-memory representation matches what a consumer would receive over the wire: symbol keys / Symbol values become strings, nested hashes and arrays are normalized recursively, and non-JSON- safe values (BigDecimal, custom objects) raise here rather than surviving in test mode only to break in production.

Also used by Zizq::Test.enqueued_raw? / enqueued_raw_count to normalize the query side so symbol-keyed assertion payloads still match the (now string-keyed) buffer.

Parameters:

  • payload (Object)

Returns:

  • (Object) 


266
267
268
 
# File 'lib/zizq/test/client.rb', line 266

def self.normalize_payload(payload) #: (untyped) -> untyped
  JSON.parse(JSON.generate(payload))
end

Instance Method Details

#apply_filters(entries, only_queues: nil, except_queues: nil, only_types: nil, except_types: nil, filter: PASS_ALL_FILTER) ⇒ Object

Returns entries matching every named filter AND the predicate. All filter kwargs are optional; unset means "don't filter on this axis." Callers must hold @mutex — public accessors do so via synchronize before calling.

  • only_queues: / except_queues: — String, Array of Strings.
  • only_types: / except_types: — String, Class, or Array of those. Class names are matched against the wire-format type string via .to_s.
  • filter: — a lambda receiving a Resources::Job, returning truthy to keep. Defaults to PASS_ALL_FILTER.

only_* and except_* AND together with the predicate.

Parameters:

  • entries (Object)
  • only_queues: (Object) (defaults to: nil)
  • except_queues: (Object) (defaults to: nil)
  • only_types: (Object) (defaults to: nil)
  • except_types: (Object) (defaults to: nil)
  • filter: (Object) (defaults to: PASS_ALL_FILTER)

Returns:

  • (Object) 


283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
 
# File 'lib/zizq/test/client.rb', line 283

def apply_filters(
  entries,
  only_queues: nil,
  except_queues: nil,
  only_types: nil,
  except_types: nil,
  filter: PASS_ALL_FILTER
) #: (Array[Entry], **untyped) -> Array[Entry]
  only_queues = normalize_filter(only_queues)
  except_queues = normalize_filter(except_queues)
  only_types = normalize_filter(only_types)
  except_types = normalize_filter(except_types)

  entries.select do |entry|
    queue = entry.data["queue"] #: String
    type = entry.data["type"] #: String

    (only_queues.empty? || only_queues.include?(queue)) &&
      (except_queues.empty? || !except_queues.include?(queue)) &&
      (only_types.empty? || only_types.include?(type)) &&
      (except_types.empty? || !except_types.include?(type)) &&
      filter.call(entry.job)
  end
end

#clear! ⇒ Object

Reset the buffer. Called between tests via Zizq::Test.reset!.

Returns:

  • (Object) 


102
103
104
 
# File 'lib/zizq/test/client.rb', line 102

def clear! #: () -> void
  @mutex.synchronize { @entries.clear }
end

#close ⇒ Object

Signature:

  • () -> void

Returns:

  • (Object) 


106
107
 
# File 'lib/zizq/test/client.rb', line 106

def close #: () -> void
end

#completed_jobs(**filters) ⇒ Object

Signature:

  • (**untyped) -> Array[Resources::Job]

Parameters:

  • filters (Object)

Returns:

  • (Object) 


93
94
95
 
# File 'lib/zizq/test/client.rb', line 93

def completed_jobs(**filters) #: (**untyped) -> Array[Resources::Job]
  filter_by_status([STATUS_COMPLETED], **filters)
end

#dead_jobs(**filters) ⇒ Object

Signature:

  • (**untyped) -> Array[Resources::Job]

Parameters:

  • filters (Object)

Returns:

  • (Object) 


97
98
99
 
# File 'lib/zizq/test/client.rb', line 97

def dead_jobs(**filters) #: (**untyped) -> Array[Resources::Job]
  filter_by_status([STATUS_DEAD], **filters)
end

#dispatch_entry(entry) ⇒ Object

Mark the entry in_flight, dispatch through the full dequeue middleware chain (same path the real worker uses, so any registered middlewares run in tests too), settle to completed or dead. A raised exception re-raises after recording — same observable behaviour as Rails' perform_enqueued_jobs.

Parameters:

  • entry (Object)

Returns:

  • (Object) 


359
360
361
362
363
364
365
366
367
368
 
# File 'lib/zizq/test/client.rb', line 359

def dispatch_entry(entry) #: (Entry) -> void
  entry.data["status"] = STATUS_IN_FLIGHT
  begin
    Zizq.configuration.dequeue_middleware.call(entry.job)
    entry.data["status"] = STATUS_COMPLETED
  rescue StandardError
    entry.data["status"] = STATUS_DEAD
    raise
  end
end

#drain(**filters) ⇒ Object

Dispatch every runnable entry (status ready, or scheduled with ready_at already elapsed) through the configured dequeue middleware chain (same path the real worker uses, so any registered middlewares run in tests too), looping until no more match the filters. Re-enqueues during dispatch fall through the loop naturally — they get drained too unless they fall outside the filter set.

The per-iteration snapshot is taken under the mutex and marked in_flight atomically. Dispatch happens outside the mutex so handlers can re-enter the client without deadlocking. On success the entry moves to completed; on a raised exception it moves to dead and the exception re-raises (matching ActiveJob's perform_enqueued_jobs + Sidekiq's drain).

Parameters:

  • filters (Object)

Returns:

  • (Object) 


123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/zizq/test/client.rb', line 123

def drain(**filters) #: (**untyped) -> Integer
  total = 0
  loop do
    snapshot = take_runnable_snapshot(**filters)
    break if snapshot.empty?

    snapshot.each { |entry| dispatch_entry(entry) }
    total += snapshot.size
  end
  total
end

#enqueue(queue:, type:, payload:, priority: nil, ready_at: nil, retry_limit: nil, backoff: nil, retention: nil, unique_key: nil, unique_while: nil, batch: nil) ⇒ void

This method returns an undefined value.



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
 
# File 'lib/zizq/test/client.rb', line 136

def enqueue(
  queue:,
  type:,
  payload:,
  priority: nil,
  ready_at: nil,
  retry_limit: nil,
  backoff: nil,
  retention: nil,
  unique_key: nil,
  unique_while: nil,
  batch: nil
)
  req =
    EnqueueRequest.new(
      queue:,
      type:,
      payload: self.class.normalize_payload(payload),
      priority:,
      ready_at:,
      retry_limit:,
      backoff:,
      retention:,
      unique_key:,
      unique_while:,
      batch:
    )
  @mutex.synchronize { record_unsynchronized(req) }.job
end

#enqueue_bulk(jobs:) ⇒ void

This method returns an undefined value.



167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
 
# File 'lib/zizq/test/client.rb', line 167

def enqueue_bulk(jobs:)
  @mutex.synchronize do
    jobs.map do |params|
      req =
        EnqueueRequest.new(
          queue: params[:queue],
          type: params[:type],
          payload: self.class.normalize_payload(params[:payload]),
          priority: params[:priority],
          ready_at: params[:ready_at],
          retry_limit: params[:retry_limit],
          backoff: params[:backoff],
          retention: params[:retention],
          unique_key: params[:unique_key],
          unique_while: params[:unique_while],
          batch: params[:batch]
        )
      record_unsynchronized(req).job
    end
  end
end

#enqueued_jobs(**filters) ⇒ Object

All buffered jobs, in submission order, optionally filtered. See apply_filters for the filter kwargs.

Parameters:

  • filters (Object)

Returns:

  • (Object) 


70
71
72
 
# File 'lib/zizq/test/client.rb', line 70

def enqueued_jobs(**filters) #: (**untyped) -> Array[Resources::Job]
  @mutex.synchronize { apply_filters(@entries, **filters).map(&:job) }
end

#enqueued_requests(**filters) ⇒ Object

Original EnqueueRequests in submission order. Useful when a test needs metadata that doesn't survive onto Resources::Job (unique_key, unique_while, delay before ready_at resolution, etc.). Same filter kwargs as enqueued_jobs.

Parameters:

  • filters (Object)

Returns:

  • (Object) 


78
79
80
 
# File 'lib/zizq/test/client.rb', line 78

def enqueued_requests(**filters) #: (**untyped) -> Array[EnqueueRequest]
  @mutex.synchronize { apply_filters(@entries, **filters).map(&:request) }
end

#filter_by_status(statuses, **filters) ⇒ Object

Signature:

  • (Array[String], **untyped) -> Array[Resources::Job]

Parameters:

  • statuses (Object)
  • filters (Object)

Returns:

  • (Object) 


308
309
310
311
312
313
314
 
# File 'lib/zizq/test/client.rb', line 308

def filter_by_status(statuses, **filters) #: (Array[String], **untyped) -> Array[Resources::Job]
  @mutex.synchronize do
    apply_filters(@entries, **filters)
      .select { |e| statuses.include?(e.data["status"]) }
      .map(&:job)
  end
end

#in_flight_jobs(**filters) ⇒ Object

Signature:

  • (**untyped) -> Array[Resources::Job]

Parameters:

  • filters (Object)

Returns:

  • (Object) 


89
90
91
 
# File 'lib/zizq/test/client.rb', line 89

def in_flight_jobs(**filters) #: (**untyped) -> Array[Resources::Job]
  filter_by_status([STATUS_IN_FLIGHT], **filters)
end

#normalize_filter(value) ⇒ Object

Accept a Class, String, or Array of those; emit an Array of Strings. Class names match the API's type string.

Parameters:

  • value (Object)

Returns:

  • (Object) 


342
343
344
345
346
347
348
349
350
351
 
# File 'lib/zizq/test/client.rb', line 342

def normalize_filter(value) #: ((String | Class | Array[String | Class])?) -> Array[String]
  case value
  when nil
    []
  when Array
    value.map { |x| x.to_s }
  else
    [value.to_s]
  end
end

#pending_jobs(**filters) ⇒ Object

Jobs awaiting dispatch — ready and scheduled entries. pending_jobs is the set drain would attempt to run on its next call (modulo ready_at for scheduled entries).

Parameters:

  • filters (Object)

Returns:

  • (Object) 


85
86
87
 
# File 'lib/zizq/test/client.rb', line 85

def pending_jobs(**filters) #: (**untyped) -> Array[Resources::Job]
  filter_by_status([STATUS_READY, STATUS_SCHEDULED], **filters)
end

#record_unsynchronized(req) ⇒ Object

Signature:

  • (EnqueueRequest) -> Entry

Parameters:

  • req (Object)

Returns:

  • (Object) 


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
 
# File 'lib/zizq/test/client.rb', line 220

def record_unsynchronized(req) #: (EnqueueRequest) -> Entry
  # Serialized format: ready_at is integer milliseconds.
  # `req.ready_at` (from `to_enqueue_params`) is fractional
  # seconds; convert here so `Resources::Job#ready_at`'s
  # ms -> seconds round-trip produces the same value the
  # caller passed in. When the client omits ready_at the server
  # assigns `now`, so we do the same.
  now_ms = (Time.now.to_f * 1000).to_i
  ready_at_ms = req.ready_at ? (req.ready_at.to_f * 1000).to_i : now_ms

  data = {
    "id" => synthetic_id(@entries.size + 1),
    "queue" => req.queue,
    "type" => req.type,
    "payload" => req.payload,
    "priority" => req.priority,
    "ready_at" => ready_at_ms,
    "retry_limit" => req.retry_limit,
    "status" => ready_at_ms > now_ms ? STATUS_SCHEDULED : STATUS_READY
  }
  entry =
    Entry.new(
      request: req,
      job: Resources::Job.new(self, data),
      data: data
    )
  @entries << entry
  entry
end

#runnable?(entry, now_ms) ⇒ Boolean

Signature:

  • (Entry, Integer) -> bool

Parameters:

  • entry (Object)
  • now_ms (Object)

Returns:

  • (Boolean) 


329
330
331
332
333
334
335
336
337
338
 
# File 'lib/zizq/test/client.rb', line 329

def runnable?(entry, now_ms) #: (Entry, Integer) -> bool
  case entry.data["status"]
  when STATUS_READY
    true
  when STATUS_SCHEDULED
    entry.data["ready_at"] <= now_ms
  else
    false
  end
end

#synthetic_id(counter) ⇒ Object

Signature:

  • (Integer) -> String

Parameters:

  • counter (Object)

Returns:

  • (Object) 


250
251
252
 
# File 'lib/zizq/test/client.rb', line 250

def synthetic_id(counter) #: (Integer) -> String
  "#{ID_PREFIX}#{counter.to_s.rjust(ID_LENGTH - ID_PREFIX.length, "0")}"
end

#take_runnable_snapshot(**filters) ⇒ Object

Lock, find runnable entries matching the filters, flip each to in_flight, return the snapshot. Holding the mutex through this is fine because we don't call user code — dispatch happens outside.

Parameters:

  • filters (Object)

Returns:

  • (Object) 


320
321
322
323
324
325
326
327
 
# File 'lib/zizq/test/client.rb', line 320

def take_runnable_snapshot(**filters) #: (**untyped) -> Array[Entry]
  now_ms = (Time.now.to_f * 1000).to_i
  @mutex.synchronize do
    apply_filters(@entries, **filters)
      .select { |e| runnable?(e, now_ms) }
      .each { |entry| entry.data["status"] = STATUS_IN_FLIGHT }
  end
end