import unittest from pathlib import Path from typing import Any from unittest.mock import Mock, patch import boto3 PATH = Path(__file__).parent.resolve() from src.great_ai import LargeFileS3 credentials = { "aws_region_name": "your_region_like_eu-west-2", "aws_access_key_id": "YOUR_ACCESS_KEY_ID", "aws_secret_access_key": "YOUR_VERY_SECRET_ACCESS_KEY", "large_files_bucket_name": "create_a_bucket_and_put_its_name_here", "aws_endpoint_url": "this is optional, for backblaze, use this: https://s3.us-west-002.backblazeb2.com", } class TestLargeFileS3(unittest.TestCase): def test_uninitialized(self) -> None: self.assertRaises(ValueError, LargeFileS3, "test-file") def test_bad_file_modes(self) -> None: self.assertRaises(ValueError, LargeFileS3, "test-file", "w", version=3) self.assertRaises(ValueError, LargeFileS3, "test-file", "wb", version=3) self.assertRaises(ValueError, LargeFileS3, "test-file", "w+r") self.assertRaises(ValueError, LargeFileS3, "test-file", "test") @patch.object(boto3, "client") def test_initialized_with_dict(self, client: Any) -> None: s3 = Mock() s3.list_objects_v2 = Mock( return_value={ "Contents": [ { "Key": "test-file/0", "Size": 30, }, { "Key": "test-file/1", "Size": 300, }, { "Key": "test-file/2", "Size": 187, }, ] } ) boto3.client = Mock(return_value=s3) LargeFileS3.configure_credentials( aws_region_name=credentials["aws_region_name"], aws_access_key_id=credentials["aws_access_key_id"], aws_secret_access_key=credentials["aws_secret_access_key"], large_files_bucket_name=credentials["large_files_bucket_name"], aws_endpoint_url=credentials["aws_endpoint_url"], ) lf = LargeFileS3("test-file") boto3.client.assert_called_once_with( "s3", aws_access_key_id=credentials["aws_access_key_id"], aws_secret_access_key=credentials["aws_secret_access_key"], region_name=credentials["aws_region_name"], endpoint_url=credentials["aws_endpoint_url"], ) s3.list_objects_v2.assert_called_once_with( Bucket=credentials["large_files_bucket_name"], Prefix="test-file" ) self.assertEqual(lf._version, 2) self.assertEqual(lf._local_name, "test-file-2") @patch.object(boto3, "client") def test_initialized_with_file(self, client: Any) -> None: s3 = Mock() s3.list_objects_v2 = Mock( return_value={ "Contents": [ { "Key": "test-file/0", "Size": 30, }, { "Key": "test-file/1", "Size": 300, }, { "Key": "test-file/2", "Size": 187, }, ] } ) boto3.client = Mock(return_value=s3) LargeFileS3.configure_credentials_from_file(PATH / "../../example_secrets.ini") lf = LargeFileS3("test-file") boto3.client.assert_called_once_with( "s3", aws_access_key_id=credentials["aws_access_key_id"], aws_secret_access_key=credentials["aws_secret_access_key"], region_name=credentials["aws_region_name"], endpoint_url=credentials["aws_endpoint_url"], ) assert s3.list_objects_v2.called s3.list_objects_v2.assert_called_once_with( Bucket=credentials["large_files_bucket_name"], Prefix="test-file" ) self.assertEqual(lf._version, 2) self.assertEqual(lf._local_name, "test-file-2")