code: def _test_create_store(
      host, base_path=None, ds_path=None, clone_path=None
  ):
      ds = Dataset(ds_path).create(force=True)
  
      subds = ds.create("sub", force=True)
      subds2 = ds.create("sub2", force=True, annex=False)
      ds.save(recursive=True)
      assert_repo_status(ds.path)
      # don't specify special remote. By default should be git-remote + "-storage"
      res = ds.create_sibling_ria(
          "ria+ssh://test-store:",
          "datastore",
          post_update_hook=True,
          new_store_ok=True,
      )
      assert_result_count(
          res, 1, status="ok", action="create-sibling-ria"
      )
  
      # remotes exist, but only in super
      siblings = ds.siblings(result_renderer="disabled")
      eq_(
          {"datastore", "datastore-storage", "here"},
          {s["name"] for s in siblings},
      )
      sub_siblings = subds.siblings(result_renderer="disabled")
      eq_({"here"}, {s["name"] for s in sub_siblings})
      sub2_siblings = subds2.siblings(result_renderer="disabled")
      eq_({"here"}, {s["name"] for s in sub2_siblings})
  
      # check bare repo:
      git_dir = Path(base_path) / ds.id[:3] / ds.id[3:]
  
      # The post-update hook was enabled.
      ok_exists(git_dir / "hooks" / "post-update")
      # And create_sibling_ria took care of an initial call to
      # git-update-server-info.
      ok_exists(git_dir / "info" / "refs")
  
      git_config = git_dir / "config"
      ok_exists(git_config)
      content = git_config.read_text()
      assert_in('[datalad "ora-remote"]', content)
      super_uuid = ds.config.get(
          "remote.{}.annex-uuid".format("datastore-storage")
      )
      assert_in("uuid = {}".format(super_uuid), content)
  
      # implicit test of success by ria-installing from store:
      ds.push(to="datastore")
      with chpwd(clone_path):
          if host:
              # note, we are not using the "test-store"-label here
              clone(
                  "ria+ssh://{}{}#{}".format(host, base_path, ds.id),
                  path="test_install",
              )
          else:
              # TODO: Whenever ria+file supports special remote config (label),
              # change here:
              clone(
                  "ria+file://{}#{}".format(base_path, ds.id),
                  path="test_install",
              )
          installed_ds = Dataset(op.join(clone_path, "test_install"))
          assert installed_ds.is_installed()
          assert_repo_status(installed_ds.repo)
          eq_(installed_ds.id, ds.id)
          # Note: get_annexed_files() always reports POSIX paths.
          assert_in(
              "ds/file1.txt", installed_ds.repo.get_annexed_files()
          )
          assert_result_count(
              installed_ds.get(op.join("ds", "file1.txt")),
              1,
              status="ok",
              action="get",
              path=op.join(installed_ds.path, "ds", "file1.txt"),
          )
      # repeat the call to ensure it doesn't crash (see #6950)
      res = ds.create_sibling_ria(
          "ria+ssh://test-store:", "datastore", on_failure="ignore"
      )
      assert_result_count(
          res,
          1,
          status="error",
          action="create-sibling-ria",
          message=(
              "a sibling %r is already configured in dataset %r",
              "datastore",
              ds.path,
          ),
      )
  
      # now, again but recursive.
      res = ds.create_sibling_ria(
          "ria+ssh://test-store:",
          "datastore",
          recursive=True,
          existing="reconfigure",
          new_store_ok=True,
      )
      assert_result_count(
          res,
          1,
          path=str(ds.pathobj),
          status="ok",
          action="create-sibling-ria",
      )
      assert_result_count(
          res,
          1,
          path=str(subds.pathobj),
          status="ok",
          action="create-sibling-ria",
      )
      assert_result_count(
          res,
          1,
          path=str(subds2.pathobj),
          status="ok",
          action="create-sibling-ria",
      )
  
      # remotes now exist in super and sub
      siblings = ds.siblings(result_renderer="disabled")
      eq_(
          {"datastore", "datastore-storage", "here"},
          {s["name"] for s in siblings},
      )
      sub_siblings = subds.siblings(result_renderer="disabled")
      eq_(
          {"datastore", "datastore-storage", "here"},
          {s["name"] for s in sub_siblings},
      )
      # but no special remote in plain git subdataset:
      sub2_siblings = subds2.siblings(result_renderer="disabled")
      eq_({"datastore", "here"}, {s["name"] for s in sub2_siblings})
  
      # for testing trust_level parameter, redo for each label:
      for trust in ["trust", "semitrust", "untrust"]:
          ds.create_sibling_ria(
              "ria+ssh://test-store:",
              "datastore",
              existing="reconfigure",
              trust_level=trust,
              new_store_ok=True,
          )
          res = ds.repo.repo_info()
          assert_in(
              "[datastore-storage]",
              [
                  r["description"]
                  for r in res["{}ed repositories".format(trust)]
              ],
          )
  
file_path: datalad/datalad/distributed/tests/test_create_sibling_ria.py
imports: import logging
  import os.path
  from functools import wraps
  from unittest.mock import patch
  from datalad import cfg
  from datalad.api import Dataset, clone
  from datalad.support.network import get_local_file_url
  from datalad.tests.utils_pytest import assert_false, assert_in, assert_raises, assert_repo_status, assert_result_count, assert_status, attr, chpwd, eq_, known_failure_githubci_win, ok_exists, skip_if_on_windows, skip_ssh, skip_wo_symlink_capability, slow, swallow_logs, with_tempfile, with_tree
  from datalad.utils import Path
label_desc: Use image node_id 3 for calling a global function with example usage: _test_create_store(host, base_path, ds_path, clone_path) without return types