mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-06 07:00:37 +02:00
saphidandClaude Opus 5.5
a9d78679ec
Adding a repository downloads and verifies its whole index, so it now runs as a job (runJob in the UI) and reports 'Trusted on first use: <fingerprint>' when no pin was given. fdroidrepos:// links pass the server check, as documented. Jobs report SourceError messages without a 'SourceError:' prefix. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
275 lines
15 KiB
Python
275 lines
15 KiB
Python
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)
|
|
search._running.clear()
|
|
search._status.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_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):
|
|
search.install('one', 'brush', 1)
|
|
obb.assert_called_once_with('org.brush', ['main.obb'])
|
|
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 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')
|