Open Source Monorail

appengine/monorail/framework/test/framework_helpers_test.py

+# Copyright 2016 The Chromium Authors. All rights reserved.
+# Use of this source code is govered by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+"""Unit tests for the framework_helpers module."""
+
+import unittest
+
+import mox
+import time
+
+from framework import framework_helpers
+from framework import framework_views
+from proto import project_pb2
+from services import service_manager
+from testing import fake
+from testing import testing_helpers
+
+
+class HelperFunctionsTest(unittest.TestCase):
+
+  def setUp(self):
+    self.mox = mox.Mox()
+    self.time = self.mox.CreateMock(framework_helpers.time)
+    framework_helpers.time = self.time  # Point to a mocked out time module.
+
+  def tearDown(self):
+    framework_helpers.time = time  # Point back to the time module.
+    self.mox.UnsetStubs()
+    self.mox.ResetAll()
+
+  def testRetryDecorator_ExceedFailures(self):
+    class Tracker(object):
+      func_called = 0
+    tracker = Tracker()
+
+    # Use a function that always fails.
+    @framework_helpers.retry(2, delay=1, backoff=2)
+    def testFunc(tracker):
+      tracker.func_called += 1
+      raise Exception('Failed')
+
+    self.time.sleep(1).AndReturn(None)
+    self.time.sleep(2).AndReturn(None)
+    self.mox.ReplayAll()
+    with self.assertRaises(Exception):
+      testFunc(tracker)
+    self.mox.VerifyAll()
+    self.assertEquals(3, tracker.func_called)
+
+  def testRetryDecorator_EventuallySucceed(self):
+    class Tracker(object):
+      func_called = 0
+    tracker = Tracker()
+
+    # Use a function that succeeds on the 2nd attempt.
+    @framework_helpers.retry(2, delay=1, backoff=2)
+    def testFunc(tracker):
+      tracker.func_called += 1
+      if tracker.func_called < 2:
+        raise Exception('Failed')
+
+    self.time.sleep(1).AndReturn(None)
+    self.mox.ReplayAll()
+    testFunc(tracker)
+    self.mox.VerifyAll()
+    self.assertEquals(2, tracker.func_called)
+
+  def testGetRoleName(self):
+    proj = project_pb2.Project()
+    proj.owner_ids.append(111L)
+    proj.committer_ids.append(222L)
+    proj.contributor_ids.append(333L)
+
+    self.assertEquals(None, framework_helpers.GetRoleName(set(), proj))
+
+    self.assertEquals(
+        'Owner', framework_helpers.GetRoleName({111L}, proj))
+    self.assertEquals(
+        'Committer', framework_helpers.GetRoleName({222L}, proj))
+    self.assertEquals(
+        'Contributor', framework_helpers.GetRoleName({333L}, proj))
+
+    self.assertEquals(
+        'Owner',
+        framework_helpers.GetRoleName({111L, 222L, 999L}, proj))
+    self.assertEquals(
+        'Committer',
+        framework_helpers.GetRoleName({222L, 333L, 999L}, proj))
+    self.assertEquals(
+        'Contributor',
+        framework_helpers.GetRoleName({333L, 999L}, proj))
+
+
+class UrlFormattingTest(unittest.TestCase):
+  """Tests for URL formatting."""
+
+  def setUp(self):
+    self.services = service_manager.Services(user=fake.UserService())
+
+  def testFormatMovedProjectURL(self):
+    """Project foo has been moved to bar.  User is visiting /p/foo/..."""
+    mr = testing_helpers.MakeMonorailRequest()
+    mr.current_page_url = '/p/foo/'
+    self.assertEqual(
+      '/p/bar/',
+      framework_helpers.FormatMovedProjectURL(mr, 'bar'))
+
+    mr.current_page_url = '/p/foo/issues/list'
+    self.assertEqual(
+      '/p/bar/issues/list',
+      framework_helpers.FormatMovedProjectURL(mr, 'bar'))
+
+    mr.current_page_url = '/p/foo/issues/detail?id=123'
+    self.assertEqual(
+      '/p/bar/issues/detail?id=123',
+      framework_helpers.FormatMovedProjectURL(mr, 'bar'))
+
+    mr.current_page_url = '/p/foo/issues/detail?id=123#c7'
+    self.assertEqual(
+      '/p/bar/issues/detail?id=123#c7',
+      framework_helpers.FormatMovedProjectURL(mr, 'bar'))
+
+  def testFormatURL(self):
+    mr = testing_helpers.MakeMonorailRequest()
+    path = '/dude/wheres/my/car'
+    url = framework_helpers.FormatURL(mr, path)
+    self.assertEqual(path, url)
+
+  def testFormatURLWithRecognizedParams(self):
+    params = {}
+    query = []
+    for name in framework_helpers.RECOGNIZED_PARAMS:
+      params[name] = name
+      query.append('%s=%s' % (name, name))
+    path = '/dude/wheres/my/car'
+    expected = '%s?%s' % (path, '&'.join(query))
+    mr = testing_helpers.MakeMonorailRequest(path=expected)
+    url = framework_helpers.FormatURL(mr, path)  # No added params.
+    self.assertEqual(expected, url)
+
+  def testFormatURLWithKeywordArgs(self):
+    params = {}
+    query_pairs = []
+    for name in framework_helpers.RECOGNIZED_PARAMS:
+      params[name] = name
+      if name is not 'can' and name is not 'start':
+        query_pairs.append('%s=%s' % (name, name))
+    path = '/dude/wheres/my/car'
+    mr = testing_helpers.MakeMonorailRequest(
+        path='%s?%s' % (path, '&'.join(query_pairs)))
+    query_pairs.append('can=yep')
+    query_pairs.append('start=486')
+    query_string = '&'.join(query_pairs)
+    expected = '%s?%s' % (path, query_string)
+    url = framework_helpers.FormatURL(mr, path, can='yep', start=486)
+    self.assertEqual(expected, url)
+
+  def testFormatURLWithKeywordArgsAndID(self):
+    params = {}
+    query_pairs = []
+    query_pairs.append('id=200')  # id should be the first parameter.
+    for name in framework_helpers.RECOGNIZED_PARAMS:
+      params[name] = name
+      if name is not 'can' and name is not 'start':
+        query_pairs.append('%s=%s' % (name, name))
+    path = '/dude/wheres/my/car'
+    mr = testing_helpers.MakeMonorailRequest(
+        path='%s?%s' % (path, '&'.join(query_pairs)))
+    query_pairs.append('can=yep')
+    query_pairs.append('start=486')
+    query_string = '&'.join(query_pairs)
+    expected = '%s?%s' % (path, query_string)
+    url = framework_helpers.FormatURL(mr, path, can='yep', start=486, id=200)
+    self.assertEqual(expected, url)
+
+  def testFormatURLWithStrangeParams(self):
+    mr = testing_helpers.MakeMonorailRequest(path='/foo?start=0')
+    url = framework_helpers.FormatURL(
+        mr, '/foo', r=0, path='/foo/bar', sketchy='/foo/ bar baz ')
+    self.assertEqual(
+        '/foo?start=0&path=/foo/bar&r=0&sketchy=/foo/%20bar%20baz%20',
+        url)
+
+  def testFormatAbsoluteURL(self):
+    _request, mr = testing_helpers.GetRequestObjects(
+        path='/p/proj/some-path',
+        headers={'Host': 'www.test.com'})
+    self.assertEqual(
+        'http://www.test.com/p/proj/some/path',
+        framework_helpers.FormatAbsoluteURL(mr, '/some/path'))
+
+  def testFormatAbsoluteURL_CommonRequestParams(self):
+    _request, mr = testing_helpers.GetRequestObjects(
+        path='/p/proj/some-path?foo=bar&can=1',
+        headers={'Host': 'www.test.com'})
+    self.assertEqual(
+        'http://www.test.com/p/proj/some/path?can=1',
+        framework_helpers.FormatAbsoluteURL(mr, '/some/path'))
+    self.assertEqual(
+        'http://www.test.com/p/proj/some/path',
+        framework_helpers.FormatAbsoluteURL(
+            mr, '/some/path', copy_params=False))
+
+  def testFormatAbsoluteURL_NoProject(self):
+    path = '/some/path'
+    _request, mr = testing_helpers.GetRequestObjects(
+        headers={'Host': 'www.test.com'}, path=path)
+    url = framework_helpers.FormatAbsoluteURL(mr, path, include_project=False)
+    self.assertEqual(url, 'http://www.test.com/some/path')
+
+
+class WordWrapSuperLongLinesTest(unittest.TestCase):
+
+  def testEmptyLogMessage(self):
+    msg = ''
+    wrapped_msg = framework_helpers.WordWrapSuperLongLines(msg)
+    self.assertEqual(wrapped_msg, '')
+
+  def testShortLines(self):
+    msg = 'one\ntwo\nthree\n'
+    wrapped_msg = framework_helpers.WordWrapSuperLongLines(msg)
+    expected = 'one\ntwo\nthree\n'
+    self.assertEqual(wrapped_msg, expected)
+
+  def testOneLongLine(self):
+    msg = ('This is a super long line that just goes on and on '
+           'and it seems like it will never stop because it is '
+           'super long and it was entered by a user who had no '
+           'familiarity with the return key.')
+    wrapped_msg = framework_helpers.WordWrapSuperLongLines(msg)
+    expected = ('This is a super long line that just goes on and on and it '
+                'seems like it will never stop because it\n'
+                'is super long and it was entered by a user who had no '
+                'familiarity with the return key.')
+    self.assertEqual(wrapped_msg, expected)
+
+    msg2 = ('This is a super long line that just goes on and on '
+            'and it seems like it will never stop because it is '
+            'super long and it was entered by a user who had no '
+            'familiarity with the return key. '
+            'This is a super long line that just goes on and on '
+            'and it seems like it will never stop because it is '
+            'super long and it was entered by a user who had no '
+            'familiarity with the return key.')
+    wrapped_msg2 = framework_helpers.WordWrapSuperLongLines(msg2)
+    expected2 = ('This is a super long line that just goes on and on and it '
+                 'seems like it will never stop because it\n'
+                 'is super long and it was entered by a user who had no '
+                 'familiarity with the return key. This is a\n'
+                 'super long line that just goes on and on and it seems like '
+                 'it will never stop because it is super\n'
+                 'long and it was entered by a user who had no familiarity '
+                 'with the return key.')
+    self.assertEqual(wrapped_msg2, expected2)
+
+  def testMixOfShortAndLong(self):
+    msg = ('[Author: mpcomplete]\n'
+           '\n'
+           # Description on one long line
+           'Fix a memory leak in JsArray and JsObject for the IE and NPAPI '
+           'ports.  Each time you call GetElement* or GetProperty* to '
+           'retrieve string or object token, the token would be leaked.  '
+           'I added a JsScopedToken to ensure that the right thing is '
+           'done when the object leaves scope, depending on the platform.\n'
+           '\n'
+           'R=zork\n'
+           'CC=google-gears-eng@googlegroups.com\n'
+           'DELTA=108  (52 added, 36 deleted, 20 changed)\n'
+           'OCL=5932446\n'
+           'SCL=5933728\n')
+    wrapped_msg = framework_helpers.WordWrapSuperLongLines(msg)
+    expected = (
+        '[Author: mpcomplete]\n'
+        '\n'
+        'Fix a memory leak in JsArray and JsObject for the IE and NPAPI '
+        'ports.  Each time you call\n'
+        'GetElement* or GetProperty* to retrieve string or object token, the '
+        'token would be leaked.  I added\n'
+        'a JsScopedToken to ensure that the right thing is done when the '
+        'object leaves scope, depending on\n'
+        'the platform.\n'
+        '\n'
+        'R=zork\n'
+        'CC=google-gears-eng@googlegroups.com\n'
+        'DELTA=108  (52 added, 36 deleted, 20 changed)\n'
+        'OCL=5932446\n'
+        'SCL=5933728\n')
+    self.assertEqual(wrapped_msg, expected)
+
+
+class ComputeListDeltasTest(unittest.TestCase):
+
+  def DoOne(self, old=None, new=None, added=None, removed=None):
+    """Run one call to the target method and check expected results."""
+    actual_added, actual_removed = framework_helpers.ComputeListDeltas(
+        old, new)
+    self.assertItemsEqual(added, actual_added)
+    self.assertItemsEqual(removed, actual_removed)
+
+  def testEmptyLists(self):
+    self.DoOne(old=[], new=[], added=[], removed=[])
+    self.DoOne(old=[1, 2], new=[], added=[], removed=[1, 2])
+    self.DoOne(old=[], new=[1, 2], added=[1, 2], removed=[])
+
+  def testUnchanged(self):
+    self.DoOne(old=[1], new=[1], added=[], removed=[])
+    self.DoOne(old=[1, 2], new=[1, 2], added=[], removed=[])
+    self.DoOne(old=[1, 2], new=[2, 1], added=[], removed=[])
+
+  def testCompleteChange(self):
+    self.DoOne(old=[1, 2], new=[3, 4], added=[3, 4], removed=[1, 2])
+
+  def testGeneralChange(self):
+    self.DoOne(old=[1, 2], new=[2], added=[], removed=[1])
+    self.DoOne(old=[1], new=[1, 2], added=[2], removed=[])
+    self.DoOne(old=[1, 2], new=[2, 3], added=[3], removed=[1])
+
+
+class UserSettingsTest(unittest.TestCase):
+
+  def testGatherUnifiedSettingsPageData(self):
+    email_options = []
+
+    class UserSettingsStub(framework_helpers.UserSettings):
+
+      # pylint: disable=unused-argument
+      @classmethod
+      def _GetEmailOptions(cls, user_view, conn_pool):
+        return email_options
+
+    mr = testing_helpers.MakeMonorailRequest()
+    mr.auth.user_view = framework_views.UserView(100, 'user@invalid', True)
+    mr.auth.user_view.profile_url = '/u/profile/url'
+    page_data = UserSettingsStub.GatherUnifiedSettingsPageData(
+        mr.auth.user_id, mr.auth.user_view, mr.auth.user_pb)
+
+    expected_keys = [
+        'api_request_reset',
+        'api_request_lifetime_limit',
+        'api_request_hard_limit',
+        'api_request_soft_limit',
+        'settings_user',
+        'settings_user_pb',
+        'settings_user_is_banned',
+        'settings_user_ignore_action_limits',
+        'self',
+        'project_creation_reset',
+        'issue_comment_reset',
+        'issue_attachment_reset',
+        'issue_bulk_edit_reset',
+        'project_creation_lifetime_limit',
+        'project_creation_soft_limit',
+        'project_creation_hard_limit',
+        'issue_comment_lifetime_limit',
+        'issue_comment_soft_limit',
+        'issue_comment_hard_limit',
+        'issue_attachment_lifetime_limit',
+        'issue_attachment_soft_limit',
+        'issue_attachment_hard_limit',
+        'issue_bulk_edit_lifetime_limit',
+        'issue_bulk_edit_hard_limit',
+        'issue_bulk_edit_soft_limit',
+        'profile_url_fragment',
+        'preview_on_hover',
+        ]
+    self.assertItemsEqual(expected_keys, page_data.keys())
+
+    self.assertEqual('profile/url', page_data['profile_url_fragment'])
+    # TODO(jrobbins): Test action limit support
+
+  # TODO(jrobbins): Test ProcessForm.
+
+
+class MurmurHash3Test(unittest.TestCase):
+
+  def testMurmurHash(self):
+    test_data = [
+        ('', 0),
+        ('agable@chromium.org', 4092810879),
+        (u'jrobbins@chromium.org', 904770043),
+        ('seanmccullough%google.com@gtempaccount.com', 1301269279),
+        ('rmistry+monorail@chromium.org', 4186878788),
+        ('jparent+foo@', 2923900874),
+        ('@example.com', 3043483168),
+    ]
+    hashes = [framework_helpers.MurmurHash3_x86_32(x)
+              for (x, _) in test_data]
+    self.assertListEqual(hashes, [e for (_, e) in test_data])
+
+  def testMurmurHashWithSeed(self):
+    test_data = [
+        ('', 1113155926, 2270882445),
+        ('agable@chromium.org', 772936925, 3995066671),
+        (u'jrobbins@chromium.org', 1519359761, 1273489513),
+        ('seanmccullough%google.com@gtempaccount.com', 49913829, 1202521153),
+        ('rmistry+monorail@chromium.org', 314860298, 3636123309),
+        ('jparent+foo@', 195791379, 332453977),
+        ('@example.com', 521490555, 257496459),
+    ]
+    hashes = [framework_helpers.MurmurHash3_x86_32(x, s)
+              for (x, s, _) in test_data]
+    self.assertListEqual(hashes, [e for (_, _, e) in test_data])
+
+
+class MakeRandomKeyTest(unittest.TestCase):
+
+  def testMakeRandomKey_Normal(self):
+    key1 = framework_helpers.MakeRandomKey()
+    key2 = framework_helpers.MakeRandomKey()
+    self.assertEqual(128, len(key1))
+    self.assertEqual(128, len(key2))
+    self.assertNotEqual(key1, key2)
+
+  def testMakeRandomKey_Length(self):
+    key = framework_helpers.MakeRandomKey()
+    self.assertEqual(128, len(key))
+    key16 = framework_helpers.MakeRandomKey(length=16)
+    self.assertEqual(16, len(key16))
+
+  def testMakeRandomKey_Chars(self):
+    key = framework_helpers.MakeRandomKey(chars='a', length=4)
+    self.assertEqual('aaaa', key)
+
+
+class IsServiceAccountTest(unittest.TestCase):
+
+  def testIsServiceAccount(self):
+    appspot = 'abc@appspot.gserviceaccount.com'
+    developer = '@developer.gserviceaccount.com'
+    bugdroid = 'bugdroid1@chromium.org'
+    user = 'test@example.com'
+
+    self.assertTrue(framework_helpers.IsServiceAccount(appspot))
+    self.assertTrue(framework_helpers.IsServiceAccount(developer))
+    self.assertTrue(framework_helpers.IsServiceAccount(bugdroid))
+    self.assertFalse(framework_helpers.IsServiceAccount(user))
+
+
+if __name__ == '__main__':
+  unittest.main()