#!/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 from fdroidserver import index 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)