code: def test_push_recursive(
      origin_path=None,
      src_path=None,
      dst_top=None,
      dst_sub=None,
      dst_subnoannex=None,
      dst_subsub=None,
  ):
      # dataset with two submodules and one subsubmodule
      origin = Dataset(origin_path).create()
      origin_subm1 = origin.create("sub m")
      origin_subm1.create("subsub m")
      origin.create("subm noannex", annex=False)
      origin.save()
      assert_repo_status(origin.path)
      # prepare src as a fresh clone with all subdatasets checkout out recursively
      # running on a clone should make the test scenario more different than
      # test_push(), even for the pieces that should be identical
      top = Clone.__call__(source=origin.path, path=src_path)
      subs = top.get(
          ".", recursive=True, get_data=False, result_xfm="datasets"
      )
      # order for '.' should not be relied upon, so sort by path
      sub, subsub, subnoannex = sorted(subs, key=lambda ds: ds.path)
  
      target_top = mk_push_target(top, "target", dst_top, annex=True)
      # subdatasets have no remote yet, so recursive publishing should fail:
      res = top.push(to="target", recursive=True, on_failure="ignore")
      check_datasets_order(res)
      assert_in_results(
          res,
          path=top.path,
          type="dataset",
          refspec=DEFAULT_REFSPEC,
          operations=["new-branch"],
          action="publish",
          status="ok",
          target="target",
      )
      for d in (sub, subsub, subnoannex):
          assert_in_results(
              res,
              status="error",
              type="dataset",
              path=d.path,
              message=("Unknown target sibling '%s'.", "target"),
          )
      # now fix that and set up targets for the submodules
      target_sub = mk_push_target(sub, "target", dst_sub, annex=True)
      target_subnoannex = mk_push_target(
          subnoannex, "target", dst_subnoannex, annex=False
      )
      target_subsub = mk_push_target(
          subsub, "target", dst_subsub, annex=True
      )
  
      # and same push call as above
      res = top.push(to="target", recursive=True)
      check_datasets_order(res)
      # topds skipped
      assert_in_results(
          res,
          path=top.path,
          type="dataset",
          action="publish",
          status="notneeded",
          target="target",
      )
      # the rest pushed
      for d in (sub, subsub, subnoannex):
          assert_in_results(
              res,
              status="ok",
              type="dataset",
              path=d.path,
              refspec=DEFAULT_REFSPEC,
          )
      # all corresponding branches match across all datasets
      for s, d in zip(
          (top, sub, subnoannex, subsub),
          (target_top, target_sub, target_subnoannex, target_subsub),
      ):
          eq_(
              list(s.repo.get_branch_commits_(DEFAULT_BRANCH)),
              list(d.get_branch_commits_(DEFAULT_BRANCH)),
          )
          if s != subnoannex:
              eq_(
                  list(s.repo.get_branch_commits_("git-annex")),
                  list(d.get_branch_commits_("git-annex")),
              )
  
      # rerun should not result in further pushes of the default branch
      res = top.push(to="target", recursive=True)
      check_datasets_order(res)
      assert_not_in_results(res, status="ok", refspec=DEFAULT_REFSPEC)
      assert_in_results(
          res, status="notneeded", refspec=DEFAULT_REFSPEC
      )
  
      # now annex a file in subsub
      test_copy_file = subsub.pathobj / "test_mod_annex_file"
      test_copy_file.write_text("Heavy stuff.")
      # save all the way up
      assert_status(
          ("ok", "notneeded"),
          top.save(message="subsub got something", recursive=True),
      )
      assert_repo_status(top.path)
      # publish straight up, should be smart by default
      res = top.push(to="target", recursive=True)
      check_datasets_order(res)
      # we see 3 out of 4 datasets pushed (sub noannex was left unchanged)
      for d in (top, sub, subsub):
          assert_in_results(
              res,
              status="ok",
              type="dataset",
              path=d.path,
              refspec=DEFAULT_REFSPEC,
          )
      # file content copied too
      assert_in_results(
          res, action="copy", status="ok", path=str(test_copy_file)
      )
      # verify it is accessible, drop and bring back
      assert_status("ok", top.drop(str(test_copy_file)))
      ok_(not subsub.repo.file_has_content("test_mod_annex_file"))
      top.get(test_copy_file)
      ok_file_has_content(test_copy_file, "Heavy stuff.")
  
      # make two modification
      (sub.pathobj / "test_mod_annex_file").write_text("annex")
      (subnoannex.pathobj / "test_mod_file").write_text("git")
      # save separately
      top.save(sub.pathobj, message="annexadd", recursive=True)
      top.save(subnoannex.pathobj, message="gitadd", recursive=True)
      # now only publish the latter one
      res = top.push(
          to="target", since=DEFAULT_BRANCH + "~1", recursive=True
      )
      # nothing copied, no reports on the other modification
      assert_not_in_results(res, action="copy")
      assert_not_in_results(res, path=sub.path)
      for d in (top, subnoannex):
          assert_in_results(
              res,
              status="ok",
              type="dataset",
              path=d.path,
              refspec=DEFAULT_REFSPEC,
          )
      # an unconditional push should now pick up the remaining changes
      res = top.push(to="target", recursive=True)
      assert_in_results(
          res,
          action="copy",
          status="ok",
          path=str(sub.pathobj / "test_mod_annex_file"),
      )
      assert_in_results(
          res,
          status="ok",
          type="dataset",
          path=sub.path,
          refspec=DEFAULT_REFSPEC,
      )
      for d in (top, subnoannex, subsub):
          assert_in_results(
              res,
              status="notneeded",
              type="dataset",
              path=d.path,
              refspec=DEFAULT_REFSPEC,
          )
  
      # if noannex target gets some annex, we still should not fail to push
      target_subnoannex.call_git(["annex", "init"])
      # just to ensure that we do need something to push
      (subnoannex.pathobj / "newfile").write_text("content")
      subnoannex.save()
      res = subnoannex.push(to="target")
      assert_in_results(res, status="ok", type="dataset")
  
imports: import logging
  import os
  import pytest
  from datalad.core.distributed.clone import Clone
  from datalad.core.distributed.push import Push
  from datalad.distribution.dataset import Dataset
  from datalad.support.annexrepo import AnnexRepo
  from datalad.support.exceptions import IncompleteResultsError, InsufficientArgumentsError
  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_not_in, assert_not_in_results, assert_raises, assert_repo_status, assert_result_count, assert_status, eq_, known_failure_githubci_osx, known_failure_githubci_win, neq_, ok_, ok_file_has_content, serve_path_via_http, skip_if_adjusted_branch, skip_if_on_windows, skip_ssh, slow, swallow_logs, with_tempfile, with_tree
  from datalad.utils import Path, chpwd, path_startswith, swallow_outputs
label_desc: Use image node_id 6 for calling a global function with example usage: test_push_recursive(origin_path, src_path, dst_top, dst_sub, dst_subnoannex, dst_subsub) without return types