Add porcelain for 'status'.

NEWS

@@ -7,6 +7,10 @@
   * Support staging symbolic links in Repo.stage.
     (Robert Brown)
 
+ IMPROVEMENTS
+
+  * Add porcelain 'status'. (Ryan Faulkner)
+
 0.9.6	2014-04-23
 
  IMPROVEMENTS

dulwich/porcelain.py

@@ -34,6 +34,7 @@ Currently implemented:
  * rev-list
  * tag
  * update-server-info
+ * status
  * symbolic-ref
 
 These functions are meant to behave similarly to the git subcommands.
@@ -42,6 +43,7 @@ Differences in behaviour are considered bugs.
 
 __docformat__ = 'restructuredText'
 
+from collections import namedtuple
 import os
 import sys
 import time
@@ -53,6 +55,7 @@ from dulwich.errors import (
     UpdateRefsError,
     )
 from dulwich.objects import (
+    Blob,
     Tag,
     parse_timezone,
     )
@@ -61,6 +64,9 @@ from dulwich.patch import write_tree_diff
 from dulwich.repo import (BaseRepo, Repo)
 from dulwich.server import update_server_info as server_update_server_info
 
+# Module level tuple definition for status output
+GitStatus = namedtuple('GitStatus', 'staged unstaged untracked')
+
 
 def open_repo(path_or_repo):
     """Open an argument that can be a repository or a path for a repository."""
@@ -485,3 +491,68 @@ def pull(repo, remote_location, refs_path,
     indexfile = r.index_path()
     tree = r["HEAD"].tree
     index.build_index_from_tree(r.path, indexfile, r.object_store, tree)
+
+
+def status(repo):
+    """Returns staged, unstaged, and untracked changes relative to the HEAD.
+
+    :param repo: Path to repository
+    :return: GitStatus tuple,
+        staged -    list of staged paths (diff index/HEAD)
+        unstaged -  list of unstaged paths (diff index/working-tree)
+        untracked - list of untracked, un-ignored & non-.git paths
+    """
+    # 1. Get status of staged
+    tracked_changes = get_tree_changes(repo)
+    # 2. Get status of unstaged
+    unstaged_changes = list(get_unstaged_changes(repo))
+    # TODO - Status of untracked - add untracked changes, need gitignore.
+    untracked_changes = []
+    return GitStatus(tracked_changes, unstaged_changes, untracked_changes)
+
+
+def get_tree_changes(repo):
+    """Return add/delete/modify changes to tree by comparing index to HEAD.
+
+    :param repo: repo path or object
+    :return: dict with lists for each type of change
+    """
+    r = open_repo(repo)
+    index = r.open_index()
+
+    # Compares the Index to the HEAD & determines changes
+    # Iterate through the changes and report add/delete/modify
+    tracked_changes = {
+        'add': [],
+        'delete': [],
+        'modify': [],
+    }
+    for change in index.changes_from_tree(r.object_store, r['HEAD'].tree):
+        if not change[0][0]:
+            tracked_changes['add'].append(change[0][1])
+        elif not change[0][1]:
+            tracked_changes['delete'].append(change[0][0])
+        elif change[0][0] == change[0][1]:
+            tracked_changes['modify'].append(change[0][0])
+        else:
+            raise AssertionError('git mv ops not yet supported')
+    return tracked_changes
+
+
+def get_unstaged_changes(repo):
+    """Walk through the index check for differences against working tree.
+
+    :param repo: repo path or object
+    :param tracked_changes: list of paths already staged
+    :yields: paths not staged
+    """
+    r = open_repo(repo)
+    index = r.open_index()
+
+    # For each entry in the index check the sha1 & ensure not staged
+    for entry in index.iteritems():
+        fp = os.path.join(r.path, entry[0])
+        with open(fp, 'rb') as f:
+            sha1 = Blob.from_string(f.read()).id
+        if sha1 != entry[1][-2]:
+            yield entry[0]

dulwich/tests/test_porcelain.py

@@ -496,3 +496,121 @@ class PullTests(PorcelainTestCase):
         # Check the target repo for pushed changes
         r = Repo(target_path)
         self.assertEquals(r['HEAD'].id, self.repo['HEAD'].id)
+
+
+class StatusTests(PorcelainTestCase):
+
+    def test_status(self):
+        """Integration test for `status` functionality."""
+
+        # Commit a dummy file then modify it
+        fullpath = os.path.join(self.repo.path, 'foo')
+        with open(fullpath, 'w') as f:
+            f.write('origstuff')
+
+        porcelain.add(repo=self.repo.path, paths=['foo'])
+        porcelain.commit(repo=self.repo.path, message='test status',
+            author='', committer='')
+
+        # modify access and modify time of path
+        os.utime(fullpath, (0, 0))
+
+        with open(fullpath, 'w') as f:
+            f.write('stuff')
+
+        # Make a dummy file and stage it
+        filename_add = 'bar'
+        fullpath = os.path.join(self.repo.path, filename_add)
+        with open(fullpath, 'w') as f:
+            f.write('stuff')
+        porcelain.add(repo=self.repo.path, paths=filename_add)
+
+        results = porcelain.status(self.repo)
+
+        self.assertEquals(results.staged['add'][0], filename_add)
+        self.assertEquals(results.unstaged, ['foo'])
+
+    def test_get_unstaged_changes(self):
+        """Unit test for get_unstaged_changes."""
+
+        # Commit a dummy file then modify it
+        filename_commit = 'foo'
+        fullpath = os.path.join(self.repo.path, filename_commit)
+        with open(fullpath, 'w') as f:
+            f.write('origstuff')
+
+
+        porcelain.add(repo=self.repo.path, paths=[filename_commit])
+        porcelain.commit(repo=self.repo.path, message='test status',
+            author='', committer='')
+
+        with open(fullpath, 'w') as f:
+            f.write('newstuff')
+
+        # modify access and modify time of path
+        os.utime(fullpath, (0, 0))
+
+        changes = porcelain.get_unstaged_changes(self.repo.path)
+
+        self.assertEquals(list(changes), [filename_commit])
+
+    def test_get_tree_changes_add(self):
+        """Unit test for get_tree_changes add."""
+
+        # Make a dummy file, stage
+        filename = 'bar'
+        with open(os.path.join(self.repo.path, filename), 'w') as f:
+            f.write('stuff')
+        porcelain.add(repo=self.repo.path, paths=filename)
+        porcelain.commit(repo=self.repo.path, message='test status',
+            author='', committer='')
+
+        filename = 'foo'
+        with open(os.path.join(self.repo.path, filename), 'w') as f:
+            f.write('stuff')
+        porcelain.add(repo=self.repo.path, paths=filename)
+        changes = porcelain.get_tree_changes(self.repo.path)
+
+        self.assertEquals(changes['add'][0], filename)
+        self.assertEquals(len(changes['add']), 1)
+        self.assertEquals(len(changes['modify']), 0)
+        self.assertEquals(len(changes['delete']), 0)
+
+    def test_get_tree_changes_modify(self):
+        """Unit test for get_tree_changes modify."""
+
+        # Make a dummy file, stage, commit, modify
+        filename = 'foo'
+        fullpath = os.path.join(self.repo.path, filename)
+        with open(fullpath, 'w') as f:
+            f.write('stuff')
+        porcelain.add(repo=self.repo.path, paths=filename)
+        porcelain.commit(repo=self.repo.path, message='test status',
+            author='', committer='')
+        with open(fullpath, 'w') as f:
+            f.write('otherstuff')
+        porcelain.add(repo=self.repo.path, paths=filename)
+        changes = porcelain.get_tree_changes(self.repo.path)
+
+        self.assertEquals(changes['modify'][0], filename)
+        self.assertEquals(len(changes['add']), 0)
+        self.assertEquals(len(changes['modify']), 1)
+        self.assertEquals(len(changes['delete']), 0)
+
+    def test_get_tree_changes_delete(self):
+        """Unit test for get_tree_changes delete."""
+
+        # Make a dummy file, stage, commit, remove
+        filename = 'foo'
+        with open(os.path.join(self.repo.path, filename), 'w') as f:
+            f.write('stuff')
+        porcelain.add(repo=self.repo.path, paths=filename)
+        porcelain.commit(repo=self.repo.path, message='test status',
+            author='', committer='')
+        porcelain.rm(repo=self.repo.path, paths=[filename])
+        changes = porcelain.get_tree_changes(self.repo.path)
+
+        self.assertEquals(changes['delete'][0], filename)
+        self.assertEquals(len(changes['add']), 0)
+        self.assertEquals(len(changes['modify']), 0)
+        self.assertEquals(len(changes['delete']), 1)