code: def test_add_archive_content(
      path_orig=None, url=None, repo_path=None
  ):
      with chpwd(repo_path):
          # TODO we need to be able to pass path into add_archive_content
          # We could mock but I mean for the API
  
          # no repo yet
          assert_raises(
              NoDatasetFound, add_archive_content, "nonexisting.tar.gz"
          )
          ds = Dataset(repo_path).create()
          res = ds.add_archive_content(
              "nonexisting.tar.gz", on_failure="ignore"
          )
          assert_in_results(
              res, action="add-archive-content", status="impossible"
          )
          repo = ds.repo
  
          # we can't add a file from outside the repo ATM
          res = ds.add_archive_content(
              Path(path_orig) / "1.tar.gz", on_failure="ignore"
          )
          assert_in_results(
              res,
              action="add-archive-content",
              status="impossible",
              type="dataset",
              message="Can not add archive outside of the dataset",
          )
  
          # Let's add first archive to the repo so we could test
          with swallow_outputs():
              repo.add_url_to_file("1.tar.gz", opj(url, "1.tar.gz"))
              for s in range(1, 5):
                  repo.add_url_to_file(
                      "%du/1.tar.gz" % s, opj(url, "%du/1.tar.gz" % s)
                  )
              repo.commit("added 1.tar.gz")
  
          key_1tar = repo.get_file_annexinfo("1.tar.gz")[
              "key"
          ]  # will be used in the test later
  
          def d1_basic_checks():
              ok_(exists("1"))
              ok_file_under_git("1", "1 f.txt", annexed=True)
              ok_file_under_git(opj("1", "d", "1d"), annexed=True)
              ok_archives_caches(repo_path, 0)
  
          # and by default it just does it, everything goes to annex
          res = add_archive_content("1.tar.gz")
          assert_in_results(
              res, action="add-archive-content", status="ok"
          )
          d1_basic_checks()
  
          # If ran again, should proceed just fine since the content is the same
          # so no changes would be made really
          res = add_archive_content("1.tar.gz")
          assert_in_results(
              res, action="add-archive-content", status="ok"
          )
  
          # But that other one carries updated file, so should fail due to
          # overwrite
          res = add_archive_content(
              Path("1u") / "1.tar.gz",
              use_current_dir=True,
              on_failure="ignore",
          )
          assert_in_results(
              res,
              action="add-archive-content",
              status="error",
          )
          assert_in(
              "exists, but would be overwritten by new file",
              res[0]["message"],
          )
          # but should do fine if overrides are allowed
          add_archive_content(
              Path("1u") / "1.tar.gz",
              existing="overwrite",
              use_current_dir=True,
          )
          add_archive_content(
              Path("2u") / "1.tar.gz",
              existing="archive-suffix",
              use_current_dir=True,
          )
          add_archive_content(
              Path("3u") / "1.tar.gz",
              existing="archive-suffix",
              use_current_dir=True,
          )
          add_archive_content(
              Path("4u") / "1.tar.gz",
              existing="archive-suffix",
              use_current_dir=True,
          )
  
          # rudimentary test
          assert_equal(
              sorted(map(basename, glob(opj(repo_path, "1", "1*")))),
              ["1 f-1.1.txt", "1 f-1.2.txt", "1 f-1.txt", "1 f.txt"],
          )
          whereis = repo.whereis(glob(opj(repo_path, "1", "1*")))
          # they all must be the same
          assert all([x == whereis[0] for x in whereis[1:]])
  
      # and we should be able to reference it while under subdirectory
      subdir = opj(repo_path, "subdir")
      with chpwd(subdir, mkdir=True):
          add_archive_content(
              opj(pardir, "1.tar.gz"),
              dataset=ds.path,
              use_current_dir=True,
          )
          d1_basic_checks()
          # or we could keep relative path and also demand to keep the archive prefix
          # while extracting under original (annex root) dir
          add_archive_content(
              opj(pardir, "1.tar.gz"),
              dataset=ds.path,
              add_archive_leading_dir=True,
          )
  
      with chpwd(opj(repo_path, "1")):
          d1_basic_checks()
  
      with chpwd(repo_path):
          # test with excludes and renames and annex options
          ds.add_archive_content(
              "1.tar.gz",
              exclude=["d"],
              rename=["/ /_", "/^1/2"],
              annex_options="-c annex.largefiles=exclude=*.txt",
              delete=True,
          )
          # no conflicts since new name
          ok_file_under_git("2", "1_f.txt", annexed=False)
          assert_false(exists(opj("2", "d")))
          assert_false(exists("1.tar.gz"))  # delete was in effect
  
      # now test ability to extract within subdir
      with chpwd(opj(repo_path, "d1"), mkdir=True):
          # Let's add first archive to the repo so we could test
          # named the same way but different content
          with swallow_outputs():
              repo.add_url_to_file(
                  "d1/1.tar.gz", opj(url, "d1", "1.tar.gz")
              )
          repo.commit("added 1.tar.gz in d1")
  
          def d2_basic_checks():
              ok_(exists("1"))
              ok_file_under_git("1", "2 f.txt", annexed=True)
              ok_file_under_git(opj("1", "d2", "2d"), annexed=True)
              ok_archives_caches(repo.path, 0)
  
          add_archive_content("1.tar.gz", dataset=ds.path)
          d2_basic_checks()
  
      # in manual tests ran into the situation of inability to obtain on a single run
      # a file from an archive which was coming from a dropped key.  I thought it was
      # tested in custom remote tests, but I guess not sufficiently well enough
      repo.drop(opj("1", "1 f.txt"))  # should be all kosher
      repo.get(opj("1", "1 f.txt"))
      ok_archives_caches(repo.path, 1, persistent=True)
      ok_archives_caches(repo.path, 0, persistent=False)
  
      repo.drop(opj("1", "1 f.txt"))  # should be all kosher
      repo.drop(
          key_1tar, key=True
      )  # is available from the URL -- should be kosher
      repo.get(opj("1", "1 f.txt"))  # that what managed to not work
  
      # TODO: check if persistent archive is there for the 1.tar.gz
  
      # We should be able to drop everything since available online
      with swallow_outputs():
          clean(dataset=ds)
      repo.drop(
          key_1tar, key=True
      )  # is available from the URL -- should be kosher
  
      ds.drop(opj("1", "1 f.txt"))  # should be all kosher
      ds.get(opj("1", "1 f.txt"))  # and should be able to get it again
  
      # bug was that dropping didn't work since archive was dropped first
      repo.call_annex(["drop", "--all"])
  
      # verify that we can't drop a file if archive key was dropped and online archive was removed or changed size! ;)
      repo.get(key_1tar, key=True)
      unlink(opj(path_orig, "1.tar.gz"))
      with assert_raises(CommandError) as e:
          repo.drop(key_1tar, key=True)
          assert_equal(e.kwargs["stdout_json"][0]["success"], False)
          assert_result_values_cond(
              e.kwargs["stdout_json"],
              "note",
              lambda x: "(Use --force to override this check, or adjust numcopies.)"
              in x,
          )
      assert exists(opj(repo.path, repo.get_contentlocation(key_1tar)))
  
imports: import os
  import tempfile
  from glob import glob
  from os import unlink
  from os.path import basename, exists
  from os.path import join
  from os.path import lexists, pardir
  from pathlib import Path
  from datalad.api import Dataset, add_archive_content, clean
  from datalad.consts import ARCHIVES_SPECIAL_REMOTE, DATALAD_SPECIAL_REMOTES_UUIDS
  from datalad.support.exceptions import CommandError, NoDatasetFound
  from datalad.support.external_versions import external_versions
  from datalad.tests.utils_pytest import assert_cwd_unchanged, assert_equal, assert_false, assert_in, assert_in_results, assert_not_in, assert_raises, assert_repo_status, assert_result_values_cond, assert_true, create_tree, eq_, integration, known_failure_windows, ok_, ok_archives_caches, ok_file_has_content, ok_file_under_git, serve_path_via_http, skip_if, skip_if_adjusted_branch, swallow_outputs, with_tempfile, with_tree, xfail_buggy_annex_info
  from datalad.utils import chpwd, find_files, get_tempfile_kwargs, getpwd, on_windows, rmtemp
label_desc: Use image node_id 2 for calling a global function with example usage: test_add_archive_content(path_orig, url, repo_path) without return types