def test_get_default_remote_version_label(self):
     url = self.remote_path
     client = GitClient(self.local_path)
     self.assertTrue(client.checkout(url))
     self.assertEqual(client.get_default_remote_version_label(), 'master')
     subprocess.check_call("git symbolic-ref HEAD refs/heads/test_branch", shell=True, cwd=self.remote_path)
     self.assertEqual(client.get_default_remote_version_label(), 'test_branch')

    def test_checkout_no_unnecessary_updates_other_branch(self):
        client = GitClient(self.local_path)
        client.fetches = 0
        client.submodules = 0
        client.fast_forwards = 0

        def ifetch(self):
            self.fetches += 1
            return True

        def iff(self, branch_parent, fetch=True, verbose=False):
            self.fast_forwards += 1
            return True

        def isubm(self, verbose=False, timeout=None):
            self.submodules += 1
            return True
        client._do_fetch = types.MethodType(ifetch, client)
        client._do_fast_forward = types.MethodType(iff, client)
        client._update_submodules = types.MethodType(isubm, client)
        url = self.remote_path
        self.assertFalse(client.path_exists())
        self.assertFalse(client.detect_presence())
        self.assertTrue(client.checkout(url, 'test_branch'))
        self.assertEqual(1, client.submodules)
        self.assertEqual(0, client.fetches)
        self.assertEqual(0, client.fast_forwards)

 def test_get_environment_metadata(self):
     # Verify that metadata is generated
     directory = tempfile.mkdtemp()
     self.directories['local'] = directory
     local_path = os.path.join(directory, "local")
     client = GitClient(local_path)
     self.assertTrue('version' in client.get_environment_metadata())

 def test_get_log_defaults(self):
     client = GitClient(self.local_path)
     log = client.get_log()
     self.assertEquals(self.n_commits + 1, len(log))
     self.assertEquals('local_%d' % (self.n_commits - 1), log[0]['message'])
     for key in ['id', 'author', 'email', 'date', 'message']:
         self.assertTrue(log[0][key] is not None, key)

    def setUpClass(self):
        GitClientTestSetups.setUpClass()

        client = GitClient(self.local_path)
        client.checkout(self.remote_path, self.readonly_version)

        self.basepath_export = os.path.join(self.root_directory, 'export')

    def testDiffRelpath(self):
        client = GitClient(self.local_path)
        self.assertTrue(client.path_exists())
        self.assertTrue(client.detect_presence())
        self.assertEquals(
            '''\
diff --git ros/added.txt ros/added.txt
new file mode 100644
index 0000000..454f6b3
--- /dev/null
+++ ros/added.txt
@@ -0,0 +1 @@
+0123456789abcdef
\\ No newline at end of file
diff --git ros/deleted-fs.txt ros/deleted-fs.txt
deleted file mode 100644
index e69de29..0000000
diff --git ros/deleted.txt ros/deleted.txt
deleted file mode 100644
index e69de29..0000000
diff --git ros/modified-fs.txt ros/modified-fs.txt
index e69de29..454f6b3 100644
--- ros/modified-fs.txt
+++ ros/modified-fs.txt
@@ -0,0 +1 @@
+0123456789abcdef
\\ No newline at end of file
diff --git ros/modified.txt ros/modified.txt
index e69de29..454f6b3 100644
--- ros/modified.txt
+++ ros/modified.txt
@@ -0,0 +1 @@
+0123456789abcdef
\\ No newline at end of file''',
            client.get_diff(basepath=os.path.dirname(self.local_path)).rstrip())

    def setUp(self):
        client = GitClient(self.local_path)
        client.checkout(self.remote_path)
        # Create some local untracking branch
        subprocess.check_call("git checkout test_tag -b localbranch", shell=True, cwd=self.local_path)
        subprocess.check_call("touch local.txt", shell=True, cwd=self.local_path)
        subprocess.check_call("git add *", shell=True, cwd=self.local_path)
        subprocess.check_call("git commit -m my_branch", shell=True, cwd=self.local_path)
        subprocess.check_call("git tag my_branch_tag", shell=True, cwd=self.local_path)
        po = subprocess.Popen("git log -n 1 --pretty=format:\"%H\"", shell=True, cwd=self.local_path, stdout=subprocess.PIPE)
        self.untracked_version = po.stdout.read().decode('UTF-8').rstrip('"').lstrip('"')

        # Go detached to create some dangling commits
        subprocess.check_call("git checkout test_tag", shell=True, cwd=self.local_path)
        # create a commit only referenced by tag
        subprocess.check_call("touch tagged.txt", shell=True, cwd=self.local_path)
        subprocess.check_call("git add *", shell=True, cwd=self.local_path)
        subprocess.check_call("git commit -m no_branch", shell=True, cwd=self.local_path)
        subprocess.check_call("git tag no_br_tag", shell=True, cwd=self.local_path)
        # create a dangling commit
        subprocess.check_call("touch dangling.txt", shell=True, cwd=self.local_path)
        subprocess.check_call("git add *", shell=True, cwd=self.local_path)
        subprocess.check_call("git commit -m dangling", shell=True, cwd=self.local_path)

        po = subprocess.Popen("git log -n 1 --pretty=format:\"%H\"", shell=True, cwd=self.local_path, stdout=subprocess.PIPE)
        self.dangling_version = po.stdout.read().decode('UTF-8').rstrip('"').lstrip('"')

        # go back to master to make head point somewhere else
        subprocess.check_call("git checkout master", shell=True, cwd=self.local_path)

 def test_checkout_dir_exists(self):
     url = self.remote_path
     client = GitClient(self.local_path)
     self.assertFalse(client.path_exists())
     os.makedirs(self.local_path)
     self.assertTrue(client.checkout(url))
     # non-empty
     self.assertFalse(client.checkout(url))

    def test_get_affected_files(self):
        client = GitClient(self.local_path)
        affected = client.get_affected_files(client.get_log()[0]['id'])

        self.assertEqual(sorted(['local_file']),
                         sorted(affected))

        self.assertEquals(['local_file'], affected)

    def test_fast_forward_simple_ref(self):
        url = self.remote_path
        client = GitClient(self.local_path)
        self.assertTrue(client.checkout(url, "master"))
        subprocess.check_call("git reset --hard test_tag", shell=True, cwd=self.local_path)
        # replace "refs/head/master" with just "master"
        subprocess.check_call("git config --replace-all branch.master.merge master", shell=True, cwd=self.local_path)

        self.assertTrue(client._get_branch_parent() is not (None, None))

    def setUp(self):
        client = GitClient(self.local_path)
        client.checkout(self.remote_path)
        # Create some local untracking branch

        subprocess.check_call("git checkout test_tag -b localbranch", shell=True, cwd=self.local_path)
        subprocess.check_call("touch local_file", shell=True, cwd=self.local_path)
        subprocess.check_call("git add local_file", shell=True, cwd=self.local_path)
        subprocess.check_call("git commit -m \"local_file\"", shell=True, cwd=self.local_path)

    def testExportRepository(self):
        client = GitClient(self.local_path)
        self.assertTrue(
          client.export_repository(self.readonly_version, self.basepath_export)
        )

        self.assertTrue(os.path.exists(self.basepath_export + '.tar.gz'))
        self.assertFalse(os.path.exists(self.basepath_export + '.tar'))
        self.assertFalse(os.path.exists(self.basepath_export))

 def test_fast_forward_diverged(self):
     url = self.remote_path
     client = GitClient(self.local_path)
     self.assertTrue(client.checkout(url, "master"))
     subprocess.check_call("git reset --hard test_tag", shell=True, cwd=self.local_path)
     subprocess.check_call("touch diverged.txt", shell=True, cwd=self.local_path)
     subprocess.check_call("git add *", shell=True, cwd=self.local_path)
     subprocess.check_call("git commit -m diverge", shell=True, cwd=self.local_path)
     # fail because we have diverged
     self.assertFalse(client.update('master'))

 def test_get_current_version_label(self):
     client = GitClient(path=self.local_path)
     # with detached local status
     client.checkout(url=self.remote_path, version='test_tag')
     self.assertEqual(client.get_current_version_label(), '<detached>')
     # when difference between local and tracking branch
     client.update(version='master')
     self.assertEqual(client.get_current_version_label(), 'master')
     # with other tracking branch
     cmd = 'git config --replace-all branch.master.merge test_branch'
     subprocess.check_call(cmd, shell=True, cwd=self.local_path)
     self.assertEqual(client.get_current_version_label(),
                      'master < test_branch')
     # with other remote
     for cmd in [
             'git remote add remote2 %s' % self.remote_path,
             'git config --replace-all branch.master.remote remote2',
             'git fetch remote2']:
         subprocess.check_call(cmd, shell=True, cwd=self.local_path)
     self.assertEqual(client.get_current_version_label(),
                      'master < remote2/test_branch')
     # return remote back to original config
     for cmd in [
          'git config --replace-all branch.master.remote origin',
          'git config --replace-all branch.master.merge refs/heads/master']:
         subprocess.check_call(cmd, shell=True, cwd=self.local_path)

 def setUp(self):
     client = GitClient(self.local_path)
     client.checkout(self.remote_path)
     subprocess.check_call("git checkout test_tag", shell=True, cwd=self.local_path)
     subprocess.check_call("echo 0 >> count.txt", shell=True, cwd=self.local_path)
     subprocess.check_call("git add count.txt", shell=True, cwd=self.local_path)
     subprocess.check_call("git commit -m modified-0", shell=True, cwd=self.local_path)
     # produce many tags to make git log command fail if all are added
     for count in range(4000):
         subprocess.check_call("git tag modified-%s" % count, shell=True, cwd=self.local_path)
     po = subprocess.Popen("git log -n 1 --pretty=format:\"%H\"", shell=True, cwd=self.local_path, stdout=subprocess.PIPE)
     self.last_version = po.stdout.read().decode('UTF-8').rstrip('"').lstrip('"')

 def test_checkout(self):
     url = self.remote_path
     client = GitClient(self.local_path)
     self.assertFalse(client.path_exists())
     self.assertFalse(client.detect_presence())
     self.assertTrue(client.checkout(url))
     self.assertTrue(client.path_exists())
     self.assertTrue(client.detect_presence())
     self.assertEqual(client.get_path(), self.local_path)
     self.assertEqual(client.get_url(), url)
     self.assertEqual(client._get_branch(), "master")
     self.assertEqual(client._get_branch_parent(), ("master", "origin"))

    def testStatus(self):
        client = GitClient(self.local_path)
        self.assertTrue(client.path_exists())
        self.assertTrue(client.detect_presence())
        self.assertEquals(
            '''\
A  ./added.txt
 D ./deleted-fs.txt
D  ./deleted.txt
 M ./modified-fs.txt
M  ./modified.txt
''',
            client.get_status(porcelain=True))

    def testStatusRelPath(self):
        client = GitClient(self.local_path)
        self.assertTrue(client.path_exists())
        self.assertTrue(client.detect_presence())
        self.assertEquals(
            '''\
A  ros/added.txt
 D ros/deleted-fs.txt
D  ros/deleted.txt
 M ros/modified-fs.txt
M  ros/modified.txt
''',
            client.get_status(basepath=os.path.dirname(self.local_path), porcelain=True))

 def test_checkout_timeout(self):
     ## SSH'ing to a mute server will hang for a very long time
     url = 'ssh://[email protected]:{0}/test'.format(self.mute_port)
     client = GitClient(self.local_path)
     start = time.time()
     self.assertFalse(client.checkout(url, timeout=2.0))
     stop = time.time()
     self.assertTrue(stop - start > 1.9)
     self.assertTrue(stop - start < 3.0)
     # the git processes will clean up the checkout dir, we have to wait
     # for them to finish in order to avoid a race condition with rmtree()
     while os.path.exists(self.local_path):
         time.sleep(0.2)

 def test_checkout_shallow(self):
     url = 'file://' + self.remote_path
     client = GitClient(self.local_path)
     self.assertFalse(client.path_exists())
     self.assertFalse(client.detect_presence())
     self.assertTrue(client.checkout(url, shallow=True))
     self.assertTrue(client.path_exists())
     self.assertTrue(client.detect_presence())
     self.assertEqual(client.get_path(), self.local_path)
     self.assertEqual(client.get_url(), url)
     self.assertEqual(client.get_branch(), "master")
     self.assertEqual(client.get_branch_parent(), "master")
     po = subprocess.Popen("git log --pretty=format:%H", shell=True, cwd=self.local_path, stdout=subprocess.PIPE)
     log = po.stdout.read().decode('UTF-8').splitlines()
     # shallow only contains last 2 commits
     self.assertEqual(2, len(log), log)