code: def test_run_inputs_outputs(src=None, path=None):
      for subds in [
          ("s0", "s1_0", "s2"),
          ("s0", "s1_1", "s2"),
          ("s0", "s1_0"),
          ("s0", "s1_1"),
          ("s0", "ss"),
          ("s0",),
      ]:
          Dataset(op.join(*((src,) + subds))).create(force=True).save()
      src_ds = Dataset(src).create(force=True)
      src_ds.save()
  
      ds = install(
          path,
          source=src,
          result_xfm="datasets",
          return_type="item-or-list",
      )
      assert_false(ds.repo.file_has_content("input.dat"))
      assert_false(ds.repo.file_has_content("extra-input.dat"))
  
      # The specified inputs and extra inputs will be retrieved before the run.
      # (Use run_command() to access the extra_inputs argument.)
      list(
          run_command(
              "{} {{inputs}} {{inputs}} >doubled.dat".format(
                  "type" if on_windows else "cat"
              ),
              dataset=ds,
              inputs=["input.dat"],
              extra_inputs=["extra-input.dat"],
          )
      )
  
      assert_repo_status(ds.path)
      ok_(ds.repo.file_has_content("input.dat"))
      ok_(ds.repo.file_has_content("extra-input.dat"))
      ok_(ds.repo.file_has_content("doubled.dat"))
      with open(op.join(path, "doubled.dat")) as fh:
          content = fh.read()
          assert_in("input", content)
          assert_not_in("extra-input", content)
  
      # Rerunning the commit will also get the input file.
      ds.repo.drop(
          ["input.dat", "extra-input.dat"], options=["--force"]
      )
      assert_false(ds.repo.file_has_content("input.dat"))
      assert_false(ds.repo.file_has_content("extra-input.dat"))
      ds.rerun()
      ok_(ds.repo.file_has_content("input.dat"))
      ok_(ds.repo.file_has_content("extra-input.dat"))
  
      # Test different combinations of globs and explicit files.
      inputs = ["a.dat", "b.dat", "c.txt", "d.txt"]
      create_tree(ds.path, {i: i for i in inputs})
  
      ds.save()
      ds.push(inputs, to=DEFAULT_REMOTE)
      ds.repo.drop(inputs, options=["--force"])
  
      test_cases = [
          (["*.dat"], ["a.dat", "b.dat"]),
          (["*.dat", "c.txt"], ["a.dat", "b.dat", "c.txt"]),
          (["*"], inputs),
      ]
  
      for idx, (inputs_arg, expected_present) in enumerate(test_cases):
          assert_false(any(ds.repo.file_has_content(i) for i in inputs))
  
          ds.run("cd .> dummy{}".format(idx), inputs=inputs_arg)
          ok_(
              all(ds.repo.file_has_content(f) for f in expected_present)
          )
          # Globs are stored unexpanded by default.
          assert_in(inputs_arg[0], last_commit_msg(ds.repo))
          ds.repo.drop(inputs, options=["--force"])
  
      # --input can be passed a subdirectory.
      create_tree(
          ds.path, {"subdir": {"a": "subdir a", "b": "subdir b"}}
      )
      ds.save("subdir")
      ds.push(["subdir/a", "subdir/b"], to=DEFAULT_REMOTE)
      ds.repo.drop("subdir", options=["--force"])
      ds.run("cd .> subdir-dummy", inputs=[op.join(ds.path, "subdir")])
      ok_(
          all(
              ds.repo.file_has_content(op.join("subdir", f))
              for f in ["a", "b"]
          )
      )
  
      # Inputs are specified relative to a dataset's subdirectory.
      ds.repo.drop(op.join("subdir", "a"), options=["--force"])
      with chpwd(op.join(path, "subdir")):
          run("cd .> subdir-dummy1", inputs=["a"])
      ok_(ds.repo.file_has_content(op.join("subdir", "a")))
  
      # --input=. runs "datalad get ."
      ds.run("cd .> dot-dummy", inputs=["."])
      eq_(
          ds.repo.get_annexed_files(),
          ds.repo.get_annexed_files(with_content_only=True),
      )
      # On rerun, we get all files, even those that weren't in the tree at the
      # time of the run.
      create_tree(ds.path, {"after-dot-run": "after-dot-run content"})
      ds.save()
      ds.push(["after-dot-run"], to=DEFAULT_REMOTE)
      ds.repo.drop(["after-dot-run"], options=["--force"])
      ds.rerun(DEFAULT_BRANCH + "^")
      ds.repo.file_has_content("after-dot-run")
  
      # --output will unlock files that are present.
      ds.repo.get("a.dat")
      ds.run("echo ' appended' >>a.dat", outputs=["a.dat"])
      with open(op.join(path, "a.dat")) as fh:
          eq_(
              fh.read(),
              "a.dat' appended' \n"
              if on_windows
              else "a.dat appended\n",
          )
  
      # --output will remove files that are not present.
      ds.repo.drop(["a.dat", "d.txt"], options=["--force"])
      ds.run("echo ' appended' >>a.dat", outputs=["a.dat"])
      if not on_windows:
          # MIH doesn't yet understand how to port this
          with open(op.join(path, "a.dat")) as fh:
              eq_(fh.read(), " appended\n")
  
      # --input can be combined with --output.
      ds.run(
          "echo ' appended' >>a.dat",
          inputs=["a.dat"],
          outputs=["a.dat"],
      )
      if not on_windows:
          # MIH doesn't yet understand how to port this
          with open(op.join(path, "a.dat")) as fh:
              eq_(fh.read(), " appended\n appended\n")
  
      ds.create("sub")
      ds.run("echo sub_orig >sub/subfile")
      ds.run("echo sub_overwrite >sub/subfile", outputs=["sub/subfile"])
      ds.drop("sub/subfile", reckless="kill")
      ds.run("echo sub_overwrite >sub/subfile", outputs=["sub/subfile"])
  
      # --input/--output globs can be stored in expanded form.
      ds.run(
          "cd .> expand-dummy",
          inputs=["a.*"],
          outputs=["b.*"],
          expand="both",
      )
      assert_in("a.dat", last_commit_msg(ds.repo))
      assert_in("b.dat", last_commit_msg(ds.repo))
  
      res = ds.rerun(report=True, return_type="item-or-list")
      eq_(res["run_info"]["inputs"], ["a.dat"])
      eq_(res["run_info"]["outputs"], ["b.dat"])
  
      # We uninstall subdatasets to fully resolve globs.
      ds.drop("s0", what="all", reckless="kill", recursive=True)
      assert_false(Dataset(op.join(path, "s0")).is_installed())
      ds.run(
          "echo {inputs} >globbed-subds", inputs=["s0/s1_*/s2/*.dat"]
      )
      ok_file_has_content(
          op.join(ds.path, "globbed-subds"),
          "'s0\\s1_0\\s2\\a.dat' 's0\\s1_1\\s2\\c.dat'"
          if on_windows
          else "s0/s1_0/s2/a.dat s0/s1_1/s2/c.dat",
          strip=True,
      )
  
      ds_ss = Dataset(op.join(path, "s0", "ss"))
      assert_false(ds_ss.is_installed())
      ds.run("echo blah >{outputs}", outputs=["s0/ss/out"])
      ok_(ds_ss.is_installed())
      ok_file_has_content(
          op.join(ds.path, "s0", "ss", "out"), "blah", strip=True
      )
  
imports: import logging
  import os.path
  import sys
  from io import StringIO
  from os import remove
  import pytest
  from unittest.mock import patch
  from datalad.api import install, run
  from datalad.core.local.run import run_command
  from datalad.core.local.tests.test_run import last_commit_msg
  from datalad.distribution.dataset import Dataset
  from datalad.local.rerun import diff_revision, get_run_info, new_or_modified
  from datalad.support.exceptions import CommandError, IncompleteResultsError
  from datalad.support.gitrepo import GitRepo
  from datalad.tests.utils_pytest import DEFAULT_BRANCH, DEFAULT_REMOTE, SkipTest, assert_dict_equal, assert_false, assert_in, assert_in_results, assert_not_in, assert_not_in_results, assert_raises, assert_repo_status, assert_result_count, assert_status, create_tree, eq_, known_failure_windows, neq_, ok_, ok_exists, ok_file_has_content, ok_file_under_git, skip_if_adjusted_branch, slow, swallow_logs, swallow_outputs, with_tempfile, with_tree
  from datalad.utils import Path, chpwd, on_windows
label_desc: Use image node_id 15 for calling a global function with example usage: test_run_inputs_outputs(src, path) without return types