def test_using_force_when_not_an_a_branch(qisrc_action, git_server):
    git_server.create_repo("foo.git")
    git_server.push_file("foo.git", "foo.txt", "this is foo")
    manifest_url = git_server.manifest_url
    qisrc_action("init", manifest_url)
    git_worktree = TestGitWorkTree()
    foo_proj = git_worktree.get_git_project("foo")
    git = qisrc.git.Git(foo_proj.path)
    git.checkout("HEAD~1")
    assert not git.get_current_branch()
    qisrc_action("checkout", "master", "--force")
    assert git.get_current_branch() == "master"

 def test_get_current_branch(self):
     bar_url = create_git_repo(self.tmp, "bar")
     work = os.path.join(self.tmp, "work")
     qibuild.sh.mkdir(work)
     bar_src = os.path.join(work, "bar")
     git = qisrc.git.Git(bar_src)
     git.clone(bar_url)
     self.assertEqual(git.get_current_branch(), "master")
     push_readme_v2(self.tmp, "bar", "master")
     git.pull()
     self.assertEqual(git.get_current_branch(), "master")
     git.checkout("-f", "HEAD~1")
     self.assertEqual(git.get_current_branch(), None)

def test_removing_forked_project(qisrc_action, git_server):
    git_server.create_repo("booz")
    git_server.switch_manifest_branch("devel")
    git_server.change_branch("booz", "devel")
    qisrc_action("init", git_server.manifest_url, "--branch", "devel")
    git_worktree = TestGitWorkTree()
    booz_proj = git_worktree.get_git_project("booz")
    git = qisrc.git.Git(booz_proj.path)
    assert git.get_current_branch() == "devel"
    git_server.change_branch("booz", "master")
    qisrc_action("sync", "-a", retcode=True)
    qisrc_action("checkout", "devel")
    assert git.get_current_branch() == "master"

def test_switch_manifest_branch(tmpdir, git_server):
    git_server.switch_manifest_branch("devel")
    assert git_server.manifest_branch == "devel"
    foo_clone = tmpdir.mkdir("foo")
    git = qisrc.git.Git(foo_clone.strpath)
    git.clone(git_server.manifest_url, "--branch", git_server.manifest_branch)
    assert git.get_current_branch() == "devel"

def do(args):
    """ Main entry point """
    git_worktree = qisrc.parsers.get_git_worktree(args)
    git_projects = qisrc.parsers.get_git_projects(git_worktree, args)
    for git_project in git_projects:
        maintainers = qisrc.maintainers.get(git_project)
        if not maintainers:
            mess = """\
The project in {src} has no maintainer.
Please edit {qiproject_xml} to silence this warning
"""
            ui.warning(mess.format(src=git_project.src, qiproject_xml=git_project.qiproject_xml), end="")
        reviewers = [x["email"] for x in maintainers]
        reviewers.extend(args.reviewers or list())
        git = qisrc.git.Git(git_project.path)
        current_branch = git.get_current_branch()
        if not current_branch:
            ui.error("Not currently on any branch")
            sys.exit(2)
        if git_project.review:
            qisrc.review.push(
                git_project,
                current_branch,
                bypass_review=(not args.review),
                dry_run=args.dry_run,
                reviewers=reviewers,
                topic=args.topic,
            )
        else:
            if args.dry_run:
                git.push("-n")
            else:
                git.push()

    def test_untracked_would_be_overwritten(self):
        bar_url = create_git_repo(self.tmp, "bar")
        work = os.path.join(self.tmp, "work")
        qibuild.sh.mkdir(work)
        bar_src = os.path.join(work, "bar")
        git = qisrc.git.Git(bar_src)
        git.clone(bar_url)

        # Create untracked, conflicting changes
        a_file = os.path.join(bar_src, "a_file")
        with open(a_file, "w") as fp:
            fp.write("a_file\n")

        upstream_src = os.path.join(self.tmp, "src", "bar")
        upstream_file = os.path.join(upstream_src, "a_file")
        with open(upstream_file, "w") as fp:
            fp.write("upstream file\n")
        upstream_git = qisrc.git.Git(upstream_src)
        upstream_git.call("add", "a_file")
        upstream_git.call("commit", "-m", "Add a file")
        upstream_git.call("push", bar_url, "master:master")

        err = git.update_branch("master", "origin")
        self.assertTrue(err)
        self.assertTrue("untracked working tree files" in err)
        self.assertEqual(git.get_current_branch(), "master")

def test_qisrc_checkout_with_branch_to_ref(qisrc_action, git_server):
    """ Test QiSrc Checkout With Branch to Ref """
    manifest_url = git_server.manifest_url
    git_server.create_repo("foo.git")
    git_server.create_repo("bar.git")
    git_server.push_file("foo.git", "a.txt", "a")
    git_server.push_tag("foo.git", "v0.1")
    git_server.push_file("foo.git", "b.txt", "b")
    git_server.push_tag("bar.git", "v0.2")
    git_server.push_file("bar.git", "c.txt", "b")
    qisrc_action("init", manifest_url)
    git_server.switch_manifest_branch("devel")
    git_server.set_fixed_ref("foo.git", "v0.1")
    git_server.set_fixed_ref("bar.git", "v0.2")
    git_worktree = TestGitWorkTree()
    foo_proj = git_worktree.get_git_project("foo")
    git = TestGit(foo_proj.path)
    assert git.get_current_branch() == "master"
    qisrc_action("checkout", "devel")
    _, sha1 = git.call("rev-parse", "HEAD", raises=False)
    expected = git.get_ref_sha1("refs/tags/v0.1")
    assert sha1 == expected
    bar_proj = git_worktree.get_git_project("bar")
    bar_git = qisrc.git.Git(bar_proj.path)
    _, sha1 = bar_git.call("rev-parse", "HEAD", raises=False)
    expected = bar_git.get_ref_sha1("refs/tags/v0.2")
    assert sha1 == expected

    def push_submodule(self, project, submodule_url, destination_dir,
                       branch="master", fast_forward=True,
                       message=None):
        """
        Push a submodule to the given project.
        It is assumed that the project has been created.
        """
        src = project.replace(".git", "")
        repo_src = self.src.join(src)
        git = qisrc.git.Git(repo_src.strpath)
        if git.get_current_branch() != branch:
            git.checkout("--force", "-B", branch)
        if not fast_forward:
            git.reset("--hard", "HEAD~1")

        to_write = repo_src.join(destination_dir)
        if to_write.exists():
            raise RuntimeError("path %s already exists" % destination_dir)

        if not message:
            message = "Add submodule %s" % destination_dir

        git.call("submodule", "add", submodule_url, destination_dir)
        git.add(destination_dir)
        git.commit("--message", message)
        if fast_forward:
            git.push("origin", "%s:%s" % (branch, branch))
        else:
            git.push("origin", "--force", "%s:%s" % (branch, branch))

def test_safe_checkout(cd_to_tmpdir, git_server):
    git_server.create_repo("foo.git")
    git = TestGit()
    git.clone(git_server.srv.join("foo.git").strpath)
    ok, mess = git.safe_checkout("devel", "origin")
    assert git.get_current_branch() == "devel"
    assert ok

def diff_worktree(git_worktree, git_projects, branch, cmd=None):
    """ Run  `git <cmd> local_branch..remote_branch` for every project

    """
    if not cmd:
        cmd = ["log"]
    remote_projects = git_worktree.get_projects_on_branch(branch)
    for git_project in git_projects:
        remote_project = remote_projects.get(git_project.src)
        if not remote_project:
            continue
        git = qisrc.git.Git(git_project.path)
        local_branch = git.get_current_branch()
        remote_branch = remote_project.default_branch.name
        remote_ref = "%s/%s" % (remote_project.default_remote.name, remote_branch)
        rc, out = git.call("merge-base", local_branch, remote_ref, raises=False)
        if rc != 0:
            continue
        merge_base = out.strip()
        full_cmd = cmd + ["%s..%s" % (merge_base, local_branch)]

        color = ui.config_color(sys.stdout)
        if color:
            full_cmd.append("--color=always")
        rc, out = git.call(*full_cmd, raises=False)
        if rc != 0:
            continue
        if not out:
            continue
        ui.info(ui.bold, git_project.src)
        ui.info(ui.bold, "-" * len(git_project.src))
        ui.info(out)
        ui.info()

def test_switching_from_fixed_ref_to_branch_local_changes(qisrc_action, git_server, record_messages):
    """ Test Swithcing From Fixed Ref To Branch Local Changes """
    git_server.create_repo("foo.git")
    git_server.push_file("foo.git", "a.txt", "a")
    git_server.push_tag("foo.git", "v0.1")
    git_server.push_file("foo.git", "b.txt", "b")
    git_server.push_tag("foo.git", "v0.2")
    git_server.push_file("foo.git", "c.txt", "c")
    git_server.set_fixed_ref("foo.git", "v0.1")
    qisrc_action("init", git_server.manifest_url)
    git_worktree = TestGitWorkTree()
    foo_proj = git_worktree.get_git_project("foo")
    git = TestGit(foo_proj.path)
    git.write_file("a.txt", "unstaged changes")
    git_server.set_branch("foo.git", "master")
    record_messages.reset()
    rc = qisrc_action("sync", retcode=True)
    # ERROR message must be displayed to warn user
    assert rc != 0
    assert record_messages.find("unstaged changes")
    _, sha1 = git.call("rev-parse", "HEAD", raises=False)
    expected = git.get_ref_sha1("refs/tags/v0.1")
    # git repo unchanged
    assert sha1 == expected
    git.call("reset", "--hard")
    qisrc_action("sync", retcode=True)
    # if modification is revert sync must be successful
    assert git.get_current_branch() == "master"

def check_state(project, untracked):
    """ Check and register the state of a project. """
    state_project = ProjectState(project)
    git = qisrc.git.Git(project.path)
    if not git.is_valid():
        state_project.valid = False
        return state_project
    state_project.clean = git.is_clean(untracked=untracked)
    if project.fixed_ref:
        state_project.ahead, state_project.behind = stat_fixed_ref(git, project.fixed_ref)
        state_project.fixed_ref = project.fixed_ref
        _set_status(git, state_project, untracked=untracked)
        return state_project
    state_project.current_branch = git.get_current_branch()
    state_project.tracking = git.get_tracking_branch()
    if project.default_remote and project.default_branch:
        state_project.manifest_branch = "%s/%s" % (project.default_remote.name, project.default_branch.name)
    if state_project.current_branch is None:
        state_project.not_on_a_branch = True
        return state_project
    if project.default_branch:
        if state_project.current_branch != project.default_branch.name:
            state_project.incorrect_proj = True
    (state_project.ahead, state_project.behind) = stat_tracking_remote(
        git,
        state_project.current_branch,
        state_project.tracking)
    if state_project.incorrect_proj:
        (state_project.ahead_manifest, state_project.behind_manifest) = stat_tracking_remote(
            git, state_project.current_branch, "%s/%s" % (
                project.default_remote.name, project.default_branch.name))
    _set_status(git, state_project, untracked=untracked)
    return state_project

def do(args):
    """ Main entry point """
    git_worktree = qisrc.parsers.get_git_worktree(args)
    git_project = qisrc.parsers.get_one_git_project(git_worktree, args)
    git = qisrc.git.Git(git_project.path)
    current_branch = git.get_current_branch()
    if not current_branch:
        ui.error("Not currently on any branch")
        sys.exit(2)
    if git_project.review:
        maintainers = qisrc.maintainers.get(git_project, warn_if_none=True)
        reviewers = [x['email'] for x in maintainers]
        reviewers.extend(args.reviewers or list())
        # Prefer gerrit logins or groups instead of e-mails
        reviewers = [x.split("@")[0] for x in reviewers]
        qisrc.review.push(git_project, current_branch,
                          bypass_review=(not args.review),
                          dry_run=args.dry_run, reviewers=reviewers,
                          topic=args.topic)
    else:
        if args.dry_run:
            git.push("-n")
        else:
            if args.review and not args.yes:
                mess = """\
The project is not under code review.
Are you sure you want to run `git push` ?
This action cannot be undone\
"""
                answer = qisys.interact.ask_yes_no(mess, default=False)
                if answer:
                    git.push()
            else:
                git.push()

    def test_wrong_branch_unstaged(self):
        bar_url = create_git_repo(self.tmp, "bar")
        work = os.path.join(self.tmp, "work")
        qibuild.sh.mkdir(work)
        bar_src = os.path.join(work, "bar")
        git = qisrc.git.Git(bar_src)
        git.clone(bar_url)

        # Checkout a 'next' branch with unstaged changes
        git.checkout("-b", "next")
        write_readme(bar_src, "bar on next\n")
        push_readme_v2(self.tmp, "bar", "master")

        err = git.update_branch("master", "origin")
        self.assertFalse(err)

        # Check we are still on next with our
        # unstaged changes back
        self.assertEqual(git.get_current_branch(), "next")
        readme = read_readme(bar_src)
        self.assertEqual(readme, "bar on next\n")

        # Check that master is up to date
        git.checkout("-f", "master")
        readme = read_readme(bar_src)
        self.assertEqual(readme, "bar v2 on master\n")

def check_state(project, untracked):
    """Check and register the state of a project."""
    state_project = ProjectState(project)

    git = qisrc.git.Git(project.path)

    if not git.is_valid():
        state_project.valid = False
        return state_project

    state_project.clean = git.is_clean(untracked = untracked)
    state_project.current_branch = git.get_current_branch()
    state_project.tracking = git.get_tracking_branch()

    #clean worktree, but is the current branch sync with the remote one?
    if state_project.clean:
        if state_project.current_branch is None:
            state_project.not_on_a_branch = True
            return state_project

        if state_project.current_branch != project.branch:
            state_project.incorrect_proj = True

        (state_project.ahead, state_project.behind) = stat_tracking_remote(git,
                state_project.current_branch, state_project.tracking)

    if not state_project.sync_and_clean:
        out = git.get_status(untracked)
        if out is not None:
            state_project.status = [ x[:3]
                    for x in out.splitlines() if len(x.strip()) > 0 ]

    return state_project

    def push_file(self, project, filename, contents,
                  branch="master", fast_forward=True,
                  message=None):
        """ Push a new file with the given contents to the given project
        It is assumed that the project has beed created

        """
        src = project.replace(".git", "")
        repo_src = self.src.join(src)
        git = qisrc.git.Git(repo_src.strpath)
        if git.get_current_branch() != branch:
            git.checkout("--force", "-B", branch)
        if not fast_forward:
            git.reset("--hard", "HEAD~1")
        to_write = repo_src.join(filename)
        if not message:
            if to_write.check(file=True):
                message = "Update %s" % filename
            else:
                message = "Add %s" % filename
        repo_src.ensure(filename, file=True)
        repo_src.join(filename).write(contents)
        git.add(filename)
        git.commit("--message", message)
        if fast_forward:
            git.push("origin", "%s:%s" % (branch, branch))
        else:
            git.push("origin", "--force", "%s:%s" % (branch, branch))

    def test_conflict(self):
        bar_url = create_git_repo(self.tmp, "bar")
        work = os.path.join(self.tmp, "work")
        qibuild.sh.mkdir(work)
        bar_src = os.path.join(work, "bar")
        git = qisrc.git.Git(bar_src)
        git.clone(bar_url)

        # Create conflicting changes
        write_readme(bar_src, "conflicting changes\n", append=True)
        git.commit("-a", "-m", "conflicting changes")
        push_readme_v2(self.tmp, "bar", "master")

        # Updating branch should fail
        err = git.update_branch("master", "origin")
        self.assertTrue(err)
        self.assertTrue("Merge conflict in README" in err, err)

        # But we should be back at our previous commit
        readme = read_readme(bar_src)
        self.assertEqual(readme, "bar\nconflicting changes\n")

        self.assertEqual(git.get_current_branch(), "master")
        rebase_apply = os.path.join(git.repo, ".git", "rebase_apply")
        self.assertFalse(os.path.exists(rebase_apply))

def test_use_correct_manifest_branch(git_worktree, git_server):
    git_server.switch_manifest_branch("devel")
    manifest_url = git_server.manifest_url
    worktree_syncer = qisrc.sync.WorkTreeSyncer(git_worktree)
    worktree_syncer.configure_manifest("default", manifest_url, branch="devel")
    local_manifest = git_worktree.tmpdir.join(".qi", "manifests", "default")
    git = qisrc.git.Git(local_manifest.strpath)
    assert git.get_current_branch() == "devel"

 def push_manifest(self, message, allow_empty=False):
     """ Push new manifest.xml version """
     manifest_repo = self.root.join("src", "manifest")
     git = qisrc.git.Git(manifest_repo.strpath)
     commit_args = ["--all", "--message", message]
     if allow_empty:
         commit_args.append("--allow-empty")
     git.commit(*commit_args)
     if git.get_current_branch() != self.manifest_branch:
         git.checkout("--force", "-B", self.manifest_branch)
     git.push("origin", "%s:%s" % (self.manifest_branch, self.manifest_branch))

def init_worktree(worktree, manifest_location, setup_review=True):
    """ (re)-intianlize a worktree given a manifest location.
    Clonie any missing repository, set the correct
    remote and tracking branch on every repository

    :param setup_review: Also set up the projects for review
    """
    errors = list()
    manifest = qisrc.manifest.load(manifest_location)
    if not manifest.projects:
        return
    project_count = len(manifest.projects)
    ui.info(ui.green, "Initializing worktree ...")
    setup_ok = True
    for i, project in enumerate(manifest.projects):
        ui.info(
            ui.green, "*", ui.reset, "(%2i/%2i)" % (i+1, project_count),
            ui.blue, project.name)
        # Use the same branch for the project as the branch
        # for the manifest, unless explicitely set:
        p_revision = project.revision
        p_url = project.fetch_url
        p_remote = project.remote
        p_src = project.path
        wt_project = worktree.get_project(p_src)
        if not wt_project:
            # Maybe we just need to add the project to the worktree:
            wt_project = add_if_missing(worktree, p_src, p_remote, p_url)
            if not wt_project:
                wt_project = clone_project(worktree, p_url, src=p_src,
                                           branch=p_revision,
                                           remote=p_remote)
        p_path = wt_project.path
        if project.review and setup_review and setup_ok:
            worktree.set_project_review(p_src, True)
            # If setup failed once, no point in trying for every project
            setup_ok = qisrc.review.setup_project(p_path, project.name,
                                                  project.review_url, p_revision)
        git = qisrc.git.Git(p_path)
        git.set_remote(p_remote, p_url)
        git.set_tracking_branch(p_revision, p_remote)
        cur_branch = git.get_current_branch()
        if cur_branch != p_revision:
            if not cur_branch:
                ui.warning("Project", project.name, "is on a detached HEAD",
                    "but should be on", p_revision)
            else:
                ui.warning("Project", project.name, "is on", cur_branch,
                    "but should be on", p_revision)
        worktree.set_git_project_config(p_src, p_remote, p_revision)
    if not setup_ok:
        qisrc.review.warn_gerrit()