SteamFramelyHomebrew--framely/tests/test_bootstrap.py

import hashlib
import importlib.util
import io
import json
import contextlib
import pathlib
import tarfile
import tempfile
import unittest
from types import SimpleNamespace
from unittest.mock import mock_open, patch

spec = importlib.util.spec_from_file_location('bootstrap', pathlib.Path(__file__).parents[1] / 'tools/bootstrap.py')
bootstrap = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bootstrap)


class BootstrapTests(unittest.TestCase):
    def test_explicit_install_and_update_skip_confirmation(self):
        with tempfile.TemporaryDirectory() as tmp:
            archive = pathlib.Path(tmp) / 'package.tar.gz'
            checksums = pathlib.Path(tmp) / 'SHA256SUMS'
            archive.touch()
            checksums.touch()
            with patch.object(bootstrap.os, 'uname', return_value=SimpleNamespace(machine='aarch64')), \
                 patch.object(bootstrap.os, 'geteuid', return_value=0), \
                 patch.object(bootstrap, 'confirm', side_effect=AssertionError('unexpected prompt')), \
                 patch.object(bootstrap, 'verify_archive'), \
                 patch.object(bootstrap, 'root_operation') as operation:
                for action in ('install', 'update'):
                    bootstrap.main([action, '--archive', str(archive), '--checksums', str(checksums)])
                self.assertEqual(operation.call_count, 2)

    def test_github_proxy_option_reaches_engine_downloads(self):
        with tempfile.TemporaryDirectory() as tmp:
            archive = pathlib.Path(tmp) / 'package.tar.gz'
            checksums = pathlib.Path(tmp) / 'SHA256SUMS'
            archive.touch()
            checksums.touch()
            with patch.dict(bootstrap.os.environ, {}, clear=True), \
                 patch.object(bootstrap.os, 'uname', return_value=SimpleNamespace(machine='aarch64')), \
                 patch.object(bootstrap.os, 'geteuid', return_value=0), \
                 patch.object(bootstrap, 'verify_archive'), \
                 patch.object(bootstrap, 'root_operation'):
                bootstrap.main(['install', '--archive', str(archive), '--checksums', str(checksums),
                                '--github-proxy', 'https://gh-proxy.com/'])
                self.assertEqual(bootstrap.configured_github_proxy(), 'https://gh-proxy.com')

    def test_explicit_uninstall_still_requires_confirmation(self):
        with patch.object(bootstrap.os, 'uname', return_value=SimpleNamespace(machine='aarch64')), \
             patch.object(bootstrap, 'confirm', return_value=False) as confirmation:
            with self.assertRaisesRegex(ValueError, 'Cancelled'):
                bootstrap.main(['uninstall'])
            confirmation.assert_called_once()

    def test_confirmation_distinguishes_no_terminal_from_decline(self):
        with patch('builtins.open', side_effect=OSError('no tty')):
            with self.assertRaisesRegex(ValueError, 'No interactive terminal.*--yes'):
                bootstrap.confirm('install')
        with patch('builtins.open', mock_open(read_data='')):
            with self.assertRaisesRegex(ValueError, 'No confirmation input.*--yes'):
                bootstrap.confirm('install')
        with patch('builtins.open', mock_open(read_data='n\n')):
            self.assertFalse(bootstrap.confirm('install'))
        with patch('builtins.open', mock_open(read_data='Y\n')):
            self.assertTrue(bootstrap.confirm('install'))

    def test_external_checksums_match_exact_name_and_reject_tampering(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = pathlib.Path(tmp)
            archive = root / 'framely-1-linux-arm64.tar.gz'
            sums = root / 'SHA256SUMS'
            archive.write_bytes(b'payload')
            entry = hashlib.sha256(b'payload').hexdigest() + '  ' + archive.name + '\n'
            sums.write_text(entry)
            bootstrap.verify_archive(archive, sums)
            archive.write_bytes(b'changed')
            with self.assertRaises(ValueError):
                bootstrap.verify_archive(archive, sums)
            sums.write_text(entry * 2)
            with self.assertRaises(ValueError):
                bootstrap.verify_archive(archive, sums)
            sums.write_text(entry.replace(archive.name, './' + archive.name))
            with self.assertRaises(ValueError):
                bootstrap.verify_archive(archive, sums)

    def archive(self, root, entries):
        archive = root / 'release.tar.gz'
        with tarfile.open(archive, 'w:gz') as out:
            for name, contents in entries:
                member = tarfile.TarInfo(name)
                if contents is None:
                    member.type = tarfile.SYMTYPE
                    member.linkname = '/etc/shadow'
                    out.addfile(member)
                else:
                    member.size = len(contents)
                    out.addfile(member, io.BytesIO(contents))
        return archive

    def test_safe_extraction_checks_structure_before_writing(self):
        entries = [('framely-1/' + name, b'1' if name == 'VERSION' else b'test') for name in ['VERSION', 'SHA256SUMS', 'install.sh', 'uninstall.sh', 'bin/framely']]
        with tempfile.TemporaryDirectory() as tmp:
            root = pathlib.Path(tmp)
            base = bootstrap.extract_archive(self.archive(root, entries), root / 'out')
            self.assertEqual(base.name, 'framely-1')
        for extra in [('framely-1/../../escape', b'bad'), ('framely-1/link', None), ('framely-1/VERSION', b'1'), ('framely-2/file', b'bad')]:
            with self.subTest(extra=extra), tempfile.TemporaryDirectory() as tmp:
                root = pathlib.Path(tmp)
                with self.assertRaises(ValueError):
                    bootstrap.extract_archive(self.archive(root, entries + [extra]), root / 'out')
                self.assertFalse((root / 'out').exists())

    def test_redirect_cannot_downgrade_https(self):
        with self.assertRaises(ValueError):
            bootstrap.HTTPSRedirect().redirect_request(None, None, 302, '', {}, 'http://example.org/package')

    def test_extraction_progress_reports_real_bytes_and_never_completes_invalid_release(self):
        entries = [('framely-1/' + name, b'1' if name == 'VERSION' else b'test') for name in ['VERSION', 'SHA256SUMS', 'install.sh', 'uninstall.sh', 'bin/framely']]
        with tempfile.TemporaryDirectory() as tmp:
            root = pathlib.Path(tmp)
            reports = []
            bootstrap.extract_archive(self.archive(root, entries), root / 'out', lambda *value: reports.append(value))
            total = sum(len(contents) for _, contents in entries)
            self.assertEqual(reports[0], ('extract', 0, total))
            self.assertEqual(reports[-1], ('extract', total, total))
        with tempfile.TemporaryDirectory() as tmp:
            root = pathlib.Path(tmp)
            reports = []
            with self.assertRaises(ValueError):
                bootstrap.extract_archive(self.archive(root, [('framely-1/VERSION', b'wrong')]), root / 'out', lambda *value: reports.append(value))
            self.assertFalse(any(done == total and total > 0 for _, done, total in reports))
        output = io.StringIO()
        with contextlib.redirect_stdout(output):
            bootstrap.emit_progress('configure')
        payload = json.loads(output.getvalue().removeprefix('FRAMELY_PROGRESS '))
        self.assertEqual(payload, {'step': 'configure'})


if __name__ == '__main__':
    unittest.main()