/*-
 * #%L
 * kylo-ui-app
 * %%
 * Copyright (C) 2017 ThinkBig Analytics
 * %%
 * 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.
 * #L%
 */
define(["angularMocks", "feed-mgr/visual-query/module-name", "feed-mgr/visual-query/module", "feed-mgr/visual-query/module-require"], function (mocks, moduleName) {
    describe("SparkDatasourceService", function () {
        // Include dependencies
        beforeEach(mocks.module("kylo", "kylo.feedmgr", moduleName));

        // toScript
        it("should produce Spark for one table", mocks.inject(function (SparkDatasourceService) {
            var spark = SparkDatasourceService.toScript({
                "nodes": [{
                    "id": 11,
                    "name": "tickit.sales",
                    "nodeAttributes": {
                        "attributes": [{"name": "salesid", "dataType": "int", "selected": false}, {"name": "buyerid", "dataType": "int", "selected": false},
                            {"name": "eventid", "dataType": "int", "selected": false}, {"name": "dateid", "dataType": "smallint", "selected": false},
                            {"name": "qtysold", "dataType": "string", "selected": true}, {"name": "pricepaid", "dataType": "double", "selected": true},
                            {"name": "commission", "dataType": "double", "selected": true}], "sql": "`tickit`.`sales`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 4}, "left": {"location": "LEFT", "id": 1}, "right": {"location": "RIGHT", "id": 2}, "top": {"location": "TOP", "id": 3}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }], "connections": []
            });
            expect(spark).toBe("val tbl11 = sqlContext.table(\"tickit.sales\").alias(\"tbl11\")\n"
                               + "var df = tbl11.select(tbl11.col(\"qtysold\"), tbl11.col(\"pricepaid\"), tbl11.col(\"commission\"))\n");
        }));

        it("should produce Spark for joined tables", mocks.inject(function (SparkDatasourceService) {
            var spark = SparkDatasourceService.toScript({
                "nodes": [{
                    "id": 10,
                    "name": "tickit.users",
                    "nodeAttributes": {
                        "attributes": [{"name": "userid", "dataType": "int", "selected": false}, {"name": "username", "dataType": "string", "selected": true},
                            {"name": "firstname", "dataType": "string", "selected": true}, {"name": "lastname", "dataType": "string", "selected": true}], "sql": "`tickit`.`users`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 8}, "left": {"location": "LEFT", "id": 5}, "right": {"location": "RIGHT", "id": 6}, "top": {"location": "TOP", "id": 7}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 11,
                    "name": "tickit.sales",
                    "nodeAttributes": {
                        "attributes": [{"name": "salesid", "dataType": "int", "selected": false}, {"name": "buyerid", "dataType": "int", "selected": false},
                            {"name": "eventid", "dataType": "int", "selected": false}, {"name": "dateid", "dataType": "smallint", "selected": false},
                            {"name": "qtysold", "dataType": "string", "selected": true}, {"name": "pricepaid", "dataType": "double", "selected": true},
                            {"name": "commission", "dataType": "double", "selected": true}], "sql": "`tickit`.`sales`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 12}, "left": {"location": "LEFT", "id": 9}, "right": {"location": "RIGHT", "id": 10}, "top": {"location": "TOP", "id": 11}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 12,
                    "name": "tickit.event",
                    "nodeAttributes": {
                        "attributes": [{"name": "eventid", "dataType": "int", "selected": false}, {"name": "dateid", "dataType": "smallint", "selected": false},
                            {"name": "eventname", "dataType": "string", "selected": true}], "sql": "`tickit`.`event`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 16}, "left": {"location": "LEFT", "id": 13}, "right": {"location": "RIGHT", "id": 14}, "top": {"location": "TOP", "id": 15}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 13,
                    "name": "tickit.venue",
                    "nodeAttributes": {
                        "attributes": [{"name": "venueid", "dataType": "int", "selected": false}, {"name": "venuename", "dataType": "string", "selected": true}],
                        "sql": "`tickit`.`venue`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 20}, "left": {"location": "LEFT", "id": 17}, "right": {"location": "RIGHT", "id": 18}, "top": {"location": "TOP", "id": 19}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }],
                "connections": [{
                    "source": {"nodeID": 11, "connectorIndex": 0, "connectorId": 9},
                    "dest": {"nodeID": 10, "connectorIndex": 0, "connectorID": 5},
                    "joinKeys": {"sourceKey": "userid", "destKey": "buyerid"},
                    "joinType": "INNER JOIN"
                }, {
                    "source": {"nodeID": 12, "connectorIndex": 0, "connectorId": 13},
                    "dest": {"nodeID": 11, "connectorIndex": 0, "connectorID": 9},
                    "joinKeys": {"sourceKey": "eventid", "destKey": "eventid"},
                    "joinType": "INNER JOIN"
                }, {
                    "source": {"nodeID": 13, "connectorIndex": 0, "connectorId": 17},
                    "dest": {"nodeID": 12, "connectorIndex": 0, "connectorID": 13},
                    "joinKeys": {"sourceKey": "venueid", "destKey": "venueid"},
                    "joinType": "INNER JOIN"
                }]
            });
            expect(spark).toBe("val tbl10 = sqlContext.table(\"tickit.users\").alias(\"tbl10\")\n"
                               + "val tbl11 = sqlContext.table(\"tickit.sales\").alias(\"tbl11\")\n"
                               + "val tbl12 = sqlContext.table(\"tickit.event\").alias(\"tbl12\")\n"
                               + "val tbl13 = sqlContext.table(\"tickit.venue\").alias(\"tbl13\")\n"
                               + "var df = tbl10.join(tbl11, tbl11.col(\"buyerid\").equalTo(tbl10.col(\"userid\")), \"inner\")"
                               + ".join(tbl12, tbl12.col(\"eventid\").equalTo(tbl11.col(\"eventid\")), \"inner\")"
                               + ".join(tbl13, tbl13.col(\"venueid\").equalTo(tbl12.col(\"venueid\")), \"inner\")"
                               + ".select(tbl10.col(\"username\"), tbl10.col(\"firstname\"), tbl10.col(\"lastname\"), tbl11.col(\"qtysold\"), tbl11.col(\"pricepaid\"), tbl11.col(\"commission\"),"
                               + " tbl12.col(\"eventname\"), tbl13.col(\"venuename\"))\n");
        }));

        it("should produce Spark for multiple tables", mocks.inject(function (SparkDatasourceService) {
            var spark = SparkDatasourceService.toScript({
                "nodes": [{
                    "id": 10,
                    "name": "sample.t1",
                    "nodeAttributes": {"attributes": [{"name": "id", "dataType": "smallint", "selected": true}, {"name": "id_1", "dataType": "smallint", "selected": true}], "sql": "`sample`.`t1`"},
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 24}, "left": {"location": "LEFT", "id": 21}, "right": {"location": "RIGHT", "id": 22}, "top": {"location": "TOP", "id": 23}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 11,
                    "name": "sample.t2",
                    "nodeAttributes": {
                        "attributes": [{"name": "id", "dataType": "smallint", "selected": true}, {"name": "t1_id", "dataType": "smallint", "selected": true},
                            {"name": "sample_t1_id", "dataType": "smallint", "selected": true}], "sql": "`sample`.`t2`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 28}, "left": {"location": "LEFT", "id": 25}, "right": {"location": "RIGHT", "id": 26}, "top": {"location": "TOP", "id": 27}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }],
                "connections": []
            });
            expect(spark).toBe("val tbl10 = sqlContext.table(\"sample.t1\").alias(\"tbl10\")\n"
                               + "val tbl11 = sqlContext.table(\"sample.t2\").alias(\"tbl11\")\n"
                               + "var df = tbl10.join(tbl11).select(tbl10.col(\"id\").alias(\"id_2\"), tbl10.col(\"id_1\"), tbl11.col(\"id\").alias(\"t2_id\"),"
                               + " tbl11.col(\"t1_id\").alias(\"t2_t1_id\"), tbl11.col(\"sample_t1_id\").alias(\"t2_sample_t1_id\"))\n");
        }));

        it("should produce Spark for pre-joined tables", mocks.inject(function (SparkDatasourceService) {
            var spark = SparkDatasourceService.toScript({
                "nodes": [{
                    "id": 10,
                    "name": "tickit.users",
                    "nodeAttributes": {
                        "attributes": [{"name": "userid", "dataType": "int", "selected": false}, {"name": "username", "dataType": "string", "selected": true},
                            {"name": "firstname", "dataType": "string", "selected": true}, {"name": "lastname", "dataType": "string", "selected": true}], "sql": "`tickit`.`users`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 32}, "left": {"location": "LEFT", "id": 29}, "right": {"location": "RIGHT", "id": 30}, "top": {"location": "TOP", "id": 31}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 11,
                    "name": "tickit.sales",
                    "nodeAttributes": {
                        "attributes": [{"name": "salesid", "dataType": "int", "selected": false}, {"name": "buyerid", "dataType": "int", "selected": false},
                            {"name": "eventid", "dataType": "int", "selected": false}, {"name": "dateid", "dataType": "smallint", "selected": false},
                            {"name": "qtysold", "dataType": "string", "selected": true}, {"name": "pricepaid", "dataType": "double", "selected": true},
                            {"name": "commission", "dataType": "double", "selected": true}], "sql": "`tickit`.`sales`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 36}, "left": {"location": "LEFT", "id": 33}, "right": {"location": "RIGHT", "id": 34}, "top": {"location": "TOP", "id": 35}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }], "connections": [{"source": {"nodeID": 11, "connectorIndex": 0, "connectorId": 33}, "dest": {"nodeID": 10, "connectorIndex": 0, "connectorID": 29}, "joinKeys": {}}]
            });
            expect(spark).toBe("val tbl10 = sqlContext.table(\"tickit.users\").alias(\"tbl10\")\n"
                               + "val tbl11 = sqlContext.table(\"tickit.sales\").alias(\"tbl11\")\n"
                               + "var df = tbl10.join(tbl11).select(tbl10.col(\"username\"), tbl10.col(\"firstname\"), tbl10.col(\"lastname\"), tbl11.col(\"qtysold\"), tbl11.col(\"pricepaid\"),"
                               + " tbl11.col(\"commission\"))\n");
        }));

        it("should produce Spark for multiple join conditions", mocks.inject(function (SparkDatasourceService) {
            var spark = SparkDatasourceService.toScript({
                "nodes": [{
                    "id": 11,
                    "name": "tickit.sales",
                    "nodeAttributes": {
                        "attributes": [{"name": "salesid", "dataType": "int", "selected": false}, {"name": "buyerid", "dataType": "int", "selected": false},
                            {"name": "eventid", "dataType": "int", "selected": false}, {"name": "dateid", "dataType": "smallint", "selected": false},
                            {"name": "qtysold", "dataType": "string", "selected": true}, {"name": "pricepaid", "dataType": "double", "selected": true},
                            {"name": "commission", "dataType": "double", "selected": true}], "sql": "`tickit`.`sales`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 40}, "left": {"location": "LEFT", "id": 37}, "right": {"location": "RIGHT", "id": 38}, "top": {"location": "TOP", "id": 39}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 12,
                    "name": "tickit.event",
                    "nodeAttributes": {
                        "attributes": [{"name": "eventid", "dataType": "int", "selected": false}, {"name": "dateid", "dataType": "smallint", "selected": false},
                            {"name": "eventname", "dataType": "string", "selected": true}], "sql": "`tickit`.`event`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 44}, "left": {"location": "LEFT", "id": 41}, "right": {"location": "RIGHT", "id": 42}, "top": {"location": "TOP", "id": 43}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 14,
                    "name": "tickit.date",
                    "nodeAttributes": {
                        "attributes": [{"name": "dateid", "dataType": "smallint", "selected": false}, {"name": "caldate", "dataType": "date", "selected": true}],
                        "sql": "`tickit`.`date`"
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 48}, "left": {"location": "LEFT", "id": 45}, "right": {"location": "RIGHT", "id": 46}, "top": {"location": "TOP", "id": 47}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }],
                "connections": [{
                    "source": {"nodeID": 12, "connectorIndex": 0, "connectorId": 41},
                    "dest": {"nodeID": 11, "connectorIndex": 0, "connectorID": 37},
                    "joinKeys": {"sourceKey": "eventid", "destKey": "eventid"},
                    "joinType": "INNER JOIN"
                }, {
                    "source": {"nodeID": 14, "connectorIndex": 0, "connectorId": 45},
                    "dest": {"nodeID": 12, "connectorIndex": 0, "connectorID": 41},
                    "joinKeys": {"sourceKey": "dateid", "destKey": "dateid"},
                    "joinType": "INNER JOIN"
                }, {
                    "source": {"nodeID": 14, "connectorIndex": 0, "connectorId": 45},
                    "dest": {"nodeID": 11, "connectorIndex": 0, "connectorID": 37},
                    "joinKeys": {"sourceKey": "dateid", "destKey": "dateid"},
                    "joinType": "INNER JOIN"
                }]
            });
            expect(spark).toBe("val tbl11 = sqlContext.table(\"tickit.sales\").alias(\"tbl11\")\n"
                               + "val tbl12 = sqlContext.table(\"tickit.event\").alias(\"tbl12\")\n"
                               + "val tbl14 = sqlContext.table(\"tickit.date\").alias(\"tbl14\")\n"
                               + "var df = tbl11.join(tbl12, tbl12.col(\"eventid\").equalTo(tbl11.col(\"eventid\")), \"inner\")"
                               + ".join(tbl14, tbl14.col(\"dateid\").equalTo(tbl11.col(\"dateid\")).and(tbl14.col(\"dateid\").equalTo(tbl12.col(\"dateid\"))), \"inner\")"
                               + ".select(tbl11.col(\"qtysold\"), tbl11.col(\"pricepaid\"), tbl11.col(\"commission\"), tbl12.col(\"eventname\"), tbl14.col(\"caldate\"))\n");
        }));

        it("should produce Spark for multiple data sources", mocks.inject(function (SparkDatasourceService) {
            var spark = SparkDatasourceService.toScript({
                "nodes": [{
                    "id": 10,
                    "name": "sample.t1",
                    "datasourceId": "HIVE",
                    "nodeAttributes": {
                        "attributes": [{"name": "id", "dataType": "smallint", "selected": true}, {"name": "id_1", "dataType": "smallint", "selected": true}],
                        "reference": ["sample", "t1"]
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 24}, "left": {"location": "LEFT", "id": 21}, "right": {"location": "RIGHT", "id": 22}, "top": {"location": "TOP", "id": 23}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }, {
                    "id": 11,
                    "name": "sample.t2",
                    "datasourceId": "0e63b63e-e1c3-4502-99fb-b86272bc6ede",
                    "nodeAttributes": {
                        "attributes": [{"name": "id", "dataType": "smallint", "selected": true}, {"name": "t1_id", "dataType": "smallint", "selected": true},
                            {"name": "sample_t1_id", "dataType": "smallint", "selected": true}], "reference": ["sample", "t2"]
                    },
                    "connectors": {"bottom": {"location": "BOTTOM", "id": 28}, "left": {"location": "LEFT", "id": 25}, "right": {"location": "RIGHT", "id": 26}, "top": {"location": "TOP", "id": 27}},
                    "inputConnectors": [{"name": ""}],
                    "outputConnectors": [{"name": ""}],
                    "width": 250
                }],
                "connections": []
            });
            expect(spark).toBe("val tbl10 = sqlContext.table(\"sample.t1\").alias(\"tbl10\")\n"
                               + "val tbl11 = datasourceProvider.getTableFromDatasource(\"sample.t2\", \"0e63b63e-e1c3-4502-99fb-b86272bc6ede\", sqlContext).alias(\"tbl11\")\n"
                               + "var df = tbl10.join(tbl11).select(tbl10.col(\"id\").alias(\"id_2\"), tbl10.col(\"id_1\"), tbl11.col(\"id\").alias(\"t2_id\"),"
                               + " tbl11.col(\"t1_id\").alias(\"t2_t1_id\"), tbl11.col(\"sample_t1_id\").alias(\"t2_sample_t1_id\"))\n");
        }));
    });
});
