saphid--frame-control/tests/test_apk_search.py

import http.client
import json
from pathlib import Path
import sys
import tempfile
import threading
import time
import types
import unittest
from unittest.mock import Mock, patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
from apk_sources import search, SourceError
import server

ENTRIES = json.loads((Path(__file__).parent / 'fixtures/apk-search/entries.json').read_text())


def fake(source_id='one', fn=None):
    return types.SimpleNamespace(KIND=source_id, sources=lambda: [dict(id=source_id, name=source_id,
                                 enabled=True, trust='official', builtin=True)],
                                 search=fn or (lambda s, q, limit=50: [e for e in ENTRIES if e['source'] == source_id]),
                                 details=lambda s, i: ENTRIES[0],
                                 download=Mock(return_value={'apk': '/fake.apk', 'obb': []}))


class SettingsTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        p = patch.object(search, 'settings_path', return_value=Path(self.tmp.name) / 'enabled.json')
        p.start()
        self.addCleanup(p.stop)
        for state in (search._running, search._pending, search._status, search._game_data):
            state.clear()


class SearchTests(SettingsTest):
    def test_group_does_not_merge_distinct_or_unknown_packages(self):
        result = search.group(ENTRIES)
        self.assertEqual(len(result), 3)
        self.assertEqual(sorted(len(a['offers']) for a in result), [1, 2, 2])
        same_name = dict(ENTRIES[0], package=None)
        self.assertEqual(len(search.group(ENTRIES[:1] + [same_name])), 2)

    def test_rank_exact_and_installable_and_verified(self):
        result = search.group(ENTRIES, 'Open Brush')
        self.assertEqual(result[0]['package'], 'org.brush')
        self.assertEqual(result[0]['offers'][0]['source'], 'one')
        self.assertEqual(result[1]['package'], 'org.other')
        self.assertEqual(len(search.group(ENTRIES, vr=False)), 1)
        self.assertEqual(len(search.group(ENTRIES, installable=True)), 1)

    def test_unknown_vr_counts_as_flat(self):
        entries = [dict(ENTRIES[3], id='a', name='Flat', vr=False), dict(ENTRIES[3], id='b', name='Unknown', vr=None),
                   dict(ENTRIES[3], id='c', name='Headset', vr=True)]
        self.assertEqual(sorted(a['name'] for a in search.group(entries, vr=False)), ['Flat', 'Unknown'])
        self.assertEqual([a['name'] for a in search.group(entries, vr=True)], ['Headset'])

    def test_browse_puts_unknown_fit_vr_first_and_blocked_last(self):
        entries = [dict(source='s', id='flat', name='Flat', package='a.flat', vr=False, min_sdk=21, abis=[]),
                   dict(source='s', id='vr', name='Headset', package='a.vr', vr=True),
                   dict(source='s', id='bad', name='Blocked', package='a.bad', vr=True, min_sdk=34, abis=[])]
        self.assertEqual([a['name'] for a in search.group(entries)], ['Headset', 'Flat', 'Blocked'])

    def test_fit_unknown_and_native_free_and_vr_hints(self):
        self.assertIsNone(search.fit({})['installable'])
        self.assertTrue(search.fit({'min_sdk': 23, 'abis': []})['installable'])
        self.assertFalse(search.fit({'min_sdk': 23, 'abis': ['x86']})['installable'])
        self.assertIn('Legacy VrApi', search.fit({'engine': 'VrApi'})['reasons'][0])

    def test_timeout_and_failure_leave_other_results(self):
        release = threading.Event()
        calls = []
        def slow(s, q, limit=50):
            calls.append(q)
            release.wait(2)
            return []
        mods = [fake(), fake('slow', slow), fake('broken', Mock(side_effect=SourceError('offline')))]
        try:
            with patch.object(search, 'modules', return_value=(mods, [])):
                started = time.monotonic()
                result = search.search(timeout=.03)
                self.assertLess(time.monotonic() - started, .3)
                self.assertTrue(result['apps'])
                self.assertEqual([s['status'] for s in result['sources']], ['ok', 'loading', 'error'])
                self.assertEqual(search.search('other', timeout=.03)['sources'][1]['status'], 'loading')
                search.search('newest', timeout=.03)
                self.assertEqual(calls, [''])
                queued = search._pending['slow']
                release.set()
                self.assertTrue(queued['event'].wait(2))
                self.assertEqual(queued['entries'], [])
                self.assertEqual(calls, ['', 'newest'])  # 'other' was superseded, never run
        finally:
            release.set()

    def test_query_arriving_as_a_search_finishes_is_not_stranded(self):
        mod = fake()
        source = mod.sources()[0]
        arrived = []

        class Event(threading.Event):
            def set(self):
                if not arrived:  # a request lands just as the first search completes
                    arrived.append(None)
                    t = threading.Thread(target=lambda: arrived.append(search._launch(mod, source, 'second', 50)))
                    t.start()
                    t.join(.3)  # blocks on search._lock if completion is published atomically
                super().set()
        with patch.object(search, 'threading', types.SimpleNamespace(Event=Event, Thread=threading.Thread)):
            search._launch(mod, source, 'first', 50)
            for _ in range(200):
                if len(arrived) == 2:
                    break
                time.sleep(.01)
            self.assertTrue(arrived[1]['event'].wait(2))
        self.assertEqual(arrived[1]['query'], ('second', 50))

    def test_set_enabled_does_not_hold_search_lock_in_source(self):
        free = []
        def set_enabled(source_id, enabled):
            t = threading.Thread(target=lambda: free.append(search._lock.acquire(timeout=1) and not search._lock.release()))
            t.start()
            t.join()
        mod = fake()
        mod.set_enabled = set_enabled
        with patch.object(search, 'modules', return_value=([mod], [])):
            search.set_enabled('one', False)
        self.assertEqual(free, [True])

    def test_stale_source_status(self):
        mod = fake()
        mod.stale = lambda source: True
        with patch.object(search, 'modules', return_value=([mod], [])):
            status = search.search(timeout=1)['sources'][0]
        self.assertEqual((status['status'], status['stale']), ('ok', True))

    def test_limited_source_status(self):
        from apk_sources import SourceLimited
        mods = [fake('busy', Mock(side_effect=SourceLimited('busy is limiting requests', 60)))]
        with patch.object(search, 'modules', return_value=(mods, [])):
            status = search.search(timeout=1)['sources'][0]
        self.assertEqual((status['status'], status['error']), ('limited', 'busy is limiting requests'))

    def test_disable_persists_and_prevents_queries_and_installs(self):
        mod = fake()
        with patch.object(search, 'modules', return_value=([mod], [])):
            search.set_enabled('one', False)
            self.assertFalse(search.sources()[0]['enabled'])
            self.assertEqual(search.search()['apps'], [])
            with self.assertRaisesRegex(SourceError, 'disabled'):
                search.install('one', 'brush')

    def test_install_passes_metadata_artwork_and_obb(self):
        mod = fake()
        mod.download.return_value.update(obb=['main.obb'], artwork={'hero': '/hero.png'}, icon_png=b'png')
        def install(apk, name=None, icon_png=None, source=None, artwork=None):
            self.assertEqual((apk, name, icon_png, source, artwork),
                             ('/fake.apk', 'Open Brush', b'png', 'one', {'hero': '/hero.png'}))
            return {'package': 'org.brush'}
        with patch.object(search, 'modules', return_value=([mod], [])), \
             patch.object(server.frame_android, 'install_obb', create=True) as obb:
            # An actual function exposes the future signature for inspection.
            with patch.object(server.frame_android, 'install', install):
                result = search.install('one', 'brush', 1)
            # The app's instance isn't running right after install, so game data is a follow-up step.
            obb.assert_not_called()
            self.assertTrue(result['game_data'])
            self.assertIn('Add game data', result['message'])
            obb.return_value = {'package': 'org.brush', 'obb': []}
            self.assertEqual(search.add_game_data('org.brush')['message'], 'Game data added')
            obb.assert_called_once_with('org.brush', ['main.obb'])
            with self.assertRaisesRegex(SourceError, 'install it again'):
                search.add_game_data('org.brush')
            mod.download.assert_called_once_with(mod.sources()[0] | {'status': 'not searched'}, 'brush', version_code=1)

    def test_discovery_and_demo_are_opt_in(self):
        module = fake()
        with patch.object(search.pkgutil, 'iter_modules', return_value=[types.SimpleNamespace(name='example')]), \
             patch.object(search.importlib, 'import_module', return_value=module), \
             patch.dict(search.os.environ, {'FRAME_APK_SEARCH_DEMO': '0'}):
            self.assertEqual(search.modules(), ([module], []))
        with patch.object(search.pkgutil, 'iter_modules', return_value=[]), \
             patch.dict(search.os.environ, {'FRAME_APK_SEARCH_DEMO': '0'}):
            self.assertEqual(search.modules(), ([], []))

    def test_newest_compatible_then_official_offer(self):
        first = dict(ENTRIES[0], verified=False, trust='community')
        newer = dict(first, source='new', version_code=2, updated='2026-01-01')
        official = dict(newer, source='official', trust='official')
        incompatible = dict(newer, source='blocked', verified=True, min_sdk=40)
        offers = search.group([first, incompatible, newer, official])[0]['offers']
        self.assertEqual([e['source'] for e in offers], ['official', 'new', 'one', 'blocked'])

    def test_missing_obb_support_stops_before_install(self):
        mod = fake()
        mod.download.return_value['obb'] = ['main.obb']
        with patch.object(search, 'modules', return_value=([mod], [])), \
             patch.object(server.frame_android, 'install') as install, \
             patch.dict(server.frame_android.__dict__):
            server.frame_android.__dict__.pop('install_obb', None)
            with self.assertRaisesRegex(SourceError, 'OBB'):
                search.install('one', 'brush')
            install.assert_not_called()

    def test_listing_cannot_download(self):
        mod = fake()
        mod.details = lambda s, i: dict(ENTRIES[0], downloadable=False)
        with patch.object(search, 'modules', return_value=([mod], [])):
            with self.assertRaisesRegex(SourceError, 'developer page'):
                search.install('one', 'brush')
            mod.download.assert_not_called()


class EndpointTests(SettingsTest):
    def setUp(self):
        super().setUp()
        self.mod = fake()
        p = patch.object(search, 'modules', return_value=([self.mod], []))
        p.start()
        self.addCleanup(p.stop)
        self.httpd = server.ThreadingHTTPServer(('127.0.0.1', 0), server.Handler)
        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
        self.addCleanup(self.httpd.server_close)
        self.addCleanup(self.httpd.shutdown)

    def request(self, method, path, body=None):
        c = http.client.HTTPConnection('127.0.0.1', self.httpd.server_port)
        c.request(method, path, json.dumps(body) if body is not None else None,
                  {'X-Frame-UI': '1', 'Content-Type': 'application/json'})
        r = c.getresponse()
        result = r.status, json.loads(r.read())
        c.close()
        return result

    def test_http_search_and_validation(self):
        self.assertEqual(self.request('GET', '/api/sources')[1]['sources'][0]['id'], 'one')
        self.assertTrue(self.request('GET', '/api/search?q=Brush&vr=true')[1]['apps'])
        self.assertEqual(self.request('GET', '/api/search?vr=invalid')[0], 400)
        self.assertEqual(self.request('GET', '/api/search?source=missing')[0], 400)
        for body in ({'source': 'one'}, {'source': 'one', 'id': 'brush', 'version_code': True}):
            self.assertEqual(self.request('POST', '/api/sources/install', body)[0], 400)

    def test_http_install_background_job(self):
        with patch.object(server.frame_android, 'install', return_value={'package': 'org.brush'}) as install:
            status, reply = self.request('POST', '/api/sources/install', {'source': 'one', 'id': 'brush'})
            self.assertEqual(status, 200)
            for _ in range(100):
                job = self.request('GET', '/api/job?id=' + reply['job'])[1]
                if job['done']:
                    break
                time.sleep(.01)
            self.assertTrue(job['done'])
            self.assertIsNone(job['error'])
            install.assert_called_once_with('/fake.apk', name='Open Brush', icon_png=None, source='one')

    def test_details_endpoint_and_real_install_stages(self):
        code, entry = self.request('GET', '/api/sources/details?source=one&id=brush')
        self.assertEqual(code, 200)
        self.assertEqual(entry['name'], 'Open Brush')
        self.assertEqual(entry['verdict']['label'], 'Ready to try on the Frame')
        self.assertIn('artwork', entry)
        self.assertEqual(self.request('GET', '/api/sources/details?source=one')[0], 400)
        stages = []
        with patch.object(server.frame_android, 'install', return_value={'package':'org.brush'}):
            search.install('one', 'brush', progress=lambda stage, percent: stages.append((stage,percent)))
        self.assertEqual(stages, [('Downloading',None),('Installing',None)])

    def test_http_repository_management(self):
        self.assertEqual(self.request('POST', '/api/sources', {'action': 'enable', 'source': 'one', 'enabled': False})[0], 200)
        code, reply = self.request('POST', '/api/sources', {'action': 'add', 'url': 'https://repo.example/repo'})
        self.assertEqual(code, 400)
        self.assertIn('not available', reply['error'])
        mod = fake('fdroid')
        added = {'id': 'fdroid-user-1', 'name': 'repo.example', 'fingerprint': 'ab' * 32, 'trust_on_first_use': True}
        mod.add_repo, mod.remove_repo, mod.set_enabled = Mock(return_value=added), Mock(), Mock()
        with patch.object(search, 'modules', return_value=([mod], [])):
            code, reply = self.request('POST', '/api/sources', {'action': 'add', 'url': 'https://repo.example/repo'})
            self.assertEqual(code, 200)
            job = self.wait(reply['job'])
            self.assertEqual(job['message'], 'Added repo.example. Trusted on first use: ' + 'AB' * 32)
            self.assertEqual(job['result']['source']['fingerprint'], 'ab' * 32)
            mod.add_repo.assert_called_once_with(url='https://repo.example/repo', fingerprint=None, name=None)
            link = 'fdroidrepos://repo.example/repo?fingerprint=' + 'ab' * 32
            mod.add_repo.return_value = dict(added, trust_on_first_use=False)
            job = self.wait(self.request('POST', '/api/sources', {'action': 'add', 'url': link})[1]['job'])
            self.assertEqual(job['message'], 'Added repo.example')
            mod.add_repo.assert_called_with(url=link, fingerprint=None, name=None)
            mod.add_repo.side_effect = SourceError('repository fingerprint mismatch')
            job = self.wait(self.request('POST', '/api/sources', {'action': 'add', 'url': link})[1]['job'])
            self.assertEqual(job['error'], 'repository fingerprint mismatch')  # no "SourceError:" prefix
            for url in ('http://repo.example/repo', 'fdroidrepo://repo.example/repo', 'https://u@repo.example/'):
                self.assertEqual(self.request('POST', '/api/sources', {'action': 'add', 'url': url})[0], 400)
            self.assertEqual(self.request('POST', '/api/sources', {'action': 'remove', 'source': 'fdroid'})[0], 200)
            mod.remove_repo.assert_called_once_with(source_id='fdroid')

    def test_http_add_game_data_job(self):
        search._game_data['org.brush'] = ['/cache/main.1.org.brush.obb']
        self.addCleanup(search._game_data.clear)
        with patch.object(server.frame_android, 'install_obb', create=True,
                          side_effect=server.frame_android.FrameError('start this app instance before installing OBB data')):
            job = self.wait(self.request('POST', '/api/sources', {'action': 'game-data', 'package': 'org.brush'})[1]['job'])
        self.assertEqual(job['error'], 'start this app instance before installing OBB data')
        with patch.object(server.frame_android, 'install_obb', create=True, return_value={'package': 'org.brush'}) as obb:
            job = self.wait(self.request('POST', '/api/sources', {'action': 'game-data', 'package': 'org.brush'})[1]['job'])
        self.assertEqual(job['message'], 'Game data added')
        obb.assert_called_once_with('org.brush', ['/cache/main.1.org.brush.obb'])

    def wait(self, job_id):
        for _ in range(200):
            job = self.request('GET', '/api/job?id=' + job_id)[1]
            if job['done']:
                return job
            time.sleep(.01)
        self.fail('job did not finish')