code: def test_target_ssh_simple(
      origin=None, src_path=None, target_rootpath=None
  ):
      ca = dict(result_renderer="disabled")
      test_fname = "test-annex.dat"
      orig = Dataset(origin).create(**ca)
      (orig.pathobj / test_fname).write_text("some")
      orig.save(**ca)
  
      port = get_ssh_port("datalad-test")
      # prepare src
      source = install(
          src_path,
          source=origin,
          result_xfm="datasets",
          return_type="item-or-list",
      )
  
      target_path = opj(target_rootpath, "basic")
      with swallow_logs(new_level=logging.ERROR) as cml:
          create_sibling(
              dataset=source,
              name="local_target",
              sshurl="ssh://datalad-test:{}".format(port),
              target_dir=target_path,
              ui=have_webui(),
          )
          assert_not_in("enableremote local_target failed", cml.out)
  
      target_gitrepo = GitRepo(
          target_path, create=False
      )  # raises if not a git repo
      assert_in("local_target", source.repo.get_remotes())
      # Both must be annex or git repositories
      src_is_annex = AnnexRepo.is_valid_repo(src_path)
      eq_(src_is_annex, AnnexRepo.is_valid_repo(target_path))
      # And target one should be known to have a known UUID within the source if annex
      if src_is_annex:
          lclcfg = AnnexRepo(src_path).config
          target_aversion = target_gitrepo.config["annex.version"]
          # basic config in place
          eq_(lclcfg.get("remote.local_target.annex-ignore"), "false")
          ok_(lclcfg.get("remote.local_target.annex-uuid"))
  
      # do it again without force, but use a different name to avoid initial checks
      # for existing remotes:
      with assert_raises(RuntimeError) as cm:
          assert_create_sshwebserver(
              dataset=source,
              name="local_target_alt",
              sshurl="ssh://datalad-test",
              target_dir=target_path,
          )
      ok_(
          str(cm.value).startswith(
              "Target path %s already exists." % target_path
          )
      )
      if src_is_annex:
          # Before we "talk" to it with git-annex directly -- we must prevent auto-upgrades
          # since the git-annex on "remote server" (e.g. docker container) could be outdated
          # and not support new git-annex repo version
          target_gitrepo.config.set(
              "annex.autoupgraderepository", "false", scope="local"
          )
          try:
              target_description = AnnexRepo(
                  target_path, create=False
              ).get_description()
          except CommandError as e:
              if "is at unsupported version" not in str(e.stderr):
                  raise
              # we just would skip this part of the test and avoid future similar query
              target_description = None
          else:
              assert (
                  target_gitrepo.config["annex.version"]
                  == target_aversion
              )
              assert_not_equal(target_description, None)
              assert_not_equal(target_description, target_path)
              # on yoh's laptop TMPDIR is under HOME, so things start to become
              # tricky since then target_path is shortened and we would need to know
              # remote $HOME.  To not over-complicate and still test, test only for
              # the basename of the target_path
              ok_endswith(target_description, basename(target_path))
      # now, with force and correct url, which is also used to determine
      # target_dir
      # Note: on windows absolute path is not url conform. But this way it's easy
      # to test, that ssh path is correctly used.
      if not on_windows:
          # add random file under target_path, to explicitly test existing=replace
          open(opj(target_path, "random"), "w").write("123")
  
          @with_testsui(responses=["yes"])
          def interactive_assert_create_sshwebserver():
              assert_create_sshwebserver(
                  dataset=source,
                  name="local_target",
                  sshurl="ssh://datalad-test" + target_path,
                  publish_by_default=DEFAULT_BRANCH,
                  existing="replace",
                  ui=have_webui(),
              )
  
          interactive_assert_create_sshwebserver()
  
          eq_(
              "ssh://datalad-test" + urlquote(target_path),
              source.repo.get_remote_url("local_target"),
          )
          ok_(
              source.repo.get_remote_url("local_target", push=True)
              is None
          )
  
          # ensure target tree actually replaced by source
          assert_false(exists(opj(target_path, "random")))
  
          if src_is_annex:
              lclcfg = AnnexRepo(src_path).config
              eq_(
                  lclcfg.get("remote.local_target.annex-ignore"),
                  "false",
              )
              # valid uuid
              eq_(
                  lclcfg.get("remote.local_target.annex-uuid").count(
                      "-"
                  ),
                  4,
              )
              # should be added too, even if URL matches prior state
              eq_(
                  lclcfg.get("remote.local_target.push"), DEFAULT_BRANCH
              )
  
          # again, by explicitly passing urls. Since we are on datalad-test, the
          # could use local path, but then it would not use "remote" git-annex
          # and thus potentially lead to incongruent result. So make URLs a bit
          # different by adding trailing /. to regular target_url
          target_url = "ssh://datalad-test" + target_path + "/."
          cpkwargs = dict(
              dataset=source,
              name="local_target",
              sshurl="ssh://datalad-test",
              target_dir=target_path,
              target_url=target_url,
              target_pushurl="ssh://datalad-test" + target_path,
              ui=have_webui(),
          )
  
          @with_testsui(responses=["yes"])
          def interactive_assert_create_sshwebserver():
              assert_create_sshwebserver(existing="replace", **cpkwargs)
  
          interactive_assert_create_sshwebserver()
  
          if src_is_annex and target_description:
              target_description = AnnexRepo(
                  target_path, create=False
              ).get_description()
              eq_(target_description, target_url)
  
          eq_(target_url, source.repo.get_remote_url("local_target"))
          eq_(
              "ssh://datalad-test" + target_path,
              source.repo.get_remote_url("local_target", push=True),
          )
  
          if have_webui():
              from datalad_deprecated.tests.test_create_sibling_webui import (
                  assert_publish_with_ui,
              )
  
              assert_publish_with_ui(target_path)
  
          # now, push should work:
          push(dataset=source, to="local_target")
  
          # and we should be able to 'reconfigure'
          def process_digests_mtimes(digests, mtimes):
              # it should have triggered a hook, which would have created log and metadata files
              check_metadata = False
              for part in "logs", "metadata":
                  metafiles = [
                      k
                      for k in digests
                      if k.startswith(_path_(".git/datalad/%s/" % part))
                  ]
                  # This is in effect ONLY if we have "compatible" datalad installed on remote
                  # end. ATM we don't have easy way to guarantee that AFAIK (yoh),
                  # so let's not check/enforce (TODO)
                  # assert(len(metafiles) >= 1)  # we might have 2 logs if timestamps do not collide ;)
                  # Let's actually do it to some degree
                  if part == "logs":
                      # always should have those:
                      assert len(metafiles) >= 1
                      with open(opj(target_path, metafiles[0])) as f:
                          if "no datalad found" not in f.read():
                              check_metadata = True
                  if part == "metadata":
                      eq_(len(metafiles), bool(check_metadata))
                  for f in metafiles:
                      digests.pop(f)
                      mtimes.pop(f)
              # and just pop some leftovers from annex
              # and ignore .git/logs content (gh-5298)
              for f in list(digests):
                  if f.startswith(
                      ".git/annex/mergedrefs"
                  ) or f.startswith(".git/logs/"):
                      digests.pop(f)
                      mtimes.pop(f)
  
          if not have_webui():
              # the rest of the test assumed that we have uploaded a UI
              return
          orig_digests, orig_mtimes = get_mtimes_and_digests(
              target_path
          )
          process_digests_mtimes(orig_digests, orig_mtimes)
  
          import time
  
          time.sleep(0.1)  # just so that mtimes change
          assert_create_sshwebserver(existing="reconfigure", **cpkwargs)
          digests, mtimes = get_mtimes_and_digests(target_path)
          process_digests_mtimes(digests, mtimes)
  
          assert_dict_equal(
              orig_digests, digests
          )  # nothing should change in terms of content
  
          # but some files should have been modified
          modified_files = {
              k
              for k in mtimes
              if orig_mtimes.get(k, 0) != mtimes.get(k, 0)
          }
          # collect which files were expected to be modified without incurring any changes
          ok_modified_files = {
              _path_(".git/hooks/post-update"),
              "index.html",
          }
          ok_modified_files.add(_path_(".git/config"))
          ok_modified_files.update(
              {
                  f
                  for f in digests
                  if f.startswith(_path_(".git/datalad/web"))
              }
          )
          # it seems that with some recent git behavior has changed a bit
          # and index might get touched
          if _path_(".git/index") in modified_files:
              ok_modified_files.add(_path_(".git/index"))
          ok_(modified_files.issuperset(ok_modified_files))
  
imports: import logging
  import os
  import stat
  import sys
  from os import chmod
  from os.path import basename, exists
  from os.path import join
  import pytest
  from datalad.api import create_sibling, install, push
  from datalad.cmd import StdOutErrCapture
  from datalad.cmd import WitlessRunner
  from datalad.distribution.create_sibling import _RunnerAdapter
  from datalad.support.annexrepo import AnnexRepo
  from datalad.support.exceptions import CommandError, InsufficientArgumentsError
  from datalad.support.gitrepo import GitRepo
  from datalad.support.network import urlquote
  from datalad.tests.utils_testdatasets import _mk_submodule_annex
  from datalad.tests.utils_pytest import DEFAULT_BRANCH, DEFAULT_REMOTE, SkipTest, assert_dict_equal, assert_false, assert_in, assert_in_results, assert_no_errors_logged, assert_not_equal, assert_not_in, assert_raises, assert_repo_status, assert_result_count, assert_status, create_tree, eq_, get_mtimes_and_digests, get_ssh_port, known_failure_windows, ok_, ok_endswith, ok_file_has_content, ok_file_under_git, skip_if_on_windows, skip_if_root, skip_ssh, slow, swallow_logs, with_tempfile, with_testsui
  from datalad.utils import Path, _path_, chpwd, on_windows
  from ..dataset import Dataset
label_desc: Use image node_id 4 for calling a global function with example usage: test_target_ssh_simple(origin, src_path, target_rootpath) without return types