Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def prepare_retry_test():
urls = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'test', 'data', 'invalid-url.yaml')
config = os.path.join(os.path.dirname(__file__), 'data', 'urlwatch.yaml')
cache = os.path.join(os.path.dirname(__file__), 'data', 'cache.db')
hooks = ''
config_storage = YamlConfigStorage(config)
cache_storage = CacheMiniDBStorage(cache)
urls_storage = UrlsYaml(urls)
urlwatch_config = TestConfig(config, urls, cache, hooks, True)
urlwatcher = Urlwatch(urlwatch_config, config_storage, cache_storage, urls_storage)
return urlwatcher, cache_storage
def prepare_retry_test():
urls = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'test', 'data', 'invalid-url.yaml')
config = os.path.join(os.path.dirname(__file__), 'data', 'urlwatch.yaml')
cache = os.path.join(os.path.dirname(__file__), 'data', 'cache.db')
hooks = ''
config_storage = YamlConfigStorage(config)
cache_storage = CacheMiniDBStorage(cache)
urls_storage = UrlsYaml(urls)
urlwatch_config = TestConfig(config, urls, cache, hooks, True)
urlwatcher = Urlwatch(urlwatch_config, config_storage, cache_storage, urls_storage)
return urlwatcher, cache_storage
def prepare_retry_test():
urls = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'test', 'data', 'invalid-url.yaml')
config = os.path.join(os.path.dirname(__file__), 'data', 'urlwatch.yaml')
cache = os.path.join(os.path.dirname(__file__), 'data', 'cache.db')
hooks = ''
config_storage = YamlConfigStorage(config)
cache_storage = CacheMiniDBStorage(cache)
urls_storage = UrlsYaml(urls)
urlwatch_config = TestConfig(config, urls, cache, hooks, True)
urlwatcher = Urlwatch(urlwatch_config, config_storage, cache_storage, urls_storage)
return urlwatcher, cache_storage
def prepare_retry_test():
urls = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'test', 'data', 'invalid-url.yaml')
config = os.path.join(os.path.dirname(__file__), 'data', 'urlwatch.yaml')
cache = os.path.join(os.path.dirname(__file__), 'data', 'cache.db')
hooks = ''
config_storage = YamlConfigStorage(config)
cache_storage = CacheMiniDBStorage(cache)
urls_storage = UrlsYaml(urls)
urlwatch_config = TestConfig(config, urls, cache, hooks, True)
urlwatcher = Urlwatch(urlwatch_config, config_storage, cache_storage, urls_storage)
return urlwatcher, cache_storage
filter = filter_tests[test_name]['filter']
data = filter_tests[test_name]['data']
expected_result = filter_tests[test_name]['expected_result']
if isinstance(filter, dict):
key = next(iter(filter))
kind, subfilter = key, filter[key]
elif isinstance(filter, str):
if ',' in filter:
raise ValueError('Only single filter allowed in this test')
elif ':' in filter:
kind, subfilter = filter.split(':', 1)
else:
kind = filter
subfilter = None
logger.info('filter kind: %s, subfilter: %s', kind, subfilter)
filtercls = FilterBase.__subclasses__.get(kind)
if filtercls is None:
raise ValueError('Unknown filter kind: %s:%s' % (filter_kind, subfilter))
result = filtercls(None, None).filter(data, subfilter)
logger.debug('Expected result:\n%s', expected_result)
logger.debug('Actual result:\n%s', result)
eq_(result, expected_result)
hooks_py = 'share/urlwatch/examples/hooks.py.example'
if os.path.exists(hooks_py):
imp.load_source('hooks', hooks_py)
def test_pep8_conformance():
"""Test that we conform to PEP-8."""
style = pycodestyle.StyleGuide(ignore=['E501', 'E402', 'W503'])
py_files = [y for x in os.walk(os.path.abspath('.')) for y in glob(os.path.join(x[0], '*.py'))]
py_files.append(os.path.abspath('urlwatch'))
result = style.check_files(py_files)
assert result.total_errors == 0, "Found #{0} code style errors".format(result.total_errors)
class TestConfig(BaseConfig):
def __init__(self, config, urls, cache, hooks, verbose):
(prefix, bindir) = os.path.split(os.path.dirname(os.path.abspath(sys.argv[0])))
super().__init__('urlwatch', os.path.dirname(__file__), config, urls, cache, hooks, verbose)
self.edit = False
self.edit_hooks = False
def teardown_func():
"tear down test fixtures"
cache = os.path.join(os.path.dirname(__file__), 'data', 'cache.db')
if os.path.exists(cache):
os.remove(cache)
@with_setup(teardown=teardown_func)
def test_run_watcher():
def test_save_load_jobs():
jobs = [
UrlJob(name='news', url='http://news.orf.at/'),
ShellJob(name='list homedir', command='ls ~'),
ShellJob(name='list proc', command='ls /proc'),
]
# tempfile.NamedTemporaryFile() doesn't work on Windows
# because the returned file object cannot be opened again
fd, name = tempfile.mkstemp()
UrlsYaml(name).save(jobs)
jobs2 = UrlsYaml(name).load()
os.chmod(name, 0o777)
jobs3 = UrlsYaml(name).load_secure()
os.close(fd)
os.remove(name)
assert len(jobs2) == len(jobs)
# Assert that the shell jobs have been removed due to secure loading
if sys.platform != 'win32':
assert len(jobs3) == 1
def test_save_load_jobs():
jobs = [
UrlJob(name='news', url='http://news.orf.at/'),
ShellJob(name='list homedir', command='ls ~'),
ShellJob(name='list proc', command='ls /proc'),
]
# tempfile.NamedTemporaryFile() doesn't work on Windows
# because the returned file object cannot be opened again
fd, name = tempfile.mkstemp()
UrlsYaml(name).save(jobs)
jobs2 = UrlsYaml(name).load()
os.chmod(name, 0o777)
jobs3 = UrlsYaml(name).load_secure()
os.close(fd)
os.remove(name)
assert len(jobs2) == len(jobs)
# Assert that the shell jobs have been removed due to secure loading
if sys.platform != 'win32':
assert len(jobs3) == 1
def test_run_watcher():
urls = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'share', 'urlwatch', 'examples', 'urls.yaml.example')
config = os.path.join(os.path.dirname(__file__), 'data', 'urlwatch.yaml')
cache = os.path.join(os.path.dirname(__file__), 'data', 'cache.db')
hooks = ''
config_storage = YamlConfigStorage(config)
urls_storage = UrlsYaml(urls)
cache_storage = CacheMiniDBStorage(cache)
try:
urlwatch_config = TestConfig(config, urls, cache, hooks, True)
urlwatcher = Urlwatch(urlwatch_config, config_storage, cache_storage, urls_storage)
urlwatcher.run_jobs()
finally:
cache_storage.close()
def test_save_load_jobs():
jobs = [
UrlJob(name='news', url='http://news.orf.at/'),
ShellJob(name='list homedir', command='ls ~'),
ShellJob(name='list proc', command='ls /proc'),
]
# tempfile.NamedTemporaryFile() doesn't work on Windows
# because the returned file object cannot be opened again
fd, name = tempfile.mkstemp()
UrlsYaml(name).save(jobs)
jobs2 = UrlsYaml(name).load()
os.chmod(name, 0o777)
jobs3 = UrlsYaml(name).load_secure()
os.close(fd)
os.remove(name)
assert len(jobs2) == len(jobs)
# Assert that the shell jobs have been removed due to secure loading
if sys.platform != 'win32':
assert len(jobs3) == 1