# -*- coding: utf-8 -*-
#
# Copyright 2016 VNG Corporation
#
# 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.
#
from helpers import LuigiTestCase, RunOnceTask

import luigi
import luigi.task
from luigi.util import inherits, requires


class BasicsTest(LuigiTestCase):
    # following tests using inherits decorator
    def test_task_ids_using_inherits(self):
        class ParentTask(luigi.Task):
            my_param = luigi.Parameter()

        luigi.namespace("blah")

        @inherits(ParentTask)
        class ChildTask(luigi.Task):
            def requires(self):
                return self.clone(ParentTask)

        luigi.namespace("")
        child_task = ChildTask(my_param="hello")
        self.assertEqual(str(child_task), "blah.ChildTask(my_param=hello)")
        self.assertIn(ParentTask(my_param="hello"), luigi.task.flatten(child_task.requires()))

    def test_task_ids_using_inherits_2(self):
        # Here we use this decorator in a unnormal way.
        # But it should still work.
        class ParentTask(luigi.Task):
            my_param = luigi.Parameter()

        decorator = inherits(ParentTask)
        luigi.namespace("blah")

        class ChildTask(luigi.Task):
            def requires(self):
                return self.clone_parent()

        luigi.namespace("")
        ChildTask = decorator(ChildTask)
        child_task = ChildTask(my_param="hello")
        self.assertEqual(str(child_task), "blah.ChildTask(my_param=hello)")
        self.assertIn(ParentTask(my_param="hello"), luigi.task.flatten(child_task.requires()))

    def test_task_ids_using_inherits_kwargs(self):
        class ParentTask(luigi.Task):
            my_param = luigi.Parameter()

        luigi.namespace("blah")

        @inherits(parent=ParentTask)
        class ChildTask(luigi.Task):
            def requires(self):
                return self.clone(ParentTask)

        luigi.namespace("")
        child_task = ChildTask(my_param="hello")
        self.assertEqual(str(child_task), "blah.ChildTask(my_param=hello)")
        self.assertIn(ParentTask(my_param="hello"), luigi.task.flatten(child_task.requires()))

    def _setup_parent_and_child_inherits(self):
        class ParentTask(luigi.Task):
            my_parameter = luigi.Parameter()
            class_variable = "notset"

            def run(self):
                self.__class__.class_variable = self.my_parameter

            def complete(self):
                return self.class_variable == "actuallyset"

        @inherits(ParentTask)
        class ChildTask(RunOnceTask):
            def requires(self):
                return self.clone_parent()

        return ParentTask

    def test_inherits_has_effect_run_child(self):
        ParentTask = self._setup_parent_and_child_inherits()
        self.assertTrue(self.run_locally_split("ChildTask --my-parameter actuallyset"))
        self.assertEqual(ParentTask.class_variable, "actuallyset")

    def test_inherits_has_effect_run_parent(self):
        ParentTask = self._setup_parent_and_child_inherits()
        self.assertTrue(self.run_locally_split("ParentTask --my-parameter actuallyset"))
        self.assertEqual(ParentTask.class_variable, "actuallyset")

    def _setup_inherits_inheritence(self):
        class InheritedTask(luigi.Task):
            pass

        class ParentTask(luigi.Task):
            pass

        @inherits(InheritedTask)
        class ChildTask(ParentTask):
            pass

        return ChildTask

    def test_inherits_has_effect_MRO(self):
        ChildTask = self._setup_inherits_inheritence()
        self.assertNotEqual(str(ChildTask.__mro__[0]), str(ChildTask.__mro__[1]))

    # following tests using requires decorator
    def test_task_ids_using_requries(self):
        class ParentTask(luigi.Task):
            my_param = luigi.Parameter()

        luigi.namespace("blah")

        @requires(ParentTask)
        class ChildTask(luigi.Task):
            pass

        luigi.namespace("")
        child_task = ChildTask(my_param="hello")
        self.assertEqual(str(child_task), "blah.ChildTask(my_param=hello)")
        self.assertIn(ParentTask(my_param="hello"), luigi.task.flatten(child_task.requires()))

    def test_task_ids_using_requries_2(self):
        # Here we use this decorator in a unnormal way.
        # But it should still work.
        class ParentTask(luigi.Task):
            my_param = luigi.Parameter()

        decorator = requires(ParentTask)
        luigi.namespace("blah")

        class ChildTask(luigi.Task):
            pass

        luigi.namespace("")
        ChildTask = decorator(ChildTask)
        child_task = ChildTask(my_param="hello")
        self.assertEqual(str(child_task), "blah.ChildTask(my_param=hello)")
        self.assertIn(ParentTask(my_param="hello"), luigi.task.flatten(child_task.requires()))

    def _setup_parent_and_child(self):
        class ParentTask(luigi.Task):
            my_parameter = luigi.Parameter()
            class_variable = "notset"

            def run(self):
                self.__class__.class_variable = self.my_parameter

            def complete(self):
                return self.class_variable == "actuallyset"

        @requires(ParentTask)
        class ChildTask(RunOnceTask):
            pass

        return ParentTask

    def test_requires_has_effect_run_child(self):
        ParentTask = self._setup_parent_and_child()
        self.assertTrue(self.run_locally_split("ChildTask --my-parameter actuallyset"))
        self.assertEqual(ParentTask.class_variable, "actuallyset")

    def test_requires_has_effect_run_parent(self):
        ParentTask = self._setup_parent_and_child()
        self.assertTrue(self.run_locally_split("ParentTask --my-parameter actuallyset"))
        self.assertEqual(ParentTask.class_variable, "actuallyset")

    def _setup_requires_inheritence(self):
        class RequiredTask(luigi.Task):
            pass

        class ParentTask(luigi.Task):
            pass

        @requires(RequiredTask)
        class ChildTask(ParentTask):
            pass

        return ChildTask

    def test_requires_has_effect_MRO(self):
        ChildTask = self._setup_requires_inheritence()
        self.assertNotEqual(str(ChildTask.__mro__[0]), str(ChildTask.__mro__[1]))

    def test_kwargs_requires_gives_named_inputs(self):
        class ParentTask(RunOnceTask):
            def output(self):
                return "Target"

        @requires(parent_1=ParentTask, parent_2=ParentTask)
        class ChildTask(RunOnceTask):
            resulting_input = "notset"

            def run(self):
                self.__class__.resulting_input = self.input()

        self.assertTrue(self.run_locally_split("ChildTask"))
        self.assertEqual(ChildTask.resulting_input, {"parent_1": "Target", "parent_2": "Target"})
