1
0
mirror of https://gitlab.com/fdroid/fdroidserver.git synced 2024-11-04 22:40:12 +01:00
fdroidserver/tests/net.TestCase

142 lines
4.7 KiB
Plaintext
Raw Normal View History

#!/usr/bin/env python3
import inspect
import logging
import optparse
import os
import random
import requests
import socket
import sys
import tempfile
import threading
import time
import unittest
from unittest.mock import MagicMock, patch
localmodule = os.path.realpath(
os.path.join(os.path.dirname(inspect.getfile(inspect.currentframe())), '..')
)
print('localmodule: ' + localmodule)
if localmodule not in sys.path:
sys.path.insert(0, localmodule)
from fdroidserver import common, net
from pathlib import Path
class RetryServer:
"""A stupid simple HTTP server that can fail to connect"""
def __init__(self, port=None, failures=3):
self.port = port
if self.port is None:
self.port = random.randint(1024, 65535)
self.failures = failures
self.stop_event = threading.Event()
threading.Thread(target=self.run_fake_server).start()
def stop(self):
self.stop_event.set()
def run_fake_server(self):
server_sock = socket.socket()
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_sock.bind(('127.0.0.1', self.port))
server_sock.listen(5)
server_sock.settimeout(5)
time.sleep(0.001) # wait for it to start
while not self.stop_event.is_set():
self.failures -= 1
conn = None
try:
conn, address = server_sock.accept()
conn.settimeout(5)
except TimeoutError:
break
if self.failures > 0:
conn.close()
continue
conn.recv(8192) # request ignored
self.reply = b"""HTTP/1.1 200 OK
Date: Mon, 26 Feb 2024 09:00:14 GMT
Connection: close
Content-Type: text/html
<HTML><BODY>Hello World!</HEAD></HTML>
"""
self.reply = self.reply.replace(b' ', b'') # dedent
conn.sendall(self.reply)
conn.shutdown(socket.SHUT_RDWR)
conn.close()
self.stop_event.wait(timeout=1)
server_sock.shutdown(socket.SHUT_RDWR)
server_sock.close()
class NetTest(unittest.TestCase):
basedir = Path(__file__).resolve().parent
def setUp(self):
logging.basicConfig(level=logging.DEBUG)
self.tempdir = tempfile.TemporaryDirectory()
os.chdir(self.tempdir.name)
Path('tmp').mkdir()
def tearDown(self):
self.tempdir.cleanup()
@patch('requests.get')
def test_download_file_url_parsing(self, requests_get):
# pylint: disable=unused-argument
def _get(url, stream, allow_redirects, headers, timeout):
return MagicMock()
requests_get.side_effect = _get
f = net.download_file('https://f-droid.org/repo/entry.jar', retries=0)
self.assertTrue(requests_get.called)
self.assertTrue(os.path.exists(f))
self.assertEqual('tmp/entry.jar', f)
f = net.download_file(
2022-10-22 23:15:13 +02:00
'https://d-05.example.com/custom/com.downloader.aegis-3175421.apk?_fn=QVBLUHVyZV92My4xNy41NF9hcGtwdXJlLmNvbS5hcGs&_p=Y29tLmFwa3B1cmUuYWVnb24&am=6avvTpfJ1dMl9-K6JYKzQw&arg=downloader%3A%2F%2Fcampaign%2F%3Futm_medium%3Ddownloader%26utm_source%3Daegis&at=1652080635&k=1f6e58465df3a441665e585719ab0b13627a117f&r=https%3A%2F%2Fdownloader.com%2Fdownloader-app.html%3Ficn%3Daegis%26ici%3Dimage_qr&uu=http%3A%2F%2F172.16.82.1%2Fcustom%2Fcom.downloader.aegis-3175421.apk%3Fk%3D3fb9c4ae0be578206f6a1c330736fac1627a117f',
retries=0,
)
self.assertTrue(requests_get.called)
self.assertTrue(os.path.exists(f))
self.assertEqual('tmp/com.downloader.aegis-3175421.apk', f)
def test_download_file_retries(self):
server = RetryServer()
f = net.download_file('http://localhost:%d/f.txt' % server.port)
# strip the HTTP headers and compare the reply
self.assertEqual(server.reply.split(b'\n\n')[1], Path(f).read_bytes())
server.stop()
def test_download_file_retries_not_forever(self):
"""The retry logic should eventually exit with an error."""
server = RetryServer(failures=5)
with self.assertRaises(requests.exceptions.ConnectionError):
net.download_file('http://localhost:%d/f.txt' % server.port)
server.stop()
if __name__ == "__main__":
os.chdir(os.path.dirname(__file__))
parser = optparse.OptionParser()
parser.add_option(
"-v",
"--verbose",
action="store_true",
default=False,
help="Spew out even more information than normal",
)
(common.options, args) = parser.parse_args(['--verbose'])
newSuite = unittest.TestSuite()
newSuite.addTest(unittest.makeSuite(NetTest))
unittest.main(failfast=False)