# Copyright (C) 2012 Anaconda, Inc
# SPDX-License-Identifier: BSD-3-Clause
import os
import random
import unittest
from os.path import join
from unittest.mock import patch
from uuid import uuid4

from conda.base.context import conda_tests_ctxt_mgmt_def_pol
from conda.common.compat import on_win
from conda.common.io import env_vars
from conda.common.serialize import yaml_round_trip_load
from conda.core.prefix_data import PrefixData
from conda.exceptions import CondaHTTPError, EnvironmentFileNotFound
from conda.models.match_spec import MatchSpec
from conda_env.env import from_environment
from tests.test_utils import is_prefix_activated_PATHwise

from . import support_file
from .utils import Commands, make_temp_envs_dir, run_command

PYTHON_BINARY = "python.exe" if on_win else "bin/python"


try:
    from io import StringIO
except ImportError:
    from StringIO import StringIO

import pytest

from conda_env import env


class FakeStream:
    def __init__(self):
        self.output = ""

    def write(self, chunk):
        self.output += chunk.decode("utf-8")


def get_environment(filename):
    return env.from_file(support_file(filename))


def get_simple_environment():
    return get_environment("simple.yml")


def get_valid_keys_environment():
    return get_environment("valid_keys.yml")


def get_invalid_keys_environment():
    return get_environment("invalid_keys.yml")


class from_file_TestCase(unittest.TestCase):
    def test_returns_Environment(self):
        e = get_simple_environment()
        self.assertIsInstance(e, env.Environment)

    def test_retains_full_filename(self):
        e = get_simple_environment()
        self.assertEqual(support_file("simple.yml"), e.filename)

    def test_with_pip(self):
        e = env.from_file(support_file("with-pip.yml"))
        assert "pip" in e.dependencies
        assert "foo" in e.dependencies["pip"]
        assert "baz" in e.dependencies["pip"]

    @pytest.mark.timeout(20)
    def test_add_pip(self):
        e = env.from_file(support_file("add-pip.yml"))
        expected = {
            "conda": ["pip", "car"],
            "pip": ["foo", "baz"],
        }
        self.assertEqual(e.dependencies, expected)

    @pytest.mark.integration
    def test_http(self):
        e = get_simple_environment()
        f = env.from_file(
            "https://raw.githubusercontent.com/conda/conda/main/tests/conda_env/support/simple.yml"
        )
        self.assertEqual(e.dependencies, f.dependencies)
        assert e.dependencies == f.dependencies

    @pytest.mark.integration
    def test_http_raises(self):
        with self.assertRaises(CondaHTTPError):
            env.from_file(
                "https://raw.githubusercontent.com/conda/conda/main/tests/conda_env/support/does-not-exist.yml"
            )

    def test_envvars(self):
        current_conda_token = os.environ.get("CONDA_TOKEN")
        os.environ["CONDA_TOKEN"] = "aaa-12345"
        os.environ["OTHER_KEY"] = "12345-aaa"
        e = get_environment("channels_with_envvars.yml")
        self.assertEqual(
            set(e.channels),
            {
                "https://localhost/t/aaa-12345/stable",
                "https://localhost/t/12345-aaa/stable",
                "conda-forge",
                "defaults",
            },
        )
        if current_conda_token:
            os.environ["CONDA_TOKEN"] = current_conda_token
        else:
            del os.environ["CONDA_TOKEN"]
        del os.environ["OTHER_KEY"]


class EnvironmentTestCase(unittest.TestCase):
    def test_has_empty_filename_by_default(self):
        e = env.Environment()
        self.assertEqual(e.filename, None)

    def test_has_filename_if_provided(self):
        r = random.randint(100, 200)
        random_filename = f"/path/to/random/environment-{r}.yml"
        e = env.Environment(filename=random_filename)
        self.assertEqual(e.filename, random_filename)

    def test_has_empty_name_by_default(self):
        e = env.Environment()
        self.assertEqual(e.name, None)

    def test_has_name_if_provided(self):
        random_name = f"random-{random.randint(100, 200)}"
        e = env.Environment(name=random_name)
        self.assertEqual(e.name, random_name)

    def test_dependencies_are_empty_by_default(self):
        e = env.Environment()
        self.assertEqual(0, len(e.dependencies))

    def test_parses_dependencies_from_raw_file(self):
        e = get_simple_environment()
        expected = {"conda": ["nltk"]}
        self.assertEqual(e.dependencies, expected)

    def test_builds_spec_from_line_raw_dependency(self):
        # TODO Refactor this inside conda to not be a raw string
        e = env.Environment(dependencies=["nltk=3.0.0=np18py27_0"])
        expected = {"conda": ["nltk==3.0.0=np18py27_0"]}
        self.assertEqual(e.dependencies, expected)

    def test_args_are_wildcarded(self):
        e = env.Environment(dependencies=["python=2.7"])
        expected = {"conda": ["python=2.7"]}
        self.assertEqual(e.dependencies, expected)

    def test_other_tips_of_dependencies_are_supported(self):
        e = env.Environment(dependencies=["nltk", {"pip": ["foo", "bar"]}])
        expected = {
            "conda": ["nltk", "pip"],
            "pip": ["foo", "bar"],
        }
        self.assertEqual(e.dependencies, expected)

    def test_channels_default_to_empty_list(self):
        e = env.Environment()
        self.assertIsInstance(e.channels, list)
        self.assertEqual(e.channels, [])

    def test_add_channels(self):
        e = env.Environment()
        e.add_channels(["dup", "dup", "unique"])
        self.assertEqual(e.channels, ["dup", "unique"])

    def test_remove_channels(self):
        e = env.Environment(channels=["channel"])
        e.remove_channels()
        self.assertEqual(e.channels, [])

    def test_channels_are_provided_by_kwarg(self):
        random_channels = (random.randint(100, 200), random)
        e = env.Environment(channels=random_channels)
        self.assertEqual(e.channels, random_channels)

    def test_to_dict_returns_dictionary_of_data(self):
        random_name = f"random{random.randint(100, 200)}"
        e = env.Environment(
            name=random_name, channels=["javascript"], dependencies=["nodejs"]
        )

        expected = {
            "name": random_name,
            "channels": ["javascript"],
            "dependencies": ["nodejs"],
        }
        self.assertEqual(e.to_dict(), expected)

    def test_to_dict_returns_just_name_if_only_thing_present(self):
        e = env.Environment(name="simple")
        expected = {"name": "simple"}
        self.assertEqual(e.to_dict(), expected)

    def test_to_yaml_returns_yaml_parseable_string(self):
        random_name = f"random{random.randint(100, 200)}"
        e = env.Environment(
            name=random_name, channels=["javascript"], dependencies=["nodejs"]
        )

        expected = {
            "name": random_name,
            "channels": ["javascript"],
            "dependencies": ["nodejs"],
        }

        actual = yaml_round_trip_load(StringIO(e.to_yaml()))
        self.assertEqual(expected, actual)

    def test_to_yaml_returns_proper_yaml(self):
        random_name = f"random{random.randint(100, 200)}"
        e = env.Environment(
            name=random_name, channels=["javascript"], dependencies=["nodejs"]
        )

        expected = "\n".join(
            [
                "name: %s" % random_name,
                "channels:",
                "  - javascript",
                "dependencies:",
                "  - nodejs",
                "",
            ]
        )

        actual = e.to_yaml()
        self.assertEqual(expected, actual)

    def test_to_yaml_takes_stream(self):
        random_name = f"random{random.randint(100, 200)}"
        e = env.Environment(
            name=random_name, channels=["javascript"], dependencies=["nodejs"]
        )

        s = FakeStream()
        e.to_yaml(stream=s)

        expected = "\n".join(
            [
                "name: %s" % random_name,
                "channels:",
                "  - javascript",
                "dependencies:",
                "  - nodejs",
                "",
            ]
        )
        assert expected == s.output

    def test_can_add_dependencies_to_environment(self):
        e = get_simple_environment()
        e.dependencies.add("bar")

        s = FakeStream()
        e.to_yaml(stream=s)

        expected = "\n".join(["name: nlp", "dependencies:", "  - nltk", "  - bar", ""])
        assert expected == s.output

    def test_dependencies_update_after_adding(self):
        e = get_simple_environment()
        assert "bar" not in e.dependencies["conda"]
        e.dependencies.add("bar")
        assert "bar" in e.dependencies["conda"]

    def test_valid_keys(self):
        e = get_valid_keys_environment()
        e_dict = e.to_dict()
        for key in env.VALID_KEYS:
            assert key in e_dict

    def test_invalid_keys(self):
        e = get_invalid_keys_environment()
        e_dict = e.to_dict()
        assert "name" in e_dict
        assert len(e_dict) == 1


class DirectoryTestCase(unittest.TestCase):
    directory = support_file("example")

    def setUp(self):
        self.original_working_dir = os.getcwd()
        self.env = env.load_from_directory(self.directory)

    def tearDown(self):
        os.chdir(self.original_working_dir)

    def test_returns_env_object(self):
        self.assertIsInstance(self.env, env.Environment)

    def test_has_expected_name(self):
        self.assertEqual("test", self.env.name)

    def test_has_dependencies(self):
        self.assertEqual(1, len(self.env.dependencies["conda"]))
        assert "numpy" in self.env.dependencies["conda"]


class load_from_directory_example_TestCase(DirectoryTestCase):
    directory = support_file("example")


class load_from_directory_example_yaml_TestCase(DirectoryTestCase):
    directory = support_file("example-yaml")


class load_from_directory_recursive_TestCase(DirectoryTestCase):
    directory = support_file("foo/bar")


class load_from_directory_recursive_two_TestCase(DirectoryTestCase):
    directory = support_file("foo/bar/baz")


class load_from_directory_trailing_slash_TestCase(DirectoryTestCase):
    directory = support_file("foo/bar/baz/")


class load_from_directory_TestCase(unittest.TestCase):
    def test_raises_when_unable_to_find(self):
        with self.assertRaises(EnvironmentFileNotFound):
            env.load_from_directory("/path/to/unknown/env-spec")

    def test_raised_exception_has_environment_yml_as_file(self):
        with self.assertRaises(EnvironmentFileNotFound) as e:
            env.load_from_directory("/path/to/unknown/env-spec")
        self.assertEqual(e.exception.filename, "environment.yml")


class LoadEnvFromFileAndSaveTestCase(unittest.TestCase):
    env_path = support_file(os.path.join("saved-env", "environment.yml"))

    def setUp(self):
        with open(self.env_path, "rb") as fp:
            self.original_file_contents = fp.read()
        self.env = env.load_from_directory(self.env_path)

    def tearDown(self):
        with open(self.env_path, "wb") as fp:
            fp.write(self.original_file_contents)

    def test_expected_default_conditions(self):
        self.assertEqual(1, len(self.env.dependencies["conda"]))

    def test(self):
        self.env.dependencies.add("numpy")
        self.env.save()

        e = env.load_from_directory(self.env_path)
        self.assertEqual(2, len(e.dependencies["conda"]))
        assert "numpy" in e.dependencies["conda"]


class EnvironmentSaveTestCase(unittest.TestCase):
    env_file = support_file("saved.yml")

    def tearDown(self):
        if os.path.exists(self.env_file):
            os.unlink(self.env_file)

    def test_creates_file_on_save(self):
        self.assertFalse(os.path.exists(self.env_file), msg="sanity check")

        e = env.Environment(filename=self.env_file, name="simple")
        e.save()

        self.assertTrue(os.path.exists(self.env_file))

    def _test_saves_yaml_representation_of_file(self):
        e = env.Environment(filename=self.env_file, name="simple")
        e.save()

        with open(self.env_file, "rb") as fp:
            actual = fp.read()

        self.assertTrue(len(actual) > 0, msg="sanity check")
        self.assertEqual(e.to_yaml(), actual)


class SaveExistingEnvTestCase(unittest.TestCase):
    @unittest.skipIf(
        not is_prefix_activated_PATHwise(),
        "You are running `pytest` outside of proper activation. "
        "The entries necessary for conda to operate correctly "
        "are not on PATH.  Please use `conda activate`",
    )
    @pytest.mark.integration
    def test_create_advanced_pip(self):
        with make_temp_envs_dir() as envs_dir:
            with env_vars(
                {
                    "CONDA_ENVS_DIRS": envs_dir,
                    "CONDA_DLL_SEARCH_MODIFICATION_ENABLE": "true",
                },
                stack_callback=conda_tests_ctxt_mgmt_def_pol,
            ):
                env_name = str(uuid4())[:8]
                run_command(Commands.CREATE, env_name, support_file("pip_argh.yml"))
                out_file = join(envs_dir, "test_env.yaml")

            # make sure that the export reconsiders the presence of pip interop being enabled
            PrefixData._cache_.clear()

            with env_vars(
                {
                    "CONDA_ENVS_DIRS": envs_dir,
                },
                stack_callback=conda_tests_ctxt_mgmt_def_pol,
            ):
                # note: out of scope of pip interop var.  Should be enabling conda pip interop itself.
                run_command(Commands.EXPORT, env_name, out_file)
                with open(out_file) as f:
                    d = yaml_round_trip_load(f)
                assert {"pip": ["argh==0.26.2"]} in d["dependencies"]


class TestFromEnvironment(unittest.TestCase):
    def test_from_history(self):
        # We're not testing that get_requested_specs_map() actually works
        # assume it gives us back a dict of MatchSpecs
        with patch("conda.history.History.get_requested_specs_map") as m:
            m.return_value = {
                "python": MatchSpec("python=3"),
                "pytest": MatchSpec("pytest!=3.7.3"),
                "mock": MatchSpec("mock"),
                "yaml": MatchSpec("yaml>=0.1"),
            }
            out = from_environment("mock_env", "mock_prefix", from_history=True)
            assert "yaml[version='>=0.1']" in out.to_dict()["dependencies"]
            assert "pytest!=3.7.3" in out.to_dict()["dependencies"]
            assert len(out.to_dict()["dependencies"]) == 4

            m.assert_called()