/*
 * Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
 * license agreements.  See the NOTICE file distributed with this work for
 * additional information regarding copyright ownership.  Crate 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.
 *
 * However, if you have executed another commercial license agreement
 * with Crate these terms will supersede the license and you may use the
 * software solely pursuant to the terms of the relevant commercial agreement.
 */

package io.crate.metadata;

import com.google.common.collect.Sets;
import io.crate.analyze.WhereClause;
import io.crate.integrationtests.SQLTransportIntegrationTest;
import io.crate.metadata.doc.DocTableInfo;
import io.crate.metadata.table.SchemaInfo;
import io.crate.metadata.table.TableInfo;
import io.crate.test.integration.CrateIntegrationTest;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.junit.Before;
import org.junit.Test;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import static org.hamcrest.Matchers.contains;
import static org.hamcrest.core.Is.is;

@CrateIntegrationTest.ClusterScope(scope = CrateIntegrationTest.Scope.GLOBAL)
public class ReferenceInfosITest extends SQLTransportIntegrationTest {

    private ReferenceInfos referenceInfos;

    @Before
    public void setUpService() {
        referenceInfos = cluster().getInstance(ReferenceInfos.class);
    }

    @Test
    public void testDocTable() throws Exception {
        execute("create table t1 (id int primary key, name string, details object(dynamic) as (size byte, created timestamp)) clustered into 10 shards with(number_of_replicas=1)");
        ensureGreen();

        DocTableInfo ti = (DocTableInfo) referenceInfos.getTableInfo(new TableIdent(null, "t1"));
        assertThat(ti.ident().name(), is("t1"));

        assertThat(ti.columns().size(), is(3));
        assertThat(ti.primaryKey().size(), is(1));
        assertThat(ti.primaryKey().get(0), is(new ColumnIdent("id")));
        assertThat(ti.clusteredBy(), is(new ColumnIdent("id")));

        Routing routing = ti.getRouting(WhereClause.MATCH_ALL);

        Set<String> nodes = routing.nodes();

        assertThat(nodes.size(), is(2));
        int numShards = 0;
        for (Map.Entry<String, Map<String, Set<Integer>>> nodeEntry : routing.locations().entrySet()) {
            for (Map.Entry<String, Set<Integer>> indexEntry : nodeEntry.getValue().entrySet()) {
                assertThat(indexEntry.getKey(), is("t1"));
                numShards += indexEntry.getValue().size();
            }
        }
        assertThat(numShards, is(10));
    }

    @Test
    public void testTableAlias() throws Exception {
        execute("create table terminator (model string, good boolean, actor object)");
        IndicesAliasesRequest request = new IndicesAliasesRequest();
        request.addAlias("entsafter", "terminator");
        client().admin().indices().aliases(request).actionGet();
        ensureGreen();

        TableInfo terminatorTable = referenceInfos.getTableInfo(new TableIdent(null, "terminator"));
        TableInfo entsafterTable = referenceInfos.getTableInfo(new TableIdent(null, "entsafter"));
        assertNotNull(terminatorTable);
        assertFalse(terminatorTable.isAlias());

        assertNotNull(entsafterTable);
        assertTrue(entsafterTable.isAlias());
    }

    @Test
    public void testAliasPartitions() throws Exception {
        execute("create table terminator (model string, good boolean, actor object)");
        execute("create table transformer (model string, good boolean, actor object)");
        IndicesAliasesRequest request = new IndicesAliasesRequest();
        request.addAlias("entsafter", "terminator");
        request.addAlias("entsafter", "transformer");
        client().admin().indices().aliases(request).actionGet();
        ensureGreen();

        TableInfo entsafterTable = referenceInfos.getTableInfo(new TableIdent(null, "entsafter"));
        assertNotNull(entsafterTable);
        assertThat(entsafterTable.concreteIndices().length, is(2));
        assertThat(Arrays.asList(entsafterTable.concreteIndices()), contains("terminator", "transformer"));
    }


    @Test
    public void testNodesTable() throws Exception {
        TableInfo ti = referenceInfos.getTableInfo(new TableIdent("sys", "nodes"));
        Routing routing = ti.getRouting(null);
        assertTrue(routing.hasLocations());
        assertEquals(2, routing.nodes().size());
        for (Map<String, Set<Integer>> indices : routing.locations().values()) {
            assertEquals(0, indices.size());
        }
    }

    @Test
    public void testShardsTable() throws Exception {
        execute("create table t2 (id int primary key) clustered into 4 shards with(number_of_replicas=0)");
        execute("create table t3 (id int primary key) clustered into 8 shards with(number_of_replicas=0)");
        ensureGreen();

        TableInfo ti = referenceInfos.getTableInfo(new TableIdent("sys", "shards"));
        Routing routing = ti.getRouting(null);

        Set<String> tables = new HashSet<>();
        Set<String> expectedTables = Sets.newHashSet("t2", "t3");
        int numShards = 0;
        for (Map.Entry<String, Map<String, Set<Integer>>> nodeEntry : routing.locations().entrySet()) {
            for (Map.Entry<String, Set<Integer>> indexEntry : nodeEntry.getValue().entrySet()) {
                tables.add(indexEntry.getKey());
                numShards += indexEntry.getValue().size();
            }
        }
        assertThat(numShards, is(12));
        assertThat(tables, is(expectedTables));
    }

    @Test
    public void testClusterTable() throws Exception {
        TableInfo ti = referenceInfos.getTableInfo(new TableIdent("sys", "cluster"));
        assertTrue(ti.getRouting(null).locations().containsKey(null));
    }

    @Test
    public void testSysSchemaTables() throws Exception {
        SchemaInfo si = referenceInfos.getSchemaInfo("sys");
        // TODO:
//        assertThat(si.tableNames(), contains("cluster", "nodes", "shards", "jobs", "jobs_log",
//            "operations", "operations_log"));
    }

    @Test
    public void testDocSchemaTables() throws Exception {
        execute("create table users (id int primary key)");
        ensureGreen();
        // TODO:
//        SchemaInfo si = referenceInfos.getSchemaInfo(null);
//        assertThat(si.tableNames(), containsInAnyOrder("users"));
    }


}

@@ -1629,16 +1629,61 @@
 ntains;%0A
+import static org.hamcrest.Matchers.isOneOf;%0A
 import s
@@ -1713,16 +1713,16 @@
 .Is.is;%0A
-
 %0A@CrateI
@@ -2128,41 +2128,110 @@
 t1 (
-id int primary key, name string, 
+%22 +%0A                  %22id int primary key, %22 +%0A                  %22name string, %22 +%0A                  %22
 deta
@@ -2283,16 +2283,37 @@
 mestamp)
+%22 +%0A                %22
 ) cluste
@@ -2327,32 +2327,33 @@
 o 10 shards with
+ 
 (number_of_repli
@@ -2912,29 +2912,141 @@
 s.size(), is
-(2));
+OneOf(1, 2)); // for the rare case%0A                                                 // where all shards are on 1 node
 %0A        int

fix flaky ReferenceInfosITest