fdroidserver/tests/deploy.TestCase

#!/usr/bin/env python3

import configparser
import inspect
import logging
import os
import shutil
import sys
import tempfile
from unicodedata import mirrored
import unittest
from pathlib import Path
from unittest import mock

import git

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()
        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')

    def test_upload_to_servergitmirror_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

        repo_section = 'repo'
        initial_branch = 'main'

        os.chdir(self.testdir)

        local_git_repo_path = Path(self.testdir)
        local_git_repo = git.Repo.init(
            local_git_repo_path, initial_branch=initial_branch
        )
        fdroid_dir = local_git_repo_path / 'fdroid'
        repo_dir = fdroid_dir / repo_section
        repo_dir.mkdir(parents=True)
        for filename in fdroidserver.deploy.INDEX_FILES:
            fake_file = repo_dir / filename
            with fake_file.open('w') as fp:
                fp.write('not a real one, but has the right filename')

        remote_git_repo_path = Path('git-mirror')
        remote_repo = git.Repo.init(remote_git_repo_path, initial_branch=initial_branch)

        mirror_config = {"url": remote_git_repo_path, "index_only": True}
        enabled_remotes = []
        ssh_cmd = 'ssh -oBatchMode=yes'
        fdroidserver.deploy.upload_to_servergitmirror(
            mirror_config=mirror_config,
            local_repo=local_git_repo,
            enabled_remotes=enabled_remotes,
            repo_section=repo_section,
            fdroid_dir=str(fdroid_dir),
            git_mirror_path=str(local_git_repo_path),
            ssh_cmd=ssh_cmd,
        )
        assert remote_repo.is_dirty


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)