test(deploy): add test cases for update_awsbucket_libcloud and fix errors

fdroidserver/deploy.py

@@ -240,7 +240,10 @@ def update_awsbucket_libcloud(repo_section):
             objs[obj.name] = obj
 
     if options.index_only:
-        files_to_upload = _get_index_includes(repo_section)
+        index_files = [f"{os.getcwd()}/{name}" for name in _get_index_file_paths(repo_section)]
+        files_to_upload = [os.path.join(root, name) for root, dirs, files in os.walk(os.path.join(os.getcwd(), repo_section)) for name in files]
+        files_to_upload = list(set(files_to_upload) & set(index_files))
+
     else:
         files_to_upload = [os.path.join(root, name) for root, dirs, files in os.walk(os.path.join(os.getcwd(), repo_section)) for name in files]
 

tests/deploy.TestCase

@@ -888,6 +888,123 @@ class DeployTest(unittest.TestCase):
                 fdroidserver.deploy.update_awsbucket_s3cmd(repo_section)
         self.assertEqual(call_iteration, 2, 'expected 2 invocations of subprocess.call')
 
+    def test_update_awsbucket_libcloud(self):
+        from libcloud.storage.drivers.s3 import S3StorageDriver
+        from libcloud.storage.base import Container
+
+        # setup parameters for this test run
+        fdroidserver.deploy.options.no_checksum = True
+        fdroidserver.deploy.config["awsbucket"] = "bucket"
+        fdroidserver.deploy.config["awsaccesskeyid"] = "accesskeyid"
+        fdroidserver.deploy.config["awssecretkey"] = "secretkey"
+        fdroidserver.deploy.config["s3cmd"] = "s3cmd"
+        fdroidserver.deploy.options.verbose = False
+        fdroidserver.deploy.options.quiet = True
+        fdroidserver.deploy.options.index_only = False
+        repo_section = 'repo'
+
+        os.chdir(self.testdir)
+        repo = Path('repo')
+        repo.mkdir(parents=True)
+        fake_apk = repo / 'Sym.apk'
+        with fake_apk.open('w') as fp:
+            fp.write('not an APK, but has the right filename')
+        fake_index = repo / fdroidserver.deploy.INDEX_FILES[0]
+        with fake_index.open('w') as fp:
+            fp.write('not an index, but has the right filename')
+
+        with mock.patch(
+            'libcloud.storage.drivers.s3.S3StorageDriver'
+        ) as mock_driver_class:
+            mock_driver = mock_driver_class.return_value
+            mock_container = mock.MagicMock(spec=Container)
+            mock_container.list_objects.return_value = [
+                mock.MagicMock(name='Sym.apk'),
+                mock.MagicMock(name=fdroidserver.deploy.INDEX_FILES[0]),
+            ]
+
+            mock_driver.get_container.return_value = mock_container
+            mock_driver.upload_object_via_stream.return_value = None
+
+            fdroidserver.deploy.update_awsbucket_libcloud(repo_section)
+
+            mock_driver.get_container.assert_called_once_with(
+                container_name=fdroidserver.deploy.config["awsbucket"]
+            )
+            mock_container.list_objects.assert_called_once_with()
+            files_to_upload = ['fdroid/repo/Sym.apk', 'fdroid/repo/entry.jar']
+            calls = [
+                mock.call(
+                    iterator=mock.ANY,
+                    container=mock_container,
+                    object_name=file,
+                    extra={'acl': 'public-read'},
+                )
+                for file in files_to_upload
+            ]
+            mock_driver.upload_object_via_stream.assert_has_calls(calls, any_order=True)
+            assert mock_driver.upload_object_via_stream.call_count == 2
+
+    def test_update_awsbucket_libcloud_in_index_only_mode(self):
+        from libcloud.storage.drivers.s3 import S3StorageDriver
+        from libcloud.storage.base import Container
+
+        # setup parameters for this test run
+        fdroidserver.deploy.options.no_checksum = True
+        fdroidserver.deploy.config["awsbucket"] = "bucket"
+        fdroidserver.deploy.config["awsaccesskeyid"] = "accesskeyid"
+        fdroidserver.deploy.config["awssecretkey"] = "secretkey"
+        fdroidserver.deploy.config["s3cmd"] = "s3cmd"
+        fdroidserver.deploy.options.verbose = False
+        fdroidserver.deploy.options.quiet = True
+        fdroidserver.deploy.options.index_only = True
+        repo_section = 'repo'
+
+        os.chdir(self.testdir)
+        repo = Path('repo')
+        repo.mkdir(parents=True)
+        fake_apk = repo / 'Sym.apk'
+        with fake_apk.open('w') as fp:
+            fp.write('not an APK, but has the right filename')
+        fake_index = repo / fdroidserver.deploy.INDEX_FILES[0]
+        with fake_index.open('w') as fp:
+            fp.write('not an index, but has the right filename')
+
+        with mock.patch(
+            'libcloud.storage.drivers.s3.S3StorageDriver'
+        ) as mock_driver_class:
+            mock_driver = mock_driver_class.return_value
+            mock_container = mock.MagicMock(spec=Container)
+            mock_container.list_objects.return_value = [
+                mock.MagicMock(name='Sym.apk'),
+                mock.MagicMock(name=fdroidserver.deploy.INDEX_FILES[0]),
+            ]
+
+            mock_driver.get_container.return_value = mock_container
+            mock_driver.upload_object_via_stream.return_value = None
+
+            fdroidserver.deploy.update_awsbucket_libcloud(repo_section)
+
+            mock_driver.get_container.assert_called_once_with(
+                container_name=fdroidserver.deploy.config["awsbucket"]
+            )
+            mock_container.list_objects.assert_called_once_with()
+            files_to_upload = ['fdroid/repo/entry.jar']
+            calls = [
+                mock.call(
+                    iterator=mock.ANY,
+                    container=mock_container,
+                    object_name=file,
+                    extra={'acl': 'public-read'},
+                )
+                for file in files_to_upload
+            ]
+            mock_driver.upload_object_via_stream.assert_has_calls(
+                calls,
+                any_order=False,
+            )
+            assert mock_driver.upload_object_via_stream.call_count == 1
+
 
 if __name__ == "__main__":
     os.chdir(os.path.dirname(__file__))