require "spec_helper"

describe Fluent::Plugin::RedisSlowlogInput do
  let(:config) do
    { tag: "redis.slowlog" }
  end

  let(:plugin) { driver.instance }
  let(:slowlog) { [] }
  let(:fake_redis) { instance_double(Redis, ping: "PONG", quit: "OK", slowlog: slowlog) }

  subject(:driver) do
    config_string = config.map { |key, value| "#{key} #{value}" }.join("\n")
    Fluent::Test::Driver::Input.new(described_class).configure(config_string)
  end

  before do
    Fluent::Test.setup
  end

  context "when redis can't be reached" do
    it "raises an error" do
      expect { driver.run }.to raise_error(Redis::CannotConnectError)
    end
  end

  context "when specifying redis connection attributes" do
    let(:config) do
      { tag: "redis.slowlog",
        url: "redis://:p4ssw0rd@10.0.1.1:6380/15",
        path: "/path/to/redis.sock",
        host: "localhost",
        port: 1234,
        password: "5iveL!fe" }
    end

    it "delegates all of them to redis-rb" do
      # In ruby 2.5+ this could be config.slice(*[])
      redis_params = config.select { |name, _| [:url, :path, :host, :port, :password].include?(name) }

      expect(Redis).to receive(:new).with(redis_params).and_return(fake_redis)

      driver.run
    end
  end

  context "when redis is available" do
    let(:config) do
      {
        tag: "redis.slowlog",
        interval: 0,
        logsize: 10
      }
    end

    before do
      allow(plugin).to receive(:redis).and_return(fake_redis)
    end

    after do
      # Wait for the thread polling redis to finish
      plugin.__send__(:watcher).join
    end

    it "does not raise errors" do
      expect { driver.run }.not_to raise_error
    end

    it "polls the slowlog with the configured interval and size" do
      expect(plugin).to receive(:sleep).with(0).ordered
      expect(fake_redis).to receive(:slowlog).with("get", 10).ordered
      expect(plugin).to receive(:sleep).with(0).ordered
      expect(fake_redis).to receive(:slowlog).with("get", 10).ordered

      # Limit to 2 cycles
      allow(plugin).to receive(:watching).thrice.and_return(true, true, false)

      driver.run
    end

    context "when the slowlog returns entries" do
      let(:slowlog) do
        [
          [25640, 1590522258, 1, %w[ping]],
          [25639, 1590522249, 1, %w[ping]],
          [25638, 1590522208, 5, %w[SCAN 0]]
        ]
      end

      let(:expected_entries) do
        slowlog.map(&method(:log_entry)).sort_by { |event| event["id"] }
      end

      let(:emitted_entries) { driver.events.map(&:last) }

      it "emits an event for each slowlog in reverse order" do
        driver.run(expect_emits: 3)

        expect(driver.events.size).to eq(3)
        expect(emitted_entries).to eq(expected_entries)
      end

      it "does not log the same event twice" do
        expect(fake_redis).to receive(:slowlog).and_return(slowlog.last(1), slowlog)

        driver.run(expect_emits: 3)

        expect(driver.events.size).to eq(3)
        expect(emitted_entries).to eq(expected_entries)
      end
    end
  end

  def log_entry(slowlog_entry)
    { "id" => slowlog_entry.first,
      "timestamp" => Time.at(slowlog_entry[1]),
      "exec_time" => slowlog_entry[2],
      "command" => slowlog_entry.last }
  end
end