hosts/testUpdateHostsFile.py

1928 lines
66 KiB
Python

#!/usr/bin/env python
# Script by gfyoung
# https://github.com/gfyoung
#
# Python script for testing updateHostFiles.py
import json
import locale
import os
import re
import shutil
import sys
import tempfile
import unittest
import updateHostsFile
from updateHostsFile import (PY3, Colors, colorize, display_exclusion_options,
domain_to_idna, exclude_domain, flush_dns_cache,
gather_custom_exclusions, get_defaults,
get_file_by_url, is_valid_domain_format,
matches_exclusions, move_hosts_file_into_place,
normalize_rule, path_join_robust, print_failure,
print_success, prompt_for_exclusions,
prompt_for_flush_dns_cache, prompt_for_move,
prompt_for_update, query_yes_no, recursive_glob,
remove_old_hosts_file, strip_rule, supports_color,
update_all_sources, update_readme_data,
update_sources_data, write_data,
write_opening_header)
if PY3:
from io import BytesIO, StringIO
import unittest.mock as mock
unicode = str
else:
from StringIO import StringIO
BytesIO = StringIO
import mock
# Test Helper Objects
class Base(unittest.TestCase):
@staticmethod
def mock_property(name):
return mock.patch(name, new_callable=mock.PropertyMock)
@property
def sep(self):
return "\\" if sys.platform == "win32" else "/"
def assert_called_once(self, mock_method):
if PY3 and sys.version_info < (3, 6):
self.assertEqual(mock_method.call_count, 1)
else:
mock_method.assert_called_once()
class BaseStdout(Base):
def setUp(self):
sys.stdout = StringIO()
def tearDown(self):
sys.stdout.close()
sys.stdout = sys.__stdout__
class BaseMockDir(Base):
@property
def dir_count(self):
return len(os.listdir(self.test_dir))
def setUp(self):
self.test_dir = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.test_dir)
def builtins():
if PY3:
return "builtins"
else:
return "__builtin__"
# End Test Helper Objects
# Project Settings
class TestGetDefaults(Base):
def test_get_defaults(self):
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = "foo"
actual = get_defaults()
expected = {"numberofrules": 0,
"datapath": "foo" + self.sep + "data",
"freshen": True,
"replace": False,
"backup": False,
"skipstatichosts": False,
"keepdomaincomments": False,
"extensionspath": "foo" + self.sep + "extensions",
"extensions": [],
"compress": False,
"minimise": False,
"outputsubfolder": "",
"hostfilename": "hosts",
"targetip": "0.0.0.0",
"sourcedatafilename": "update.json",
"sourcesdata": [],
"readmefilename": "readme.md",
"readmetemplate": ("foo" + self.sep +
"readme_template.md"),
"readmedata": {},
"readmedatafilename": ("foo" + self.sep +
"readmeData.json"),
"exclusionpattern": "([a-zA-Z\d-]+\.){0,}",
"exclusionregexs": [],
"exclusions": [],
"commonexclusions": ["hulu.com"],
"blacklistfile": "foo" + self.sep + "blacklist",
"whitelistfile": "foo" + self.sep + "whitelist"}
self.assertDictEqual(actual, expected)
# End Project Settings
# Prompt the User
class TestPromptForUpdate(BaseStdout, BaseMockDir):
def setUp(self):
BaseStdout.setUp(self)
BaseMockDir.setUp(self)
def test_no_freshen_no_new_file(self):
hosts_file = os.path.join(self.test_dir, "hosts")
hosts_data = "This data should not be overwritten"
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
with open(hosts_file, "w") as f:
f.write(hosts_data)
for update_auto in (False, True):
dir_count = self.dir_count
prompt_for_update(freshen=False, update_auto=update_auto)
output = sys.stdout.getvalue()
self.assertEqual(output, "")
sys.stdout = StringIO()
self.assertEqual(self.dir_count, dir_count)
with open(hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, hosts_data)
def test_no_freshen_new_file(self):
hosts_file = os.path.join(self.test_dir, "hosts")
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
dir_count = self.dir_count
prompt_for_update(freshen=False, update_auto=False)
output = sys.stdout.getvalue()
self.assertEqual(output, "")
sys.stdout = StringIO()
self.assertEqual(self.dir_count, dir_count + 1)
with open(hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, "")
@mock.patch(builtins() + ".open")
def test_no_freshen_fail_new_file(self, mock_open):
for exc in (IOError, OSError):
mock_open.side_effect = exc("failed open")
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
prompt_for_update(freshen=False, update_auto=False)
output = sys.stdout.getvalue()
expected = ("ERROR: No 'hosts' file in the folder. "
"Try creating one manually.")
self.assertIn(expected, output)
sys.stdout = StringIO()
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def test_freshen_no_update(self, _):
hosts_file = os.path.join(self.test_dir, "hosts")
hosts_data = "This data should not be overwritten"
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
with open(hosts_file, "w") as f:
f.write(hosts_data)
dir_count = self.dir_count
update_sources = prompt_for_update(freshen=True, update_auto=False)
self.assertFalse(update_sources)
output = sys.stdout.getvalue()
expected = ("OK, we'll stick with "
"what we've got locally.")
self.assertIn(expected, output)
sys.stdout = StringIO()
self.assertEqual(self.dir_count, dir_count)
with open(hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, hosts_data)
@mock.patch("updateHostsFile.query_yes_no", return_value=True)
def test_freshen_update(self, _):
hosts_file = os.path.join(self.test_dir, "hosts")
hosts_data = "This data should not be overwritten"
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
with open(hosts_file, "w") as f:
f.write(hosts_data)
dir_count = self.dir_count
for update_auto in (False, True):
update_sources = prompt_for_update(freshen=True,
update_auto=update_auto)
self.assertTrue(update_sources)
output = sys.stdout.getvalue()
self.assertEqual(output, "")
sys.stdout = StringIO()
self.assertEqual(self.dir_count, dir_count)
with open(hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, hosts_data)
def tearDown(self):
BaseStdout.tearDown(self)
# BaseStdout.tearDown(self)
class TestPromptForExclusions(BaseStdout):
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testSkipPrompt(self, mock_query):
gather_exclusions = prompt_for_exclusions(skip_prompt=True)
self.assertFalse(gather_exclusions)
output = sys.stdout.getvalue()
self.assertEqual(output, "")
mock_query.assert_not_called()
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testNoSkipPromptNoDisplay(self, mock_query):
gather_exclusions = prompt_for_exclusions(skip_prompt=False)
self.assertFalse(gather_exclusions)
output = sys.stdout.getvalue()
expected = ("OK, we'll only exclude "
"domains in the whitelist.")
self.assertIn(expected, output)
self.assert_called_once(mock_query)
@mock.patch("updateHostsFile.query_yes_no", return_value=True)
def testNoSkipPromptDisplay(self, mock_query):
gather_exclusions = prompt_for_exclusions(skip_prompt=False)
self.assertTrue(gather_exclusions)
output = sys.stdout.getvalue()
self.assertEqual(output, "")
self.assert_called_once(mock_query)
class TestPromptForFlushDnsCache(Base):
@mock.patch("updateHostsFile.flush_dns_cache", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testFlushCache(self, mock_query, mock_flush):
for prompt_flush in (False, True):
prompt_for_flush_dns_cache(flush_cache=True,
prompt_flush=prompt_flush)
mock_query.assert_not_called()
self.assert_called_once(mock_flush)
mock_query.reset_mock()
mock_flush.reset_mock()
@mock.patch("updateHostsFile.flush_dns_cache", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testNoFlushCacheNoPrompt(self, mock_query, mock_flush):
prompt_for_flush_dns_cache(flush_cache=False,
prompt_flush=False)
mock_query.assert_not_called()
mock_flush.assert_not_called()
@mock.patch("updateHostsFile.flush_dns_cache", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testNoFlushCachePromptNoFlush(self, mock_query, mock_flush):
prompt_for_flush_dns_cache(flush_cache=False,
prompt_flush=True)
self.assert_called_once(mock_query)
mock_flush.assert_not_called()
@mock.patch("updateHostsFile.flush_dns_cache", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=True)
def testNoFlushCachePromptFlush(self, mock_query, mock_flush):
prompt_for_flush_dns_cache(flush_cache=False,
prompt_flush=True)
self.assert_called_once(mock_query)
self.assert_called_once(mock_flush)
class TestPromptForMove(Base):
def setUp(self):
Base.setUp(self)
self.final_file = "final.txt"
def prompt_for_move(self, **move_params):
return prompt_for_move(self.final_file, **move_params)
@mock.patch("updateHostsFile.move_hosts_file_into_place", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testSkipStaticHosts(self, mock_query, mock_move):
for replace in (False, True):
for auto in (False, True):
move_file = self.prompt_for_move(replace=replace, auto=auto,
skipstatichosts=True)
self.assertFalse(move_file)
mock_query.assert_not_called()
mock_move.assert_not_called()
mock_query.reset_mock()
mock_move.reset_mock()
@mock.patch("updateHostsFile.move_hosts_file_into_place", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testReplaceNoSkipStaticHosts(self, mock_query, mock_move):
for auto in (False, True):
move_file = self.prompt_for_move(replace=True, auto=auto,
skipstatichosts=False)
self.assertTrue(move_file)
mock_query.assert_not_called()
self.assert_called_once(mock_move)
mock_query.reset_mock()
mock_move.reset_mock()
@mock.patch("updateHostsFile.move_hosts_file_into_place", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testAutoNoSkipStaticHosts(self, mock_query, mock_move):
for replace in (False, True):
move_file = self.prompt_for_move(replace=replace, auto=True,
skipstatichosts=True)
self.assertFalse(move_file)
mock_query.assert_not_called()
mock_move.assert_not_called()
mock_query.reset_mock()
mock_move.reset_mock()
@mock.patch("updateHostsFile.move_hosts_file_into_place", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=False)
def testPromptNoMove(self, mock_query, mock_move):
move_file = self.prompt_for_move(replace=False, auto=False,
skipstatichosts=False)
self.assertFalse(move_file)
self.assert_called_once(mock_query)
mock_move.assert_not_called()
mock_query.reset_mock()
mock_move.reset_mock()
@mock.patch("updateHostsFile.move_hosts_file_into_place", return_value=0)
@mock.patch("updateHostsFile.query_yes_no", return_value=True)
def testPromptMove(self, mock_query, mock_move):
move_file = self.prompt_for_move(replace=False, auto=False,
skipstatichosts=False)
self.assertTrue(move_file)
self.assert_called_once(mock_query)
self.assert_called_once(mock_move)
mock_query.reset_mock()
mock_move.reset_mock()
# End Prompt the User
# Exclusion Logic
class TestDisplayExclusionsOptions(Base):
@mock.patch("updateHostsFile.query_yes_no", return_value=0)
@mock.patch("updateHostsFile.exclude_domain", return_value=None)
@mock.patch("updateHostsFile.gather_custom_exclusions", return_value=None)
def test_no_exclusions(self, mock_gather, mock_exclude, _):
common_exclusions = []
display_exclusion_options(common_exclusions, "foo", [])
mock_gather.assert_not_called()
mock_exclude.assert_not_called()
@mock.patch("updateHostsFile.query_yes_no", side_effect=[1, 1, 0])
@mock.patch("updateHostsFile.exclude_domain", return_value=None)
@mock.patch("updateHostsFile.gather_custom_exclusions", return_value=None)
def test_only_common_exclusions(self, mock_gather, mock_exclude, _):
common_exclusions = ["foo", "bar"]
display_exclusion_options(common_exclusions, "foo", [])
mock_gather.assert_not_called()
exclude_calls = [mock.call("foo", "foo", []),
mock.call("bar", "foo", None)]
mock_exclude.assert_has_calls(exclude_calls)
@mock.patch("updateHostsFile.query_yes_no", side_effect=[0, 0, 1])
@mock.patch("updateHostsFile.exclude_domain", return_value=None)
@mock.patch("updateHostsFile.gather_custom_exclusions", return_value=None)
def test_gather_exclusions(self, mock_gather, mock_exclude, _):
common_exclusions = ["foo", "bar"]
display_exclusion_options(common_exclusions, "foo", [])
mock_exclude.assert_not_called()
self.assert_called_once(mock_gather)
@mock.patch("updateHostsFile.query_yes_no", side_effect=[1, 0, 1])
@mock.patch("updateHostsFile.exclude_domain", return_value=None)
@mock.patch("updateHostsFile.gather_custom_exclusions", return_value=None)
def test_mixture_gather_exclusions(self, mock_gather, mock_exclude, _):
common_exclusions = ["foo", "bar"]
display_exclusion_options(common_exclusions, "foo", [])
mock_exclude.assert_called_once_with("foo", "foo", [])
self.assert_called_once(mock_gather)
class TestGatherCustomExclusions(BaseStdout):
# Can only test in the invalid domain case
# because of the settings global variable.
@mock.patch("updateHostsFile.raw_input", side_effect=["foo", "no"])
@mock.patch("updateHostsFile.is_valid_domain_format", return_value=False)
def test_basic(self, *_):
gather_custom_exclusions("foo", [])
expected = "Do you have more domains you want to enter? [Y/n]"
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("updateHostsFile.raw_input", side_effect=["foo", "yes",
"bar", "no"])
@mock.patch("updateHostsFile.is_valid_domain_format", return_value=False)
def test_multiple(self, *_):
gather_custom_exclusions("foo", [])
expected = ("Do you have more domains you want to enter? [Y/n] "
"Do you have more domains you want to enter? [Y/n]")
output = sys.stdout.getvalue()
self.assertIn(expected, output)
class TestExcludeDomain(Base):
def test_invalid_exclude_domain(self):
exclusion_regexes = []
exclusion_pattern = "*.com"
for domain in ["google.com", "hulu.com", "adaway.org"]:
self.assertRaises(re.error, exclude_domain, domain,
exclusion_pattern, exclusion_regexes)
self.assertListEqual(exclusion_regexes, [])
def test_valid_exclude_domain(self):
exp_count = 0
expected_regexes = []
exclusion_regexes = []
exclusion_pattern = "[a-z]\."
for domain in ["google.com", "hulu.com", "adaway.org"]:
self.assertEqual(len(exclusion_regexes), exp_count)
exclusion_regexes = exclude_domain(domain, exclusion_pattern,
exclusion_regexes)
expected_regex = re.compile(exclusion_pattern + domain)
expected_regexes.append(expected_regex)
exp_count += 1
self.assertEqual(len(exclusion_regexes), exp_count)
self.assertListEqual(exclusion_regexes, expected_regexes)
class TestMatchesExclusions(Base):
def test_no_match_empty_list(self):
exclusion_regexes = []
for domain in ["1.2.3.4 localhost", "5.6.7.8 hulu.com",
"9.1.2.3 yahoo.com", "4.5.6.7 cloudfront.net"]:
self.assertFalse(matches_exclusions(domain, exclusion_regexes))
def test_no_match_list(self):
exclusion_regexes = [".*\.org", ".*\.edu"]
exclusion_regexes = [re.compile(regex) for regex in exclusion_regexes]
for domain in ["1.2.3.4 localhost", "5.6.7.8 hulu.com",
"9.1.2.3 yahoo.com", "4.5.6.7 cloudfront.net"]:
self.assertFalse(matches_exclusions(domain, exclusion_regexes))
def test_match_list(self):
exclusion_regexes = [".*\.com", ".*\.org", ".*\.edu"]
exclusion_regexes = [re.compile(regex) for regex in exclusion_regexes]
for domain in ["5.6.7.8 hulu.com", "9.1.2.3 yahoo.com",
"4.5.6.7 adaway.org", "8.9.1.2 education.edu"]:
self.assertTrue(matches_exclusions(domain, exclusion_regexes))
# End Exclusion Logic
# Update Logic
class TestUpdateSourcesData(Base):
def setUp(self):
Base.setUp(self)
self.data_path = "data"
self.extensions_path = "extensions"
self.source_data_filename = "update.json"
self.update_kwargs = dict(datapath=self.data_path,
extensionspath=self.extensions_path,
sourcedatafilename=self.source_data_filename)
def update_sources_data(self, sources_data, extensions):
return update_sources_data(sources_data[:], extensions=extensions,
**self.update_kwargs)
@mock.patch("updateHostsFile.recursive_glob", return_value=[])
@mock.patch("updateHostsFile.path_join_robust", return_value="dirpath")
@mock.patch(builtins() + ".open", return_value=mock.Mock())
def test_no_update(self, mock_open, mock_join_robust, _):
extensions = []
sources_data = [{"source": "source1.txt"}, {"source": "source2.txt"}]
new_sources_data = self.update_sources_data(sources_data, extensions)
self.assertEqual(new_sources_data, sources_data)
mock_join_robust.assert_not_called()
mock_open.assert_not_called()
extensions = [".json", ".txt"]
new_sources_data = self.update_sources_data(sources_data, extensions)
self.assertEqual(new_sources_data, sources_data)
join_calls = [mock.call(self.extensions_path, ".json"),
mock.call(self.extensions_path, ".txt")]
mock_join_robust.assert_has_calls(join_calls)
mock_open.assert_not_called()
@mock.patch("updateHostsFile.recursive_glob",
side_effect=[[], ["update1.txt", "update2.txt"]])
@mock.patch("json.load", return_value={"mock_source": "mock_source.ext"})
@mock.patch(builtins() + ".open", return_value=mock.Mock())
@mock.patch("updateHostsFile.path_join_robust", return_value="dirpath")
def test_update_only_extensions(self, mock_join_robust, *_):
extensions = [".json"]
sources_data = [{"source": "source1.txt"}, {"source": "source2.txt"}]
new_sources_data = self.update_sources_data(sources_data, extensions)
expected = sources_data + [{"mock_source": "mock_source.ext"}] * 2
self.assertEqual(new_sources_data, expected)
self.assert_called_once(mock_join_robust)
@mock.patch("updateHostsFile.recursive_glob",
side_effect=[["update1.txt", "update2.txt"],
["update3.txt", "update4.txt"]])
@mock.patch("json.load", side_effect=[{"mock_source": "mock_source.txt"},
{"mock_source": "mock_source2.txt"},
{"mock_source": "mock_source3.txt"},
{"mock_source": "mock_source4.txt"}])
@mock.patch(builtins() + ".open", return_value=mock.Mock())
@mock.patch("updateHostsFile.path_join_robust", return_value="dirpath")
def test_update_both_pathways(self, mock_join_robust, *_):
extensions = [".json"]
sources_data = [{"source": "source1.txt"}, {"source": "source2.txt"}]
new_sources_data = self.update_sources_data(sources_data, extensions)
expected = sources_data + [{"mock_source": "mock_source.txt"},
{"mock_source": "mock_source2.txt"},
{"mock_source": "mock_source3.txt"},
{"mock_source": "mock_source4.txt"}]
self.assertEqual(new_sources_data, expected)
self.assert_called_once(mock_join_robust)
class TestUpdateAllSources(BaseStdout):
def setUp(self):
BaseStdout.setUp(self)
self.source_data_filename = "data.json"
self.host_filename = "hosts.txt"
@mock.patch(builtins() + ".open")
@mock.patch("updateHostsFile.recursive_glob", return_value=[])
def test_no_sources(self, _, mock_open):
update_all_sources(self.source_data_filename, self.host_filename)
mock_open.assert_not_called()
@mock.patch(builtins() + ".open", return_value=mock.Mock())
@mock.patch("json.load", return_value={"url": "example.com"})
@mock.patch("updateHostsFile.recursive_glob", return_value=["foo"])
@mock.patch("updateHostsFile.write_data", return_value=0)
@mock.patch("updateHostsFile.get_file_by_url", return_value="file_data")
def test_one_source(self, mock_get, mock_write, *_):
update_all_sources(self.source_data_filename, self.host_filename)
self.assert_called_once(mock_write)
self.assert_called_once(mock_get)
output = sys.stdout.getvalue()
expected = "Updating source from example.com"
self.assertIn(expected, output)
@mock.patch(builtins() + ".open", return_value=mock.Mock())
@mock.patch("json.load", return_value={"url": "example.com"})
@mock.patch("updateHostsFile.recursive_glob", return_value=["foo"])
@mock.patch("updateHostsFile.write_data", return_value=0)
@mock.patch("updateHostsFile.get_file_by_url",
return_value=Exception("fail"))
def test_source_fail(self, mock_get, mock_write, *_):
update_all_sources(self.source_data_filename, self.host_filename)
mock_write.assert_not_called()
self.assert_called_once(mock_get)
output = sys.stdout.getvalue()
expecteds = ["Updating source from example.com",
"Error in updating source: example.com"]
for expected in expecteds:
self.assertIn(expected, output)
@mock.patch(builtins() + ".open", return_value=mock.Mock())
@mock.patch("json.load", side_effect=[{"url": "example.com"},
{"url": "example2.com"}])
@mock.patch("updateHostsFile.recursive_glob", return_value=["foo", "bar"])
@mock.patch("updateHostsFile.write_data", return_value=0)
@mock.patch("updateHostsFile.get_file_by_url",
side_effect=[Exception("fail"), "file_data"])
def test_sources_fail_succeed(self, mock_get, mock_write, *_):
update_all_sources(self.source_data_filename, self.host_filename)
self.assert_called_once(mock_write)
get_calls = [mock.call("example.com"), mock.call("example2.com")]
mock_get.assert_has_calls(get_calls)
output = sys.stdout.getvalue()
expecteds = ["Updating source from example.com",
"Error in updating source: example.com",
"Updating source from example2.com"]
for expected in expecteds:
self.assertIn(expected, output)
# End Update Logic
# File Logic
class TestNormalizeRule(BaseStdout):
def test_no_match(self):
kwargs = dict(target_ip="0.0.0.0", keep_domain_comments=False)
for rule in ["foo", "128.0.0.1", "bar.com/usa", "0.0.0 google",
"0.1.2.3.4 foo/bar", "twitter.com"]:
self.assertEqual(normalize_rule(rule, **kwargs), (None, None))
output = sys.stdout.getvalue()
sys.stdout = StringIO()
expected = "==>" + rule + "<=="
self.assertIn(expected, output)
def test_no_comments(self):
for target_ip in ("0.0.0.0", "127.0.0.1", "8.8.8.8"):
rule = "127.0.0.1 1.google.com foo"
expected = ("1.google.com", str(target_ip) + " 1.google.com\n")
actual = normalize_rule(rule, target_ip=target_ip,
keep_domain_comments=False)
self.assertEqual(actual, expected)
# Nothing gets printed if there's a match.
output = sys.stdout.getvalue()
self.assertEqual(output, "")
sys.stdout = StringIO()
def test_with_comments(self):
for target_ip in ("0.0.0.0", "127.0.0.1", "8.8.8.8"):
for comment in ("foo", "bar", "baz"):
rule = "127.0.0.1 1.google.co.uk " + comment
expected = ("1.google.co.uk",
(str(target_ip) + " 1.google.co.uk # " +
comment + "\n"))
actual = normalize_rule(rule, target_ip=target_ip,
keep_domain_comments=True)
self.assertEqual(actual, expected)
# Nothing gets printed if there's a match.
output = sys.stdout.getvalue()
self.assertEqual(output, "")
sys.stdout = StringIO()
def test_two_ips(self):
for target_ip in ("0.0.0.0", "127.0.0.1", "8.8.8.8"):
rule = "127.0.0.1 11.22.33.44 foo"
expected = ("11.22.33.44", str(target_ip) + " 11.22.33.44\n")
actual = normalize_rule(rule, target_ip=target_ip,
keep_domain_comments=False)
self.assertEqual(actual, expected)
# Nothing gets printed if there's a match.
output = sys.stdout.getvalue()
self.assertEqual(output, "")
sys.stdout = StringIO()
class TestStripRule(Base):
def test_strip_empty(self):
for line in ["0.0.0.0", "domain.com", "foo"]:
output = strip_rule(line)
self.assertEqual(output, "")
def test_strip_exactly_two(self):
for line in ["0.0.0.0 twitter.com", "127.0.0.1 facebook.com",
"8.8.8.8 google.com", "1.2.3.4 foo.bar.edu"]:
output = strip_rule(line)
self.assertEqual(output, line)
def test_strip_more_than_two(self):
for line in ["0.0.0.0 twitter.com", "127.0.0.1 facebook.com",
"8.8.8.8 google.com", "1.2.3.4 foo.bar.edu"]:
output = strip_rule(line + " # comments here galore")
self.assertEqual(output, line)
class TestWriteOpeningHeader(BaseMockDir):
def setUp(self):
super(TestWriteOpeningHeader, self).setUp()
self.final_file = BytesIO()
def test_missing_keyword(self):
kwargs = dict(extensions="", outputsubfolder="",
numberofrules=5, skipstatichosts=False)
for k in kwargs.keys():
bad_kwargs = kwargs.copy()
bad_kwargs.pop(k)
self.assertRaises(KeyError, write_opening_header,
self.final_file, **bad_kwargs)
def test_basic(self):
kwargs = dict(extensions="", outputsubfolder="",
numberofrules=5, skipstatichosts=True)
write_opening_header(self.final_file, **kwargs)
contents = self.final_file.getvalue()
contents = contents.decode("UTF-8")
# Expected contents.
for expected in (
"# This hosts file is a merged collection",
"# with a dash of crowd sourcing via Github",
"# Number of unique domains: {count}".format(
count=kwargs["numberofrules"]),
"Fetch the latest version of this file:",
"Project home page: https://github.com/StevenBlack/hosts",
):
self.assertIn(expected, contents)
# Expected non-contents.
for expected in (
"# Extensions added to this file:",
"127.0.0.1 localhost",
"127.0.0.1 local",
"127.0.0.53",
"127.0.1.1",
):
self.assertNotIn(expected, contents)
def test_basic_include_static_hosts(self):
kwargs = dict(extensions="", outputsubfolder="",
numberofrules=5, skipstatichosts=False)
with self.mock_property("platform.system") as obj:
obj.return_value = "Windows"
write_opening_header(self.final_file, **kwargs)
contents = self.final_file.getvalue()
contents = contents.decode("UTF-8")
# Expected contents.
for expected in (
"127.0.0.1 local",
"127.0.0.1 localhost",
"# This hosts file is a merged collection",
"# with a dash of crowd sourcing via Github",
"# Number of unique domains: {count}".format(
count=kwargs["numberofrules"]),
"Fetch the latest version of this file:",
"Project home page: https://github.com/StevenBlack/hosts",
):
self.assertIn(expected, contents)
# Expected non-contents.
for expected in (
"# Extensions added to this file:",
"127.0.0.53",
"127.0.1.1",
):
self.assertNotIn(expected, contents)
def test_basic_include_static_hosts_linux(self):
kwargs = dict(extensions="", outputsubfolder="",
numberofrules=5, skipstatichosts=False)
with self.mock_property("platform.system") as system:
system.return_value = "Linux"
with self.mock_property("socket.gethostname") as hostname:
hostname.return_value = "steven-hosts"
write_opening_header(self.final_file, **kwargs)
contents = self.final_file.getvalue()
contents = contents.decode("UTF-8")
# Expected contents.
for expected in (
"127.0.1.1",
"127.0.0.53",
"steven-hosts",
"127.0.0.1 local",
"127.0.0.1 localhost",
"# This hosts file is a merged collection",
"# with a dash of crowd sourcing via Github",
"# Number of unique domains: {count}".format(
count=kwargs["numberofrules"]),
"Fetch the latest version of this file:",
"Project home page: https://github.com/StevenBlack/hosts",
):
self.assertIn(expected, contents)
# Expected non-contents.
expected = "# Extensions added to this file:"
self.assertNotIn(expected, contents)
def test_extensions(self):
kwargs = dict(extensions=["epsilon", "gamma", "mu", "phi"],
outputsubfolder="", numberofrules=5,
skipstatichosts=True)
write_opening_header(self.final_file, **kwargs)
contents = self.final_file.getvalue()
contents = contents.decode("UTF-8")
# Expected contents.
for expected in (
", ".join(kwargs["extensions"]),
"# Extensions added to this file:",
"# This hosts file is a merged collection",
"# with a dash of crowd sourcing via Github",
"# Number of unique domains: {count}".format(
count=kwargs["numberofrules"]),
"Fetch the latest version of this file:",
"Project home page: https://github.com/StevenBlack/hosts",
):
self.assertIn(expected, contents)
# Expected non-contents.
for expected in (
"127.0.0.1 localhost",
"127.0.0.1 local",
"127.0.0.53",
"127.0.1.1",
):
self.assertNotIn(expected, contents)
def test_no_preamble(self):
# We should not even attempt to read this, as it is a directory.
hosts_dir = os.path.join(self.test_dir, "myhosts")
os.mkdir(hosts_dir)
kwargs = dict(extensions="", outputsubfolder="",
numberofrules=5, skipstatichosts=True)
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
write_opening_header(self.final_file, **kwargs)
contents = self.final_file.getvalue()
contents = contents.decode("UTF-8")
# Expected contents.
for expected in (
"# This hosts file is a merged collection",
"# with a dash of crowd sourcing via Github",
"# Number of unique domains: {count}".format(
count=kwargs["numberofrules"]),
"Fetch the latest version of this file:",
"Project home page: https://github.com/StevenBlack/hosts",
):
self.assertIn(expected, contents)
# Expected non-contents.
for expected in (
"# Extensions added to this file:",
"127.0.0.1 localhost",
"127.0.0.1 local",
"127.0.0.53",
"127.0.1.1",
):
self.assertNotIn(expected, contents)
def test_preamble(self):
hosts_file = os.path.join(self.test_dir, "myhosts")
with open(hosts_file, "w") as f:
f.write("peter-piper-picked-a-pepper")
kwargs = dict(extensions="", outputsubfolder="",
numberofrules=5, skipstatichosts=True)
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
write_opening_header(self.final_file, **kwargs)
contents = self.final_file.getvalue()
contents = contents.decode("UTF-8")
# Expected contents.
for expected in (
"peter-piper-picked-a-pepper",
"# This hosts file is a merged collection",
"# with a dash of crowd sourcing via Github",
"# Number of unique domains: {count}".format(
count=kwargs["numberofrules"]),
"Fetch the latest version of this file:",
"Project home page: https://github.com/StevenBlack/hosts",
):
self.assertIn(expected, contents)
# Expected non-contents.
for expected in (
"# Extensions added to this file:",
"127.0.0.1 localhost",
"127.0.0.1 local",
"127.0.0.53",
"127.0.1.1",
):
self.assertNotIn(expected, contents)
def tearDown(self):
super(TestWriteOpeningHeader, self).tearDown()
self.final_file.close()
class TestUpdateReadmeData(BaseMockDir):
def setUp(self):
super(TestUpdateReadmeData, self).setUp()
self.readme_file = os.path.join(self.test_dir, "readmeData.json")
def test_missing_keyword(self):
kwargs = dict(extensions="", outputsubfolder="",
numberofrules="", sourcesdata="")
for k in kwargs.keys():
bad_kwargs = kwargs.copy()
bad_kwargs.pop(k)
self.assertRaises(KeyError, update_readme_data,
self.readme_file, **bad_kwargs)
def test_add_fields(self):
with open(self.readme_file, "w") as f:
json.dump({"foo": "bar"}, f)
kwargs = dict(extensions=None, outputsubfolder="foo",
numberofrules=5, sourcesdata="hosts")
update_readme_data(self.readme_file, **kwargs)
expected = {
"base": {
"location": "foo" + self.sep,
"sourcesdata": "hosts",
"entries": 5,
},
"foo": "bar"
}
with open(self.readme_file, "r") as f:
actual = json.load(f)
self.assertEqual(actual, expected)
def test_modify_fields(self):
with open(self.readme_file, "w") as f:
json.dump({"base": "soprano"}, f)
kwargs = dict(extensions=None, outputsubfolder="foo",
numberofrules=5, sourcesdata="hosts")
update_readme_data(self.readme_file, **kwargs)
expected = {
"base": {
"location": "foo" + self.sep,
"sourcesdata": "hosts",
"entries": 5,
}
}
with open(self.readme_file, "r") as f:
actual = json.load(f)
self.assertEqual(actual, expected)
def test_set_extensions(self):
with open(self.readme_file, "w") as f:
json.dump({}, f)
kwargs = dict(extensions=["com", "org"], outputsubfolder="foo",
numberofrules=5, sourcesdata="hosts")
update_readme_data(self.readme_file, **kwargs)
expected = {
"com-org": {
"location": "foo" + self.sep,
"sourcesdata": "hosts",
"entries": 5,
}
}
with open(self.readme_file, "r") as f:
actual = json.load(f)
self.assertEqual(actual, expected)
class TestMoveHostsFile(BaseStdout):
@mock.patch("os.path.abspath", side_effect=lambda f: f)
def test_move_hosts_no_name(self, _):
with self.mock_property("os.name"):
os.name = "foo"
mock_file = mock.Mock(name="foo")
move_hosts_file_into_place(mock_file)
expected = ""
output = sys.stdout.getvalue()
self.assertEqual(output, expected)
@mock.patch("os.path.abspath", side_effect=lambda f: f)
def test_move_hosts_windows(self, _):
with self.mock_property("os.name"):
os.name = "nt"
mock_file = mock.Mock(name="foo")
move_hosts_file_into_place(mock_file)
expected = ("Automatically moving the hosts "
"file in place is not yet supported.\n"
"Please move the generated file to "
"%SystemRoot%\system32\drivers\etc\hosts")
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("os.path.abspath", side_effect=lambda f: f)
@mock.patch("subprocess.call", return_value=0)
def test_move_hosts_posix(self, *_):
with self.mock_property("os.name"):
os.name = "posix"
mock_file = mock.Mock(name="foo")
move_hosts_file_into_place(mock_file)
expected = ("Moving the file requires administrative "
"privileges. You might need to enter your password.")
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("os.path.abspath", side_effect=lambda f: f)
@mock.patch("subprocess.call", return_value=1)
def test_move_hosts_posix_fail(self, *_):
with self.mock_property("os.name"):
os.name = "posix"
mock_file = mock.Mock(name="foo")
move_hosts_file_into_place(mock_file)
expected = "Moving the file failed."
output = sys.stdout.getvalue()
self.assertIn(expected, output)
class TestFlushDnsCache(BaseStdout):
@mock.patch("subprocess.call", return_value=0)
def test_flush_darwin(self, _):
with self.mock_property("platform.system") as obj:
obj.return_value = "Darwin"
flush_dns_cache()
expected = ("Flushing the DNS cache to utilize new hosts "
"file...\nFlushing the DNS cache requires "
"administrative privileges. You might need to "
"enter your password.")
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("subprocess.call", return_value=1)
def test_flush_darwin_fail(self, _):
with self.mock_property("platform.system") as obj:
obj.return_value = "Darwin"
flush_dns_cache()
expected = "Flushing the DNS cache failed."
output = sys.stdout.getvalue()
self.assertIn(expected, output)
def test_flush_windows(self):
with self.mock_property("platform.system") as obj:
obj.return_value = "win32"
with self.mock_property("os.name"):
os.name = "nt"
flush_dns_cache()
expected = ("Automatically flushing the DNS cache is "
"not yet supported.\nPlease copy and paste "
"the command 'ipconfig /flushdns' in "
"administrator command prompt after running "
"this script.")
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("os.path.isfile", return_value=False)
def test_flush_no_tool(self, _):
with self.mock_property("platform.system") as obj:
obj.return_value = "Linux"
with self.mock_property("os.name"):
os.name = "posix"
flush_dns_cache()
expected = "Unable to determine DNS management tool."
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("os.path.isfile", side_effect=[True] + [False] * 10)
@mock.patch("subprocess.call", return_value=0)
def test_flush_posix(self, *_):
with self.mock_property("platform.system") as obj:
obj.return_value = "Linux"
with self.mock_property("os.name"):
os.name = "posix"
flush_dns_cache()
expected = ("Flushing the DNS cache by "
"restarting nscd succeeded")
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("os.path.isfile", side_effect=[True] + [False] * 10)
@mock.patch("subprocess.call", return_value=1)
def test_flush_posix_fail(self, *_):
with self.mock_property("platform.system") as obj:
obj.return_value = "Linux"
with self.mock_property("os.name"):
os.name = "posix"
flush_dns_cache()
expected = ("Flushing the DNS cache by "
"restarting nscd failed")
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@mock.patch("os.path.isfile", side_effect=[True, False,
True] + [False] * 10)
@mock.patch("subprocess.call", side_effect=[1, 0])
def test_flush_posix_fail_then_succeed(self, *_):
with self.mock_property("platform.system") as obj:
obj.return_value = "Linux"
with self.mock_property("os.name"):
os.name = "posix"
flush_dns_cache()
output = sys.stdout.getvalue()
for expected in [("Flushing the DNS cache by "
"restarting nscd failed"),
("Flushing the DNS cache by restarting "
"NetworkManager.service succeeded")]:
self.assertIn(expected, output)
def mock_path_join_robust(*args):
# We want to hard-code the backup hosts filename
# instead of parametrizing based on current time.
if len(args) == 2 and args[1].startswith("hosts-"):
return os.path.join(args[0], "hosts-new")
else:
return os.path.join(*args)
class TestRemoveOldHostsFile(BaseMockDir):
def setUp(self):
super(TestRemoveOldHostsFile, self).setUp()
self.hosts_file = os.path.join(self.test_dir, "hosts")
def test_remove_hosts_file(self):
old_dir_count = self.dir_count
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
remove_old_hosts_file(backup=False)
new_dir_count = old_dir_count + 1
self.assertEqual(self.dir_count, new_dir_count)
with open(self.hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, "")
def test_remove_hosts_file_exists(self):
with open(self.hosts_file, "w") as f:
f.write("foo")
old_dir_count = self.dir_count
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
remove_old_hosts_file(backup=False)
new_dir_count = old_dir_count
self.assertEqual(self.dir_count, new_dir_count)
with open(self.hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, "")
@mock.patch("updateHostsFile.path_join_robust",
side_effect=mock_path_join_robust)
def test_remove_hosts_file_backup(self, _):
with open(self.hosts_file, "w") as f:
f.write("foo")
old_dir_count = self.dir_count
with self.mock_property("updateHostsFile.BASEDIR_PATH"):
updateHostsFile.BASEDIR_PATH = self.test_dir
remove_old_hosts_file(backup=True)
new_dir_count = old_dir_count + 1
self.assertEqual(self.dir_count, new_dir_count)
with open(self.hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, "")
new_hosts_file = self.hosts_file + "-new"
with open(new_hosts_file, "r") as f:
contents = f.read()
self.assertEqual(contents, "foo")
# 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)
self.domains = [b'\xc9\xa2oogle.com', b'www.huala\xc3\xb1e.cl']
self.expected_domains = ['xn--oogle-wmc.com', 'www.xn--hualae-0wa.cl']
def test_empty_line(self):
data = ["", "\r", "\n"]
for empty in data:
expected = empty
actual = domain_to_idna(empty)
self.assertEqual(actual, expected)
def test_commented_line(self):
data = "# Hello World"
expected = data
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
def test_simple_line(self):
# Test with a space as separator.
for i in range(len(self.domains)):
data = (b"0.0.0.0 " + self.domains[i]).decode('utf-8')
expected = "0.0.0.0 " + self.expected_domains[i]
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
# Test with a tabulation as separator.
for i in range(len(self.domains)):
data = (b"0.0.0.0\t" + self.domains[i]).decode('utf-8')
expected = "0.0.0.0\t" + self.expected_domains[i]
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
def test_multiple_space_as_separator(self):
# Test with multiple space as separator.
for i in range(len(self.domains)):
data = (b"0.0.0.0 " + self.domains[i]).decode('utf-8')
expected = "0.0.0.0 " + self.expected_domains[i]
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
def test_multiple_tabs_as_separator(self):
# Test with multiple tabls as separator.
for i in range(len(self.domains)):
data = (b"0.0.0.0\t\t\t\t\t\t" + self.domains[i]).decode('utf-8')
expected = "0.0.0.0\t\t\t\t\t\t" + self.expected_domains[i]
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
def test_line_with_comment_at_the_end(self):
# Test with a space as separator.
for i in range(len(self.domains)):
data = (b"0.0.0.0 " + self.domains[i] + b" # Hello World") \
.decode('utf-8')
expected = "0.0.0.0 " + self.expected_domains[i] + " # Hello World"
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
# Test with a tabulation as separator.
for i in range(len(self.domains)):
data = (b"0.0.0.0\t" + self.domains[i] + b" # Hello World") \
.decode('utf-8')
expected = "0.0.0.0\t" + self.expected_domains[i] + \
" # Hello World"
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
# Test with tabulation as separator of domain and comment.
for i in range(len(self.domains)):
data = (b"0.0.0.0\t" + self.domains[i] + b"\t # Hello World") \
.decode('utf-8')
expected = "0.0.0.0\t" + self.expected_domains[i] + \
"\t # Hello World"
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
# Test with space as separator of domain and tabulation as separator
# of comments.
for i in range(len(self.domains)):
data = (b"0.0.0.0 " + self.domains[i] + b" \t # Hello World") \
.decode('utf-8')
expected = "0.0.0.0 " + self.expected_domains[i] + \
" \t # Hello World"
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
# Test with multiple space as seprator of domain and space and
# tabulation as separator or comments.
for i in range(len(self.domains)):
data = (b"0.0.0.0 " + self.domains[i] + b" \t # Hello World") \
.decode('utf-8')
expected = "0.0.0.0 " + self.expected_domains[i] + \
" \t # Hello World"
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
# Test with multiple tabulations as seprator of domain and space and
# tabulation as separator or comments.
for i in range(len(self.domains)):
data = (b"0.0.0.0\t\t\t" +
self.domains[i] +
b" \t # Hello World") \
.decode('utf-8')
expected = "0.0.0.0\t\t\t" + self.expected_domains[i] + \
" \t # Hello World"
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
def test_line_without_prefix(self):
for i in range(len(self.domains)):
data = self.domains[i].decode('utf-8')
expected = self.expected_domains[i]
actual = domain_to_idna(data)
self.assertEqual(actual, expected)
class GetFileByUrl(BaseStdout):
@mock.patch("updateHostsFile.urlopen",
side_effect=mock_url_open)
def test_read_url(self, _):
url = b"www.google.com"
expected = "www.google.com"
actual = get_file_by_url(url)
self.assertEqual(actual, expected)
@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))
expected = "Problem getting file:"
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@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))
expected = "Problem getting file:"
output = sys.stdout.getvalue()
self.assertIn(expected, output)
@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))
expected = "Problem getting file:"
output = sys.stdout.getvalue()
self.assertIn(expected, output)
class TestWriteData(Base):
def test_write_basic(self):
f = BytesIO()
data = "foo"
write_data(f, data)
expected = b"foo"
actual = f.getvalue()
self.assertEqual(actual, expected)
def test_write_unicode(self):
f = BytesIO()
data = u"foo"
write_data(f, data)
expected = b"foo"
actual = f.getvalue()
self.assertEqual(actual, expected)
class TestQueryYesOrNo(BaseStdout):
def test_invalid_default(self):
for invalid_default in ["foo", "bar", "baz", 1, 2, 3]:
self.assertRaises(ValueError, query_yes_no, "?", invalid_default)
@mock.patch("updateHostsFile.raw_input", side_effect=["yes"] * 3)
def test_valid_default(self, _):
for valid_default, expected in [(None, "[y/n]"), ("yes", "[Y/n]"),
("no", "[y/N]")]:
self.assertTrue(query_yes_no("?", valid_default))
output = sys.stdout.getvalue()
sys.stdout = StringIO()
self.assertIn(expected, output)
@mock.patch("updateHostsFile.raw_input", side_effect=([""] * 2))
def test_use_valid_default(self, _):
for valid_default in ["yes", "no"]:
expected = (valid_default == "yes")
actual = query_yes_no("?", valid_default)
self.assertEqual(actual, expected)
@mock.patch("updateHostsFile.raw_input", side_effect=["no", "NO", "N",
"n", "No", "nO"])
def test_valid_no(self, _):
self.assertFalse(query_yes_no("?", None))
@mock.patch("updateHostsFile.raw_input", side_effect=["yes", "YES", "Y",
"yeS", "y", "YeS",
"yES", "YEs"])
def test_valid_yes(self, _):
self.assertTrue(query_yes_no("?", None))
@mock.patch("updateHostsFile.raw_input", side_effect=["foo", "yes",
"foo", "no"])
def test_invalid_then_valid(self, _):
expected = "Please respond with 'yes' or 'no'"
# The first time, we respond "yes"
self.assertTrue(query_yes_no("?", None))
output = sys.stdout.getvalue()
self.assertIn(expected, output)
sys.stdout = StringIO()
# The second time, we respond "no"
self.assertFalse(query_yes_no("?", None))
output = sys.stdout.getvalue()
self.assertIn(expected, output)
class TestIsValidDomainFormat(BaseStdout):
def test_empty_domain(self):
self.assertFalse(is_valid_domain_format(""))
output = sys.stdout.getvalue()
expected = "You didn't enter a domain. Try again."
self.assertTrue(expected in output)
def test_invalid_domain(self):
expected = ("Do not include www.domain.com or "
"http(s)://domain.com. Try again.")
for invalid_domain in ["www.subdomain.domain", "https://github.com",
"http://www.google.com"]:
self.assertFalse(is_valid_domain_format(invalid_domain))
output = sys.stdout.getvalue()
sys.stdout = StringIO()
self.assertIn(expected, output)
def test_valid_domain(self):
for valid_domain in ["github.com", "travis.org", "twitter.com"]:
self.assertTrue(is_valid_domain_format(valid_domain))
output = sys.stdout.getvalue()
sys.stdout = StringIO()
self.assertEqual(output, "")
def mock_walk(stem):
"""
Mock method for `os.walk`.
Please refer to the documentation of `os.walk` for information about
the provided parameters.
"""
files = ["foo.txt", "bar.bat", "baz.py", "foo/foo.c", "foo/bar.doc",
"foo/baz/foo.py", "bar/foo/baz.c", "bar/bar/foo.bat"]
if stem == ".":
stem = ""
matches = []
for f in files:
if not stem or f.startswith(stem + "/"):
matches.append(("", "_", [f]))
return matches
class TestRecursiveGlob(Base):
@staticmethod
def sorted_recursive_glob(stem, file_pattern):
actual = recursive_glob(stem, file_pattern)
actual.sort()
return actual
@mock.patch("os.walk", side_effect=mock_walk)
def test_all_match(self, _):
with self.mock_property("sys.version_info"):
sys.version_info = (2, 6)
expected = ["bar.bat", "bar/bar/foo.bat",
"bar/foo/baz.c", "baz.py",
"foo.txt", "foo/bar.doc",
"foo/baz/foo.py", "foo/foo.c"]
actual = self.sorted_recursive_glob("*", "*")
self.assertListEqual(actual, expected)
expected = ["bar/bar/foo.bat", "bar/foo/baz.c"]
actual = self.sorted_recursive_glob("bar", "*")
self.assertListEqual(actual, expected)
expected = ["foo/bar.doc", "foo/baz/foo.py", "foo/foo.c"]
actual = self.sorted_recursive_glob("foo", "*")
self.assertListEqual(actual, expected)
@mock.patch("os.walk", side_effect=mock_walk)
def test_file_ending(self, _):
with self.mock_property("sys.version_info"):
sys.version_info = (2, 6)
expected = ["foo/baz/foo.py"]
actual = self.sorted_recursive_glob("foo", "*.py")
self.assertListEqual(actual, expected)
expected = ["bar/foo/baz.c", "foo/foo.c"]
actual = self.sorted_recursive_glob("*", "*.c")
self.assertListEqual(actual, expected)
expected = []
actual = self.sorted_recursive_glob("*", ".xlsx")
self.assertListEqual(actual, expected)
def mock_path_join(*_):
"""
Mock method for `os.path.join`.
Please refer to the documentation of `os.path.join` for information about
the provided parameters.
"""
raise UnicodeDecodeError("foo", b"", 1, 5, "foo")
class TestPathJoinRobust(Base):
def test_basic(self):
expected = "path1"
actual = path_join_robust("path1")
self.assertEqual(actual, expected)
actual = path_join_robust(u"path1")
self.assertEqual(actual, expected)
def test_join(self):
for i in range(1, 4):
paths = ["pathNew"] * i
expected = "path1" + (self.sep + "pathNew") * i
actual = path_join_robust("path1", *paths)
self.assertEqual(actual, expected)
def test_join_unicode(self):
for i in range(1, 4):
paths = [u"pathNew"] * i
expected = "path1" + (self.sep + "pathNew") * i
actual = path_join_robust("path1", *paths)
self.assertEqual(actual, expected)
@mock.patch("os.path.join", side_effect=mock_path_join)
def test_join_error(self, _):
self.assertRaises(locale.Error, path_join_robust, "path")
# Colors
class TestSupportsColor(BaseStdout):
def test_posix(self):
with self.mock_property("sys.platform"):
sys.platform = "Linux"
with self.mock_property("sys.stdout.isatty") as obj:
obj.return_value = True
self.assertTrue(supports_color())
def test_pocket_pc(self):
with self.mock_property("sys.platform"):
sys.platform = "Pocket PC"
self.assertFalse(supports_color())
def test_windows_no_ansicon(self):
with self.mock_property("sys.platform"):
sys.platform = "win32"
with self.mock_property("os.environ"):
os.environ = []
self.assertFalse(supports_color())
def test_windows_ansicon(self):
with self.mock_property("sys.platform"):
sys.platform = "win32"
with self.mock_property("os.environ"):
os.environ = ["ANSICON"]
with self.mock_property("sys.stdout.isatty") as obj:
obj.return_value = True
self.assertTrue(supports_color())
def test_no_isatty_attribute(self):
with self.mock_property("sys.platform"):
sys.platform = "Linux"
with self.mock_property("sys.stdout"):
sys.stdout = list()
self.assertFalse(supports_color())
def test_no_isatty(self):
with self.mock_property("sys.platform"):
sys.platform = "Linux"
with self.mock_property("sys.stdout.isatty") as obj:
obj.return_value = False
self.assertFalse(supports_color())
class TestColorize(Base):
def setUp(self):
self.text = "house"
self.colors = ["red", "orange", "yellow",
"green", "blue", "purple"]
@mock.patch("updateHostsFile.supports_color", return_value=False)
def test_colorize_no_support(self, _):
for color in self.colors:
expected = self.text
actual = colorize(self.text, color)
self.assertEqual(actual, expected)
@mock.patch("updateHostsFile.supports_color", return_value=True)
def test_colorize_support(self, _):
for color in self.colors:
expected = color + self.text + Colors.ENDC
actual = colorize(self.text, color)
self.assertEqual(actual, expected)
class TestPrintSuccess(BaseStdout):
def setUp(self):
super(TestPrintSuccess, self).setUp()
self.text = "house"
@mock.patch("updateHostsFile.supports_color", return_value=False)
def test_print_success_no_support(self, _):
print_success(self.text)
expected = self.text + "\n"
actual = sys.stdout.getvalue()
self.assertEqual(actual, expected)
@mock.patch("updateHostsFile.supports_color", return_value=True)
def test_print_success_support(self, _):
print_success(self.text)
expected = Colors.SUCCESS + self.text + Colors.ENDC + "\n"
actual = sys.stdout.getvalue()
self.assertEqual(actual, expected)
class TestPrintFailure(BaseStdout):
def setUp(self):
super(TestPrintFailure, self).setUp()
self.text = "house"
@mock.patch("updateHostsFile.supports_color", return_value=False)
def test_print_failure_no_support(self, _):
print_failure(self.text)
expected = self.text + "\n"
actual = sys.stdout.getvalue()
self.assertEqual(actual, expected)
@mock.patch("updateHostsFile.supports_color", return_value=True)
def test_print_failure_support(self, _):
print_failure(self.text)
expected = Colors.FAIL + self.text + Colors.ENDC + "\n"
actual = sys.stdout.getvalue()
self.assertEqual(actual, expected)
# End Helper Functions
if __name__ == "__main__":
unittest.main()