# -*- coding: utf-8 -*- from __future__ import absolute_import, division, print_function, unicode_literals from itertools import chain import os from os.path import join from tempfile import gettempdir from unittest import TestCase import pytest from conda._vendor.auxlib.collection import AttrDict from conda._vendor.auxlib.ish import dals from conda._vendor.toolz.itertoolz import concat from conda.base.constants import PathConflict, ChannelPriority from conda.base.context import context, reset_context, conda_tests_ctxt_mgmt_def_pol from conda.common.compat import odict, iteritems from conda.common.configuration import ValidationError, YamlRawParameter from conda.common.io import env_var, env_vars from conda.common.path import expand, win_path_backout from conda.common.url import join_url, path_to_url from conda.common.serialize import yaml_load from conda.core.package_cache_data import PackageCacheData from conda.gateways.disk.create import mkdir_p, create_package_cache_directory from conda.gateways.disk.delete import rm_rf from conda.gateways.disk.permissions import make_read_only from conda.gateways.disk.update import touch from conda.models.channel import Channel from conda.models.match_spec import MatchSpec from conda.utils import on_win from ..helpers import tempdir class ContextCustomRcTests(TestCase): def setUp(self): string = dals(""" custom_channels: darwin: https://some.url.somewhere/stuff chuck: http://another.url:8080/with/path custom_multichannels: michele: - https://do.it.with/passion - learn_from_every_thing steve: - more-downloads migrated_custom_channels: darwin: s3://just/cant chuck: file:///var/lib/repo/ migrated_channel_aliases: - https://conda.anaconda.org channel_alias: ftp://new.url:8082 conda-build: root-dir: /some/test/path proxy_servers: http: http://user:pass@corp.com:8080 https: none ftp: sftp: '' ftps: false rsync: 'false' aggressive_update_packages: [] channel_priority: false """) reset_context(()) rd = odict(testdata=YamlRawParameter.make_raw_parameters('testdata', yaml_load(string))) context._set_raw_data(rd) def tearDown(self): reset_context() def test_migrated_custom_channels(self): assert Channel('https://some.url.somewhere/stuff/darwin/noarch/a-mighty-fine.tar.bz2').canonical_name == 'darwin' assert Channel('s3://just/cant/darwin/noarch/a-mighty-fine.tar.bz2').canonical_name == 'darwin' assert Channel('s3://just/cant/darwin/noarch/a-mighty-fine.tar.bz2').urls() == [ 'https://some.url.somewhere/stuff/darwin/noarch'] def test_old_channel_alias(self): platform = context.subdir cf_urls = ["ftp://new.url:8082/conda-forge/%s" % platform, "ftp://new.url:8082/conda-forge/noarch"] assert Channel('conda-forge').urls() == cf_urls url = "https://conda.anaconda.org/conda-forge/osx-64/some-great-package.tar.bz2" assert Channel(url).canonical_name == 'conda-forge' assert Channel(url).base_url == 'ftp://new.url:8082/conda-forge' assert Channel(url).urls() == [ 'ftp://new.url:8082/conda-forge/osx-64', 'ftp://new.url:8082/conda-forge/noarch' ] assert Channel("https://conda.anaconda.org/conda-forge/label/dev/linux-64/" "some-great-package.tar.bz2").urls() == [ "ftp://new.url:8082/conda-forge/label/dev/linux-64", "ftp://new.url:8082/conda-forge/label/dev/noarch", ] def test_client_ssl_cert(self): string = dals(""" client_ssl_cert_key: /some/key/path """) reset_context() rd = odict(testdata=YamlRawParameter.make_raw_parameters('testdata', yaml_load(string))) context._set_raw_data(rd) pytest.raises(ValidationError, context.validate_configuration) def test_conda_envs_path(self): saved_envs_path = os.environ.get('CONDA_ENVS_PATH') beginning = "C:" + os.sep if on_win else os.sep path1 = beginning + os.sep.join(['my', 'envs', 'dir', '1']) path2 = beginning + os.sep.join(['my', 'envs', 'dir', '2']) try: os.environ['CONDA_ENVS_PATH'] = path1 reset_context() assert context.envs_dirs[0] == path1 os.environ['CONDA_ENVS_PATH'] = os.pathsep.join([path1, path2]) reset_context() assert context.envs_dirs[0] == path1 assert context.envs_dirs[1] == path2 finally: if saved_envs_path: os.environ['CONDA_ENVS_PATH'] = saved_envs_path else: del os.environ['CONDA_ENVS_PATH'] def test_conda_bld_path(self): conda_bld_path = join(gettempdir(), 'conda-bld') conda_bld_url = path_to_url(conda_bld_path) try: mkdir_p(conda_bld_path) with env_var('CONDA_BLD_PATH', conda_bld_path, stack_callback=conda_tests_ctxt_mgmt_def_pol): assert len(context.conda_build_local_paths) >= 1 assert context.conda_build_local_paths[0] == conda_bld_path channel = Channel('local') assert channel.channel_name == "local" assert channel.channel_location is None assert channel.platform is None assert channel.package_filename is None assert channel.auth is None assert channel.token is None assert channel.scheme is None assert channel.canonical_name == "local" assert channel.url() is None urls = list(concat(( join_url(url, context.subdir), join_url(url, 'noarch'), ) for url in context.conda_build_local_urls)) assert channel.urls() == urls channel = Channel(conda_bld_url) assert channel.canonical_name == "local" assert channel.platform is None assert channel.package_filename is None assert channel.auth is None assert channel.token is None assert channel.scheme == "file" assert channel.urls() == [ join_url(conda_bld_url, context.subdir), join_url(conda_bld_url, 'noarch'), ] assert channel.url() == join_url(conda_bld_url, context.subdir) assert channel.channel_name.lower() == win_path_backout(conda_bld_path).lstrip('/').lower() assert channel.channel_location == '' # location really is an empty string; all path information is in channel_name assert channel.canonical_name == "local" finally: rm_rf(conda_bld_path) def test_custom_multichannels(self): assert context.custom_multichannels['michele'] == ( Channel('passion'), Channel('learn_from_every_thing'), ) def test_restore_free_channel(self): assert 'https://repo.anaconda.com/pkgs/free' not in context.default_channels with env_var("CONDA_RESTORE_FREE_CHANNEL", 'true', stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.default_channels.index('https://repo.anaconda.com/pkgs/free') == 1 def test_proxy_servers(self): assert context.proxy_servers['http'] == 'http://user:pass@corp.com:8080' assert context.proxy_servers['https'] is None assert context.proxy_servers['ftp'] is None assert context.proxy_servers['sftp'] == '' assert context.proxy_servers['ftps'] == 'False' assert context.proxy_servers['rsync'] == 'false' def test_conda_build_root_dir(self): assert context.conda_build['root-dir'] == "/some/test/path" def test_clobber_enum(self): with env_var("CONDA_PATH_CONFLICT", 'prevent', stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.path_conflict == PathConflict.prevent def test_context_parameter_map(self): all_parameter_names = context.list_parameters() all_mapped_parameter_names = tuple(chain.from_iterable(context.category_map.values())) unmapped_parameter_names = set(all_parameter_names) - set(all_mapped_parameter_names) assert not unmapped_parameter_names, unmapped_parameter_names assert len(all_parameter_names) == len(all_mapped_parameter_names) def test_context_parameters_have_descriptions(self): skip_categories = ('CLI-only', 'Hidden and Undocumented') documented_parameter_names = chain.from_iterable(( parameter_names for category, parameter_names in iteritems(context.category_map) if category not in skip_categories )) from pprint import pprint for name in documented_parameter_names: description = context.get_descriptions()[name] pprint(context.describe_parameter(name)) def test_local_build_root_custom_rc(self): assert context.local_build_root == "C:\\some\\test\\path" if on_win else "/some/test/path" test_path_1 = join(os.getcwd(), 'test_path_1') with env_var("CONDA_CROOT", test_path_1, stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.local_build_root == test_path_1 test_path_2 = join(os.getcwd(), 'test_path_2') with env_var("CONDA_BLD_PATH", test_path_2, stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.local_build_root == test_path_2 def test_default_target_is_root_prefix(self): assert context.target_prefix == context.root_prefix def test_target_prefix(self): with tempdir() as prefix: mkdir_p(join(prefix, 'first', 'envs')) mkdir_p(join(prefix, 'second', 'envs')) create_package_cache_directory(join(prefix, 'first', 'pkgs')) create_package_cache_directory(join(prefix, 'second', 'pkgs')) envs_dirs = (join(prefix, 'first', 'envs'), join(prefix, 'second', 'envs')) with env_var('CONDA_ENVS_DIRS', os.pathsep.join(envs_dirs), stack_callback=conda_tests_ctxt_mgmt_def_pol): # with both dirs writable, choose first reset_context((), argparse_args=AttrDict(name='blarg', func='create')) assert context.target_prefix == join(envs_dirs[0], 'blarg') # with first dir read-only, choose second PackageCacheData._cache_.clear() make_read_only(join(envs_dirs[0], '.conda_envs_dir_test')) reset_context((), argparse_args=AttrDict(name='blarg', func='create')) assert context.target_prefix == join(envs_dirs[1], 'blarg') # if first dir is read-only but environment exists, choose first PackageCacheData._cache_.clear() mkdir_p(join(envs_dirs[0], 'blarg')) touch(join(envs_dirs[0], 'blarg', 'history')) reset_context((), argparse_args=AttrDict(name='blarg', func='create')) assert context.target_prefix == join(envs_dirs[0], 'blarg') def test_aggressive_update_packages(self): assert context.aggressive_update_packages == tuple() specs = ['certifi', 'openssl>=1.1'] with env_var('CONDA_AGGRESSIVE_UPDATE_PACKAGES', ','.join(specs), stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.aggressive_update_packages == tuple(MatchSpec(s) for s in specs) def test_channel_priority(self): assert context.channel_priority == ChannelPriority.DISABLED def test_cuda_detection(self): # confirm that CUDA detection doesn't raise exception version = context.cuda_version assert version is None or isinstance(version, str) def test_cuda_override(self): with env_var('CONDA_OVERRIDE_CUDA', '4.5'): version = context.cuda_version assert version == '4.5' def test_cuda_override_none(self): with env_var('CONDA_OVERRIDE_CUDA', ''): version = context.cuda_version assert version is None def test_threads(self): default_value = None assert context.default_threads == default_value assert context.repodata_threads == default_value assert context.verify_threads == 1 assert context.execute_threads == 1 with env_var('CONDA_DEFAULT_THREADS', '3', stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.default_threads == 3 assert context.verify_threads == 3 assert context.repodata_threads == 3 assert context.execute_threads == 3 with env_var('CONDA_VERIFY_THREADS', '3', stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.default_threads == default_value assert context.verify_threads == 3 assert context.repodata_threads == default_value assert context.execute_threads == 1 with env_var('CONDA_REPODATA_THREADS', '3', stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.default_threads == default_value assert context.verify_threads == 1 assert context.repodata_threads == 3 assert context.execute_threads == 1 with env_var('CONDA_EXECUTE_THREADS', '3', stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.default_threads == default_value assert context.verify_threads == 1 assert context.repodata_threads == default_value assert context.execute_threads == 3 with env_vars({'CONDA_EXECUTE_THREADS': '3', 'CONDA_DEFAULT_THREADS': '1'}, stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.default_threads == 1 assert context.verify_threads == 1 assert context.repodata_threads == 1 assert context.execute_threads == 3 class ContextDefaultRcTests(TestCase): def test_subdirs(self): assert context.subdirs == (context.subdir, 'noarch') subdirs = ('linux-highest', 'linux-64', 'noarch') with env_var('CONDA_SUBDIRS', ','.join(subdirs), stack_callback=conda_tests_ctxt_mgmt_def_pol): assert context.subdirs == subdirs def test_local_build_root_default_rc(self): if context.root_writable: assert context.local_build_root == join(context.root_prefix, 'conda-bld') else: assert context.local_build_root == expand('~/conda-bld')