# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import json
import time
import unittest
import datetime as dt
from os import path
import six
import ming
from ming.odm import session
from bson import ObjectId
from webob import Request
from mock import Mock, patch
from pygments import highlight
from pygments.lexers import get_lexer_for_filename
import pytest
from tg import config
import html5lib
import html5lib.treewalkers
from alluratest.controller import setup_unit_test
from allura import model as M
from allura.lib import utils
from allura.lib import helpers as h
@patch.dict('allura.lib.utils.tg.config', clear=True, foo='bar', baz='true')
class TestConfigProxy(unittest.TestCase):
def setup_method(self, method):
self.cp = utils.ConfigProxy(mybaz="baz")
def test_getattr(self):
self.assertEqual(self.cp.foo, "bar")
self.assertEqual(self.cp.mybaz, "true")
def test_get(self):
self.assertEqual(self.cp.get("foo"), "bar")
self.assertEqual(self.cp.get("mybaz"), "true")
self.assertEqual(self.cp.get("fake"), None)
self.assertEqual(self.cp.get("fake", "default"), "default")
def test_get_bool(self):
self.assertEqual(self.cp.get_bool("mybaz"), True)
self.assertEqual(self.cp.get_bool("fake"), False)
class TestChunkedIterator(unittest.TestCase):
def setup_method(self, method):
setup_unit_test()
config = {
'ming.main.uri': 'mim://host/allura_test',
}
ming.configure(**config)
for i in range(10):
M.User.upsert('sample-user-%d' % i)
def test_can_iterate(self):
chunks = list(utils.chunked_find(M.User, {}, 2))
assert len(chunks) > 1, chunks
assert len(chunks[0]) == 2, chunks[0]
def test_filter_on_sort_key(self):
query = {'username':
{'$in': ['sample-user-1', 'sample-user-2', 'sample-user-3']}}
chunks = list(utils.chunked_find(M.User,
query,
2,
sort_key='username'))
assert len(chunks) == 2, chunks
assert len(chunks[0]) == 2, chunks[0]
assert len(chunks[1]) == 1, chunks[1]
assert chunks[0][0].username == 'sample-user-1'
assert chunks[0][1].username == 'sample-user-2'
assert chunks[1][0].username == 'sample-user-3'
class TestChunkedList(unittest.TestCase):
def test_chunked_list(self):
l = list(range(10))
chunks = list(utils.chunked_list(l, 3))
self.assertEqual(len(chunks), 4)
self.assertEqual(len(chunks[0]), 3)
self.assertEqual([el for sublist in chunks for el in sublist], l)
class TestAntispam(unittest.TestCase):
def setup_method(self, method):
setup_unit_test()
self.a = utils.AntiSpam()
def test_generate_fields(self):
fields = '\n'.join(self.a.extra_fields())
assert 'name="timestamp"' in fields, fields
assert 'name="spinner"' in fields, fields
assert ('class="%s"' % self.a.honey_class) in fields, fields
def test_invalid_old(self):
form = dict(a='1', b='2')
r = Request.blank('/', POST=self._encrypt_form(**form))
self.assertRaises(
ValueError,
utils.AntiSpam.validate_request,
r, now=time.time() + 24 * 60 * 60 * 4 + 1)
def test_valid_submit(self):
form = dict(a='1', b='2')
r = Request.blank('/', POST=self._encrypt_form(**form),
environ={'remote_addr': '127.0.0.1'}
)
validated = utils.AntiSpam.validate_request(r)
assert dict(a='1', b='2') == validated, validated
def test_invalid_future(self):
form = dict(a='1', b='2')
r = Request.blank('/', POST=self._encrypt_form(**form))
self.assertRaises(
ValueError,
utils.AntiSpam.validate_request,
r, now=time.time() - 10)
def test_invalid_spinner(self):
form = dict(a='1', b='2')
eform = self._encrypt_form(**form)
eform['spinner'] += 'a'
r = Request.blank('/', POST=eform)
self.assertRaises(ValueError, utils.AntiSpam.validate_request, r)
def test_invalid_honey(self):
form = dict(a='1', b='2', honey0='a')
eform = self._encrypt_form(**form)
r = Request.blank('/', POST=eform)
self.assertRaises(ValueError, utils.AntiSpam.validate_request, r)
def test_missing_honey(self):
form = dict(a='1', b='2')
eform = self._encrypt_form(**form)
del eform[self.a.enc('honey0')]
r = Request.blank('/', POST=eform)
self.assertRaises(ValueError, utils.AntiSpam.validate_request, r)
def _encrypt_form(self, **kwargs):
encrypted_form = {
self.a.enc(k): v for k, v in kwargs.items()}
encrypted_form.setdefault(self.a.enc('honey0'), '')
encrypted_form.setdefault(self.a.enc('honey1'), '')
encrypted_form['spinner'] = self.a.spinner_text
encrypted_form['timestamp'] = self.a.timestamp_text
return encrypted_form
class TestTruthyCallable(unittest.TestCase):
def test_everything(self):
def wrapper_func(bool_flag):
def predicate(bool_flag=bool_flag):
return bool_flag
return utils.TruthyCallable(predicate)
true_predicate = wrapper_func(True)
false_predicate = wrapper_func(False)
assert true_predicate(True) is True
assert false_predicate(False) is False
assert true_predicate() is True
assert false_predicate() is False
assert bool(true_predicate) is True
assert bool(false_predicate) is False
t, f = True, False # use variables because '== True' would generate warnings, and we do want '==' not 'is'
assert true_predicate == t
assert false_predicate == f
class TestCaseInsensitiveDict(unittest.TestCase):
def test_everything(self):
d = utils.CaseInsensitiveDict(Foo=5)
assert d['foo'] == d['Foo'] == d['FOO'] == 5
d['bAr'] = 6
assert d['bar'] == d['Bar'] == 6
d['bar'] = 7
assert d['bar'] == d['bAr'] == 7
del d['bar']
assert len(d) == 1, d
assert d.get('foo') == 5
d.update(foo=1, Bar=2)
assert d.get('FOO') == 1
assert d == dict(foo=1, bar=2)
assert d != dict(Foo=1, bar=2)
assert d == utils.CaseInsensitiveDict(Foo=1, bar=2)
class TestLineAnchorCodeHtmlFormatter(unittest.TestCase):
def test_render(self):
code = '#!/usr/bin/env python\n'\
'print "Hello, world!"'
formatter = utils.LineAnchorCodeHtmlFormatter(cssclass='codehilite',
linenos='inline')
lexer = get_lexer_for_filename("some.py", encoding='chardet')
hl_code = highlight(code, lexer, formatter)
assert '<div class="codehilite">' in hl_code
assert '<div id="l1" class="code_block">' in hl_code
try:
# older pygments
assert '<span class="lineno">1 </span>' in hl_code
except AssertionError:
# newer pygments
assert '<span class="linenos">1</span>' in hl_code
class TestIsTextFile(unittest.TestCase):
def test_is_text_file(self):
here_dir = path.dirname(__file__)
text_file = path.join(here_dir, 'data/test_mime/text_file.txt')
assert utils.is_text_file(open(text_file, 'rb').read())
bin_file = path.join(here_dir, 'data/test_mime/bin_file')
assert not utils.is_text_file(open(bin_file, 'rb').read())
class TestCodeStats(unittest.TestCase):
def setup_method(self, method):
setup_unit_test()
def test_generate_code_stats(self):
blob = Mock()
blob.text = \
"""class Person(object):
def __init__(self, name='Alice'):
self.name = name
def greetings(self):
print "Hello, %s" % self.name
\t\t"""
blob.size = len(blob.text)
stats = utils.generate_code_stats(blob)
assert stats['line_count'] == 8
assert stats['data_line_count'] == 5
assert stats['code_size'] == len(blob.text)
class TestHTMLSanitizer(unittest.TestCase):
def walker_from_text(self, text):
parsed = html5lib.parseFragment(text)
TreeWalker = html5lib.treewalkers.getTreeWalker("etree")
walker = TreeWalker(parsed)
return walker
def simple_tag_list(self, sanitizer):
# no attrs, no close tag flag check, just real simple
return [
t['name'] for t in sanitizer if t.get('name')
]
def test_html_sanitizer_iframe(self):
walker = self.walker_from_text('<div><iframe></iframe></div>')
p = utils.ForgeHTMLSanitizerFilter(walker)
assert self.simple_tag_list(p) == ['div', 'div']
def test_html_sanitizer_youtube_iframe(self):
walker = self.walker_from_text(
'<div><iframe src="https://www.youtube.com/embed/kOLpSPEA72U?feature=oembed"></iframe></div>')
p = utils.ForgeHTMLSanitizerFilter(walker)
assert self.simple_tag_list(p) == ['div', 'iframe', 'iframe', 'div']
walker = self.walker_from_text(
'<div><iframe src="https://www.youtube-nocookie.com/embed/kOLpSPEA72U?feature=oembed"></iframe></div>')
p = utils.ForgeHTMLSanitizerFilter(walker)
assert self.simple_tag_list(p) == ['div', 'iframe', 'iframe', 'div']
def test_html_sanitizer_form_elements(self):
walker = self.walker_from_text('<p>test</p><form method="post" action="http://localhost/foo.php"><input type=file><input type=text><textarea>asdf</textarea></form>')
p = utils.ForgeHTMLSanitizerFilter(walker)
assert self.simple_tag_list(p) == ['p', 'p']
def test_html_sanitizer_checkbox(self):
walker = self.walker_from_text('<p><input type="checkbox" disabled/><input type="text" disabled/><input type="checkbox" disabled checked/></p>')
p = utils.ForgeHTMLSanitizerFilter(walker)
assert self.simple_tag_list(p) == ['p', 'input', 'input', 'p']
def test_html_sanitizer_summary(self):
walker = self.walker_from_text('<details open="open"><summary>An Summary</summary><ul><li>Bullet Item</li></ul></details>')
p = utils.ForgeHTMLSanitizerFilter(walker)
assert self.simple_tag_list(p) == ['details', 'summary', 'summary', 'ul', 'li', 'li', 'ul', 'details']
def test_ip_address():
req = Mock()
req.remote_addr = '1.2.3.4'
req.headers = {}
assert (utils.ip_address(req) ==
'1.2.3.4')
def test_ip_address_header():
req = Mock()
req.remote_addr = '1.2.3.4'
req.headers = {'X_FORWARDED_FOR': '5.6.7.8'}
with h.push_config(config, **{'ip_address_header': 'X_FORWARDED_FOR'}):
assert (utils.ip_address(req) ==
'5.6.7.8')
def test_ip_address_header_not_set():
req = Mock()
req.remote_addr = '1.2.3.4'
req.headers = {}
with h.push_config(config, **{'ip_address_header': 'X_FORWARDED_FOR'}):
assert (utils.ip_address(req) ==
'1.2.3.4')
def test_empty_cursor():
"""EmptyCursors conforms to specification of Ming's ODMCursor"""
cursor = utils.EmptyCursor()
assert cursor.count() == 0
assert cursor.first() is None
assert cursor.all() == []
assert cursor.limit(10) == cursor
assert cursor.skip(10) == cursor
assert cursor.sort('name', 1) == cursor
assert cursor.hint('index') == cursor
assert cursor.extensions == []
assert cursor.options(arg1='val1', arg2='val2') == cursor
pytest.raises(ValueError, cursor.one)
pytest.raises(StopIteration, cursor.next)
pytest.raises(StopIteration, cursor._next_impl)
def test_DateJSONEncoder():
data = {'message': 'Hi!',
'date': dt.datetime(2015, 1, 30, 13, 13, 13)}
result = json.dumps(data, cls=utils.DateJSONEncoder)
assert result in ['{"date": "2015-01-30T13:13:13Z", "message": "Hi!"}',
'{"message": "Hi!", "date": "2015-01-30T13:13:13Z"}'], result
def test_clean_phone_number():
clean = utils.clean_phone_number
assert clean('123456789') == '123456789'
assert clean('+123 456:789') == '123456789'
assert clean('555-555-5555') == '5555555555'
assert clean('1-555-555-5555') == '15555555555'
def test_phone_number_hash():
hash = utils.phone_number_hash
assert hash('1234567890') == hash('+123 456:7890')
assert hash('1234567890') != hash('1234567891')
def test_skip_mod_date():
with utils.skip_mod_date(M.Artifact):
assert getattr(session(M.Artifact)._get(), 'skip_mod_date', None) is True
assert getattr(session(M.Artifact)._get(), 'skip_mod_date', None) is False
class FakeAttachment:
def __init__(self, filename):
self._id = ObjectId()
self.filename = filename
def __repr__(self):
return f'{self._id} {self.filename}'
def unique_attachments():
ua = utils.unique_attachments
assert [] == ua(None)
assert [] == ua([])
pic1 = FakeAttachment('pic.png')
pic2 = FakeAttachment('pic.png')
file1 = FakeAttachment('file.txt')
file2 = FakeAttachment('file.txt')
other = FakeAttachment('other')
attachments = [pic1, file1, pic2, file2, pic2, other]
expected = [file2, other, pic2]
assert expected == ua(attachments)
def test_is_nofollow_url():
with patch.dict(config, {'domain': 'localhost'}):
assert not utils.is_nofollow_url('relative/path')
assert not utils.is_nofollow_url('http://localhost/path')
assert utils.is_nofollow_url('http://google.com/')
assert utils.is_nofollow_url('https://google.com/')
with patch.dict(config, {'domain': 'localhost',
'nofollow_exempt_domains': 'foo.com, bar.io'}):
assert utils.is_nofollow_url('http://google.com/')
assert utils.is_nofollow_url('http://xfoo.com/')
assert not utils.is_nofollow_url('http://foo.com/')
assert not utils.is_nofollow_url('http://bar.io/')
assert utils.is_nofollow_url('http://bar.iot/')
def test_close_ipv4_addrs():
assert utils.close_ipv4_addrs('1.2.3.4', '1.2.3.4')
assert utils.close_ipv4_addrs('1.2.3.4', '1.2.3.255')
assert not utils.close_ipv4_addrs('1.2.3.4', '1.2.4.4')
def test_urlencode():
# dict - a simple one so arbitrary ordering doesn't cause problems on py2
assert (utils.urlencode({'a': 'hello'}) ==
'a=hello')
# list of pairs - including unicode and bytes
assert (utils.urlencode([('a', 1), ('b', 'ƒ'), ('c', 'ƒ'.encode())]) ==
'a=1&b=%C6%92&c=%C6%92')