/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.flink.api.java.typeutils;

import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.Map;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.junit.Test;

import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.BOOLEAN_TYPE_INFO;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.DOUBLE_TYPE_INFO;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.FLOAT_TYPE_INFO;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO;
import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

/**
 * Tests for extracting {@link org.apache.flink.api.common.typeinfo.TypeInformation} from types
 * using a {@link org.apache.flink.api.common.typeinfo.TypeInfoFactory}
 */
public class TypeInfoFactoryTest {

	@Test
	public void testSimpleType() {
		TypeInformation<?> ti = TypeExtractor.createTypeInfo(IntLike.class);
		assertEquals(INT_TYPE_INFO, ti);

		ti = TypeExtractor.getForClass(IntLike.class);
		assertEquals(INT_TYPE_INFO, ti);

		ti = TypeExtractor.getForObject(new IntLike());
		assertEquals(INT_TYPE_INFO, ti);
	}

	@Test
	public void testMyEitherGenericType() {
		MapFunction<Boolean, MyEither<Boolean, String>> f = new MyEitherMapper<>();
		TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(f, BOOLEAN_TYPE_INFO);
		assertTrue(ti instanceof EitherTypeInfo);
		EitherTypeInfo eti = (EitherTypeInfo) ti;
		assertEquals(BOOLEAN_TYPE_INFO, eti.getLeftType());
		assertEquals(STRING_TYPE_INFO, eti.getRightType());
	}

	@Test
	public void testMyOptionGenericType() {
		TypeInformation<MyOption<Tuple2<Boolean, String>>> inTypeInfo = new MyOptionTypeInfo<>(
			new TupleTypeInfo<Tuple2<Boolean, String>>(BOOLEAN_TYPE_INFO, STRING_TYPE_INFO));
		MapFunction<MyOption<Tuple2<Boolean, String>>, MyOption<Tuple2<Boolean, Boolean>>> f = new MyOptionMapper<>();
		TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(f, inTypeInfo);
		assertTrue(ti instanceof MyOptionTypeInfo);
		MyOptionTypeInfo oti = (MyOptionTypeInfo) ti;
		assertTrue(oti.getInnerType() instanceof TupleTypeInfo);
		TupleTypeInfo tti = (TupleTypeInfo) oti.getInnerType();
		assertEquals(BOOLEAN_TYPE_INFO, tti.getTypeAt(0));
		assertEquals(BOOLEAN_TYPE_INFO, tti.getTypeAt(1));
	}

	@Test
	public void testMyTuple() {
		TypeInformation<Tuple1<MyTuple<Double, String>>> inTypeInfo = new TupleTypeInfo<>(
			new MyTupleTypeInfo(DOUBLE_TYPE_INFO, STRING_TYPE_INFO));
		MapFunction<Tuple1<MyTuple<Double, String>>, Tuple1<MyTuple<Boolean, Double>>> f = new MyTupleMapperL2<>();
		TypeInformation<?> ti = TypeExtractor.getMapReturnTypes(f, inTypeInfo);
		assertTrue(ti instanceof TupleTypeInfo);
		TupleTypeInfo<?> tti = (TupleTypeInfo<?>) ti;
		assertTrue(tti.getTypeAt(0) instanceof MyTupleTypeInfo);
		MyTupleTypeInfo mtti = (MyTupleTypeInfo) tti.getTypeAt(0);
		assertEquals(BOOLEAN_TYPE_INFO, mtti.getField0());
		assertEquals(DOUBLE_TYPE_INFO, mtti.getField1());
	}

	@Test
	public void testMyTupleHierarchy() {
		TypeInformation<?> ti = TypeExtractor.createTypeInfo(MyTuple2.class);
		assertTrue(ti instanceof MyTupleTypeInfo);
		MyTupleTypeInfo<?, ?> mtti = (MyTupleTypeInfo) ti;
		assertEquals(STRING_TYPE_INFO, mtti.getField0());
		assertEquals(BOOLEAN_TYPE_INFO, mtti.getField1());
	}

	@Test
	public void testMyTupleHierarchyWithInference() {
		TypeInformation<Tuple1<MyTuple3<Tuple1<Float>>>> inTypeInfo = new TupleTypeInfo<>(new MyTupleTypeInfo<>(
			new TupleTypeInfo<Tuple1<Float>>(FLOAT_TYPE_INFO), BOOLEAN_TYPE_INFO));
		MapFunction<Tuple1<MyTuple3<Tuple1<Float>>>, Tuple1<MyTuple3<Tuple2<Float, String>>>> f = new MyTuple3Mapper<>();
		TypeInformation ti = TypeExtractor.getMapReturnTypes(f, inTypeInfo);
		assertTrue(ti instanceof TupleTypeInfo);
		TupleTypeInfo<?> tti = (TupleTypeInfo) ti;
		assertTrue(tti.getTypeAt(0) instanceof MyTupleTypeInfo);
		MyTupleTypeInfo mtti = (MyTupleTypeInfo) tti.getTypeAt(0);
		assertEquals(new TupleTypeInfo<>(FLOAT_TYPE_INFO, STRING_TYPE_INFO), mtti.getField0());
		assertEquals(BOOLEAN_TYPE_INFO, mtti.getField1());
	}

	@Test(expected = InvalidTypesException.class)
	public void testMissingTypeInfo() {
		MapFunction f = new MyFaultyMapper();
		TypeExtractor.getMapReturnTypes(f, INT_TYPE_INFO);
	}

	@Test(expected = InvalidTypesException.class)
	public void testMissingTypeInference() {
		MapFunction f = new MyFaultyMapper2();
		TypeExtractor.getMapReturnTypes(f, new MyFaultyTypeInfo());
	}

	// --------------------------------------------------------------------------------------------
	//  Utilities
	// --------------------------------------------------------------------------------------------

	public static class MyTuple3Mapper<Y> implements MapFunction<Tuple1<MyTuple3<Tuple1<Y>>>, Tuple1<MyTuple3<Tuple2<Y, String>>>> {
		@Override
		public Tuple1<MyTuple3<Tuple2<Y, String>>> map(Tuple1<MyTuple3<Tuple1<Y>>> value) throws Exception {
			return null;
		}
	}

	public static class MyTuple3<T> extends MyTuple<T, Boolean> {
		// empty
	}

	public static class MyTuple2 extends MyTuple<String, Boolean> {
		// empty
	}

	public static class MyFaultyMapper2<T> implements MapFunction<MyFaulty<T>, MyFaulty<T>> {
		@Override
		public MyFaulty<T> map(MyFaulty<T> value) throws Exception {
			return null;
		}
	}

	public static class MyFaultyMapper<T> implements MapFunction<T, MyFaulty<T>> {
		@Override
		public MyFaulty<T> map(T value) throws Exception {
			return null;
		}
	}

	@TypeInfo(FaultyTypeInfoFactory.class)
	public static class MyFaulty<Y> {
		// empty
	}

	public static class FaultyTypeInfoFactory extends TypeInfoFactory {
		@Override
		public TypeInformation createTypeInfo(Type t, Map genericParameters) {
			return null;
		}
	}

	public static class MyFaultyTypeInfo extends TypeInformation<MyFaulty> {
		@Override
		public boolean isBasicType() {
			return false;
		}

		@Override
		public boolean isTupleType() {
			return false;
		}

		@Override
		public int getArity() {
			return 0;
		}

		@Override
		public int getTotalFields() {
			return 0;
		}

		@Override
		public Class<MyFaulty> getTypeClass() {
			return null;
		}

		@Override
		public boolean isKeyType() {
			return false;
		}

		@Override
		public TypeSerializer<MyFaulty> createSerializer(ExecutionConfig config) {
			return null;
		}

		@Override
		public String toString() {
			return null;
		}

		@Override
		public boolean equals(Object obj) {
			return false;
		}

		@Override
		public int hashCode() {
			return 0;
		}

		@Override
		public boolean canEqual(Object obj) {
			return false;
		}
	}

	public static class MyTupleMapperL1<A, B> implements MapFunction<Tuple1<MyTuple<A, String>>, Tuple1<MyTuple<B, A>>> {
		@Override
		public Tuple1<MyTuple<B, A>> map(Tuple1<MyTuple<A, String>> value) throws Exception {
			return null;
		}
	}

	public static class MyTupleMapperL2<C> extends MyTupleMapperL1<C, Boolean> {
		// empty
	}

	@TypeInfo(MyTupleTypeInfoFactory.class)
	public static class MyTuple<T0, T1> {
		// empty
	}

	public static class MyTupleTypeInfoFactory extends TypeInfoFactory<MyTuple> {
		@Override
		@SuppressWarnings("unchecked")
		public TypeInformation<MyTuple> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParameters) {
			return new MyTupleTypeInfo(genericParameters.get("T0"), genericParameters.get("T1"));
		}
	}

	public static class MyTupleTypeInfo<T0, T1> extends TypeInformation<MyTuple<T0, T1>> {
		private TypeInformation field0;
		private TypeInformation field1;

		public TypeInformation getField0() {
			return field0;
		}

		public TypeInformation getField1() {
			return field1;
		}

		public MyTupleTypeInfo(TypeInformation field0, TypeInformation field1) {
			this.field0 = field0;
			this.field1 = field1;
		}

		@Override
		public boolean isBasicType() {
			return false;
		}

		@Override
		public boolean isTupleType() {
			return false;
		}

		@Override
		public int getArity() {
			return 0;
		}

		@Override
		public int getTotalFields() {
			return 0;
		}

		@Override
		public Class<MyTuple<T0, T1>> getTypeClass() {
			return null;
		}

		@Override
		public boolean isKeyType() {
			return false;
		}

		@Override
		public TypeSerializer<MyTuple<T0, T1>> createSerializer(ExecutionConfig config) {
			return null;
		}

		@Override
		public String toString() {
			return null;
		}

		@Override
		public boolean equals(Object obj) {
			return false;
		}

		@Override
		public int hashCode() {
			return 0;
		}

		@Override
		public boolean canEqual(Object obj) {
			return false;
		}

		@Override
		public Map<String, TypeInformation<?>> getGenericParameters() {
			Map<String, TypeInformation<?>> map = new HashMap<>(2);
			map.put("T0", field0);
			map.put("T1", field1);
			return map;
		}
	}

	public static class MyOptionMapper<T> implements MapFunction<MyOption<Tuple2<T, String>>, MyOption<Tuple2<T, T>>> {
		@Override
		public MyOption<Tuple2<T, T>> map(MyOption<Tuple2<T, String>> value) throws Exception {
			return null;
		}
	}

	@TypeInfo(MyOptionTypeInfoFactory.class)
	public static class MyOption<T> {
		// empty
	}

	public static class MyOptionTypeInfoFactory<T> extends TypeInfoFactory<MyOption<T>> {
		@Override
		@SuppressWarnings("unchecked")
		public TypeInformation<MyOption<T>> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParams) {
			return new MyOptionTypeInfo(genericParams.get("T"));
		}
	}

	public static class MyOptionTypeInfo<T> extends TypeInformation<MyOption<T>> {

		private final TypeInformation<T> innerType;

		public MyOptionTypeInfo(TypeInformation<T> innerType) {
			this.innerType = innerType;
		}

		public TypeInformation<T> getInnerType() {
			return innerType;
		}

		@Override
		public boolean isBasicType() {
			return false;
		}

		@Override
		public boolean isTupleType() {
			return false;
		}

		@Override
		public int getArity() {
			return 0;
		}

		@Override
		public int getTotalFields() {
			return 1;
		}

		@Override
		public Class<MyOption<T>> getTypeClass() {
			return null;
		}

		@Override
		public boolean isKeyType() {
			return false;
		}

		@Override
		public TypeSerializer<MyOption<T>> createSerializer(ExecutionConfig config) {
			return null;
		}

		@Override
		public String toString() {
			return null;
		}

		@Override
		public boolean equals(Object obj) {
			return false;
		}

		@Override
		public int hashCode() {
			return 0;
		}

		@Override
		public boolean canEqual(Object obj) {
			return false;
		}

		@Override
		public Map<String, TypeInformation<?>> getGenericParameters() {
			Map<String, TypeInformation<?>> map = new HashMap<>(1);
			map.put("T", innerType);
			return map;
		}
	}

	public static class MyEitherMapper<T> implements MapFunction<T, MyEither<T, String>> {
		@Override
		public MyEither<T, String> map(T value) throws Exception {
			return null;
		}
	}

	@TypeInfo(MyEitherTypeInfoFactory.class)
	public static class MyEither<A, B> {
		// empty
	}

	public static class MyEitherTypeInfoFactory<A, B> extends TypeInfoFactory<MyEither<A, B>> {
		@Override
		@SuppressWarnings("unchecked")
		public TypeInformation<MyEither<A,B>> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParams) {
			return new EitherTypeInfo(genericParams.get("A"), genericParams.get("B"));
		}
	}

	@TypeInfo(IntLikeTypeInfoFactory.class)
	public static class IntLike {
		// empty
	}

	public static class IntLikeTypeInfoFactory extends TypeInfoFactory<IntLike> {
		@Override
		@SuppressWarnings("unchecked")
		public TypeInformation<IntLike> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParams) {
			return (TypeInformation) INT_TYPE_INFO;
		}
	}

}