Repository: spotify/luigi
File: test/contrib/s3_test.py (Python)

Code before:
        tmp_file = tempfile.NamedTemporaryFile(mode='wb', delete=True)
        tmp_file_path = tmp_file.name
        tmp_file.write(file_contents)
        tmp_file.flush()

        s3_client = S3Client(AWS_ACCESS_KEY, AWS_SECRET_KEY)

        for i in range(n):
            file_path = s3_dir + str(i)
            s3_client.put_multipart(tmp_file_path, file_path)
            self.assertTrue(s3_client.exists(file_path))

        s3_dest = 's3://mybucket/copydir_new/'
        response = s3_client.copy(s3_dir, s3_dest, threads=10, part_size=copy_part_size)

        self._run_copy_response_test(response, expected_num=n, expected_size=(n * file_size))

        for i in range(n):
            original_size = s3_client.get_key(s3_dir + str(i)).size
            copy_size = s3_client.get_key(s3_dest + str(i)).size
            self.assertEqual(original_size, copy_size)

    def test__path_to_bucket_and_key(self):
        self.assertEqual([('bucket', 'noquestionmarkhere'), ('bucket', 'questionmark?here')],
                         [S3Client._path_to_bucket_and_key('s3://bucket/noquestionmarkhere'),
                          S3Client._path_to_bucket_and_key('s3://bucket/questionmark?here')])

    @mock_s3
    def _run_copy_test(self, put_method, is_multipart=False):
        create_bucket()
        # Run the method to put the file into s3 into the first place
        expected_num, expected_size = put_method()

        # As all the multipart put methods use `self._run_multipart_test`
        # we can just use this key
        original = 's3://mybucket/putMe'
        copy = 's3://mybucket/putMe_copy'

        # Copy the file from old location to new
        s3_client = S3Client(AWS_ACCESS_KEY, AWS_SECRET_KEY)
        if is_multipart:
            # 5MB is minimum part size, use it here so we don't have to generate huge files to test
            # the multipart upload in moto
            part_size = (1024 ** 2) * 5
            response = s3_client.copy(original, copy, part_size=part_size, threads=4)
        else:
            response = s3_client.copy(original, copy, threads=4)

        self._run_copy_response_test(response, expected_num=expected_num, expected_size=expected_size)

        # We can't use etags to compare between multipart and normal keys,

Review comment (bug):
While this does what you want it to, i believe it's generally better to split unittest tests into very small tests or just having a couple quick/easy assertions. Such as

```python
def test__path_to_bucket_and_key(self):
    self.assertEqual(('bucket', 'dir'), S3Client._path_to_bucket_and_key('s3://bucket/dir))

def test__path_to_bucket_and_key_with_question_mark(self):
    self.assertEqual(('bucket', 'dir?arg'), S3Client._path_to_bucket_and_key('s3://bucket/dir?arg))
```

or even within 1 unittest test, with multiple assertions:

```python
def test__path_to_bucket_and_key(self):
    self.assertEqual(('bucket', 'dir'), S3Client._path_to_bucket_and_key('s3://bucket/dir))
    self.assertEqual(('bucket', 'dir?arg'), S3Client._path_to_bucket_and_key('s3://bucket/dir?arg))
```

The first example above would present easier identification during a test failure (bc the test named `test__path_to_bucket_and_key_with_question_mark` would fail and that's very clear what the problem is). The latter may be easier to read, but test failure wouldn't be as explicit.

Code after:
        tmp_file = tempfile.NamedTemporaryFile(mode='wb', delete=True)
        tmp_file_path = tmp_file.name
        tmp_file.write(file_contents)
        tmp_file.flush()

        s3_client = S3Client(AWS_ACCESS_KEY, AWS_SECRET_KEY)

        for i in range(n):
            file_path = s3_dir + str(i)
            s3_client.put_multipart(tmp_file_path, file_path)
            self.assertTrue(s3_client.exists(file_path))

        s3_dest = 's3://mybucket/copydir_new/'
        response = s3_client.copy(s3_dir, s3_dest, threads=10, part_size=copy_part_size)

        self._run_copy_response_test(response, expected_num=n, expected_size=(n * file_size))

        for i in range(n):
            original_size = s3_client.get_key(s3_dir + str(i)).size
            copy_size = s3_client.get_key(s3_dest + str(i)).size
            self.assertEqual(original_size, copy_size)

    def test__path_to_bucket_and_key(self):
        self.assertEqual(('bucket', 'key'), S3Client._path_to_bucket_and_key('s3://bucket/key'))

    def test__path_to_bucket_and_key_with_question_mark(self):
        self.assertEqual(('bucket', 'key?blade'), S3Client._path_to_bucket_and_key('s3://bucket/key?blade'))

    @mock_s3
    def _run_copy_test(self, put_method, is_multipart=False):
        create_bucket()
        # Run the method to put the file into s3 into the first place
        expected_num, expected_size = put_method()

        # As all the multipart put methods use `self._run_multipart_test`
        # we can just use this key
        original = 's3://mybucket/putMe'
        copy = 's3://mybucket/putMe_copy'

        # Copy the file from old location to new
        s3_client = S3Client(AWS_ACCESS_KEY, AWS_SECRET_KEY)
        if is_multipart:
            # 5MB is minimum part size, use it here so we don't have to generate huge files to test
            # the multipart upload in moto
            part_size = (1024 ** 2) * 5
            response = s3_client.copy(original, copy, part_size=part_size, threads=4)
        else:
            response = s3_client.copy(original, copy, threads=4)

        self._run_copy_response_test(response, expected_num=expected_num, expected_size=expected_size)