mirror of
https://github.com/StevenBlack/hosts.git
synced 2026-10-05 22:05:46 +01:00
Merge pull request #1392 from AlexanderCecile/introduce-requests
Introduce the Requests library
This commit is contained in:
Steven Black
+1
-2
@@ -1,3 +1,2 @@
|
||||
lxml>=4.2.4,<=5.0
|
||||
beautifulsoup4>=4.6.1,<=5.0
|
||||
requests
|
||||
flake8>=3.8,<=4.0
|
||||
|
||||
+32
-100
@@ -17,6 +17,8 @@ import unittest
|
||||
import unittest.mock as mock
|
||||
from io import BytesIO, StringIO
|
||||
|
||||
import requests
|
||||
|
||||
import updateHostsFile
|
||||
from updateHostsFile import (
|
||||
Colors,
|
||||
@@ -1406,77 +1408,6 @@ class TestRemoveOldHostsFile(BaseMockDir):
|
||||
# End File Logic
|
||||
|
||||
|
||||
# Helper Functions
|
||||
def mock_url_open(url):
|
||||
"""
|
||||
Mock of `urlopen` that returns the url in a `BytesIO` stream.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
url : str
|
||||
The URL associated with the file to open.
|
||||
|
||||
Returns
|
||||
-------
|
||||
bytes_stream : BytesIO
|
||||
The `url` input wrapped in a `BytesIO` stream.
|
||||
"""
|
||||
|
||||
return BytesIO(url)
|
||||
|
||||
|
||||
def mock_url_open_fail(_):
|
||||
"""
|
||||
Mock of `urlopen` that fails with an Exception.
|
||||
"""
|
||||
|
||||
raise Exception()
|
||||
|
||||
|
||||
def mock_url_open_read_fail(_):
|
||||
"""
|
||||
Mock of `urlopen` that returns an object that fails on `read`.
|
||||
|
||||
Returns
|
||||
-------
|
||||
file_mock : mock.Mock
|
||||
A mock of a file object that fails when reading.
|
||||
"""
|
||||
|
||||
def fail_read():
|
||||
raise Exception()
|
||||
|
||||
m = mock.Mock()
|
||||
|
||||
m.read = fail_read
|
||||
return m
|
||||
|
||||
|
||||
def mock_url_open_decode_fail(_):
|
||||
"""
|
||||
Mock of `urlopen` that returns an object that fails on during decoding
|
||||
the output of `urlopen`.
|
||||
|
||||
Returns
|
||||
-------
|
||||
file_mock : mock.Mock
|
||||
A mock of a file object that fails when decoding the output.
|
||||
"""
|
||||
|
||||
def fail_decode(_):
|
||||
raise Exception()
|
||||
|
||||
def read():
|
||||
s = mock.Mock()
|
||||
s.decode = fail_decode
|
||||
|
||||
return s
|
||||
|
||||
m = mock.Mock()
|
||||
m.read = read
|
||||
return m
|
||||
|
||||
|
||||
class DomainToIDNA(Base):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(DomainToIDNA, self).__init__(*args, **kwargs)
|
||||
@@ -1616,44 +1547,45 @@ class DomainToIDNA(Base):
|
||||
|
||||
|
||||
class GetFileByUrl(BaseStdout):
|
||||
@mock.patch("updateHostsFile.urlopen", side_effect=mock_url_open)
|
||||
def test_read_url(self, _):
|
||||
url = b"www.google.com"
|
||||
def test_basic(self):
|
||||
raw_resp_content = "hello, ".encode("ascii") + "world".encode("utf-8")
|
||||
resp_obj = requests.Response()
|
||||
resp_obj.__setstate__({"_content": raw_resp_content})
|
||||
|
||||
expected = "www.google.com"
|
||||
actual = get_file_by_url(url, delay=0)
|
||||
expected = "hello, world"
|
||||
|
||||
self.assertEqual(actual, expected)
|
||||
with mock.patch("requests.get", return_value=resp_obj):
|
||||
actual = get_file_by_url("www.test-url.com")
|
||||
|
||||
@mock.patch("updateHostsFile.urlopen", side_effect=mock_url_open_fail)
|
||||
def test_read_url_fail(self, _):
|
||||
url = b"www.google.com"
|
||||
self.assertIsNone(get_file_by_url(url, delay=0))
|
||||
self.assertEqual(expected, actual)
|
||||
|
||||
expected = "Problem getting file:"
|
||||
output = sys.stdout.getvalue()
|
||||
def test_with_idna(self):
|
||||
raw_resp_content = b"www.huala\xc3\xb1e.cl"
|
||||
resp_obj = requests.Response()
|
||||
resp_obj.__setstate__({"_content": raw_resp_content})
|
||||
|
||||
self.assertIn(expected, output)
|
||||
expected = "www.xn--hualae-0wa.cl"
|
||||
|
||||
@mock.patch("updateHostsFile.urlopen", side_effect=mock_url_open_read_fail)
|
||||
def test_read_url_read_fail(self, _):
|
||||
url = b"www.google.com"
|
||||
self.assertIsNone(get_file_by_url(url, delay=0))
|
||||
with mock.patch("requests.get", return_value=resp_obj):
|
||||
actual = get_file_by_url("www.test-url.com")
|
||||
|
||||
expected = "Problem getting file:"
|
||||
output = sys.stdout.getvalue()
|
||||
self.assertEqual(expected, actual)
|
||||
|
||||
self.assertIn(expected, output)
|
||||
def test_connect_unknown_domain(self):
|
||||
test_url = "http://doesnotexist.google.com" # leads to exception: ConnectionError
|
||||
with mock.patch("requests.get", side_effect=requests.exceptions.ConnectionError):
|
||||
return_value = get_file_by_url(test_url)
|
||||
self.assertIsNone(return_value)
|
||||
printed_output = sys.stdout.getvalue()
|
||||
self.assertEqual(printed_output, "Error retrieving data from {}\n".format(test_url))
|
||||
|
||||
@mock.patch("updateHostsFile.urlopen", side_effect=mock_url_open_decode_fail)
|
||||
def test_read_url_decode_fail(self, _):
|
||||
url = b"www.google.com"
|
||||
self.assertIsNone(get_file_by_url(url, delay=0))
|
||||
|
||||
expected = "Problem getting file:"
|
||||
output = sys.stdout.getvalue()
|
||||
|
||||
self.assertIn(expected, output)
|
||||
def test_invalid_url(self):
|
||||
test_url = "http://fe80::5054:ff:fe5a:fc0" # leads to exception: InvalidURL
|
||||
with mock.patch("requests.get", side_effect=requests.exceptions.ConnectionError):
|
||||
return_value = get_file_by_url(test_url)
|
||||
self.assertIsNone(return_value)
|
||||
printed_output = sys.stdout.getvalue()
|
||||
self.assertEqual(printed_output, "Error retrieving data from {}\n".format(test_url))
|
||||
|
||||
|
||||
class TestWriteData(Base):
|
||||
|
||||
+28
-28
@@ -21,17 +21,20 @@ import tempfile
|
||||
import time
|
||||
from glob import glob
|
||||
|
||||
import lxml # noqa: F401
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
# Detecting Python 3 for version-dependent implementations
|
||||
PY3 = sys.version_info >= (3, 0)
|
||||
|
||||
if PY3:
|
||||
from urllib.request import urlopen
|
||||
else:
|
||||
if not PY3:
|
||||
raise Exception("We do not support Python 2 anymore.")
|
||||
|
||||
|
||||
try:
|
||||
import requests
|
||||
except ImportError:
|
||||
raise ImportError("This project's dependencies have changed. The Requests library ("
|
||||
"https://requests.readthedocs.io/en/master/) is now required.")
|
||||
|
||||
|
||||
# Syntactic sugar for "sudo" command in UNIX / Linux
|
||||
if platform.system() == "OpenBSD":
|
||||
SUDO = ["/usr/bin/doas"]
|
||||
@@ -1469,40 +1472,37 @@ def maybe_copy_example_file(file_path):
|
||||
shutil.copyfile(example_file_path, file_path)
|
||||
|
||||
|
||||
def get_file_by_url(url, retries=3, delay=10):
|
||||
def get_file_by_url(url, params=None, **kwargs):
|
||||
"""
|
||||
Get a file data located at a particular URL.
|
||||
Retrieve the contents of the hosts file at the URL, then pass it through domain_to_idna().
|
||||
|
||||
Parameters are passed to the requests.get() function.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
url : str
|
||||
The URL at which to access the data.
|
||||
url : str or bytes
|
||||
URL for the new Request object.
|
||||
params :
|
||||
Dictionary, list of tuples or bytes to send in the query string for the Request.
|
||||
kwargs :
|
||||
Optional arguments that request takes.
|
||||
|
||||
Returns
|
||||
-------
|
||||
url_data : str or None
|
||||
The data retrieved at that URL from the file. Returns None if the
|
||||
attempted retrieval is unsuccessful.
|
||||
|
||||
Note
|
||||
----
|
||||
- BeautifulSoup is used in this case to avoid having to search in which
|
||||
format we have to encode or decode data before parsing it to UTF-8.
|
||||
"""
|
||||
|
||||
while retries:
|
||||
try:
|
||||
with urlopen(url) as f:
|
||||
soup = BeautifulSoup(f.read(), "lxml").get_text()
|
||||
return "\n".join(list(map(domain_to_idna, soup.split("\n"))))
|
||||
except Exception as e:
|
||||
if 'failure in name resolution' in str(e):
|
||||
print('No internet connection! Retrying in {} seconds'.format(delay))
|
||||
time.sleep(delay)
|
||||
retries -= 1
|
||||
continue
|
||||
break
|
||||
print("Problem getting file: ", url)
|
||||
try:
|
||||
req = requests.get(url=url, params=params, **kwargs)
|
||||
except requests.exceptions.RequestException:
|
||||
print("Error retrieving data from {}".format(url))
|
||||
return None
|
||||
|
||||
req.encoding = req.apparent_encoding
|
||||
res_text = "\n".join([domain_to_idna(line) for line in req.text.split("\n")])
|
||||
return res_text
|
||||
|
||||
|
||||
def write_data(f, data):
|
||||
|
||||
Reference in New Issue
Block a user