/**
 * Copyright 2012 Twitter, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package parquet.cascading;

import cascading.flow.FlowProcess;
import cascading.scheme.SinkCall;
import cascading.tap.Tap;
import com.twitter.data.proto.tutorial.thrift.AddressBook;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.RecordReader;
import org.junit.Test;

import static org.junit.Assert.*;
import static org.mockito.Mockito.mock;

import parquet.hadoop.api.ReadSupport;
import parquet.hadoop.thrift.ThriftReadSupport;
import parquet.io.api.RecordMaterializer;
import parquet.schema.MessageType;
import parquet.thrift.TBaseRecordConverter;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class TestParquetTBaseScheme {

  @Test
  public void testGetRecordMaterializer() throws Exception {
    ThriftReadSupport<AddressBook> readSupport = new ThriftReadSupport<AddressBook>();
    ParquetTBaseScheme tBaseScheme = new ParquetTBaseScheme(AddressBook.class);
    String descriptorKey = "thrift.descriptor";
    String mockDescriptor = "{\"id\":\"STRUCT\"}";
    Map<String, String> metaData = new HashMap<String, String>();
    metaData.put(descriptorKey, mockDescriptor);
    FlowProcess<JobConf> flowProcess = mock(FlowProcess.class);
    Tap<JobConf, RecordReader, OutputCollector> tap = mock(Tap.class);
    JobConf jobConf = new JobConf();
    jobConf.set(ThriftReadSupport.THRIFT_READ_CLASS_KEY, AddressBook.class.getName());
    MessageType fileSchema = new MessageType("Test");
    ReadSupport.ReadContext readContext = new ReadSupport.ReadContext(fileSchema);

    //Mock the method call by cascading
    tBaseScheme.sourceConfInit(flowProcess, tap, jobConf);

    RecordMaterializer recordMaterializer = readSupport.prepareForRead(jobConf, metaData, fileSchema, readContext);
    assertNotNull(recordMaterializer);
    assertEquals(TBaseRecordConverter.class, recordMaterializer.getClass());
  }

  @Test
  public void testNotSupportSink() {
    ParquetTBaseScheme<AddressBook> scheme = new ParquetTBaseScheme<AddressBook>(AddressBook.class);
    assertFalse(scheme.isSink());
  }

  @Test(expected = UnsupportedOperationException.class)
  public void testNotSinkConfInit() {
    ParquetTBaseScheme<AddressBook> scheme = new ParquetTBaseScheme<AddressBook>(AddressBook.class);
    scheme.sinkConfInit(mock(FlowProcess.class),
            mock(Tap.class),
            mock(JobConf.class));
  }

  @Test(expected = UnsupportedOperationException.class)
  public void testNotSink() throws IOException {
    ParquetTBaseScheme<AddressBook> scheme = new ParquetTBaseScheme<AddressBook>(AddressBook.class);
    scheme.sink(mock(FlowProcess.class), mock(SinkCall.class));
  }
}

@@ -2515,779 +2515,7 @@
 %0A  %7D
-%0A%0A  @Test%0A  public void testNotSupportSink() %7B%0A    ParquetTBaseScheme%3CAddressBook%3E scheme = new ParquetTBaseScheme%3CAddressBook%3E(AddressBook.class);%0A    assertFalse(scheme.isSink());%0A  %7D%0A%0A  @Test(expected = UnsupportedOperationException.class)%0A  public void testNotSinkConfInit() %7B%0A    ParquetTBaseScheme%3CAddressBook%3E scheme = new ParquetTBaseScheme%3CAddressBook%3E(AddressBook.class);%0A    scheme.sinkConfInit(mock(FlowProcess.class),%0A            mock(Tap.class),%0A            mock(JobConf.class));%0A  %7D%0A%0A  @Test(expected = UnsupportedOperationException.class)%0A  public void testNotSink() throws IOException %7B%0A    ParquetTBaseScheme%3CAddressBook%3E scheme = new ParquetTBaseScheme%3CAddressBook%3E(AddressBook.class);%0A    scheme.sink(mock(FlowProcess.class), mock(SinkCall.class));%0A  %7D
 %0A%7D%0A

remove tests that check that TBaseScheme doesn't support writes