#!/usr/bin/env python3 import configparser import inspect import logging import os import shutil import sys import tempfile import unittest from pathlib import Path from unittest import mock localmodule = os.path.realpath( os.path.join(os.path.dirname(inspect.getfile(inspect.currentframe())), '..') ) if localmodule not in sys.path: sys.path.insert(0, localmodule) import fdroidserver.common import fdroidserver.deploy from fdroidserver.exception import FDroidException from testcommon import TmpCwd, mkdtemp, parse_args_for_test class Options: quiet = False verbose = False class DeployTest(unittest.TestCase): '''fdroidserver/deploy.py''' def setUp(self): logging.basicConfig(level=logging.DEBUG) self.basedir = os.path.join(localmodule, 'tests') os.chdir(self.basedir) self._td = mkdtemp() self.testdir = self._td.name fdroidserver.deploy.options = mock.Mock() fdroidserver.deploy.config = {} fdroidserver.deploy.USER_RCLONE_CONF = False def tearDown(self): self._td.cleanup() def test_update_serverwebroots_bad_None(self): with self.assertRaises(TypeError): fdroidserver.deploy.update_serverwebroots(None, 'repo') def test_update_serverwebroots_bad_int(self): with self.assertRaises(TypeError): fdroidserver.deploy.update_serverwebroots(9, 'repo') def test_update_serverwebroots_bad_float(self): with self.assertRaises(TypeError): fdroidserver.deploy.update_serverwebroots(1.0, 'repo') def test_update_serverwebroots(self): """rsync works with file paths, so this test uses paths for the URLs""" os.chdir(self.testdir) repo = Path('repo') repo.mkdir() fake_apk = repo / 'fake.apk' with fake_apk.open('w') as fp: fp.write('not an APK, but has the right filename') url0 = Path('url0/fdroid') url0.mkdir(parents=True) url1 = Path('url1/fdroid') url1.mkdir(parents=True) # setup parameters for this test run fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.config['make_current_version_link'] = False dest_apk0 = url0 / fake_apk dest_apk1 = url1 / fake_apk self.assertFalse(dest_apk0.is_file()) self.assertFalse(dest_apk1.is_file()) fdroidserver.deploy.update_serverwebroots( [ {'url': str(url0)}, {'url': str(url1)}, ], str(repo), ) self.assertTrue(dest_apk0.is_file()) self.assertTrue(dest_apk1.is_file()) def test_update_serverwebroots_url_does_not_end_with_fdroid(self): with self.assertRaises(SystemExit): fdroidserver.deploy.update_serverwebroots([{'url': 'url'}], 'repo') def test_update_serverwebroots_bad_ssh_url(self): with self.assertRaises(SystemExit): fdroidserver.deploy.update_serverwebroots( [{'url': 'f@b.ar::/path/to/fdroid'}], 'repo' ) def test_update_serverwebroots_unsupported_ssh_url(self): with self.assertRaises(SystemExit): fdroidserver.deploy.update_serverwebroots([{'url': 'ssh://nope'}], 'repo') @unittest.skipUnless(shutil.which('rclone'), '/usr/bin/rclone') def test_update_remote_storage_with_rclone(self): os.chdir(self.testdir) repo = Path('repo') repo.mkdir(parents=True, exist_ok=True) fake_apk = repo / 'another_fake.apk' with fake_apk.open('w') as fp: fp.write('not an APK, but has the right filename') # write out rclone config for test use rclone_config = configparser.ConfigParser() rclone_config.add_section("test-local-config") rclone_config.set("test-local-config", "type", "local") rclone_config_path = Path('rclone_config_path') rclone_config_path.mkdir(parents=True, exist_ok=True) rclone_file = rclone_config_path / 'rclone.conf' with open(rclone_file, 'w') as configfile: rclone_config.write(configfile) # setup parameters for this test run fdroidserver.deploy.config['awsbucket'] = 'test_bucket_folder' fdroidserver.deploy.config['rclone'] = True fdroidserver.deploy.config['rclone_config'] = 'test-local-config' fdroidserver.deploy.config['path_to_custom_rclone_config'] = str(rclone_file) fdroidserver.deploy.options = Options # write out destination path destination = Path('some_bucket_folder/fdroid') destination.mkdir(parents=True, exist_ok=True) dest_path = Path(destination) / fake_apk self.assertFalse(dest_path.is_file()) repo_section = str(repo) # fdroidserver.deploy.USER_RCLONE_CONF = str(rclone_file) fdroidserver.deploy.update_remote_storage_with_rclone(repo_section) self.assertFalse(dest_path.is_file()) def test_update_serverwebroot(self): """rsync works with file paths, so this test uses paths for the URLs""" os.chdir(self.testdir) repo = Path('repo') repo.mkdir(parents=True) fake_apk = repo / 'fake.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') url = Path('url') url.mkdir() # setup parameters for this test run fdroidserver.deploy.options = mock.Mock() fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.config['make_current_version_link'] = False dest_apk = Path(url) / fake_apk dest_index = Path(url) / fake_index self.assertFalse(dest_apk.is_file()) self.assertFalse(dest_index.is_file()) fdroidserver.deploy.update_serverwebroot({'url': str(url)}, 'repo') self.assertTrue(dest_apk.is_file()) self.assertTrue(dest_index.is_file()) def test_update_serverwebroot_in_index_only_mode(self): os.chdir(self.testdir) repo = Path('repo') repo.mkdir(parents=True) fake_apk = repo / 'fake.apk' with fake_apk.open('w') as fp: fp.write('not an APK, but has the right filename') for i in fdroidserver.deploy.INDEX_FILES: fake_index = repo / i with fake_index.open('w') as fp: fp.write('not an index, but has the right filename') url = Path('url') url.mkdir() # setup parameters for this test run fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.config['make_current_version_link'] = False dest_apk = Path(url) / fake_apk dest_index = Path(url) / fake_index self.assertFalse(dest_apk.is_file()) self.assertFalse(dest_index.is_file()) fdroidserver.deploy.update_serverwebroot( {'url': str(url), 'index_only': True}, 'repo' ) self.assertFalse(dest_apk.is_file()) self.assertTrue(dest_index.is_file()) @mock.patch.dict(os.environ, clear=True) def test_update_serverwebroot_no_rsync_error(self): os.environ['PATH'] = self.testdir os.chdir(self.testdir) with self.assertRaises(FDroidException): fdroidserver.deploy.update_serverwebroot('serverwebroot', 'repo') def test_update_serverwebroot_make_cur_version_link(self): self.maxDiff = None # setup parameters for this test run fdroidserver.common.options = mock.Mock() fdroidserver.common.options.no_checksum = True fdroidserver.common.options.identity_file = None fdroidserver.common.options.verbose = False fdroidserver.common.options.quiet = True fdroidserver.deploy.options.index_only = False fdroidserver.deploy.config = {'make_current_version_link': True} url = "example.com:/var/www/fdroid" repo_section = 'repo' # setup function for asserting subprocess.call invocations call_iteration = 0 def update_server_webroot_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--quiet', '--exclude', 'repo/altstore-index.json', '--exclude', 'repo/entry.jar', '--exclude', 'repo/entry.json', '--exclude', 'repo/entry.json.asc', '--exclude', 'repo/index-v1.jar', '--exclude', 'repo/index-v1.json', '--exclude', 'repo/index-v1.json.asc', '--exclude', 'repo/index-v2.json', '--exclude', 'repo/index-v2.json.asc', '--exclude', 'repo/index.jar', '--exclude', 'repo/index.xml', 'repo', 'example.com:/var/www/fdroid', ], ) elif call_iteration == 1: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--quiet', 'repo', url, ], ) elif call_iteration == 2: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--quiet', 'Sym.apk', 'Sym.apk.asc', 'Sym.apk.sig', 'example.com:/var/www/fdroid', ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir): os.mkdir('repo') os.symlink('repo/com.example.sym.apk', 'Sym.apk') os.symlink('repo/com.example.sym.apk.asc', 'Sym.apk.asc') os.symlink('repo/com.example.sym.apk.sig', 'Sym.apk.sig') with mock.patch('subprocess.call', side_effect=update_server_webroot_call): fdroidserver.deploy.update_serverwebroot({'url': url}, repo_section) self.assertEqual(call_iteration, 3, 'expected 3 invocations of subprocess.call') def test_update_serverwebroot_make_cur_version_link_in_index_only_mode(self): # setup parameters for this test run fdroidserver.deploy.options = mock.Mock() fdroidserver.deploy.options.no_checksum = True fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.options.verbose = False fdroidserver.deploy.options.quiet = True fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.config['make_current_version_link'] = True url = "example.com:/var/www/fdroid" repo_section = 'repo' # setup function for asserting subprocess.call invocations call_iteration = 0 def update_server_webroot_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--quiet', 'repo/entry.jar', 'repo/entry.json', 'repo/entry.json.asc', 'repo/index-v1.jar', 'repo/index-v1.json', 'repo/index-v1.json.asc', 'repo/index-v2.json', 'repo/index-v2.json.asc', 'repo/index.jar', 'repo/index.xml', 'example.com:/var/www/fdroid/repo/', ], ) elif call_iteration == 1: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--quiet', 'repo', url, ], ) elif call_iteration == 2: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--quiet', 'Sym.apk', 'Sym.apk.asc', 'Sym.apk.sig', 'example.com:/var/www/fdroid', ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir): os.mkdir('repo') os.symlink('repo/com.example.sym.apk', 'Sym.apk') os.symlink('repo/com.example.sym.apk.asc', 'Sym.apk.asc') os.symlink('repo/com.example.sym.apk.sig', 'Sym.apk.sig') with mock.patch('subprocess.call', side_effect=update_server_webroot_call): fdroidserver.deploy.update_serverwebroot( {'url': url, 'index_only': True}, repo_section ) self.assertEqual(call_iteration, 1, 'expected 1 invocations of subprocess.call') def test_update_serverwebroot_with_id_file(self): # setup parameters for this test run fdroidserver.common.options = mock.Mock() fdroidserver.common.options.identity_file = None fdroidserver.common.options.no_checksum = True fdroidserver.common.options.verbose = True fdroidserver.common.options.quiet = False fdroidserver.common.options.identity_file = None fdroidserver.deploy.options.index_only = False fdroidserver.deploy.config = {'identity_file': './id_rsa'} url = "example.com:/var/www/fdroid" repo_section = 'archive' # setup function for asserting subprocess.call invocations call_iteration = 0 def update_server_webroot_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--verbose', '-e', 'ssh -oBatchMode=yes -oIdentitiesOnly=yes -i ' + fdroidserver.deploy.config['identity_file'], '--exclude', 'archive/altstore-index.json', '--exclude', 'archive/entry.jar', '--exclude', 'archive/entry.json', '--exclude', 'archive/entry.json.asc', '--exclude', 'archive/index-v1.jar', '--exclude', 'archive/index-v1.json', '--exclude', 'archive/index-v1.json.asc', '--exclude', 'archive/index-v2.json', '--exclude', 'archive/index-v2.json.asc', '--exclude', 'archive/index.jar', '--exclude', 'archive/index.xml', 'archive', url, ], ) elif call_iteration == 1: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--verbose', '-e', 'ssh -oBatchMode=yes -oIdentitiesOnly=yes -i ' + fdroidserver.deploy.config['identity_file'], 'archive', url, ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with mock.patch('subprocess.call', side_effect=update_server_webroot_call): fdroidserver.deploy.update_serverwebroot({'url': url}, repo_section) self.assertEqual(call_iteration, 2, 'expected 2 invocations of subprocess.call') def test_update_serverwebroot_with_id_file_in_index_only_mode(self): # setup parameters for this test run fdroidserver.deploy.options.no_chcksum = False fdroidserver.deploy.options.verbose = True fdroidserver.deploy.options.quiet = False fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.config['identity_file'] = './id_rsa' fdroidserver.deploy.config['make_current_version_link'] = False url = "example.com:/var/www/fdroid" repo_section = 'archive' # setup function for asserting subprocess.call invocations call_iteration = 0 def update_server_webroot_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--verbose', '-e', 'ssh -oBatchMode=yes -oIdentitiesOnly=yes -i ' + fdroidserver.deploy.config['identity_file'], 'archive/entry.jar', 'archive/entry.json', 'archive/entry.json.asc', 'archive/index-v1.jar', 'archive/index-v1.json', 'archive/index-v1.json.asc', 'archive/index-v2.json', 'archive/index-v2.json.asc', 'archive/index.jar', 'archive/index.xml', "example.com:/var/www/fdroid/archive/", ], ) elif call_iteration == 1: self.assertListEqual( cmd, [ 'rsync', '--archive', '--delete-after', '--safe-links', '--verbose', '-e', 'ssh -oBatchMode=yes -oIdentitiesOnly=yes -i ' + fdroidserver.deploy.config['identity_file'], "example.com:/var/www/fdroid/archive/", ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with mock.patch('subprocess.call', side_effect=update_server_webroot_call): fdroidserver.deploy.update_serverwebroot( {'url': url, 'index_only': True}, repo_section ) self.assertEqual(call_iteration, 1, 'expected 1 invocations of subprocess.call') @unittest.skipIf( not os.getenv('VIRUSTOTAL_API_KEY'), 'VIRUSTOTAL_API_KEY is not set' ) def test_upload_to_virustotal(self): fdroidserver.common.options.verbose = True virustotal_apikey = os.getenv('VIRUSTOTAL_API_KEY') fdroidserver.deploy.upload_to_virustotal('repo', virustotal_apikey) def test_remote_hostname_regex(self): for remote_url, name in ( ('git@github.com:guardianproject/fdroid-repo', 'github'), ('git@gitlab.com:guardianproject/fdroid-repo', 'gitlab'), ('https://github.com:guardianproject/fdroid-repo', 'github'), ('https://gitlab.com/guardianproject/fdroid-repo', 'gitlab'), ('https://salsa.debian.org/foo/repo', 'salsa'), ): self.assertEqual( name, fdroidserver.deploy.REMOTE_HOSTNAME_REGEX.sub(r'\1', remote_url) ) def test_update_awsbucket_s3cmd(self): # setup parameters for this test run fdroidserver.deploy.options = mock.Mock() fdroidserver.deploy.options.no_checksum = True fdroidserver.deploy.options.verbose = False fdroidserver.deploy.options.quiet = True config = {} fdroidserver.common.fill_config_defaults(config) fdroidserver.deploy.config = config fdroidserver.deploy.config["awsbucket"] = "bucket" fdroidserver.deploy.config["awsaccesskeyid"] = "accesskeyid" fdroidserver.deploy.config["awssecretkey"] = "secretkey" fdroidserver.deploy.config["s3cmd"] = "s3cmd" repo_section = 'repo' # setup function for asserting subprocess.call invocations call_iteration = 0 def update_awsbucket_s3cmd_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 's3cmd', f"--config={fdroidserver.deploy.AUTO_S3CFG}", 'info', f"s3://{fdroidserver.deploy.config['awsbucket']}", ], ) elif call_iteration == 1: self.assertListEqual( cmd, [ 's3cmd', f"--config={fdroidserver.deploy.AUTO_S3CFG}", 'sync', '--acl-public', '--quiet', '--exclude', 'repo/entry.jar', '--exclude', 'repo/entry.json', '--exclude', 'repo/entry.json.asc', '--exclude', 'repo/index-v1.jar', '--exclude', 'repo/index-v1.json', '--exclude', 'repo/index-v1.json.asc', '--exclude', 'repo/index-v2.json', '--exclude', 'repo/index-v2.json.asc', '--exclude', 'repo/index.jar', '--exclude', 'repo/index.xml', '--no-check-md5', '--skip-existing', repo_section, f"s3://{fdroidserver.deploy.config['awsbucket']}/fdroid/", ], ) elif call_iteration == 2: self.assertListEqual( cmd, [ 's3cmd', f"--config={fdroidserver.deploy.AUTO_S3CFG}", 'sync', '--acl-public', '--quiet', '--exclude', 'repo/entry.jar', '--exclude', 'repo/entry.json', '--exclude', 'repo/entry.json.asc', '--exclude', 'repo/index-v1.jar', '--exclude', 'repo/index-v1.json', '--exclude', 'repo/index-v1.json.asc', '--exclude', 'repo/index-v2.json', '--exclude', 'repo/index-v2.json.asc', '--exclude', 'repo/index.jar', '--exclude', 'repo/index.xml', '--no-check-md5', repo_section, f"s3://{fdroidserver.deploy.config['awsbucket']}/fdroid/", ], ) elif call_iteration == 3: self.assertListEqual( cmd, [ 's3cmd', f"--config={fdroidserver.deploy.AUTO_S3CFG}", 'sync', '--acl-public', '--quiet', '--delete-removed', '--delete-after', '--no-check-md5', repo_section, f"s3://{fdroidserver.deploy.config['awsbucket']}/fdroid/", ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir): os.mkdir('repo') os.symlink('repo/com.example.sym.apk', 'Sym.apk') os.symlink('repo/com.example.sym.apk.asc', 'Sym.apk.asc') os.symlink('repo/com.example.sym.apk.sig', 'Sym.apk.sig') with mock.patch('subprocess.call', side_effect=update_awsbucket_s3cmd_call): fdroidserver.deploy.update_awsbucket_s3cmd(repo_section) self.assertEqual(call_iteration, 4, 'expected 4 invocations of subprocess.call') def test_update_awsbucket_s3cmd_in_index_only_mode(self): # setup parameters for this test run fdroidserver.deploy.options = mock.Mock() fdroidserver.deploy.options.no_checksum = True fdroidserver.deploy.options.verbose = False fdroidserver.deploy.options.quiet = True config = {} fdroidserver.common.fill_config_defaults(config) fdroidserver.deploy.config = config fdroidserver.deploy.config["awsbucket"] = "bucket" fdroidserver.deploy.config["awsaccesskeyid"] = "accesskeyid" fdroidserver.deploy.config["awssecretkey"] = "secretkey" fdroidserver.deploy.config["s3cmd"] = "s3cmd" repo_section = 'repo' # setup function for asserting subprocess.call invocations call_iteration = 0 def update_awsbucket_s3cmd_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 's3cmd', f"--config={fdroidserver.deploy.AUTO_S3CFG}", 'info', f"s3://{fdroidserver.deploy.config['awsbucket']}", ], ) elif call_iteration == 1: self.assertListEqual( cmd, [ 's3cmd', f"--config={fdroidserver.deploy.AUTO_S3CFG}", 'sync', '--acl-public', '--quiet', '--include', 'repo/entry.jar', '--include', 'repo/entry.json', '--include', 'repo/entry.json.asc', '--include', 'repo/index-v1.jar', '--include', 'repo/index-v1.json', '--include', 'repo/index-v1.json.asc', '--include', 'repo/index-v2.json', '--include', 'repo/index-v2.json.asc', '--include', 'repo/index.jar', '--include', 'repo/index.xml', '--delete-removed', '--delete-after', '--no-check-md5', repo_section, f"s3://{fdroidserver.deploy.config['awsbucket']}/fdroid/", ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with tempfile.TemporaryDirectory() as tmpdir, TmpCwd(tmpdir): os.mkdir('repo') os.symlink('repo/com.example.sym.apk', 'Sym.apk') os.symlink('repo/com.example.sym.apk.asc', 'Sym.apk.asc') os.symlink('repo/com.example.sym.apk.sig', 'Sym.apk.sig') with mock.patch('subprocess.call', side_effect=update_awsbucket_s3cmd_call): fdroidserver.deploy.update_awsbucket_s3cmd( repo_section, index_only=True ) self.assertEqual(call_iteration, 2, 'expected 2 invocations of subprocess.call') def test_update_awsbucket_libcloud(self): from libcloud.storage.base import Container # setup parameters for this test run fdroidserver.deploy.options = mock.Mock() fdroidserver.deploy.options.no_checksum = True fdroidserver.deploy.options.verbose = False fdroidserver.deploy.options.quiet = True config = {} fdroidserver.common.fill_config_defaults(config) fdroidserver.deploy.config = config fdroidserver.deploy.config["awsbucket"] = "bucket" fdroidserver.deploy.config["awsaccesskeyid"] = "accesskeyid" fdroidserver.deploy.config["awssecretkey"] = "secretkey" fdroidserver.deploy.config["s3cmd"] = "s3cmd" 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.base import Container # setup parameters for this test run fdroidserver.deploy.options = mock.Mock() fdroidserver.deploy.options.no_checksum = True fdroidserver.deploy.options.verbose = False fdroidserver.deploy.options.quiet = True config = {} fdroidserver.common.fill_config_defaults(config) fdroidserver.deploy.config = config fdroidserver.deploy.config["awsbucket"] = "bucket" fdroidserver.deploy.config["awsaccesskeyid"] = "accesskeyid" fdroidserver.deploy.config["awssecretkey"] = "secretkey" fdroidserver.deploy.config["s3cmd"] = "s3cmd" 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, index_only=True) 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 def test_update_servergitmirrors(self): # setup parameters for this test run fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.options.no_keep_git_mirror_archive = False fdroidserver.deploy.options.verbose = False fdroidserver.deploy.options.quiet = True config = {} fdroidserver.common.fill_config_defaults(config) fdroidserver.deploy.config = config fdroidserver.deploy.config["servergitmirrors"] = [] repo_section = 'repo' # setup function for asserting subprocess.call invocations call_iteration = 0 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') def update_servergitmirrors_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 'rsync', '--recursive', '--safe-links', '--times', '--perms', '--one-file-system', '--delete', '--chmod=Da+rx,Fa-x,a+r,u+w', '--quiet', 'repo/', "git-mirror/fdroid/repo/", ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with mock.patch('subprocess.call', side_effect=update_servergitmirrors_call): fdroidserver.deploy.update_servergitmirrors([], repo_section) self.assertEqual(call_iteration, 1, 'expected 1 invocations of subprocess.call') def test_update_servergitmirrors_in_index_only_mode(self): # setup parameters for this test run fdroidserver.deploy.options.identity_file = None fdroidserver.deploy.options.no_keep_git_mirror_archive = False fdroidserver.deploy.options.verbose = False fdroidserver.deploy.options.quiet = True config = {} fdroidserver.common.fill_config_defaults(config) fdroidserver.deploy.config = config fdroidserver.deploy.config["servergitmirrors"] = [] repo_section = 'repo' # setup function for asserting subprocess.call invocations call_iteration = 0 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') def update_servergitmirrors_call(cmd): nonlocal call_iteration if call_iteration == 0: self.assertListEqual( cmd, [ 'rsync', '--recursive', '--safe-links', '--times', '--perms', '--one-file-system', '--delete', '--chmod=Da+rx,Fa-x,a+r,u+w', '--quiet', 'repo/', "git-mirror/fdroid/repo/", ], ) else: self.fail('unexpected subprocess.call invocation') call_iteration += 1 return 0 with mock.patch('subprocess.call', side_effect=update_servergitmirrors_call): fdroidserver.deploy.update_servergitmirrors([], repo_section) self.assertEqual(call_iteration, 1, 'expected 1 invocations of subprocess.call') if __name__ == "__main__": os.chdir(os.path.dirname(__file__)) import argparse parser = argparse.ArgumentParser() parser.add_argument( "-v", "--verbose", action="store_true", default=False, help="Spew out even more information than normal", ) parse_args_for_test(parser, sys.argv) newSuite = unittest.TestSuite() newSuite.addTest(unittest.makeSuite(DeployTest)) unittest.main(failfast=False)