Skip to content

Commit

Permalink
Factorized fetching network files in test suite
Browse files Browse the repository at this point in the history
  • Loading branch information
claudep committed Dec 24, 2021
1 parent 52d987b commit ec0b143
Showing 1 changed file with 41 additions and 48 deletions.
89 changes: 41 additions & 48 deletions tests/test_samples.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
from http.client import HTTPSConnection
from os.path import dirname, splitext, exists, join, basename, getsize
from urllib.parse import quote, unquote, urlparse
from urllib.request import urlopen

from reportlab.lib.utils import haveImages
from reportlab.graphics import renderPDF, renderPM
Expand All @@ -40,6 +39,41 @@ def found_uniconv():
return len(res) > 0


def fetch_file(url, mime_accept='text/svg', uncompress=True, raise_exc=False):
"""
Get given URL content using http.client module, uncompress if needed and
`uncompress` is True.
"""

parsed = urlparse(url)
conn = HTTPSConnection(parsed.netloc)
conn.request('GET', parsed.path, headers={
'Host': parsed.netloc,
'Accept': mime_accept,
'User-Agent': 'Python/http.client',
})
response = conn.getresponse()
if (response.status, response.reason) == (200, "OK"):
data = response.read()
if uncompress and response.getheader("content-encoding") == "gzip":
zbuf = io.BytesIO(data)
zfile = gzip.GzipFile(mode="rb", fileobj=zbuf)
data = zfile.read()
zfile.close()
if 'text' in mime_accept:
data = data.decode('utf-8')
else:
if raise_exc:
conn.close()
raise Exception(
f"Unable to fetch file {url}, got {response.status} response ({response.reason})"
)
data = None
conn.close()

return data


class TestSVGSamples:
"Tests on misc. sample SVG files included in this test suite."

Expand Down Expand Up @@ -82,22 +116,6 @@ def test_create_pdf_uniconv(self):
class TestWikipediaSymbols:
"Tests on sample symbol SVG files from wikipedia.org."

def fetch_file(self, server, path):
"Fetch file using httplib module."

print(f"downloading https://{server}{path}")

req = HTTPSConnection(server)
req.putrequest('GET', path)
req.putheader('Host', server)
req.putheader('Accept', 'text/svg')
req.endheaders()
r1 = req.getresponse()
data = r1.read().decode('utf-8')
req.close()

return data

def setup_method(self):
"Check if files exists, else download and unpack it."

Expand Down Expand Up @@ -131,7 +149,7 @@ def setup_method(self):
p = join(os.getcwd(), self.folder_path, basename(path))
if not exists(p):
try:
data = self.fetch_file(server, path)
data = fetch_file(f'https://{server}{path}')
except Exception:
print("Check your internet connection and try again!")
break
Expand Down Expand Up @@ -179,31 +197,6 @@ def test_convert_pdf_uniconv(self):
class TestWikipediaFlags:
"Tests using SVG flags from Wikipedia.org."

def fetch_file(self, url, raise_exc=False):
"Get content with some given URL, uncompress if needed."

parsed = urlparse(url)
conn = HTTPSConnection(parsed.netloc)
conn.request("GET", parsed.path)
r1 = conn.getresponse()
if (r1.status, r1.reason) == (200, "OK"):
data = r1.read()
if r1.getheader("content-encoding") == "gzip":
zbuf = io.BytesIO(data)
zfile = gzip.GzipFile(mode="rb", fileobj=zbuf)
data = zfile.read()
zfile.close()
data = data.decode('utf-8')
else:
if raise_exc:
conn.close()
raise Exception(f"Unable to fetch file {url}, got {r1.status} response ({r1.reason})")
else:
data = None
conn.close()

return data

def flag_url2filename(self, url):
"""Convert given flag URL into a local filename.
Expand Down Expand Up @@ -233,7 +226,7 @@ def setup_method(self):
path = join(self.folder_path, "flags.html")
if not exists(path):
u = "https://en.wikipedia.org/wiki/Gallery_of_sovereign_state_flags"
data = self.fetch_file(u)
data = fetch_file(u)
if data:
with open(path, "w", encoding='UTF-8') as f:
f.write(data)
Expand All @@ -254,7 +247,7 @@ def setup_method(self):
for i, fn in enumerate(flag_names):
# load single flag HTML page, like
# https://en.wikipedia.org/wiki/Image:Flag_of_Bhutan.svg
flag_html = self.fetch_file(prefix + quote(fn))
flag_html = fetch_file(prefix + quote(fn))

# search link to single SVG file to download, like
# https://upload.wikimedia.org/wikipedia/commons/9/91/Flag_of_Bhutan.svg
Expand All @@ -276,7 +269,7 @@ def setup_method(self):
path = join(self.folder_path, self.flag_url2filename(flag_url))
if not exists(path):
print(f"fetch {flag_url}")
flag_svg = self.fetch_file(flag_url, raise_exc=True)
flag_svg = fetch_file(flag_url, raise_exc=True)
with open(path, "w", encoding='UTF-8') as f:
f.write(flag_svg)

Expand Down Expand Up @@ -323,7 +316,7 @@ class TestW3CSVG:
def setup_method(self):
"Check if testsuite archive exists, else download and unpack it."

server = "http://www.w3.org"
server = "https://www.w3.org"
path = "/Graphics/SVG/Test/20070907/W3C_SVG_12_TinyTestSuite.tar.gz"
url = server + path

Expand All @@ -335,7 +328,7 @@ def setup_method(self):
if not exists(join(TEST_ROOT, "samples", archive_path)):
print(f"downloading {url}")
try:
data = urlopen(url).read()
data = fetch_file(url, mime_accept='application/gzip', uncompress=False)
except OSError as details:
print(details)
print("Check your internet connection and try again!")
Expand Down

0 comments on commit ec0b143

Please sign in to comment.