# Encoding: utf-8

require 'java'
require 'askari/spec_helpers/key_value'
require 'askari/spec_helpers/hbase'
require 'askari/client/client_errors'
require 'askari/client/relation'
require 'askari/region_observer/dsl'
require 'askari/region_observer/runner'
require 'askari/region_observer/execution_context'

describe Askari::RegionObserver::ExecutionContext do

  include_context 'hbase'

  let(:schema_text) do <<-YAML
    name: my-app
    format: 0
    current_version_number: 0

    versions:
      - number: 0
        tables:
          posts:
            relations:
              authors:
                row_key_type: {author_id: bytes}
                version_column: [a, v]
                families:
                  a:
                    qualifiers:
                      v:
                        qualifier_type: v
                        value_type: {version: i8}
              posts:
                row_key_type: [{author_id: bytes}, {post_id: i32}]
                version_column: [m, v]
                families:
                  m:
                    qualifiers:
                      v:
                        qualifier_type: v
                        value_type: {version: i8}
                      t:
                        qualifier_type: t
                        value_type: {title: bytes}
                  p:
                    qualifiers:
                      b:
                        qualifier_type: b
                        value_type: {body: bytes}
          another_table:
            relations:
              another_relation:
                row_key_type: {id: bytes}
                version_column: [a, v]
                families:
                  a:
                    qualifiers:
                      v:
                        qualifier_type: v
                        value_type: {version: i8}
      - number: 1
        tables:
          posts:
            relations:
              authors:
                row_key_type: {author_id: bytes}
                version_column: [a, v]
                families:
                  a:
                    qualifiers:
                      v:
                        qualifier_type: v
                        value_type: {version: i8}
  YAML
  end

  schema :schema_text

  let(:table_name) { 'posts' }
  let(:relation_name) { 'authors' }
  let(:family_name) { 'a' }
  let(:table_name_bytes) { table_name.to_java_bytes }
  let(:region_info) { double('HRegionInfo', table_name: table_name_bytes) }
  let(:region) { double('HRegion', region_info: region_info) }
  let(:table) { double('HTable') }
  let(:event) { :post_put }

  include_context 'key-value', 'authors'

  let(:configuration) { double('HBaseConfiguration', get: nil) }

  let(:environment) do
    double(
      'RegionCoprocessorEnvironment',
      configuration: configuration,
      region: region,
      get_table: table
    )
  end

  let(:context) { double('ObserverContext', environment: environment) }

  def new_region_observer(&block)
    the_schema = schema_text
    Module.new do
      include Askari::RegionObserver::DSL
      schema :inline, the_schema
      module_eval(&block) if block
    end
  end

  let(:region_observer) { new_region_observer }
  let(:application) { region_observer.application }
  let(:runner) { Askari::RegionObserver::Runner.new(application) }

  let(:execution_context) do
    described_class.new(runner).tap do |execution_context|
      execution_context.context = context
    end
  end

  describe '#initialize' do
    let(:handler_support_module) { Module.new }

    let(:region_observer) do
      support_module = handler_support_module
      new_region_observer do
        include support_module
      end
    end

    it 'takes a runner' do
      expect { described_class.new(runner) }.to_not raise_error
    end

    it 'mixes in application handler modules' do
      expect(described_class.new(runner)).to be_a(handler_support_module)
    end
  end

  describe '#runner' do
    it 'is the runner passed at initialization' do
      expect(execution_context.runner).to eq(runner)
    end
  end

  describe '#application_schema' do
    it "is the runner's application schema" do
      expect(execution_context.application_schema).to eq(runner.application_schema)
    end
  end

  describe '#version_schema' do
    it "is the runner's version schema" do
      expect(execution_context.version_schema).to eq(runner.version_schema)
    end
  end

  describe '#log' do
    let(:log) { double('log') }

    it 'is a log4j logger' do
      expect(execution_context.log).to be_a(Java::OrgApacheLog4j::Logger)
    end

    it 'is named after the application and coprocessor' do
      coprocessor_name = 'coprocessor'
      configuration.should_receive(:get).with('askari.name').and_return(coprocessor_name)
      expect(execution_context.log.name).to eq("#{schema.name}:#{coprocessor_name}")
    end
  end

  describe '#context' do
    it 'is a read-write attribute' do
      execution_context.context = :context
      expect(execution_context.context).to eq(:context)
    end
  end

  describe '#environment' do
    it 'is the coprocessor environment available from the observer context' do
      expect(execution_context.environment).to eq(context.environment)
    end

    it 'is aliased as #env' do
      expect(execution_context.env).to eq(execution_context.environment)
    end
  end

  describe '#configuration' do
    it 'is the hbase configuration available from the coprocessor environment' do
      expect(execution_context.configuration).to eq(environment.configuration)
    end

    it 'is aliased as #config' do
      expect(execution_context.config).to eq(execution_context.configuration)
    end
  end

  describe '#region' do
    it 'is the local region available from the coprocessor environment' do
      expect(execution_context.region).to eq(environment.region)
    end
  end

  describe '#region_info' do
    it 'is the region info from the local region' do
      expect(execution_context.region_info).to eq(region.region_info)
    end
  end

  shared_examples 'relation schema detection' do
    stub_key_value(
      qualifier_name: 'v',
      row_key: { author_id: 'foo' },
      qualifier: 'v',
      value: { version: 0 },
      timestamp: 1337
    )

    let(:family_map) do
      Java::JavaUtil::HashMap.new.tap do |map|
        map.put(family_encoded, Java::JavaUtil::TreeSet.new)
      end
    end

    # All HBase operations seem to expose getFamilyMap.
    let(:operation) { double('operation', family_map: family_map) }

    it 'can take a symbolic relation name' do
      relation = create_from(relation_name.to_sym)
      expect(relation).to be_a(Askari::Client::Relation)
      expect(relation.relation_schema.name).to eq(relation_name)
    end

    it 'can take a key_value instead of a relation name' do
      relation = create_from(key_value)
      expect(relation).to be_a(Askari::Client::Relation)
      expect(relation.relation_schema.name).to eq(relation_name)
    end

    it 'can take the family names from a client operation' do
      relation = create_from(operation)
      expect(relation).to be_a(Askari::Client::Relation)
      expect(relation.relation_schema.name).to eq(relation_name)
    end

    it 'complains if the operation family names span multiple relations' do
      family_map.put('m'.to_java_bytes, Java::JavaUtil::TreeSet.new)
      expect { create_from(operation) }.to raise_error(Askari::Client::NoRelationError)
    end

    it "complains if the family isn't in the context's table" do
      key_value.should_receive(:family).and_return('x'.to_java_bytes)
      expect { create_from(key_value) }.to raise_error(Askari::Client::NoRelationError)
    end

    it "complains if it can't detect a family on the argument" do
      expect do
        create_from(7)
      end.to raise_error(Askari::Client::NoRelationError, /Can't find relation for 7/)
    end

    it "complains if it can't find the relation by name" do
      expect do
        create_from('nerf_herder')
      end.to raise_error(
        Askari::Client::NoRelationError, /Can't find relation "nerf_herder"/
      )
    end

    context 'when a namespace is in effect' do
      let(:region_info) do
        Java::OrgApacheHadoopHbase::HRegionInfo.new("my_#{table_name}".to_java_bytes)
      end

      it 'finds the namespaced table' do
        configuration.stub(:get).with('askari.namespace').and_return('my_')
        expect(create_from(operation).relation_schema.name).to eq(relation_name)
      end
    end
  end

  describe '#relation' do
    include_examples 'relation schema detection'

    def create_from(argument)
      execution_context.relation(argument)
    end

    let(:relation) { create_from(relation_name) }

    it 'takes a relation name and returns a relation client' do
      environment.should_receive(:get_table) do |name_bytes|
        expect(name_bytes.to_s).to eq(table_name_bytes.to_s)
        table
      end

      expect(relation).to be_a(Askari::Client::Relation)
      expect(relation.htable_client).to eq(table)
    end

    it 'memoizes' do
      one = execution_context.relation(key_value)
      another = execution_context.relation(key_value)

      expect(one).to eq(another)
    end
  end

  describe '#local_relation' do
    let(:region_info) do
      Java::OrgApacheHadoopHbase::HRegionInfo.new(table_name.to_java_bytes)
    end

    let(:region) do
      the_region_info = region_info
      Java::OrgApacheHadoopHbaseRegionserver::HRegion.new.tap do |region|
        region.extend(
          Module.new do
            # There really is no easy way to set the member with java reflection.
            # So monkey patch it out here instead.
            define_method :region_info do
              the_region_info
            end
          end
        )
      end
    end

    def create_from(argument)
      execution_context.local_relation(argument)
    end

    let(:relation) { create_from(relation_name) }

    include_examples 'relation schema detection'

    it 'is a relation' do
      expect(relation).to be_a(Askari::Client::Relation)
    end

    it 'is driven by the local HRegion dressed as an HTable client' do
      expect(relation.htable_client).to be_a(Java::AskariClient::HRegionAsHTable)
      expect(relation.htable_client.region).to be(region)
    end

    it 'rejects relations not in the current table' do
      expect do
        execution_context.local_relation('another_relation')
      end.to raise_error(
        Askari::Client::NoRelationError, /"another_relation" is not in "posts"/
      )
    end
  end

end
