请求协助编写触发AWS Glue作业的Lambda Python Mock单元测试用例

Got it, let's put together a solid mock unit test for your Lambda function that triggers an AWS Glue job. I'll walk you through the approach, dependencies, and full test code below.

Prerequisites #

First, install the necessary packages for mocking and testing:

pip install pytest-mock boto3-stubs[glue]

• pytest-mock simplifies mocking with pytest
• boto3-stubs provides type hints for boto3 clients to make testing cleaner (optional but helpful)

Full Test Code (Using Pytest) #

Assume your Lambda code is saved in lambda_function.py. Create a test file named test_lambda_function.py with the following code:

import pytest
import logging
from lambda_function import lambda_handler, glueJobName, client

def test_lambda_handler_triggers_glue_job(mocker, caplog):
    # Configure logging to capture output
    caplog.set_level(logging.INFO)
    
    # Mock the Glue client's start_job_run method to avoid real AWS calls
    mock_start_job_run = mocker.patch.object(client, 'start_job_run')
    mock_job_run_id = 'jr-1234567890abcdef'
    mock_start_job_run.return_value = {'JobRunId': mock_job_run_id}
    
    # Build a test event matching the structure your Lambda expects
    test_event = {
        'detail': {'event_source': 's3', 'object_key': 'test-file.csv'}
    }
    # Create a mock context object (Lambda passes this at runtime)
    test_context = mocker.Mock()
    
    # Execute the Lambda handler
    response = lambda_handler(test_event, test_context)
    
    # Verify the Glue job was triggered with the correct job name
    mock_start_job_run.assert_called_once_with(JobName=glueJobName)
    
    # Verify the response matches our mocked return value
    assert response == {'JobRunId': mock_job_run_id}
    
    # Check that expected logs were generated
    assert '## TRIGGERED BY EVENT: ' in caplog.text
    assert str(test_event['detail']) in caplog.text
    assert f'## STARTED GLUE JOB: {glueJobName}' in caplog.text
    assert f'## GLUE JOB RUN ID: {mock_job_run_id}' in caplog.text

Test Code Explanation #

Let's break down what each part does:

• Mocking the Glue Client: We use mocker.patch.object to replace the global client's start_job_run method. This ensures we don't make real API calls to AWS during testing.
• Test Event & Context: We create a dummy event with a detail field (since your Lambda accesses event['detail']) and a mock context object (required for Lambda handler execution).
• Validation:
  • Confirm start_job_run was called exactly once with the correct job name (MyTestJob).
  • Ensure the Lambda returns the mocked JobRunId as expected.
  • Verify all critical log messages are generated (using pytest's caplog fixture to capture logs).

Alternative: Using Python's Built-in Unittest #

If you prefer using the standard library instead of pytest, here's an equivalent test:

import unittest
import logging
from unittest.mock import patch, MagicMock
from lambda_function import lambda_handler, glueJobName, client

class TestGlueTriggerLambda(unittest.TestCase):
    def setUp(self):
        # Set up logging capture
        self.log_handler = logging.handlers.MemoryHandler(1024)
        logger = logging.getLogger('lambda_function')
        logger.addHandler(self.log_handler)
        logger.setLevel(logging.INFO)

    def tearDown(self):
        # Clean up logging handler
        logger = logging.getLogger('lambda_function')
        logger.removeHandler(self.log_handler)

    def test_lambda_handler(self):
        # Mock the start_job_run method
        mock_job_run_id = 'jr-9876543210fedcba'
        with patch.object(client, 'start_job_run') as mock_start:
            mock_start.return_value = {'JobRunId': mock_job_run_id}
            
            # Test event and context
            test_event = {'detail': {'trigger': 'manual_test'}}
            test_context = MagicMock()
            
            # Execute handler
            response = lambda_handler(test_event, test_context)
            
            # Validate function calls and response
            mock_start.assert_called_once_with(JobName=glueJobName)
            self.assertEqual(response, {'JobRunId': mock_job_run_id})
            
            # Validate logs
            log_records = [record.getMessage() for record in self.log_handler.buffer]
            self.assertIn('## TRIGGERED BY EVENT: ', log_records[0])
            self.assertIn(str(test_event['detail']), log_records[1])
            self.assertIn(f'## STARTED GLUE JOB: {glueJobName}', log_records[2])
            self.assertIn(f'## GLUE JOB RUN ID: {mock_job_run_id}', log_records[3])

if __name__ == '__main__':
    unittest.main()

Key Notes #

• Since your Lambda creates the Glue client globally (outside the handler), we mock the existing client instance instead of mocking boto3.client directly. If you moved the client creation inside lambda_handler, you'd mock boto3.client('glue') instead.
• Always validate both the side effects (like calling the Glue job) and the output (return value, logs) to ensure your Lambda behaves as expected.

内容的提问来源于stack exchange，提问作者diwakar g