code: def test_ria_http(lcl=None, storepath=None, url=None):
      # create a local dataset with a subdataset
      lcl = Path(lcl)
      storepath = Path(storepath)
      subds = Dataset(lcl / "ds" / "subdir" / "subds").create(
          force=True
      )
      subds.save()
      ds = Dataset(lcl / "ds").create(force=True)
      ds.save(version_tag="original")
      assert_repo_status(ds.path)
      for d in (ds, subds):
          _move2store(storepath, d)
      # location of superds in store
      storeds_loc = str(storepath / ds.id[:3] / ds.id[3:])
      # now we should be able to clone from a ria+http url
      # the super
      riaclone = clone(
          "ria+{}#{}".format(url, ds.id),
          lcl / "clone",
      )
  
      # due to default configuration, clone() should automatically look for the
      # subdataset in the store, too -- if not the following would fail, because
      # we never configured a proper submodule URL
      riaclonesub = riaclone.get(
          op.join("subdir", "subds"),
          get_data=False,
          result_xfm="datasets",
          return_type="item-or-list",
      )
  
      # both datasets came from the store and must be set up in an identical
      # fashion
      for origds, cloneds in ((ds, riaclone), (subds, riaclonesub)):
          eq_(origds.id, cloneds.id)
          if not ds.repo.is_managed_branch():
              # test logic cannot handle adjusted branches
              eq_(origds.repo.get_hexsha(), cloneds.repo.get_hexsha())
          ok_(
              cloneds.config.get(
                  f"remote.{DEFAULT_REMOTE}.url"
              ).startswith(url)
          )
          eq_(
              cloneds.config.get(
                  f"remote.{DEFAULT_REMOTE}.annex-ignore"
              ),
              "true",
          )
          eq_(
              cloneds.config.get(
                  "datalad.get.subdataset-source-candidate-200origin"
              ),
              "ria+%s#{id}" % url,
          )
  
      # now advance the source dataset
      (ds.pathobj / "newfile.txt").write_text("new")
      ds.save()
      ds.push(to="store")
      Runner(cwd=storeds_loc).run(["git", "update-server-info"])
      # re-clone as before
      riaclone2 = clone(
          "ria+{}#{}".format(url, ds.id),
          lcl / "clone2",
      )
      # and now clone a specific version, here given be the tag name
      riaclone_orig = clone(
          "ria+{}#{}@{}".format(url, ds.id, "original"),
          lcl / "clone_orig",
      )
      if not ds.repo.is_managed_branch():
          # test logic cannot handle adjusted branches
          # we got the precise version we wanted
          eq_(
              riaclone.repo.get_hexsha(),
              riaclone_orig.repo.get_hexsha(),
          )
          # and not the latest
          eq_(riaclone2.repo.get_hexsha(), ds.repo.get_hexsha())
          neq_(
              riaclone2.repo.get_hexsha(),
              riaclone_orig.repo.get_hexsha(),
          )
  
      # attempt to clone a version that doesn't exist
      with swallow_logs():
          with assert_raises(IncompleteResultsError) as cme:
              clone(
                  "ria+{}#{}@impossible".format(url, ds.id),
                  lcl / "clone_failed",
              )
          assert_in("not found in upstream", str(cme.value))
  
      # lastly test if URL rewriting is in effect
      # on the surface we clone from an SSH source identified by some custom
      # label, no full URL, but URL rewriting setup maps it back to the
      # HTTP URL used above
      with patch_config(
          {"url.ria+{}#.insteadof".format(url): "ria+ssh://somelabel#"}
      ):
          cloned_by_label = clone(
              "ria+ssh://somelabel#{}".format(origds.id),
              lcl / "cloned_by_label",
          )
      # so we get the same setup as above, but....
      eq_(origds.id, cloned_by_label.id)
      if not ds.repo.is_managed_branch():
          # test logic cannot handle adjusted branches
          eq_(
              origds.repo.get_hexsha(),
              cloned_by_label.repo.get_hexsha(),
          )
      ok_(
          cloned_by_label.config.get(
              f"remote.{DEFAULT_REMOTE}.url"
          ).startswith(url)
      )
      eq_(
          cloned_by_label.config.get(
              f"remote.{DEFAULT_REMOTE}.annex-ignore"
          ),
          "true",
      )
      # ... the clone candidates go with the label-based URL such that
      # future get() requests acknowledge a (system-wide) configuration
      # update
      eq_(
          cloned_by_label.config.get(
              "datalad.get.subdataset-source-candidate-200origin"
          ),
          "ria+ssh://somelabel#{id}",
      )
  
      if not has_symlink_capability():
          return
      # place a symlink in the store to serve as a dataset alias
      (storepath / "alias").mkdir()
      (storepath / "alias" / "myname").symlink_to(storeds_loc)
      with chpwd(lcl):
          cloned_by_alias = clone("ria+{}#~{}".format(url, "myname"))
      # still get the same data
      eq_(cloned_by_alias.id, ds.id)
      # more sensible default install path
      eq_(cloned_by_alias.pathobj.name, "myname")
  
imports: import logging
  import os.path
  import stat
  from unittest.mock import patch
  import pytest
  from datalad import consts
  from datalad.api import clone, create, remove
  from datalad.cmd import GitWitlessRunner
  from datalad.cmd import WitlessRunner
  from datalad.config import ConfigManager
  from datalad.core.distributed.clone_utils import _get_installationpath_from_url, decode_source_spec
  from datalad.distribution.dataset import Dataset
  from datalad.support.annexrepo import AnnexRepo
  from datalad.support.exceptions import IncompleteResultsError
  from datalad.support.external_versions import external_versions
  from datalad.support.gitrepo import GitRepo
  from datalad.support.network import get_local_file_url
  from datalad.tests.utils_pytest import DEFAULT_BRANCH, DEFAULT_REMOTE, SkipTest, assert_false, assert_in, assert_in_results, assert_message, assert_not_in, assert_not_is_instance, assert_raises, assert_repo_status, assert_result_count, assert_result_values_equal, assert_status, create_tree, eq_, get_datasets_topdir, has_symlink_capability, integration, known_failure, known_failure_githubci_win, known_failure_osx, known_failure_windows, neq_, nok_, ok_, ok_file_has_content, ok_startswith, on_github, on_osx, on_travis, patch_config, serve_path_via_http, set_date, skip_if, skip_if_adjusted_branch, skip_if_no_network, skip_if_on_windows, skip_ssh, slow, swallow_logs, with_sameas_remote, with_tempfile, with_tree, xfail_buggy_annex_info
  from datalad.utils import Path, chpwd, get_home_envvars, on_windows, rmtree
label_desc: Use image node_id 28 for calling a global function with example usage: test_ria_http(lcl, storepath, url) without return types