code: def check_push(annex, src_path, dst_path):
      # prepare src
      src = Dataset(src_path).create(annex=annex)
      src_repo = src.repo
      # push should not add branches to the local dataset
      orig_branches = src_repo.get_branches()
      assert_not_in("synced/" + DEFAULT_BRANCH, orig_branches)
  
      res = src.push(on_failure="ignore")
      assert_result_count(res, 1)
      assert_in_results(
          res,
          status="impossible",
          message="No push target given, and none could be auto-detected, "
          "please specify via --to",
      )
      eq_(orig_branches, src_repo.get_branches())
      # target sibling
      target = mk_push_target(src, "target", dst_path, annex=annex)
      eq_(orig_branches, src_repo.get_branches())
  
      res = src.push(to="target")
      eq_(orig_branches, src_repo.get_branches())
      assert_result_count(res, 2 if annex else 1)
      assert_in_results(
          res,
          action="publish",
          status="ok",
          target="target",
          refspec=DEFAULT_REFSPEC,
          operations=["new-branch"],
      )
  
      assert_repo_status(src_repo, annex=annex)
      eq_(
          list(target.get_branch_commits_(DEFAULT_BRANCH)),
          list(src_repo.get_branch_commits_(DEFAULT_BRANCH)),
      )
  
      # configure a default merge/upstream target
      src.config.set(
          "branch.{}.remote".format(DEFAULT_BRANCH),
          "target",
          scope="local",
      )
      src.config.set(
          "branch.{}.merge".format(DEFAULT_BRANCH),
          DEFAULT_BRANCH,
          scope="local",
      )
  
      # don't fail when doing it again, no explicit target specification
      # needed anymore
      res = src.push()
      eq_(orig_branches, src_repo.get_branches())
      # and nothing is pushed
      assert_status("notneeded", res)
  
      assert_repo_status(src_repo, annex=annex)
      eq_(
          list(target.get_branch_commits_(DEFAULT_BRANCH)),
          list(src_repo.get_branch_commits_(DEFAULT_BRANCH)),
      )
  
      # some modification:
      (src.pathobj / "test_mod_file").write_text(
          "Some additional stuff."
      )
      src.save(to_git=True, message="Modified.")
      (src.pathobj / "test_mod_annex_file").write_text("Heavy stuff.")
      src.save(to_git=not annex, message="Modified again.")
      assert_repo_status(src_repo, annex=annex)
  
      # we could say since='HEAD~2' to make things fast, or we are lazy
      # and say since='^' to indicate the state of the tracking remote
      # which is the same, because we made to commits since the last push.
      res = src.push(to="target", since="^", jobs=2)
      assert_in_results(
          res,
          action="publish",
          status="ok",
          target="target",
          refspec=DEFAULT_REFSPEC,
          # we get to see what happened
          operations=["fast-forward"],
      )
      if annex:
          # we got to see the copy result for the annexed files
          assert_in_results(
              res,
              action="copy",
              status="ok",
              path=str(src.pathobj / "test_mod_annex_file"),
          )
          # we published, so we can drop and reobtain
          ok_(src_repo.file_has_content("test_mod_annex_file"))
          src_repo.drop("test_mod_annex_file")
          ok_(not src_repo.file_has_content("test_mod_annex_file"))
          src_repo.get("test_mod_annex_file")
          ok_(src_repo.file_has_content("test_mod_annex_file"))
          ok_file_has_content(
              src_repo.pathobj / "test_mod_annex_file", "Heavy stuff."
          )
  
      eq_(
          list(target.get_branch_commits_(DEFAULT_BRANCH)),
          list(src_repo.get_branch_commits_(DEFAULT_BRANCH)),
      )
      if not (annex and src_repo.is_managed_branch()):
          # the following doesn't make sense in managed branches, because
          # a commit that could be amended is no longer the last commit
          # of a branch after a sync has happened (which did happen
          # during the last push above
  
          # amend and change commit msg in order to test for force push:
          src_repo.commit("amended", options=["--amend"])
          # push should be rejected (non-fast-forward):
          res = src.push(
              to="target", since="HEAD~2", on_failure="ignore"
          )
          # fails before even touching the annex branch
          assert_in_results(
              res,
              action="publish",
              status="error",
              target="target",
              refspec=DEFAULT_REFSPEC,
              operations=["rejected", "error"],
          )
          # push with force=True works:
          res = src.push(to="target", since="HEAD~2", force="gitpush")
          assert_in_results(
              res,
              action="publish",
              status="ok",
              target="target",
              refspec=DEFAULT_REFSPEC,
              operations=["forced-update"],
          )
          eq_(
              list(target.get_branch_commits_(DEFAULT_BRANCH)),
              list(src_repo.get_branch_commits_(DEFAULT_BRANCH)),
          )
  
      # we do not have more branches than we had in the beginning
      # in particular no 'synced/<default branch>'
      eq_(orig_branches, src_repo.get_branches())
  
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 3 for calling a global function with example usage: check_push(annex, src_path, dst_path) without return types