steam-frame/saphid--frame-control
1
0 Fork 0
mirror of https://github.com/saphid/frame-control.git synced 2026-10-06 00:00:21 +02:00
Code Issues Packages Projects Releases 6 Wiki Activity
Files
main
saphid--frame-control/tests/test_fdroid_sources.py
T
saphidandGPT-6 Astra 8fca0fe0a2 Merge origin/main into apk-store-fixes, preserving store and headset features 
Keep the union of server routes, desktop resources and responsive controls. Preserve OpenXR install defaults and telemetry hooks alongside library artwork. Adapt the resource test to single-file entries and avoid a completed-refresh race in the F-Droid test.

Co-Authored-By: GPT-6 Astra <noreply@openai.com>
2026-09-29 12:07:09 +10:00

538 lines
26 KiB
Python
Raw Permalink Blame History

"""Offline authenticated repository fixtures; no tests contact a server."""
import io
import json
import os
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
import urllib.error
import zipfile
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
from apk_sources import SourceError, SourceLimited, _web, fdroid
FIXTURES = Path(__file__).parent / 'fixtures' / 'fdroid'
PIN = (FIXTURES / 'fingerprint.txt').read_text().strip()
URL = 'https://example.org/repo/'
_KEY = []
def signed_jar(member, content, digest='sha256'):
"""A JAR signed like fdroidserver's (no CMS signed attributes) with a throwaway test key."""
import base64, hashlib
from frame_apk_sign import certificate, der, integer, sequence, signing_key
if not _KEY:
with tempfile.TemporaryDirectory() as tmp:
_KEY.append(signing_key(Path(tmp) / 'key.json'))
key = _KEY[0]
label = 'SHA1' if digest == 'sha1' else 'SHA-256'
b64 = lambda data: base64.b64encode(hashlib.new(digest, data).digest()).decode()
manifest = ('Manifest-Version: 1.0\r\n\r\nName: %s\r\n%s-Digest: %s\r\n\r\n' % (member, label, b64(content))).encode()
sf = ('Signature-Version: 1.0\r\n%s-Digest-Manifest: %s\r\n\r\n' % (label, b64(manifest))).encode()
oid, prefix = next((bytes.fromhex(o), bytes.fromhex(p)) for o, (d, p) in fdroid._DIGESTS.items() if d == digest)
alg = sequence(der(6, oid), der(5, b''))
size = (key['n'].bit_length() + 7) // 8
value = prefix + hashlib.new(digest, sf).digest()
padded = b'\0\1' + b'\xff' * (size - len(value) - 3) + b'\0' + value
signature = pow(int.from_bytes(padded, 'big'), key['d'], key['n']).to_bytes(size, 'big')
cert = certificate(key)
issuer = fdroid._der_parts(fdroid._der_parts(fdroid._der_parts(cert)[0][1])[0][1])[3][2]
signer = sequence(integer(1), sequence(issuer, integer(1)), alg,
sequence(der(6, bytes.fromhex('2a864886f70d010101')), der(5, b'')), der(4, signature))
signed = sequence(integer(1), der(0x31, alg), sequence(der(6, bytes.fromhex('2a864886f70d010701'))),
der(0xa0, cert), der(0x31, signer))
block = sequence(der(6, bytes.fromhex('2a864886f70d010702')), der(0xa0, signed))
stream = io.BytesIO()
with zipfile.ZipFile(stream, 'w') as z:
for name, data in (('META-INF/MANIFEST.MF', manifest), ('META-INF/TEST.SF', sf),
('META-INF/TEST.RSA', block), (member, content)):
z.writestr(name, data)
return stream.getvalue(), fdroid.hashlib.sha256(cert).hexdigest()
def entry_jar(timestamp, digest='sha256'):
entry = json.loads(zipfile.ZipFile(FIXTURES / 'entry.jar').read('entry.json'))
return signed_jar('entry.json', json.dumps(dict(entry, timestamp=timestamp)).encode(), digest)
class Repositories(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.root = Path(tmp.name)
for name, value in [('data_dir', lambda *p: self.root.joinpath('data', *p)),
('cache_dir', lambda *p: self.root.joinpath('cache', *p))]:
mock = patch.object(fdroid.frame_host, name, value)
mock.start()
self.addCleanup(mock.stop)
net = patch.object(fdroid.urllib.request, 'build_opener', side_effect=AssertionError('network forbidden'))
net.start()
self.addCleanup(net.stop)
mock = self.fetch_patch = patch.object(fdroid, '_fetch', side_effect=self.fetch)
self.fetch_mock = mock.start()
self.addCleanup(mock.stop)
self.v1 = False
self.corrupt = None
self.files = {}
for state in (_web._limited, fdroid._stale, fdroid._retry_at, fdroid._refreshing):
state.clear()
self.addCleanup(state.clear)
def fetch(self, url, path, maximum):
name = url.rsplit('/', 1)[-1]
if self.v1 and name == 'entry.jar':
raise urllib.error.HTTPError(url, 404, 'missing', None, None)
payload = self.files.get(name) or (FIXTURES / ('example.apk' if name.endswith('.apk') else name)).read_bytes()
if name == self.corrupt:
payload += b'tampered'
Path(path).write_bytes(payload)
def add(self):
return fdroid.add_repo(URL, PIN)
def test_add_search_details_download_cache(self):
source = self.add()
self.assertEqual(source['fingerprint'], PIN)
self.assertFalse(source['trust_on_first_use'])
result = fdroid.search(source, 'example offline')
self.assertEqual(len(result), 1)
self.assertNotIn('versions', result[0])
self.assertEqual(result[0]['version_code'], 2)
self.assertEqual([v['version_code'] for v in fdroid.details(source, 'org.example.app')['versions']], [2, 1])
downloaded = fdroid.download(source, 'org.example.app', 1)
self.assertTrue(downloaded['verified'])
self.assertEqual(Path(downloaded['apk']).read_bytes(), (FIXTURES / 'example.apk').read_bytes())
count = self.fetch_mock.call_count
os.utime(downloaded['apk'], (1, 1))
fdroid.download(source, 'org.example.app', 1)
self.assertEqual(self.fetch_mock.call_count, count)
self.assertGreater(Path(downloaded['apk']).stat().st_mtime, 1) # reuse counts as recent use
def test_wrong_pin_is_not_saved(self):
with self.assertRaisesRegex(SourceError, 'fingerprint mismatch'):
fdroid.add_repo(URL, '0' * 64)
self.assertEqual(fdroid.user_repos(), [])
def test_tampered_index_is_not_saved(self):
self.corrupt = 'index-v2.json'
with self.assertRaisesRegex(SourceError, 'SHA-256'):
self.add()
self.assertEqual(fdroid.user_repos(), [])
def test_tampered_apk_is_not_cached(self):
source = self.add()
self.corrupt = 'example2.apk'
with self.assertRaisesRegex(SourceError, 'APK SHA-256'):
fdroid.download(source, 'org.example.app')
self.assertEqual(list(self.root.rglob('*.apk')), [])
self.assertEqual(list(self.root.rglob('*.part')), [])
def test_v1_fallback(self):
self.v1 = True
source = self.add()
self.assertEqual(fdroid.search(source, 'Example')[0]['version_code'], 1)
self.assertTrue(fdroid.download(source, 'org.example.app')['verified'])
def test_bad_v2_never_downgrades(self):
self.corrupt = 'index-v2.json'
with self.assertRaises(SourceError):
self.add()
self.assertFalse(any(c.args[0].endswith('index-v1.jar') for c in self.fetch_mock.call_args_list))
def test_transient_error_never_downgrades(self):
self.fetch_mock.side_effect = urllib.error.HTTPError(URL, 503, 'unavailable', None, None)
with self.assertRaises(SourceError):
self.add()
self.assertEqual(self.fetch_mock.call_count, 1)
def test_tofu_preserves_pin_and_settings(self):
source = fdroid.add_repo('fdroidrepos://example.org/repo')
self.assertTrue(source['trust_on_first_use'])
self.assertEqual(source['fingerprint'], PIN)
fdroid.add_repo(URL)
self.assertEqual(len(fdroid.user_repos()), 1)
with self.assertRaisesRegex(SourceError, 'different pinned'):
fdroid.add_repo(URL, '0' * 64)
fdroid.set_enabled(source['id'], False)
self.assertEqual(fdroid.search(fdroid.user_repos()[0], ''), [])
with self.assertRaisesRegex(SourceError, 'disabled'):
fdroid.download(fdroid.user_repos()[0], 'org.example.app')
fdroid.set_enabled('fdroid', False)
self.assertFalse(fdroid.sources()[0]['enabled'])
fdroid.remove_repo(source['id'])
self.assertEqual(fdroid.user_repos(), [])
with self.assertRaises(SourceError):
fdroid.remove_repo('fdroid')
def test_urls(self):
self.assertEqual(fdroid._url(URL + '?fingerprint=' + PIN.upper()), (URL, PIN))
for url in ['http://example.org/repo', 'fdroidrepo://example.org', 'https://u:p@example.org', URL+'?other=x']:
with self.subTest(url=url), self.assertRaises(SourceError):
fdroid._url(url)
with self.assertRaisesRegex(SourceError, 'conflicting'):
fdroid._url(URL + '?fingerprint=' + PIN, '0' * 64)
self.assertEqual(fdroid._child(URL, '/app/en-US/phoneScreenshots/#0 a.png'),
URL + 'app/en-US/phoneScreenshots/%230%20a.png') # real F-Droid screenshot name
for name in ['../x.apk', '%2e%2e/x.apk', 'https://evil.org/a.apk', '//evil.org/../x', 'x?token=y', 'x\\y']:
with self.subTest(name=name), self.assertRaises(SourceError):
fdroid._child(URL, name)
def test_recorded_real_signature(self):
content, fingerprint = fdroid._jar(FIXTURES / 'izzy-entry.jar', 'entry.json', fdroid.IZZY_PIN, strong=True)
self.assertEqual(fingerprint, fdroid.IZZY_PIN)
self.assertIn('index', json.loads(content))
def test_tampering_each_signature_layer(self):
for member in ['entry.json', 'META-INF/MANIFEST.MF', 'META-INF/TEST.SF', 'META-INF/TEST.RSA']:
stream = io.BytesIO()
with zipfile.ZipFile(FIXTURES / 'entry.jar') as src, zipfile.ZipFile(stream, 'w') as dst:
for item in src.infolist():
data = src.read(item.filename)
if item.filename == member:
data = data[:-1] + bytes([data[-1] ^ 1])
dst.writestr(item.filename, data)
with self.subTest(member=member), self.assertRaises(SourceError):
fdroid._jar(io.BytesIO(stream.getvalue()), 'entry.json', PIN)
def test_duplicate_jar_member_rejected(self):
stream = io.BytesIO((FIXTURES / 'entry.jar').read_bytes())
import warnings
with warnings.catch_warnings():
warnings.simplefilter('ignore', UserWarning)
with zipfile.ZipFile(stream, 'a') as z:
z.writestr('entry.json', '{}')
stream.seek(0)
with self.assertRaisesRegex(SourceError, 'duplicate'):
fdroid._jar(stream, 'entry.json', PIN)
def test_corrupt_cache_refetches_verified_index(self):
source = self.add()
fdroid.frame_host.cache_dir('apk-sources', source['id'] + '.json').write_text('{')
self.assertEqual(len(fdroid.search(source, 'example')), 1)
self.assertEqual(self.fetch_mock.call_count, 4)
def test_artwork_v1_and_v2_survives_source_cache(self):
source = self.add()
for version in ('v1', 'v2'):
with self.subTest(version=version):
raw = FIXTURES / ('artwork-' + version + '.json')
if version == 'v1':
normalized = self.root / 'normalized.json'
fdroid._v1(raw.read_bytes(), normalized)
raw = normalized
apps = fdroid._reduce(raw, source)
cache = fdroid.frame_host.cache_dir('apk-sources', source['id'] + '.json')
fdroid._write(cache, {'version': fdroid.CACHE_VERSION, 'url': URL,
'fingerprint': PIN, 'apps': apps})
before = self.fetch_mock.call_count
result = fdroid.search(source, 'example offline')[0]
self.assertEqual(result['developer'], 'Example Developer')
self.assertEqual(result['summary'], 'Offline fixture & music. One line.')
self.assertEqual(result['icon'], URL + 'org.example.app/en-US/icon.png')
self.assertEqual(result['images'], {
'icon': result['icon'],
'banner': URL + 'org.example.app/fr/featureGraphic.png',
'screenshots': [URL + 'org.example.app/en-US/phoneScreenshots/' + str(i) + '.png' for i in range(1, 5)] +
[URL + 'org.example.app/fr/sevenInchScreenshots/' + str(i) + '.png' for i in range(1, 3)]})
self.assertEqual(fdroid.details(source, result['id'])['images'], result['images'])
self.assertEqual(self.fetch_mock.call_count, before)
def test_missing_artwork_is_not_invented(self):
result = fdroid.search(self.add(), 'example')[0]
self.assertEqual(result['images'], {'icon': None, 'banner': None, 'screenshots': []})
self.assertIsNone(result['icon'])
self.assertIsNone(result['developer'])
def test_v1_legacy_icon_and_tablet_fallback(self):
index = json.loads((FIXTURES / 'artwork-v1.json').read_text())
app = index['apps'][0]
app['localized'] = {'fr': {'sevenInchScreenshots': ['tablet.png']}}
app['icon'] = 'legacy.1.png'
raw = self.root / 'legacy.json'
fdroid._v1(json.dumps(index).encode(), raw)
result = fdroid._reduce(raw, {'id': 'test', 'url': URL})['org.example.app']
self.assertEqual(result['icon'], URL + 'icons/legacy.1.png')
self.assertEqual(result['images']['screenshots'], [URL + 'org.example.app/fr/sevenInchScreenshots/tablet.png'])
def test_v2_legacy_screenshot_keys_and_limit(self):
meta = {'phoneScreenshots': {'fr': [{'name': '/phone/' + str(i) + '.png'} for i in range(8)]},
'sevenInchScreenshots': {'en-US': [{'name': '/tablet.png'}]}}
images = fdroid._images(meta, URL)
self.assertEqual(images['screenshots'], [URL + 'phone/' + str(i) + '.png' for i in range(6)])
meta.pop('phoneScreenshots')
self.assertEqual(fdroid._images(meta, URL)['screenshots'], [URL + 'tablet.png'])
def test_old_cache_refreshes_for_artwork(self):
source = self.add()
path = fdroid.frame_host.cache_dir('apk-sources', source['id'] + '.json')
saved = json.loads(path.read_text())
saved.pop('version')
for app in saved['apps'].values():
app.pop('images')
fdroid._write(path, saved)
self.assertIn('images', fdroid.search(source, 'example')[0])
self.assertEqual(self.fetch_mock.call_count, 4)
def test_slow_download_blocks_neither_settings_nor_other_repos(self):
import threading
source = self.add()
other = dict(source, id='other-repo')
started, release = threading.Event(), threading.Event()
def fetch(url, path, maximum):
if threading.current_thread().name == 'slow':
started.set()
release.wait(5)
self.fetch(url, path, maximum)
self.fetch_mock.side_effect = fetch
slow = threading.Thread(target=fdroid._load, args=(other, True), name='slow')
slow.start()
try:
self.assertTrue(started.wait(2))
results = []
# The settings lock is free and another repo still loads while this one downloads.
check = threading.Thread(target=lambda: results.append(
(fdroid.set_enabled('fdroid', False), len(fdroid._load(source, force=True)[0]))))
check.start()
check.join(2)
self.assertEqual(results, [(None, 1)])
finally:
release.set()
slow.join()
def test_rollback_to_older_index_is_refused(self):
self.files['entry.jar'], pin = entry_jar(2000)
source = fdroid.add_repo(URL)
self.assertEqual(source['fingerprint'], pin)
self.files['entry.jar'], _ = entry_jar(1000)
with self.assertRaisesRegex(SourceError, 'older'):
fdroid._load(source, force=True)
for timestamp in (2000, 3000): # unchanged and newer indexes are fine
self.files['entry.jar'], _ = entry_jar(timestamp)
self.assertEqual(len(fdroid._load(source, force=True)[0]), 1)
self.files['entry.jar'], _ = entry_jar(2000)
with self.assertRaisesRegex(SourceError, 'older'):
fdroid._load(source, force=True)
fdroid.remove_repo(source['id']) # a deliberate re-add starts over
self.assertEqual(fdroid.add_repo(URL)['fingerprint'], pin)
def test_concurrent_processes_cannot_publish_an_older_index_last(self):
# Threads with their own in-memory locks, as separate processes would have; each
# _state_file_lock() opens its own file description, so flock contends for real.
import threading
self.files['entry.jar'], _ = entry_jar(50)
source = fdroid.add_repo(URL)
jars = {'older': entry_jar(100)[0], 'newer': entry_jar(200)[0]}
def fetch(url, path, maximum):
name = threading.current_thread().name
if name in jars and url.endswith('entry.jar'):
Path(path).write_bytes(jars[name])
else:
self.fetch(url, path, maximum)
self.fetch_mock.side_effect = fetch
inside, go, order = threading.Event(), threading.Event(), []
real_write = fdroid._write
def write(path, value):
if path.name.endswith('.json') and 'apps' in value and threading.current_thread().name == 'older':
inside.set() # the older load has passed its locked recheck; hold it there
go.wait(5)
order.append(threading.current_thread().name)
real_write(path, value)
errors = {}
def load():
try:
fdroid._load(source, force=True)
except SourceError as e:
errors[threading.current_thread().name] = str(e)
with patch.object(fdroid, '_source_lock', lambda source_id: threading.Lock()), \
patch.object(fdroid, '_write', write):
older = threading.Thread(target=load, name='older')
older.start()
self.assertTrue(inside.wait(5))
newer = threading.Thread(target=load, name='newer')
newer.start()
newer.join(.5)
self.assertTrue(newer.is_alive()) # blocked on the file lock, not publishing
go.set()
older.join(5)
newer.join(5)
self.assertEqual(errors, {})
self.assertEqual(order, ['older', 'older', 'newer', 'newer']) # cache+state, one load at a time
self.assertEqual(fdroid._state(source)['timestamp'], 200)
def test_overlapping_v1_load_cannot_replace_accepted_v2(self):
import threading
v1 = json.loads(zipfile.ZipFile(FIXTURES / 'index-v1.jar').read('index-v1.json'))
v1['repo'] = {'timestamp': 100}
self.files['index-v1.jar'], pin = signed_jar('index-v1.json', json.dumps(v1).encode())
self.files['entry.jar'], _ = entry_jar(100) # the same timestamp as the v1 index
source = dict(id='overlap', name='Overlap', url=URL, fingerprint=None)
self.v1 = True
inner = []
def fetch(url, path, maximum):
if url.endswith('index-v1.jar') and not inner:
inner.append(1) # the v1 load passed its fallback check; a v2 load finishes now
self.v1 = False
t = threading.Thread(target=lambda: inner.append(fdroid._load(source, force=True)))
with patch.object(fdroid, '_source_lock', lambda source_id: threading.Lock()):
t.start()
t.join(5)
self.v1 = True
self.fetch(url, path, maximum)
self.fetch_mock.side_effect = fetch
with self.assertRaisesRegex(SourceError, 'v2'):
fdroid._load(source, force=True)
self.assertEqual(inner[1][1], pin)
self.assertTrue(fdroid._state(source)['v2'])
self.v1 = False
count = self.fetch_mock.call_count
apps, _ = fdroid._load(dict(source, fingerprint=pin))
self.assertEqual((apps['org.example.app']['version_code'], self.fetch_mock.call_count), (2, count)) # v2 cache stayed
def test_apk_removed_during_cache_check_is_downloaded_again(self):
source = self.add()
first = fdroid.download(source, 'org.example.app', 1)
count = self.fetch_mock.call_count
real = fdroid._sha256
def removed_first(path):
if str(path) == first['apk'] and not hashed:
hashed.append(1)
os.remove(first['apk']) # deleted between the existence check and the open
return real(path)
hashed = []
with patch.object(fdroid, '_sha256', removed_first):
again = fdroid.download(source, 'org.example.app', 1)
self.assertEqual(self.fetch_mock.call_count, count + 1)
self.assertEqual(Path(again['apk']).read_bytes(), (FIXTURES / 'example.apk').read_bytes())
def test_cached_apk_is_touched_before_hashing(self):
source = self.add()
first = fdroid.download(source, 'org.example.app', 1)
os.utime(first['apk'], (1, 1))
count = self.fetch_mock.call_count
real = fdroid._sha256
def prune_first(path):
if str(path) == first['apk']:
with patch.object(_web, 'APK_CAP', 0):
_web.prune() # a pruner running now sees a just-used APK
return real(path)
with patch.object(fdroid, '_sha256', prune_first):
fdroid.download(source, 'org.example.app', 1)
self.assertEqual(self.fetch_mock.call_count, count)
self.assertTrue(Path(first['apk']).exists())
def test_cli_search_waits_for_background_refresh(self):
source = self.add()
self.expire(source)
before = self.fetch_mock.call_count
out = io.StringIO()
with patch.object(sys, 'argv', ['fdroid.py', 'search', source['id'], 'example']), \
patch('sys.stdout', out):
fdroid.main()
self.assertEqual(json.loads(out.getvalue())[0]['id'], 'org.example.app')
self.assertEqual(self.fetch_mock.call_count, before + 2) # the refresh finished before exit
self.assertFalse(fdroid.stale(source))
self.assertNotIn(source['id'], fdroid._refreshing)
def test_cli_error_still_waits_for_background_refresh(self):
import threading
source = self.add()
self.expire(source)
release = threading.Event()
def slow(url, path, maximum):
release.wait(5)
self.fetch(url, path, maximum)
self.fetch_mock.side_effect = slow
threading.Timer(.3, release.set).start()
with patch.object(sys, 'argv', ['fdroid.py', 'download', source['id'], 'org.missing']), \
patch('sys.stderr', io.StringIO()) as err, self.assertRaises(SystemExit):
fdroid.main()
self.assertIn('no Lepton-compatible version', err.getvalue())
self.assertTrue(release.is_set())
self.assertEqual(fdroid._refreshing, {}) # joined before exiting
self.assertFalse(fdroid.stale(source))
def test_no_v1_fallback_once_v2_accepted(self):
source = self.add()
self.v1 = True
with self.assertRaisesRegex(SourceError, 'v2'):
fdroid._load(source, force=True)
self.assertFalse(any(c.args[0].endswith('index-v1.jar') for c in self.fetch_mock.call_args_list))
def test_v1_then_v2_upgrade_is_allowed(self):
self.v1 = True
source = self.add()
self.v1 = False
self.assertEqual(fdroid._load(source, force=True)[0]['org.example.app']['version_code'], 2)
def test_sha1_only_entry_jar_rejected(self):
self.files['entry.jar'], _ = entry_jar(1, 'sha1')
with self.assertRaisesRegex(SourceError, 'SHA-1'):
fdroid.add_repo(URL)
self.assertEqual(fdroid.user_repos(), [])
stream = io.BytesIO(self.files['entry.jar'])
self.assertIn(b'index', fdroid._jar(stream, 'entry.json', None)[0]) # index-v1.jar may still use SHA-1
def test_rate_limited_host_backs_off(self):
source = dict(self.add(), name='My repo')
self.fetch_patch.stop()
error = urllib.error.HTTPError(URL, 429, 'slow down', {'Retry-After': '300'}, None)
with patch.object(fdroid.urllib.request, 'build_opener') as opener:
opener.return_value.open.side_effect = error
with self.assertRaisesRegex(SourceLimited, '^My repo is limiting requests; try again in 5 minutes$'):
fdroid._load(source, force=True)
with self.assertRaisesRegex(SourceLimited, 'My repo'):
fdroid.download(source, 'org.example.app', 1)
self.assertEqual(opener.return_value.open.call_count, 1)
self.fetch_mock = self.fetch_patch.start()
def expire(self, source):
cache = fdroid.frame_host.cache_dir('apk-sources', source['id'] + '.json')
old = cache.stat().st_mtime - fdroid.MAX_AGE - 1
fdroid.os.utime(str(cache), (old, old))
def test_expired_index_served_stale_while_refreshing(self):
source = self.add()
self.expire(source)
before = self.fetch_mock.call_count
release = __import__('threading').Event()
def slow(url, path, maximum):
release.wait(5)
self.fetch(url, path, maximum)
self.fetch_mock.side_effect = slow
self.assertEqual(len(fdroid.search(source, 'example')), 1) # immediately, from the old index
self.assertTrue(fdroid.stale(source))
refresh = fdroid._refreshing[source['id']]
fdroid.search(source, 'example')
self.assertIs(fdroid._refreshing.get(source['id']), refresh) # one refresh at a time
release.set()
refresh.join(5)
self.assertEqual(self.fetch_mock.call_count, before + 2)
self.assertFalse(fdroid.stale(source))
def test_failed_refresh_keeps_serving_stale_index(self):
source = self.add()
self.expire(source)
self.fetch_mock.side_effect = urllib.error.HTTPError(URL, 503, 'unavailable', None, None)
self.assertEqual(len(fdroid.search(source, 'example')), 1)
# A fast failed refresh may already have removed itself from the registry.
refresh = fdroid._refreshing.get(source['id'])
if refresh is not None:
refresh.join(5)
calls = self.fetch_mock.call_count
self.assertEqual(fdroid.details(source, 'org.example.app')['version_code'], 2)
self.assertTrue(fdroid.stale(source))
self.assertNotIn(source['id'], fdroid._refreshing) # failed refresh waits before retrying
self.assertEqual(self.fetch_mock.call_count, calls)
def test_cached_index_does_not_cross_pins(self):
source = self.add()
source['fingerprint'] = '0' * 64
with self.assertRaisesRegex(SourceError, 'fingerprint mismatch'):
fdroid.search(source, '')
if __name__ == '__main__':
unittest.main()
Reference in new issue View Git Blame Copy Permalink