mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-06 06:00:33 +02:00
208 lines
11 KiB
Python
208 lines
11 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_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', 'timed out', 'error'])
|
|
search.search('other', timeout=.03)
|
|
self.assertEqual(calls, [''])
|
|
finally:
|
|
release.set()
|
|
|
|
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')
|
|
mod.add_repo, mod.remove_repo, mod.set_enabled = Mock(), Mock(), Mock()
|
|
with patch.object(search, 'modules', return_value=([mod], [])):
|
|
self.assertEqual(self.request('POST', '/api/sources', {'action': 'add', 'url': 'https://repo.example/repo'})[0], 200)
|
|
mod.add_repo.assert_called_once_with(url='https://repo.example/repo', fingerprint=None, name=None)
|
|
self.assertEqual(self.request('POST', '/api/sources', {'action': 'remove', 'source': 'fdroid'})[0], 200)
|
|
mod.remove_repo.assert_called_once_with(source_id='fdroid')
|