diff --git a/gallery_dl/extractor/loveisover.py b/gallery_dl/extractor/loveisover.py index 1d7bd0a9..d4d833b0 100644 --- a/gallery_dl/extractor/loveisover.py +++ b/gallery_dl/extractor/loveisover.py @@ -18,5 +18,5 @@ class LoveisoverThreadExtractor(chan.FoolfuukaThreadExtractor): pattern = [r"(?:https?://)?(?:archive\.)?loveisover\.me" r"/([^/]+)/thread/(\d+)"] test = [("https://archive.loveisover.me/c/thread/2898043/", { - "url": "1340ba9d04d4e591d62e5ae5df3fd732f06a99fd", + "url": "f1531a84de4e25ed3ae768384f6be43a0f02db71", })] diff --git a/scripts/run_tests.sh b/scripts/run_tests.sh index 82991b1a..55ef84d4 100755 --- a/scripts/run_tests.sh +++ b/scripts/run_tests.sh @@ -1,6 +1,6 @@ #!/bin/bash -TESTS_CORE=(config cookies extractor oauth text util) +TESTS_CORE=(config cookies downloader extractor oauth text util) TESTS_RESULTS=(results) diff --git a/test/test_downloader.py b/test/test_downloader.py new file mode 100644 index 00000000..19cecb8b --- /dev/null +++ b/test/test_downloader.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +# Copyright 2018 Mike Fährmann +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 as +# published by the Free Software Foundation. + +import re +import base64 +import os.path +import tempfile +import unittest +import threading +import http.server + +import requests + +import gallery_dl.downloader as downloader +import gallery_dl.extractor as extractor +import gallery_dl.config as config +from gallery_dl.downloader.common import DownloaderBase +from gallery_dl.output import NullOutput +from gallery_dl.util import PathFormat + + +class TestDownloaderBase(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.dir = tempfile.TemporaryDirectory() + cls.fnum = 0 + config.set(("base-directory",), cls.dir.name) + + @classmethod + def tearDownClass(cls): + cls.dir.cleanup() + config.clear() + + @classmethod + def _prepare_destination(cls, content=None, part=True, extension=None): + name = "file-{}".format(cls.fnum) + cls.fnum += 1 + + kwdict = { + "category": "test", + "subcategory": "test", + "name": name, + "extension": extension, + } + pathfmt = PathFormat(extractor.find("test:")) + pathfmt.set_directory(kwdict) + pathfmt.set_keywords(kwdict) + + if content: + path = pathfmt.realpath + (".part" if part else "") + mode = "w" + ("b" if isinstance(content, bytes) else "") + with open(path, mode) as file: + file.write(content) + + return pathfmt + + def _run_test(self, url, input, output, + extension, expected_extension=None): + pathfmt = self._prepare_destination(input, extension=extension) + success = self.downloader.download(url, pathfmt) + path = pathfmt.realpath + + # test successful download + self.assertTrue(success, "downloading '{}' failed".format(url)) + + # test content + mode = "r" + ("b" if isinstance(output, bytes) else "") + with open(path, mode) as file: + content = file.read() + self.assertEqual(content, output) + + # test filename extension + self.assertEqual( + os.path.splitext(path)[1][1:], + expected_extension, + ) + + +class TestHTTPDownloader(TestDownloaderBase): + + @classmethod + def setUpClass(cls): + TestDownloaderBase.setUpClass() + cls.downloader = downloader.find("http")( + requests.session(), NullOutput()) + + port = 8088 + cls.address = "http://127.0.0.1:{}".format(port) + cls._jpg = cls.address + "/image.jpg" + cls._png = cls.address + "/image.png" + cls._gif = cls.address + "/image.gif" + + server = http.server.HTTPServer(("", port), HttpRequestHandler) + threading.Thread(target=server.serve_forever, daemon=True).start() + + def test_http_download(self): + self._run_test(self._jpg, None, DATA_JPG, "jpg", "jpg") + self._run_test(self._png, None, DATA_PNG, "png", "png") + self._run_test(self._gif, None, DATA_GIF, "gif", "gif") + + def test_http_offset(self): + self._run_test(self._jpg, DATA_JPG[:123], DATA_JPG, "jpg", "jpg") + self._run_test(self._png, DATA_PNG[:12] , DATA_PNG, "png", "png") + self._run_test(self._gif, DATA_GIF[:1] , DATA_GIF, "gif", "gif") + + def test_http_extension(self): + self._run_test(self._jpg, None, DATA_JPG, None, "jpg") + self._run_test(self._png, None, DATA_PNG, None, "png") + self._run_test(self._gif, None, DATA_GIF, None, "gif") + + def test_http_adjust_extension(self): + self._run_test(self._jpg, None, DATA_JPG, "png", "jpg") + self._run_test(self._png, None, DATA_PNG, "gif", "png") + self._run_test(self._gif, None, DATA_GIF, "jpg", "gif") + + +class TestTextDownloader(TestDownloaderBase): + + @classmethod + def setUpClass(cls): + TestDownloaderBase.setUpClass() + cls.downloader = downloader.find("text")(None, NullOutput()) + + def test_text_download(self): + self._run_test("text:foobar", None, "foobar", "txt", "txt") + + def test_text_offset(self): + self._run_test("text:foobar", "foo", "foobar", "txt", "txt") + + def test_text_extension(self): + self._run_test("text:foobar", None, "foobar", None, "txt") + + def test_text_empty(self): + self._run_test("text:", None, "", "txt", "txt") + + +class FakeDownloader(DownloaderBase): + scheme = "fake" + + def __init__(self, session, output): + DownloaderBase.__init__(self, session, output) + + def connect(self, url, offset): + pass + + def receive(self, file): + pass + + def reset(self): + pass + + def get_extension(self): + pass + + @staticmethod + def _check_extension(file, pathfmt): + pass + + +class HttpRequestHandler(http.server.BaseHTTPRequestHandler): + + def do_GET(self): + if self.path == "/image.jpg": + content_type = "image/jpeg" + output = DATA_JPG + elif self.path == "/image.png": + content_type = "image/png" + output = DATA_PNG + elif self.path == "/image.gif": + content_type = "image/gif" + output = DATA_GIF + else: + self.send_response(404) + self.wfile.write(self.path.encode()) + return + + headers = { + "Content-Type": content_type, + "Content-Length": len(output), + } + + if "Range" in self.headers: + status = 206 + + match = re.match(r"bytes=(\d+)-", self.headers["Range"]) + start = int(match.group(1)) + + headers["Content-Range"] = "bytes {}-{}/{}".format( + start, len(output)-1, len(output)) + output = output[start:] + else: + status = 200 + + self.send_response(status) + for key, value in headers.items(): + self.send_header(key, value) + self.end_headers() + self.wfile.write(output) + + +DATA_JPG = base64.standard_b64decode(""" +/9j/4AAQSkZJRgABAQEASABIAAD/2wBD +AAEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB +AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB +AQEBAQEBAQEBAQEBAQEBAQH/2wBDAQEB +AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB +AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB +AQEBAQEBAQEBAQEBAQH/wAARCAABAAED +AREAAhEBAxEB/8QAFAABAAAAAAAAAAAA +AAAAAAAACv/EABQQAQAAAAAAAAAAAAAA +AAAAAAD/xAAUAQEAAAAAAAAAAAAAAAAA +AAAA/8QAFBEBAAAAAAAAAAAAAAAAAAAA +AP/aAAwDAQACEQMRAD8AfwD/2Q==""") + + +DATA_PNG = base64.standard_b64decode(""" +iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB +CAAAAAA6fptVAAAACklEQVQIHWP4DwAB +AQEANl9ngAAAAABJRU5ErkJggg==""") + + +DATA_GIF = base64.standard_b64decode(""" +R0lGODdhAQABAIAAAP///////ywAAAAA +AQABAAACAkQBADs=""") + + +if __name__ == "__main__": + unittest.main()