/*
 * Copyright (C) 2015 Google 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 com.google.cloud.dataflow.sdk.coders;

import com.google.api.client.util.Preconditions;
import com.google.cloud.dataflow.sdk.values.TypeDescriptor;

import org.hamcrest.Matchers;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

import java.io.Serializable;

/**
 * Tests of Coder defaults.
 */
@RunWith(JUnit4.class)
@SuppressWarnings("serial")
public class DefaultCoderTest {

  @Rule
  public ExpectedException thrown = ExpectedException.none();

  @DefaultCoder(AvroCoder.class)
  private static class AvroRecord {
  }

  private static class SerializableBase implements Serializable {
  }

  @DefaultCoder(SerializableCoder.class)
  private static class SerializableRecord extends SerializableBase {
  }

  @DefaultCoder(CustomCoder.class)
  private static class CustomRecord extends SerializableBase {
  }

  private static class Unknown {
  }

  private static class CustomCoder extends SerializableCoder<CustomRecord> {
    // Extending SerializableCoder isn't trivial, but it can be done.
    @SuppressWarnings("unchecked")
    public static <T extends Serializable> SerializableCoder<T> of(Class<T> recordType) {
       Preconditions.checkArgument(
           CustomRecord.class.isAssignableFrom(recordType));
       return (SerializableCoder<T>) new CustomCoder();
    }

    protected CustomCoder() {
      super(CustomRecord.class);
    }
  }

  @Test
  public void testDefaultCoders() throws Exception {
    checkDefault(AvroRecord.class, AvroCoder.class);
    checkDefault(SerializableBase.class, SerializableCoder.class);
    checkDefault(SerializableRecord.class, SerializableCoder.class);
    checkDefault(CustomRecord.class, CustomCoder.class);
  }

  @Test
  public void testUnknown() throws Exception {
    thrown.expect(CannotProvideCoderException.class);
    CoderRegistry registry = new CoderRegistry();
    registry.getDefaultCoder(Unknown.class);
  }

  /**
   * Checks that the default Coder for {@code valueType} is an instance of
   * {@code expectedCoder}.
   */
  private void checkDefault(Class<?> valueType,
      Class<?> expectedCoder) throws Exception {
    CoderRegistry registry = new CoderRegistry();
    Coder<?> coder = registry.getDefaultCoder(TypeDescriptor.of(valueType));
    Assert.assertThat(coder, Matchers.instanceOf(expectedCoder));
  }
}

@@ -636,16 +636,153 @@
 oders;%0A%0A
+import static org.hamcrest.Matchers.equalTo;%0Aimport static org.hamcrest.Matchers.instanceOf;%0Aimport static org.junit.Assert.assertThat;%0A%0A
 import c
@@ -898,50 +898,20 @@
 org.
-hamcrest.Matchers;%0Aimport org.junit.Assert
+junit.Before
 ;%0Aim
@@ -1091,16 +1091,39 @@
 lizable;
+%0Aimport java.util.List;
 %0A%0A/**%0A *
@@ -1307,16 +1307,167 @@
 one();%0A%0A
+  public CoderRegistry registry = new CoderRegistry();%0A%0A  @Before%0A  public void registerStandardCoders() %7B%0A    registry.registerStandardCoders();%0A  %7D%0A%0A
   @Defau
@@ -2399,24 +2399,30 @@
 DefaultCoder
+Classe
 s() throws E
@@ -2436,28 +2436,51 @@
 n %7B%0A    
-check
+assertThat(registry.get
 Default
+Coder
 (AvroRec
@@ -2488,18 +2488,30 @@
 rd.class
+)
 , 
+instanceOf(
 AvroCode
@@ -2518,34 +2518,58 @@
 r.class)
+)
 ;%0A    
-check
+assertThat(registry.get
 Default
+Coder
 (Seriali
@@ -2579,26 +2579,46 @@
 leBase.class
-, 
+),%0A        instanceOf(
 Serializable
@@ -2629,34 +2629,58 @@
 r.class)
+)
 ;%0A    
-check
+assertThat(registry.get
 Default
+Coder
 (Seriali
@@ -2696,18 +2696,38 @@
 rd.class
-, 
+),%0A        instanceOf(
 Serializ
@@ -2746,26 +2746,50 @@
 ass)
+)
 ;%0A    
-check
+assertThat(registry.get
 Default
+Coder
 (Cus
@@ -2803,18 +2803,30 @@
 rd.class
+)
 , 
+instanceOf(
 CustomCo
@@ -2831,24 +2831,25 @@
 Coder.class)
+)
 ;%0A  %7D%0A%0A  @Te
@@ -2869,22 +2869,39 @@
 oid test
-Unknow
+DefaultCoderInCollectio
 n() thro
@@ -2923,560 +2923,498 @@
     
-thrown.expect(CannotProvideCoderException.class);%0A    CoderRegistry registry = new CoderRegistry();%0A    registry.getDefaultCoder(Unknown.class);%0A  %7D%0A%0A  /**%0A   * Checks that the default Coder for %7B@code valueType%7D is an instance of%0A   * %7B@code expectedCoder%7D.%0A   */%0A  private void checkDefault(Class%3C?%3E valueType,%0A      Class%3C?%3E expectedCoder) throws Exception %7B%0A    CoderRegistry registry = new CoderRegistry();%0A    Coder%3C?%3E coder = registry.getDefaultCoder(TypeDescriptor.of(valueType));%0A    Assert.assertThat(coder, Matchers.instanceOf(expectedCoder)
+assertThat(registry.getDefaultCoder(new TypeDescriptor%3CList%3CAvroRecord%3E%3E()%7B%7D),%0A        instanceOf(ListCoder.class));%0A    assertThat(registry.getDefaultCoder(new TypeDescriptor%3CList%3CSerializableRecord%3E%3E()%7B%7D),%0A        equalTo((Coder%3CList%3CSerializableRecord%3E%3E)%0A            ListCoder.of(SerializableCoder.of(SerializableRecord.class))));%0A  %7D%0A%0A  @Test%0A  public void testUnknown() throws Exception %7B%0A    thrown.expect(CannotProvideCoderException.class);%0A    registry.getDefaultCoder(Unknown.class
 );%0A 

Test @DefaultCoder for collection contents