# SPDX-License-Identifier: Apache-2.0
#
#  The OpenSearch Contributors require contributions made to
#  this file be licensed under the Apache-2.0 license or a
#  compatible open source license.
#
#  Modifications Copyright OpenSearch Contributors. See
#  GitHub history for details.

require_relative "../../../spec/opensearch_spec_helper"
require "logstash/outputs/opensearch"
require "stringio"

RSpec::Matchers.define :a_valid_gzip_encoded_string do
  match { |data|
    expect { Zlib::GzipReader.new(StringIO.new(data)).read }.not_to raise_error
  }
end


describe "indexing with http_compression turned on", :integration => true do
  let(:event) { LogStash::Event.new("message" => "Hello World!", "type" => type) }
  let(:index) { 10.times.collect { rand(10).to_s }.join("") }
  let(:type) { "_doc" }
  let(:event_count) { 10000 + rand(500) }
  let(:events) { event_count.times.map { event }.to_a }
  let(:config) {
    {
      "hosts" => get_host_port,
      "index" => index,
      "http_compression" => true
    }
  }
  subject { LogStash::Outputs::OpenSearch.new(config) }

  let(:opensearch_url) { "http://#{get_host_port}" }
  let(:index_url) {"#{opensearch_url}/#{index}"}
  let(:http_client_options) { {} }
  let(:http_client) do
    Manticore::Client.new(http_client_options)
  end

  before do
    subject.register
    subject.multi_receive([])
  end

  shared_examples "an indexer" do
    it "ships events" do
      subject.multi_receive(events)

      http_client.post("#{opensearch_url}/_refresh").call

      response = http_client.get("#{index_url}/_count?q=*")
      result = LogStash::Json.load(response.body)
      cur_count = result["count"]
      expect(cur_count).to eq(event_count)

      response = http_client.get("#{index_url}/_search?q=*&size=1000")
      result = LogStash::Json.load(response.body)
      result["hits"]["hits"].each do |doc|
        # FIXME This checks for OpenSearch 1.x or OpenDistro which has version 7.10.x
        # need a cleaner way to check this.
        if OpenSearchHelper.check_version?("< 2") || OpenSearchHelper.check_version?("> 7")
          expect(doc["_type"]).to eq(type)
        else
          expect(doc).not_to include("_type")
        end
        expect(doc["_index"]).to eq(index)
      end
    end
  end

  it "sets the correct content-encoding header and body is compressed" do
    expect(subject.client.pool.adapter.client).to receive(:send).
      with(anything, anything, {:headers=>{"Content-Encoding"=>"gzip", "content-type"=>"application/json"}, :body => a_valid_gzip_encoded_string}).
      and_call_original
    subject.multi_receive(events)
  end

  it_behaves_like("an indexer")
end