text: # -*- coding: utf-8 -*-
  # -----------------------------------------------------------------------------
  # Copyright (c) 2016, Anaconda, Inc. All rights reserved.
  #
  # Licensed under the terms of the BSD 3-Clause License.
  # The full license is in the file LICENSE.txt, distributed with this software.
  # -----------------------------------------------------------------------------
  from __future__ import absolute_import
  
  import os
  
  import pytest
  
  from anaconda_project.internal.test.tmpfile_utils import (with_directory_contents, tmp_script_commandline,
                                                            with_directory_contents_completing_project_file)
  from anaconda_project.local_state_file import LocalStateFile, DEFAULT_LOCAL_STATE_FILENAME
  from anaconda_project.requirements_registry.provider import (Provider, ProvideContext, EnvVarProvider, ProvideResult,
                                                               shutdown_service_run_state)
  from anaconda_project.requirements_registry.registry import RequirementsRegistry
  from anaconda_project.requirements_registry.requirement import EnvVarRequirement, UserConfigOverrides
  from anaconda_project.project import Project
  from anaconda_project.provide import PROVIDE_MODE_DEVELOPMENT
  from anaconda_project.project_file import DEFAULT_PROJECT_FILENAME
  from anaconda_project.prepare import (prepare_without_interaction, unprepare)
  from anaconda_project.frontend import NullFrontend
  from anaconda_project.test.project_utils import project_no_dedicated_env
  from anaconda_project.test.environ_utils import minimal_environ
  from anaconda_project.internal import (keyring, conda_api)
  
  conda_env_var = conda_api.conda_prefix_variable()
  
  
  def test_find_provider_by_class_name():
      registry = RequirementsRegistry()
      found = registry.find_provider_by_class_name(class_name="CondaEnvProvider")
      assert found is not None
      assert found.__class__.__name__ == "CondaEnvProvider"
  
  
  def test_find_provider_by_class_name_not_found():
      registry = RequirementsRegistry()
      with pytest.raises(ValueError):
          registry.find_provider_by_class_name(class_name="NotAThing")
  
  
  def test_provider_default_method_implementations():
      class UselessProvider(Provider):
          @property
          def title(self):
              return ""
  
          def read_config(self, requirement, environ, local_state_file, default_env_spec_name, overrides):
              return dict()
  
          def provide(self, requirement, context):
              pass
  
          def unprovide(self, requirement, local_state_file, requirement_status=None):
              pass
  
          def missing_env_vars_to_configure(self, requirement, environ, local_state_file):
              return ()
  
          def missing_env_vars_to_provide(self, requirement, environ, local_state_file):
              return ()
  
      provider = UselessProvider()
      # this method is supposed to do nothing by default (ignore
      # unknown names, in particular)
      provider.set_config_values_as_strings(
          requirement=None,
          environ=None,
          local_state_file=None,
          default_env_spec_name='default',
          overrides=None,
          values=dict())
  
  
  def _load_env_var_requirement(dirname, env_var):
      project = Project(dirname)
  
      for requirement in project.requirements(project.default_env_spec_name):
          if isinstance(requirement, EnvVarRequirement) and requirement.env_var == env_var:
              return requirement
      assert [] == project.problems
      raise RuntimeError("No requirement for %s was in the project file, only %r" % (env_var, project.requirements))
  
  
  def test_env_var_provider_with_no_value():
      def check_env_var_provider(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO")
          local_state_file = LocalStateFile.load_for_directory(dirname)
          status = requirement.check_status(dict(), local_state_file, 'default', UserConfigOverrides())
          context = ProvideContext(
              environ=dict(),
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
  
          provider.provide(requirement, context=context)
          assert 'FOO' not in context.environ
  
      with_directory_contents_completing_project_file({
          DEFAULT_PROJECT_FILENAME: """
  variables:
    - FOO
  """
      }, check_env_var_provider)
  
  
  def test_env_var_provider_with_default_value_in_project_file():
      def check_env_var_provider(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO")
          assert dict(default='from_default') == requirement.options
          local_state_file = LocalStateFile.load_for_directory(dirname)
          status = requirement.check_status(dict(), local_state_file, 'default', UserConfigOverrides())
          context = ProvideContext(
              environ=dict(),
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          result = provider.provide(requirement, context=context)
          assert [] == result.errors
          assert 'FOO' in context.environ
          assert 'from_default' == context.environ['FOO']
  
      with_directory_contents_completing_project_file(
          {
              DEFAULT_PROJECT_FILENAME: """
  variables:
    FOO:
      default: from_default
  """
          }, check_env_var_provider)
  
  
  def test_env_var_provider_with_unencrypted_default_value_in_project_file_for_encrypted_requirement():
      # the idea here is that if you want to put an unencrypted
      # password in the file, we aren't going to be annoying and
      # stop you.
      def check_env_var_provider(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO_SECRET")
          assert requirement.encrypted
          assert dict(default='from_default') == requirement.options
          local_state_file = LocalStateFile.load_for_directory(dirname)
          environ = dict()
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          result = provider.provide(requirement, context=context)
          assert [] == result.errors
          assert 'FOO_SECRET' in context.environ
          assert 'from_default' == context.environ['FOO_SECRET']
  
      with_directory_contents_completing_project_file(
          {
              DEFAULT_PROJECT_FILENAME: """
  variables:
    FOO_SECRET:
      default: from_default
  """
          }, check_env_var_provider)
  
  
  def test_env_var_provider_with_value_set_in_environment():
      def check_env_var_provider(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO")
          local_state_file = LocalStateFile.load_for_directory(dirname)
          environ = dict(FOO='from_environ')
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          assert dict(source='environ', value='from_environ') == status.analysis.config
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          result = provider.provide(requirement, context=context)
          assert [] == result.errors
          assert 'FOO' in context.environ
          assert 'from_environ' == context.environ['FOO']
  
      # set a default to be sure we prefer 'environ' instead
      with_directory_contents_completing_project_file(
          {
              DEFAULT_PROJECT_FILENAME: """
  variables:
    FOO:
      default: from_default
  """
          }, check_env_var_provider)
  
  
  def test_env_var_provider_with_value_set_in_local_state():
      def check_env_var_provider(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO")
          local_state_file = LocalStateFile.load_for_directory(dirname)
          # set in environ to be sure we override it with local state
          environ = dict(FOO='from_environ')
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          assert dict(value="from_local_state", source="variables") == status.analysis.config
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          result = provider.provide(requirement, context=context)
          assert [] == result.errors
          assert 'FOO' in context.environ
          assert 'from_local_state' == context.environ['FOO']
  
      with_directory_contents_completing_project_file(
          {
              DEFAULT_PROJECT_FILENAME: """
  variables:
    FOO:
      default: from_default
      """,
              DEFAULT_LOCAL_STATE_FILENAME: """
  variables:
    FOO: from_local_state
  """
          }, check_env_var_provider)
  
  
  def test_env_var_provider_with_encrypted_value_set_in_local_state():
      def check_env_var_provider(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO_PASSWORD")
          assert requirement.encrypted
          local_state_file = LocalStateFile.load_for_directory(dirname)
          # set in environ to be sure we override it with local state
          environ = dict(FOO_PASSWORD='from_environ')
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          assert dict(value="from_local_state", source="variables") == status.analysis.config
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          result = provider.provide(requirement, context=context)
          assert [] == result.errors
          assert 'FOO_PASSWORD' in context.environ
          assert 'from_local_state' == context.environ['FOO_PASSWORD']
  
      with_directory_contents_completing_project_file(
          {
              DEFAULT_PROJECT_FILENAME: """
  variables:
    FOO_PASSWORD:
      default: from_default
      """,
              DEFAULT_LOCAL_STATE_FILENAME: """
  variables:
    FOO_PASSWORD: from_local_state
  """
          }, check_env_var_provider)
  
  
  def test_env_var_provider_configure_local_state_value():
      def check_env_var_provider_config_local_state(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO")
          local_state_file = LocalStateFile.load_for_directory(dirname)
          status = requirement.check_status(dict(), local_state_file, 'default', UserConfigOverrides())
          assert dict(source='unset') == status.analysis.config
  
          assert local_state_file.get_value(['variables', 'FOO']) is None
  
          environ = dict()
  
          provider.set_config_values_as_strings(requirement, environ, local_state_file, 'default', UserConfigOverrides(),
                                                dict(value="bar"))
  
          assert local_state_file.get_value(['variables', 'FOO']) == "bar"
          local_state_file.save()
  
          local_state_file_2 = LocalStateFile.load_for_directory(dirname)
          assert local_state_file_2.get_value(['variables', 'FOO']) == "bar"
  
          # setting empty string = unset
          provider.set_config_values_as_strings(requirement, environ, local_state_file, 'default', UserConfigOverrides(),
                                                dict(value=""))
          assert local_state_file.get_value(['variables', 'FOO']) is None
  
          local_state_file.save()
  
          local_state_file_3 = LocalStateFile.load_for_directory(dirname)
          assert local_state_file_3.get_value(['variables', 'FOO']) is None
  
      with_directory_contents_completing_project_file({
          DEFAULT_PROJECT_FILENAME: """
  variables:
    - FOO
  """
      }, check_env_var_provider_config_local_state)
  
  
  def test_env_var_provider_configure_encrypted_value():
      def check_env_var_provider_config_local_state(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO_PASSWORD")
          assert requirement.encrypted
          local_state_file = LocalStateFile.load_for_directory(dirname)
          status = requirement.check_status(dict(), local_state_file, 'default', UserConfigOverrides())
          assert dict(source='unset') == status.analysis.config
  
          assert local_state_file.get_value(['variables', 'FOO_PASSWORD']) is None
          assert set(keyring.fallback_data().values()) == set()
  
          environ = dict(CONDA_DEFAULT_ENV='/pretend/env', CONDA_ENV_PATH='/pretend/env', CONDA_PREFIX='/pretend/env')
  
          provider.set_config_values_as_strings(requirement, environ, local_state_file, 'default', UserConfigOverrides(),
                                                dict(value="bar"))
  
          # should not have affected local state, should use keyring
          assert local_state_file.get_value(['variables', 'FOO_PASSWORD']) is None
          assert set(keyring.fallback_data().values()) == set(["bar"])
  
          # setting empty string = unset
          provider.set_config_values_as_strings(requirement, environ, local_state_file, 'default', UserConfigOverrides(),
                                                dict(value=""))
          assert local_state_file.get_value(['variables', 'FOO_PASSWORD']) is None
          assert set(keyring.fallback_data().values()) == set()
  
      keyring.enable_fallback_keyring()
      try:
          with_directory_contents_completing_project_file({
              DEFAULT_PROJECT_FILENAME: """
  variables:
    - FOO_PASSWORD
  """
          }, check_env_var_provider_config_local_state)
      finally:
          keyring.disable_fallback_keyring()
  
  
  def test_env_var_provider_configure_disabled_local_state_value():
      def check_env_var_provider_config_disabled_local_state(dirname):
          provider = EnvVarProvider()
          requirement = _load_env_var_requirement(dirname, "FOO")
          local_state_file = LocalStateFile.load_for_directory(dirname)
          status = requirement.check_status(dict(), local_state_file, 'default', UserConfigOverrides())
          assert dict(source='unset') == status.analysis.config
  
          assert local_state_file.get_value(['variables', 'FOO']) is None
          assert local_state_file.get_value(['disabled_variables', 'FOO']) is None
  
          environ = dict()
  
          # source=environ should mean we set disabled_variables instead of variables
          provider.set_config_values_as_strings(requirement, environ, local_state_file, 'default', UserConfigOverrides(),
                                                dict(source='environ', value="bar"))
  
          assert local_state_file.get_value(['variables', 'FOO']) is None
          assert local_state_file.get_value(['disabled_variables', 'FOO']) == "bar"
  
          config = provider.read_config(requirement, environ, local_state_file, 'default', UserConfigOverrides())
          assert config == dict(source='unset', value='bar')
  
      with_directory_contents_completing_project_file({
          DEFAULT_PROJECT_FILENAME: """
  variables:
    - FOO
  """
      }, check_env_var_provider_config_disabled_local_state)
  
  
  def test_env_var_provider_prepare_unprepare():
      def check_env_var_provider_prepare(dirname):
          project = project_no_dedicated_env(dirname)
          result = prepare_without_interaction(project, environ=minimal_environ(FOO='bar'))
          assert result
          status = unprepare(project, result)
          assert status
          assert status.status_description == 'Success.'
          assert project.frontend.logs == [
              "Nothing to clean up for FOO.", ("Current environment is not in %s, no need to delete it." % dirname)
          ]
  
      with_directory_contents_completing_project_file({
          DEFAULT_PROJECT_FILENAME: """
  variables:
    FOO: null
  """
      }, check_env_var_provider_prepare)
  
  
  def test_provide_context_properties():
      def check_provide_contents(dirname):
          environ = dict(foo='bar')
          local_state_file = LocalStateFile.load_for_directory(dirname)
          requirement = EnvVarRequirement(RequirementsRegistry(), env_var="FOO")
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          assert dict(foo='bar') == context.environ
          assert context.status is status
  
      with_directory_contents(dict(), check_provide_contents)
  
  
  def test_provide_result_properties():
      empty = ProvideResult.empty()
      assert [] == empty.errors
  
      full = ProvideResult(['c', 'd'])
      assert ['c', 'd'] == full.errors
  
      unchanged = full.copy_with_additions()
      assert ['c', 'd'] == unchanged.errors
  
      unchanged2 = full.copy_with_additions([])
      assert ['c', 'd'] == unchanged2.errors
  
      extended = full.copy_with_additions(['z'])
      assert ['c', 'd', 'z'] == extended.errors
  
  
  def test_provide_context_ensure_service_directory():
      def check_provide_contents(dirname):
          environ = dict()
          local_state_file = LocalStateFile.load_for_directory(dirname)
          requirement = EnvVarRequirement(RequirementsRegistry(), env_var="FOO")
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          workpath = context.ensure_service_directory("foo")
          assert os.path.isdir(workpath)
          assert workpath.endswith("foo")
          parent = os.path.dirname(workpath)
          assert parent.endswith("services")
          parent = os.path.dirname(parent)
          assert parent == dirname
  
          # be sure we can create if it already exists
          workpath2 = context.ensure_service_directory("foo")
          assert os.path.isdir(workpath2)
          assert workpath == workpath2
  
      with_directory_contents(dict(), check_provide_contents)
  
  
  def test_provide_context_ensure_service_directory_cannot_create(monkeypatch):
      def mock_makedirs(path, mode=0):
          raise IOError("this is not EEXIST")
  
      monkeypatch.setattr("os.makedirs", mock_makedirs)
  
      def check_provide_contents(dirname):
          environ = dict()
          local_state_file = LocalStateFile.load_for_directory(dirname)
          requirement = EnvVarRequirement(RequirementsRegistry(), env_var="FOO")
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
          with pytest.raises(IOError) as excinfo:
              context.ensure_service_directory("foo")
          assert "this is not EEXIST" in repr(excinfo.value)
  
      with_directory_contents(dict(), check_provide_contents)
  
  
  def test_provide_context_transform_service_run_state():
      def check_provide_contents(dirname):
          environ = dict()
          local_state_file = LocalStateFile.load_for_directory(dirname)
          local_state_file.set_service_run_state("myservice", dict(port=42))
          requirement = EnvVarRequirement(RequirementsRegistry(), env_var="FOO")
          status = requirement.check_status(environ, local_state_file, 'default', UserConfigOverrides())
          context = ProvideContext(
              environ=environ,
              local_state_file=local_state_file,
              default_env_spec_name='default',
              status=status,
              mode=PROVIDE_MODE_DEVELOPMENT,
              frontend=NullFrontend())
  
          def transform_it(state):
              assert 42 == state['port']
              state['port'] = 43
              state['foo'] = 'bar'
              return 1234
  
          result = context.transform_service_run_state("myservice", transform_it)
          assert 1234 == result
          assert dict(port=43, foo='bar') == local_state_file.get_service_run_state("myservice")
  
      with_directory_contents(dict(), check_provide_contents)
  
  
  def test_shutdown_service_run_state_nothing_to_do():
      def check(dirname):
          local_state_file = LocalStateFile.load_for_directory(dirname)
          status = shutdown_service_run_state(local_state_file, 'foo')
          assert status
          assert status.status_description == 'Nothing to do to shut down foo.'
  
      with_directory_contents(dict(), check)
  
  
  def test_shutdown_service_run_state_command_success():
      def check(dirname):
          local_state_file = LocalStateFile.load_for_directory(dirname)
          true_commandline = tmp_script_commandline("""import sys
  sys.exit(0)
  """)
          local_state_file.set_service_run_state('FOO', {'shutdown_commands': [true_commandline]})
          status = shutdown_service_run_state(local_state_file, 'FOO')
          assert status
          assert status.status_description == "Successfully shut down FOO."
  
      with_directory_contents(dict(), check)
  
  
  def test_shutdown_service_run_state_command_failure():
      def check(dirname):
          local_state_file = LocalStateFile.load_for_directory(dirname)
          false_commandline = tmp_script_commandline("""import sys
  sys.exit(1)
  """)
          local_state_file.set_service_run_state('FOO', {'shutdown_commands': [false_commandline]})
          status = shutdown_service_run_state(local_state_file, 'FOO')
          assert not status
          assert status.status_description == "Shutdown commands failed for FOO."
          assert status.errors == ["Shutting down FOO, command %r failed with code 1." % false_commandline]
  
      with_directory_contents(dict(), check)
  