text: # Copyright 2013-2021 Lawrence Livermore National Security, LLC and other
  # Spack Project Developers. See the top-level COPYRIGHT file for details.
  #
  # SPDX-License-Identifier: (Apache-2.0 OR MIT)
  
  import filecmp
  import os
  import pytest
  
  import spack.repo
  import spack.mirror
  import spack.util.executable
  from spack.spec import Spec
  from spack.stage import Stage
  from spack.util.executable import which
  
  from llnl.util.filesystem import resolve_link_target_relative_to_the_link
  
  pytestmark = pytest.mark.usefixtures('mutable_config', 'mutable_mock_repo')
  
  # paths in repos that shouldn't be in the mirror tarballs.
  exclude = ['.hg', '.git', '.svn']
  
  
  repos = {}
  
  
  def set_up_package(name, repository, url_attr):
      """Set up a mock package to be mirrored.
      Each package needs us to:
  
      1. Set up a mock repo/archive to fetch from.
      2. Point the package's version args at that repo.
      """
      # Set up packages to point at mock repos.
      spec = Spec(name)
      spec.concretize()
      # Get the package and fix its fetch args to point to a mock repo
      pkg = spack.repo.get(spec)
  
      repos[name] = repository
  
      # change the fetch args of the first (only) version.
      assert len(pkg.versions) == 1
      v = next(iter(pkg.versions))
  
      pkg.versions[v][url_attr] = repository.url
  
  
  def check_mirror():
      with Stage('spack-mirror-test') as stage:
          mirror_root = os.path.join(stage.path, 'test-mirror')
          # register mirror with spack config
          mirrors = {'spack-mirror-test': 'file://' + mirror_root}
          with spack.config.override('mirrors', mirrors):
              with spack.config.override('config:checksum', False):
                  specs = [Spec(x).concretized() for x in repos]
                  spack.mirror.create(mirror_root, specs)
  
              # Stage directory exists
              assert os.path.isdir(mirror_root)
  
              for spec in specs:
                  fetcher = spec.package.fetcher[0]
                  per_package_ref = os.path.join(
                      spec.name, '-'.join([spec.name, str(spec.version)]))
                  mirror_paths = spack.mirror.mirror_archive_paths(
                      fetcher,
                      per_package_ref)
                  expected_path = os.path.join(
                      mirror_root, mirror_paths.storage_path)
                  assert os.path.exists(expected_path)
  
              # Now try to fetch each package.
              for name, mock_repo in repos.items():
                  spec = Spec(name).concretized()
                  pkg = spec.package
  
                  with spack.config.override('config:checksum', False):
                      with pkg.stage:
                          pkg.do_stage(mirror_only=True)
  
                          # Compare the original repo with the expanded archive
                          original_path = mock_repo.path
                          if 'svn' in name:
                              # have to check out the svn repo to compare.
                              original_path = os.path.join(
                                  mock_repo.path, 'checked_out')
  
                              svn = which('svn', required=True)
                              svn('checkout', mock_repo.url, original_path)
  
                          dcmp = filecmp.dircmp(
                              original_path, pkg.stage.source_path)
  
                          # make sure there are no new files in the expanded
                          # tarball
                          assert not dcmp.right_only
                          # and that all original files are present.
                          assert all(left in exclude for left in dcmp.left_only)
  
  
  def test_url_mirror(mock_archive):
      set_up_package('trivial-install-test-package', mock_archive, 'url')
      check_mirror()
      repos.clear()
  
  
  @pytest.mark.skipif(
      not which('git'), reason='requires git to be installed')
  def test_git_mirror(mock_git_repository):
      set_up_package('git-test', mock_git_repository, 'git')
      check_mirror()
      repos.clear()
  
  
  @pytest.mark.skipif(
      not which('svn') or not which('svnadmin'),
      reason='requires subversion to be installed')
  def test_svn_mirror(mock_svn_repository):
      set_up_package('svn-test', mock_svn_repository, 'svn')
      check_mirror()
      repos.clear()
  
  
  @pytest.mark.skipif(
      not which('hg'), reason='requires mercurial to be installed')
  def test_hg_mirror(mock_hg_repository):
      set_up_package('hg-test', mock_hg_repository, 'hg')
      check_mirror()
      repos.clear()
  
  
  @pytest.mark.skipif(
      not all([which('svn'), which('hg'), which('git')]),
      reason='requires subversion, git, and mercurial to be installed')
  def test_all_mirror(
          mock_git_repository,
          mock_svn_repository,
          mock_hg_repository,
          mock_archive):
  
      set_up_package('git-test', mock_git_repository, 'git')
      set_up_package('svn-test', mock_svn_repository, 'svn')
      set_up_package('hg-test', mock_hg_repository, 'hg')
      set_up_package('trivial-install-test-package', mock_archive, 'url')
      check_mirror()
      repos.clear()
  
  
  def test_mirror_archive_paths_no_version(mock_packages, config, mock_archive):
      spec = Spec('trivial-install-test-package@nonexistingversion')
      fetcher = spack.fetch_strategy.URLFetchStrategy(mock_archive.url)
      spack.mirror.mirror_archive_paths(fetcher, 'per-package-ref', spec)
  
  
  def test_mirror_with_url_patches(mock_packages, config, monkeypatch):
      spec = Spec('patch-several-dependencies')
      spec.concretize()
  
      files_cached_in_mirror = set()
  
      def record_store(_class, fetcher, relative_dst, cosmetic_path=None):
          files_cached_in_mirror.add(os.path.basename(relative_dst))
  
      def successful_fetch(_class):
          with open(_class.stage.save_filename, 'w'):
              pass
  
      def successful_expand(_class):
          expanded_path = os.path.join(_class.stage.path,
                                       spack.stage._source_path_subdir)
          os.mkdir(expanded_path)
          with open(os.path.join(expanded_path, 'test.patch'), 'w'):
              pass
  
      def successful_apply(*args, **kwargs):
          pass
  
      with Stage('spack-mirror-test') as stage:
          mirror_root = os.path.join(stage.path, 'test-mirror')
  
          monkeypatch.setattr(spack.fetch_strategy.URLFetchStrategy, 'fetch',
                              successful_fetch)
          monkeypatch.setattr(spack.fetch_strategy.URLFetchStrategy,
                              'expand', successful_expand)
          monkeypatch.setattr(spack.patch, 'apply_patch', successful_apply)
          monkeypatch.setattr(spack.caches.MirrorCache, 'store', record_store)
  
          with spack.config.override('config:checksum', False):
              spack.mirror.create(mirror_root, list(spec.traverse()))
  
          assert not (set([
              'abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234',
              'abcdabcdabcdabcdabcdabcdabcdabcdabcdabcdabcdabcdabcdabcdabcdabcd.gz'  # NOQA: ignore=E501
          ]) - files_cached_in_mirror)
  
  
  class MockFetcher(object):
      """Mock fetcher object which implements the necessary functionality for
         testing MirrorCache
      """
      @staticmethod
      def archive(dst):
          with open(dst, 'w'):
              pass
  
  
  @pytest.mark.regression('14067')
  def test_mirror_cache_symlinks(tmpdir):
      """Confirm that the cosmetic symlink created in the mirror cache (which may
         be relative) targets the storage path correctly.
      """
      cosmetic_path = 'zlib/zlib-1.2.11.tar.gz'
      global_path = '_source-cache/archive/c3/c3e5.tar.gz'
      cache = spack.caches.MirrorCache(str(tmpdir), False)
      reference = spack.mirror.MirrorReference(cosmetic_path, global_path)
  
      cache.store(MockFetcher(), reference.storage_path)
      cache.symlink(reference)
  
      link_target = resolve_link_target_relative_to_the_link(
          os.path.join(cache.root, reference.cosmetic_path))
      assert os.path.exists(link_target)
      assert (os.path.normpath(link_target) ==
              os.path.join(cache.root, reference.storage_path))
  