#!/usr/bin/env python3 import inspect import logging 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 Hello World! """ 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) requests_get.assert_called() self.assertTrue(os.path.exists(f)) self.assertEqual('tmp/entry.jar', f) f = net.download_file( '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__)) import argparse parser = argparse.ArgumentParser() parser.add_argument( "-v", "--verbose", action="store_true", default=False, help="Spew out even more information than normal", ) common.options = common.parse_args(parser) newSuite = unittest.TestSuite() newSuite.addTest(unittest.makeSuite(NetTest)) unittest.main(failfast=False)