module ActiveJob::TestHelper

def after_teardown # :nodoc:
  super
  queue_adapter_changed_jobs.each { |klass| klass.disable_test_adapter }
end

def assert_enqueued_jobs(number, only: nil, except: nil, queue: nil, &block)
  require_active_job_test_adapter!("assert_enqueued_jobs")
  if block_given?
    original_jobs = enqueued_jobs_with(only: only, except: except, queue: queue)
    _assert_nothing_raised_or_warn("assert_enqueued_jobs", &block)
    new_jobs = enqueued_jobs_with(only: only, except: except, queue: queue)
    actual_count = (new_jobs - original_jobs).count
  else
    actual_count = enqueued_jobs_with(only: only, except: except, queue: queue).count
  end
  assert_equal number, actual_count, "#{number} jobs expected, but #{actual_count} were enqueued"
end

def assert_enqueued_with(job: nil, args: nil, at: nil, queue: nil, priority: nil, &block)
  require_active_job_test_adapter!("assert_enqueued_with")
  expected = { job: job, args: args, at: at, queue: queue, priority: priority }.compact
  expected_args = prepare_args_for_assertion(expected)
  potential_matches = []
  if block_given?
    original_enqueued_jobs = enqueued_jobs.dup
    _assert_nothing_raised_or_warn("assert_enqueued_with", &block)
    jobs = enqueued_jobs - original_enqueued_jobs
  else
    jobs = enqueued_jobs
  end
  matching_job = jobs.find do |enqueued_job|
    deserialized_job = deserialize_args_for_assertion(enqueued_job)
    potential_matches << deserialized_job
    expected_args.all? do |key, value|
      if value.respond_to?(:call)
        value.call(deserialized_job[key])
      else
        value == deserialized_job[key]
      end
    end
  end
  matching_class = potential_matches.select do |enqueued_job|
    enqueued_job["job_class"] == job.to_s
  end
  message = +"No enqueued job found with #{expected}"
  if potential_matches.empty?
    message << "\n\nNo jobs were enqueued"
  elsif matching_class.empty?
    message << "\n\nNo jobs of class #{expected[:job]} were enqueued, job classes enqueued: "
    message << potential_matches.map { |job| job["job_class"] }.join(", ")
  else
    message << "\n\nPotential matches: #{matching_class.join("\n")}"
  end
  assert matching_job, message
  instantiate_job(matching_job)
end

def assert_no_enqueued_jobs(only: nil, except: nil, queue: nil, &block)
  require_active_job_test_adapter!("assert_no_enqueued_jobs")
  assert_enqueued_jobs 0, only: only, except: except, queue: queue, &block
end

def assert_no_performed_jobs(only: nil, except: nil, queue: nil, &block)
  require_active_job_test_adapter!("assert_no_performed_jobs")
  assert_performed_jobs 0, only: only, except: except, queue: queue, &block
end

def assert_performed_jobs(number, only: nil, except: nil, queue: nil, &block)
  require_active_job_test_adapter!("assert_performed_jobs")
  if block_given?
    original_count = performed_jobs.size
    perform_enqueued_jobs(only: only, except: except, queue: queue, &block)
    new_count = performed_jobs.size
    performed_jobs_size = new_count - original_count
  else
    performed_jobs_size = performed_jobs_with(only: only, except: except, queue: queue).count
  end
  assert_equal number, performed_jobs_size, "#{number} jobs expected, but #{performed_jobs_size} were performed"
end

def assert_performed_with(job: nil, args: nil, at: nil, queue: nil, priority: nil, &block)
  require_active_job_test_adapter!("assert_performed_with")
  expected = { job: job, args: args, at: at, queue: queue, priority: priority }.compact
  expected_args = prepare_args_for_assertion(expected)
  potential_matches = []
  if block_given?
    original_performed_jobs_count = performed_jobs.count
    perform_enqueued_jobs(&block)
    jobs = performed_jobs.drop(original_performed_jobs_count)
  else
    jobs = performed_jobs
  end
  matching_job = jobs.find do |enqueued_job|
    deserialized_job = deserialize_args_for_assertion(enqueued_job)
    potential_matches << deserialized_job
    expected_args.all? do |key, value|
      if value.respond_to?(:call)
        value.call(deserialized_job[key])
      else
        value == deserialized_job[key]
      end
    end
  end
  matching_class = potential_matches.select do |enqueued_job|
    enqueued_job["job_class"] == job.to_s
  end
  message = +"No performed job found with #{expected}"
  if potential_matches.empty?
    message << "\n\nNo jobs were performed"
  elsif matching_class.empty?
    message << "\n\nNo jobs of class #{expected[:job]} were performed, job classes performed: "
    message << potential_matches.map { |job| job["job_class"] }.join(", ")
  else
    message << "\n\nPotential matches: #{matching_class.join("\n")}"
  end
  assert matching_job, message
  instantiate_job(matching_job)
end

def before_setup # :nodoc:
  queue_adapter_specific_to_this_test_class = queue_adapter_for_test
  queue_adapter_changed_jobs.each do |klass|
    if queue_adapter_specific_to_this_test_class
      klass.enable_test_adapter(queue_adapter_specific_to_this_test_class)
    elsif klass._queue_adapter.nil?
      klass.enable_test_adapter(ActiveJob::QueueAdapters::TestAdapter.new)
    end
  end
  clear_enqueued_jobs
  clear_performed_jobs
  super
end

def clear_enqueued_jobs
  enqueued_jobs.clear if using_test_adapter?
end

def clear_performed_jobs
  performed_jobs.clear if using_test_adapter?
end

def deserialize_args_for_assertion(job)
  job.dup.tap do |new_job|
    new_job[:at] = Time.at(new_job[:at]) if new_job[:at]
    new_job[:args] = ActiveJob::Arguments.deserialize(new_job[:args]) if new_job[:args]
  end
end

def enqueued_jobs_with(only: nil, except: nil, queue: nil, at: nil, &block)
  jobs_with(enqueued_jobs, only: only, except: except, queue: queue, at: at, &block)
end

def filter_as_proc(filter)
  return filter if filter.is_a?(Proc)
  ->(job) { Array(filter).include?(job.fetch(:job)) }
end

def flush_enqueued_jobs(only: nil, except: nil, queue: nil, at: nil)
  enqueued_jobs_with(only: only, except: except, queue: queue, at: at) do |payload|
    queue_adapter.enqueued_jobs.delete(payload)
    queue_adapter.performed_jobs << payload
    instantiate_job(payload, skip_deserialize_arguments: true).perform_now
  end.count
end

def instantiate_job(payload, skip_deserialize_arguments: false)
  job = payload[:job].deserialize(payload)
  job.scheduled_at = Time.at(payload[:at]) if payload.key?(:at)
  job.send(:deserialize_arguments_if_needed) unless skip_deserialize_arguments
  job
end

def jobs_with(jobs, only: nil, except: nil, queue: nil, at: nil)
  validate_option(only: only, except: except)
  jobs.dup.select do |job|
    job_class = job.fetch(:job)
    if only
      next false unless filter_as_proc(only).call(job)
    elsif except
      next false if filter_as_proc(except).call(job)
    end
    if queue
      next false unless queue.to_s == job.fetch(:queue, job_class.queue_name)
    end
    if at && job[:at]
      next false if job[:at] > at.to_f
    end
    yield job if block_given?
    true
  end
end

def perform_enqueued_jobs(only: nil, except: nil, queue: nil, at: nil, &block)
  unless block_given?
    require_active_job_test_adapter!("perform_enqueued_jobs (without a block)")
    return flush_enqueued_jobs(only: only, except: except, queue: queue, at: at)
  end
  return _assert_nothing_raised_or_warn("perform_enqueued_jobs", &block) unless using_test_adapter?
  validate_option(only: only, except: except)
  old_perform_enqueued_jobs = queue_adapter.perform_enqueued_jobs
  old_perform_enqueued_at_jobs = queue_adapter.perform_enqueued_at_jobs
  old_filter = queue_adapter.filter
  old_reject = queue_adapter.reject
  old_queue = queue_adapter.queue
  old_at = queue_adapter.at
  begin
    queue_adapter.perform_enqueued_jobs = true
    queue_adapter.perform_enqueued_at_jobs = true
    queue_adapter.filter = only
    queue_adapter.reject = except
    queue_adapter.queue = queue
    queue_adapter.at = at
    _assert_nothing_raised_or_warn("perform_enqueued_jobs", &block)
  ensure
    queue_adapter.perform_enqueued_jobs = old_perform_enqueued_jobs
    queue_adapter.perform_enqueued_at_jobs = old_perform_enqueued_at_jobs
    queue_adapter.filter = old_filter
    queue_adapter.reject = old_reject
    queue_adapter.queue = old_queue
    queue_adapter.at = old_at
  end
end

def performed_jobs_with(only: nil, except: nil, queue: nil, &block)
  jobs_with(performed_jobs, only: only, except: except, queue: queue, &block)
end

def prepare_args_for_assertion(args)
  args.dup.tap do |arguments|
    if arguments[:queue].is_a?(Symbol)
      arguments[:queue] = arguments[:queue].to_s
    end
    if arguments[:at].acts_like?(:time)
      at_range = arguments[:at] - 1..arguments[:at] + 1
      arguments[:at] = ->(at) { at_range.cover?(at) }
    end
  end
end

def queue_adapter
  ActiveJob::Base.queue_adapter
end

def queue_adapter_changed_jobs
  (ActiveJob::Base.descendants << ActiveJob::Base).select do |klass|
    # only override explicitly set adapters, a quirk of `class_attribute`
    klass.singleton_class.public_instance_methods(false).include?(:_queue_adapter)
  end
end

def queue_adapter_for_test
end

def require_active_job_test_adapter!(method)
  unless using_test_adapter?
    raise ArgumentError.new("#{method} requires the Active Job test adapter, you're using #{queue_adapter.class.name}.")
  end
end

def using_test_adapter?
  queue_adapter.is_a?(ActiveJob::QueueAdapters::TestAdapter)
end

def validate_option(only: nil, except: nil)
  raise ArgumentError, "Cannot specify both `:only` and `:except` options." if only && except
end