Set the git revision in deploys if it's available locally (#64)

airbrake/notifier.py

@@ -90,6 +90,8 @@ def __init__(self, project_id=None, api_key=None, host=None, timeout=None,
         self.root_directory = config.get("root_directory")
         self.timeout = timeout or self.AIRBRAKE_TIMEOUT_DEFAULT
 
+        self.local_revision = utils.get_local_git_revision()
+
         self.whitelist_keys = config.get("whitelist_keys", [])
         self.blacklist_keys = config.get("blacklist_keys", [])
         self.filter_chain = [self.filter_whitelist, self.filter_blacklist]
@@ -329,6 +331,8 @@ def deploy(self, env=None, username=None, repository=None,
         environment and project. Any errors that still exist will get created
         again when they recur.
         """
+        if not revision and self.local_revision:
+            revision = self.local_revision
         payload = {"environment": env or self.environment,
                    "username": username,
                    "repository": repository,

airbrake/utils.py

@@ -6,9 +6,11 @@
     # Py2 legacy fix
     from Queue import Queue, Full, Empty
 
+import os
 import traceback
 import types
 import json
+import subprocess
 
 try:
     TypeType = types.TypeType
@@ -138,3 +140,54 @@ def non_empty_keys(data):
         elif val and val != 'None':
             non_empty[key] = val
     return non_empty
+
+
+def get_local_git_revision():
+    """Find the commit hash of the latest local commit."""
+    rev = _git_revision_with_binary()
+    if not rev:
+        rev = _git_revision_from_file()
+    return rev
+
+
+def _git_revision_with_binary():
+    """Get the latest git hash using the git binary."""
+    try:
+        rev = subprocess.check_output(["git", "rev-parse", "HEAD"])
+        return str(rev.strip())
+    except (OSError, subprocess.CalledProcessError):
+        return None
+
+
+def _git_revision_from_file():
+    """Get the latest git hash from file in .git/refs/heads/master."""
+    path = _get_git_path()
+    if os.path.exists(path):
+        return str(_get_git_ref_revision(path))
+
+
+def _get_git_ref_revision(path):
+    """Get the latest git hash from file."""
+    head_ref_path_file = os.path.join(path, "HEAD")
+    if not os.path.exists(head_ref_path_file):
+        return None
+    with open(head_ref_path_file, 'r') as ref_path_file:
+        ref = ref_path_file.read().strip()
+        if 'ref:' in ref:
+            ref_path = ref.partition('ref:')[-1].strip()
+            rev_file = os.path.join(path, ref_path)
+            if not os.path.exists(rev_file):
+                return None
+            with open(rev_file, 'r') as rev_file_handler:
+                return rev_file_handler.read().strip()
+        elif len(ref) == 40:
+            return ref
+        else:
+            return None
+
+
+def _get_git_path():
+    """Get the path to the local git repo."""
+    package_dir = os.path.dirname(__file__)
+    root_dir = os.path.normpath(os.path.join(package_dir, os.pardir))
+    return os.path.join(root_dir, '.git')

tests/test_notifier.py

@@ -265,6 +265,20 @@ def test_deploy_payload(self):
             )
             self.assertEqual(expected_call_args, requests_post.call_args)
 
+    def test_deploy_revision(self):
+        with mock.patch('requests.post') as requests_post:
+            ab = Airbrake(project_id=1234, api_key='fake', environment='test')
+            ab.deploy('test',
+                      'user1',
+                      'https://github.com/airbrake/airbrake',
+                      None,
+                      'v2.0')
+
+            data = json.loads(requests_post.call_args[1]['data'])
+            version = airbrake.utils.get_local_git_revision()
+
+            self.assertEqual(version, data['revision'])
+
     def check_timeout(self, timeout=None, expected_timeout=None):
         ab = Airbrake(project_id=1234,
                       api_key='fake',

tests/test_utils.py

@@ -1,4 +1,6 @@
+import os
 import unittest
+import subprocess
 
 import airbrake.utils
 
@@ -25,3 +27,27 @@ def test_non_empty_keys(self):
         clean_data = airbrake.utils.non_empty_keys(data)
 
         self.assertEqual(expected_data, clean_data)
+
+    def test_get_local_git_revision(self):
+        rev = airbrake.utils.get_local_git_revision()
+        self.assertIsNotNone(rev)
+
+        has_fs_access = None
+        head_ref_path_file = os.path.join(airbrake.utils._get_git_path(),
+                                          "HEAD")
+        with open(head_ref_path_file, 'r') as test_file:
+            ref_path = test_file.read().strip()
+            if ref_path:
+                has_fs_access = True
+
+        if has_fs_access:
+            rev_file = airbrake.utils._git_revision_from_file()
+            self.assertIsNotNone(rev_file)
+
+        rev = subprocess.check_output(["git", "rev-parse", "HEAD"])
+        if rev:
+            rev_binary = airbrake.utils._git_revision_with_binary()
+            self.assertIsNotNone(rev_binary)
+
+        if has_fs_access and rev:
+            self.assertTrue(rev_file, rev_binary)