feat: Add /health/version endpoint to verify deployed commit

Author: cfsmp3

mod_health/controllers.py

@@ -1,7 +1,9 @@
 """Health check endpoints for deployment verification and monitoring."""
 
+import os
+import subprocess
 from datetime import datetime
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, Optional, Tuple
 
 from flask import Blueprint, current_app, jsonify
 
@@ -112,3 +114,70 @@ def readiness_check() -> Tuple[Any, int]:
     :rtype: Tuple[Any, int]
     """
     return health_check()
+
+
+def get_git_info() -> Dict[str, Optional[str]]:
+    """
+    Get git repository information for the deployed version.
+
+    :return: Dictionary with git commit hash, short hash, and branch
+    :rtype: Dict[str, Optional[str]]
+    """
+    result: Dict[str, Optional[str]] = {
+        'commit': None,
+        'short': None,
+        'branch': None,
+    }
+
+    try:
+        # Get the installation folder from config, fallback to current directory
+        install_folder = current_app.config.get('INSTALL_FOLDER', os.getcwd())
+
+        # Get full commit hash
+        commit = subprocess.check_output(
+            ['git', 'rev-parse', 'HEAD'],
+            cwd=install_folder,
+            stderr=subprocess.DEVNULL
+        ).decode().strip()
+        result['commit'] = commit
+        result['short'] = commit[:7]
+
+        # Get current branch
+        branch = subprocess.check_output(
+            ['git', 'rev-parse', '--abbrev-ref', 'HEAD'],
+            cwd=install_folder,
+            stderr=subprocess.DEVNULL
+        ).decode().strip()
+        result['branch'] = branch
+
+    except (subprocess.CalledProcessError, FileNotFoundError, OSError):
+        # Git not available or not a git repository
+        current_app.logger.warning('Could not retrieve git version information')
+
+    return result
+
+
+@mod_health.route('/health/version')
+def version_check() -> Tuple[Any, int]:
+    """
+    Version endpoint to verify deployed commit.
+
+    Returns the current git commit hash, useful for verifying
+    that a deployment has completed successfully.
+
+    :return: JSON response with version information
+    :rtype: Tuple[Any, int]
+    """
+    git_info = get_git_info()
+
+    response = {
+        'timestamp': datetime.utcnow().isoformat() + 'Z',
+        'git': git_info,
+    }
+
+    # Return 200 if we have version info, 503 if we couldn't get it
+    if git_info['commit']:
+        return jsonify(response), 200
+    else:
+        response['error'] = 'Could not retrieve version information'
+        return jsonify(response), 503

tests/test_health/test_controllers.py

@@ -130,3 +130,97 @@ def test_check_config_missing_keys(self):
             result = check_config()
             self.assertEqual(result['status'], 'error')
             self.assertIn('GITHUB_TOKEN', result['message'])
+
+
+class TestVersionEndpoint(BaseTestCase):
+    """Test version endpoint."""
+
+    @mock.patch('mod_health.controllers.get_git_info')
+    def test_version_endpoint_returns_200_with_git_info(self, mock_git_info):
+        """Test that /health/version returns 200 when git info is available."""
+        mock_git_info.return_value = {
+            'commit': 'abc123def456789012345678901234567890abcd',
+            'short': 'abc123d',
+            'branch': 'master',
+        }
+
+        response = self.app.test_client().get('/health/version')
+        self.assertEqual(response.status_code, 200)
+
+        data = json.loads(response.data)
+        self.assertIn('timestamp', data)
+        self.assertIn('git', data)
+        self.assertEqual(data['git']['commit'], 'abc123def456789012345678901234567890abcd')
+        self.assertEqual(data['git']['short'], 'abc123d')
+        self.assertEqual(data['git']['branch'], 'master')
+
+    @mock.patch('mod_health.controllers.get_git_info')
+    def test_version_endpoint_returns_503_when_git_unavailable(self, mock_git_info):
+        """Test that /health/version returns 503 when git info unavailable."""
+        mock_git_info.return_value = {
+            'commit': None,
+            'short': None,
+            'branch': None,
+        }
+
+        response = self.app.test_client().get('/health/version')
+        self.assertEqual(response.status_code, 503)
+
+        data = json.loads(response.data)
+        self.assertIn('error', data)
+        self.assertIn('git', data)
+        self.assertIsNone(data['git']['commit'])
+
+
+class TestGetGitInfo(BaseTestCase):
+    """Test get_git_info function."""
+
+    @mock.patch('subprocess.check_output')
+    def test_get_git_info_success(self, mock_subprocess):
+        """Test get_git_info returns correct values when git is available."""
+        from mod_health.controllers import get_git_info
+
+        # Mock both git commands
+        mock_subprocess.side_effect = [
+            b'abc123def456789012345678901234567890abcd\n',  # git rev-parse HEAD
+            b'master\n',  # git rev-parse --abbrev-ref HEAD
+        ]
+
+        with self.app.app_context():
+            result = get_git_info()
+
+        self.assertEqual(result['commit'], 'abc123def456789012345678901234567890abcd')
+        self.assertEqual(result['short'], 'abc123d')
+        self.assertEqual(result['branch'], 'master')
+
+    @mock.patch('subprocess.check_output')
+    def test_get_git_info_git_not_available(self, mock_subprocess):
+        """Test get_git_info handles missing git gracefully."""
+        import subprocess
+
+        from mod_health.controllers import get_git_info
+
+        mock_subprocess.side_effect = FileNotFoundError('git not found')
+
+        with self.app.app_context():
+            result = get_git_info()
+
+        self.assertIsNone(result['commit'])
+        self.assertIsNone(result['short'])
+        self.assertIsNone(result['branch'])
+
+    @mock.patch('subprocess.check_output')
+    def test_get_git_info_not_a_repo(self, mock_subprocess):
+        """Test get_git_info handles non-repo directory gracefully."""
+        import subprocess
+
+        from mod_health.controllers import get_git_info
+
+        mock_subprocess.side_effect = subprocess.CalledProcessError(128, 'git')
+
+        with self.app.app_context():
+            result = get_git_info()
+
+        self.assertIsNone(result['commit'])
+        self.assertIsNone(result['short'])
+        self.assertIsNone(result['branch'])