import os import sys import unittest from unittest.mock import patch, MagicMock # Add the project root to the Python path project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if project_root not in sys.path: sys.path.insert(0, project_root) from diagnosis.dataloader.s3loader import S3Loader class TestS3Loader(unittest.TestCase): @patch('boto3.resource') def test_get_file_url_and_headers(self, mock_boto3): # Setup mock S3 resource and objects mock_s3 = MagicMock() mock_boto3.return_value = mock_s3 # Setup mock response mock_response = { 'Body': MagicMock(read=MagicMock(return_value=b'test content')) } # Configure the mock object to return our mock response mock_s3.Object.return_value.get.return_value = mock_response # Create S3Loader instance bucket_name = 'test-bucket' s3_loader = S3Loader(bucket_name) # Call the method we want to test test_key = 'test/file.txt' content = s3_loader.get_file(test_key) # Verify the content is as expected self.assertEqual(content, b'test content') # Get the arguments passed to the get() method mock_s3.Object.assert_called_once_with(bucket_name, test_key) mock_object = mock_s3.Object.return_value mock_object.get.assert_called_once() # Get the actual request parameters get_call_args = mock_object.get.call_args[1] # Print detailed request information print("\n" + "="*80) print("S3 GET REQUEST DETAILS") print("="*80) # Get the endpoint URL (this will show the actual endpoint used in the code) if os.environ.get("AWS_ENDPOINT_S3"): endpoint_url = os.environ["AWS_ENDPOINT_S3"] print(f"Using custom endpoint from AWS_ENDPOINT_S3: {endpoint_url}") else: endpoint_url = 'https://s3.amazonaws.com' # Default AWS S3 endpoint print(f"Using default AWS S3 endpoint: {endpoint_url}") print(f"\nBucket: {bucket_name}") print(f"Key: {test_key}") print(f"Full URL: {endpoint_url}/{bucket_name}/{test_key}") # Print the actual method call arguments print("\nMethod call arguments:") for k, v in get_call_args.items(): print(f" {k}: {v}") # Print the actual boto3 client configuration print("\nClient configuration:") client = mock_s3.meta.client print(f" Client class: {client.__class__.__name__}") print(f" Region name: {client.meta.region_name}") print("="*80 + "\n") if __name__ == '__main__': unittest.main()