more unit tests

This commit is contained in:
klemek
2020-05-01 13:25:44 +02:00
parent 6748073048
commit 62afce22a6
3 changed files with 35 additions and 23 deletions
+1 -1
View File
@@ -82,7 +82,7 @@ def compute_part(*args: str, input_data: Optional[bytes] = None,
if len(args) <= 1:
return None, 'Image: received no input data nor URL'
else:
input_data, err = utils.read_web(args[1], max_file_size=max_file_size)
input_data, err = utils.read_web_file(args[1], max_file_size=max_file_size)
if input_data is None:
return None, 'Image: ' + err
img = img_factory.build_image_only(input_data)
+12 -12
View File
@@ -2,7 +2,7 @@ import re
import select
import sys
from urllib.request import urlopen
from urllib.error import URLError
from urllib.error import URLError, HTTPError
from urllib.parse import urlparse
import os.path as path
from typing import List, Optional, Union, Tuple, BinaryIO
@@ -218,7 +218,7 @@ def safe_index(src: Union[str, list], pattern, start: int = 0):
# endregion
# region bytes utils
# region stream utils
def is_stdin_ready() -> bool:
@@ -235,25 +235,25 @@ def read_stream(stream: BinaryIO) -> bytes:
return output_data
def read_web(url: str, *, timeout: int = 5,
# endregion
# region web utils
def read_web_file(url: str, *, timeout: float = 5,
max_file_size: Optional[int] = None) -> Tuple[Optional[bytes], Optional[str]]:
if not validate_url(url):
return None, 'Invalid URL'
try:
with urlopen(url, None, timeout) as web_file:
if web_file.getcode() != 200:
return None, 'Invalid return code'
if max_file_size is not None and int(web_file.info()['Content-Length']) > max_file_size:
return None, 'File too big'
return web_file.read(), None
except HTTPError as e:
return None, f'Could not connect: {e}'
except URLError:
return None, 'Could not connect'
# endregion
# region URL utils
return None, f'Could not connect to server'
def validate_url(url: str) -> bool:
+21 -9
View File
@@ -167,17 +167,29 @@ class TestUtilsFormat(TestCase):
self.assertEqual([5, 9, 15, 18], utils.place_line_breaks([5.2, 14.3, 14.5, 15.2], [3, 5, 9, 15, 18, 20]))
class TestUtilsStream(TestCase):
def test_read_stream(self):
pass # TODO
def test_read_web(self):
pass # TODO
class TestUtilsUrl(TestCase):
class TestUtilsWeb(TestCase):
def test_validate_url(self):
self.assertTrue(utils.validate_url("https://google.com/page#anchor?key=value&query"))
self.assertFalse(utils.validate_url("https:google.com/page#anchor?key=value&query"))
self.assertFalse(utils.validate_url(""))
self.assertFalse(utils.validate_url("google.com"))
def test_read_web_file(self):
out, err = utils.read_web_file("http:invalid.url")
self.assertIsNone(out)
self.assertEqual('Invalid URL', err)
out, err = utils.read_web_file("http://unknown.domain/")
self.assertIsNone(out)
self.assertEqual('Could not connect to server', err)
out, err = utils.read_web_file("http://httpbin.org/status/418")
self.assertIsNone(out)
self.assertEqual('Could not connect: HTTP Error 418: I\'M A TEAPOT', err)
out, err = utils.read_web_file("http://httpbin.org/bytes/1024", max_file_size=1000)
self.assertIsNone(out)
self.assertEqual('File too big', err)
out, err = utils.read_web_file("http://httpbin.org/delay/1", timeout=0.1)
self.assertIsNone(out)
self.assertEqual('Could not connect to server', err)
out, err = utils.read_web_file("http://httpbin.org/base64/dGVzdA==")
self.assertIsNone(err)
self.assertEqual('test', out.decode("utf-8"))