kallithea Changeset - 48a00daba2f2

Changeset - 48a00daba2f2

Parent rev.

Child rev.

[Not reviewed]

default

0 12 0

Lars Kruse - 9 years ago 2017-08-25 14:35:28
devel@sumpfralle.de

codingstyle: replace comparison for equality against None with "is" expression

Both style and correctness.

Reported by flake8.

7 files changed:

kallithea/lib/db_manage.py

kallithea/lib/ipaddr.py

kallithea/tests/api/api_base.py

kallithea/tests/functional/test_admin_repos.py

kallithea/tests/functional/test_admin_user_groups.py

kallithea/tests/functional/test_login.py

kallithea/tests/functional/test_pullrequests.py

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)

kallithea/lib/db_manage.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.db_manage
 ~~~~~~~~~~~~~~~~~~~~~~~
 Database creation, and setup module for Kallithea. Used for creation
 of database as well as for migration operations
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 10, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import sys
 import time
 import uuid
 import logging
 import sqlalchemy
 from os.path import dirname
 import alembic.config
 import alembic.command
 from kallithea.model.user import UserModel
 from kallithea.model.base import init_model
 from kallithea.model.db import User, Permission, Ui, \
     Setting, UserToPerm, RepoGroup, \
     UserRepoGroupToPerm, CacheInvalidation, Repository
 from sqlalchemy.engine import create_engine
 from kallithea.model.repo_group import RepoGroupModel
 #from kallithea.model import meta
 from kallithea.model.meta import Session, Base
 from kallithea.model.repo import RepoModel
 from kallithea.model.permission import PermissionModel
 log = logging.getLogger(__name__)
 def notify(msg):
     """
     Notification for migrations messages
     """
     ml = len(msg) + (4 * 2)
     print('\n%s\n*** %s ***\n%s' % ('*' * ml, msg, '*' * ml)).upper()
 class DbManage(object):
     def __init__(self, dbconf, root, tests=False, SESSION=None, cli_args=None):
         self.dbname = dbconf.split('/')[-1]
         self.tests = tests
         self.root = root
         self.dburi = dbconf
         self.db_exists = False
         self.cli_args = cli_args or {}
         self.init_db(SESSION=SESSION)
     def _ask_ok(self, msg):
         """Invoke ask_ok unless the force_ask option provides the answer"""
         force_ask = self.cli_args.get('force_ask')
         if force_ask is not None:
             return force_ask
         from kallithea.lib.paster_commands.common import ask_ok
         return ask_ok(msg)
     def init_db(self, SESSION=None):
         if SESSION:
             self.sa = SESSION
         else:
             # init new sessions
             engine = create_engine(self.dburi)
             init_model(engine)
             self.sa = Session()
     def create_tables(self, override=False):
         """
         Create a auth database
         """
         log.info("Any existing database is going to be destroyed")
         if self.tests:
             destroy = True
         else:
             destroy = self._ask_ok('Are you sure to destroy old database ? [y/n]')
         if not destroy:
             print 'Nothing done.'
             sys.exit(0)
         if destroy:
             # drop and re-create old schemas
             url = sqlalchemy.engine.url.make_url(self.dburi)
             database = url.database
             # Some databases enforce foreign key constraints and Base.metadata.drop_all() doesn't work
             if url.drivername == 'mysql':
                 url.database = None  # don't connect to the database (it might not exist)
                 engine = sqlalchemy.create_engine(url)
                 with engine.connect() as conn:
                     conn.execute('DROP DATABASE IF EXISTS ' + database)
                     conn.execute('CREATE DATABASE ' + database)
             elif url.drivername == 'postgresql':
                 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
                 url.database = 'postgres'  # connect to the system database (as the real one might not exist)
                 engine = sqlalchemy.create_engine(url)
                 with engine.connect() as conn:
                     conn.connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
                     conn.execute('DROP DATABASE IF EXISTS ' + database)
                     conn.execute('CREATE DATABASE ' + database)
             else:
                 # known to work on SQLite - possibly not on other databases with strong referential integrity
                 Base.metadata.drop_all()
         checkfirst = not override
         Base.metadata.create_all(checkfirst=checkfirst)
         # Create an Alembic configuration and generate the version table,
         # "stamping" it with the most recent Alembic migration revision, to
         # tell Alembic that all the schema upgrades are already in effect.
         alembic_cfg = alembic.config.Config()
         alembic_cfg.set_main_option('script_location', 'kallithea:alembic')
         alembic_cfg.set_main_option('sqlalchemy.url', self.dburi)
         # This command will give an error in an Alembic multi-head scenario,
         # but in practice, such a scenario should not come up during database
         # creation, even during development.
         alembic.command.stamp(alembic_cfg, 'head')
         log.info('Created tables for %s', self.dbname)
     def fix_repo_paths(self):
         """
         Fixes a old kallithea version path into new one without a '*'
         """
         paths = Ui.query() \
                 .filter(Ui.ui_key == '/') \
                 .scalar()
         paths.ui_value = paths.ui_value.replace('*', '')
         self.sa.commit()
     def fix_default_user(self):
         """
         Fixes a old default user with some 'nicer' default values,
         used mostly for anonymous access
         """
         def_user = User.query().filter_by(is_default_user=True).one()
         def_user.name = 'Anonymous'
         def_user.lastname = 'User'
         def_user.email = 'anonymous@kallithea-scm.org'
         self.sa.commit()
     def fix_settings(self):
         """
         Fixes kallithea settings adds ga_code key for google analytics
         """
         hgsettings3 = Setting('ga_code', '')
         self.sa.add(hgsettings3)
         self.sa.commit()
     def admin_prompt(self, second=False):
         if not self.tests:
             import getpass
             username = self.cli_args.get('username')
             password = self.cli_args.get('password')
             email = self.cli_args.get('email')
             def get_password():
                 password = getpass.getpass('Specify admin password '
                                            '(min 6 chars):')
                 confirm = getpass.getpass('Confirm password:')
                 if password != confirm:
                     log.error('passwords mismatch')
                     return False
                 if len(password) < 6:
                     log.error('password is to short use at least 6 characters')
                     return False
                 return password
             if username is None:
                 username = raw_input('Specify admin username:')
             if password is None:
                 password = get_password()
                 if not password:
                     # second try
                     password = get_password()
                     if not password:
                         sys.exit()
             if email is None:
                 email = raw_input('Specify admin email:')
             self.create_user(username, password, email, True)
         else:
             log.info('creating admin and regular test users')
             from kallithea.tests.base import TEST_USER_ADMIN_LOGIN, \
             TEST_USER_ADMIN_PASS, TEST_USER_ADMIN_EMAIL, \
             TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, \
             TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR2_LOGIN, \
             TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL
             self.create_user(TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS,
                              TEST_USER_ADMIN_EMAIL, True)
             self.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                              TEST_USER_REGULAR_EMAIL, False)
             self.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS,
                              TEST_USER_REGULAR2_EMAIL, False)
     def create_ui_settings(self, repo_store_path):
         """
         Creates ui settings, fills out hooks
         """
         # HOOKS
         hooks1_key = Ui.HOOK_UPDATE
         hooks1_ = Ui.query() \
             .filter(Ui.ui_key == hooks1_key).scalar()
         hooks1 = Ui() if hooks1_ is None else hooks1_
         hooks1.ui_section = 'hooks'
         hooks1.ui_key = hooks1_key
         hooks1.ui_value = 'hg update >&2'
         hooks1.ui_active = False
         self.sa.add(hooks1)
         hooks2_key = Ui.HOOK_REPO_SIZE
         hooks2_ = Ui.query() \
             .filter(Ui.ui_key == hooks2_key).scalar()
         hooks2 = Ui() if hooks2_ is None else hooks2_
         hooks2.ui_section = 'hooks'
         hooks2.ui_key = hooks2_key
         hooks2.ui_value = 'python:kallithea.lib.hooks.repo_size'
         self.sa.add(hooks2)
         hooks3 = Ui()
         hooks3.ui_section = 'hooks'
         hooks3.ui_key = Ui.HOOK_PUSH
         hooks3.ui_value = 'python:kallithea.lib.hooks.log_push_action'
         self.sa.add(hooks3)
         hooks4 = Ui()
         hooks4.ui_section = 'hooks'
         hooks4.ui_key = Ui.HOOK_PRE_PUSH
         hooks4.ui_value = 'python:kallithea.lib.hooks.pre_push'
         self.sa.add(hooks4)
         hooks5 = Ui()
         hooks5.ui_section = 'hooks'
         hooks5.ui_key = Ui.HOOK_PULL
         hooks5.ui_value = 'python:kallithea.lib.hooks.log_pull_action'
         self.sa.add(hooks5)
         hooks6 = Ui()
         hooks6.ui_section = 'hooks'
         hooks6.ui_key = Ui.HOOK_PRE_PULL
         hooks6.ui_value = 'python:kallithea.lib.hooks.pre_pull'
         self.sa.add(hooks6)
         # enable largefiles
         largefiles = Ui()
         largefiles.ui_section = 'extensions'
         largefiles.ui_key = 'largefiles'
         largefiles.ui_value = ''
         self.sa.add(largefiles)
         # set default largefiles cache dir, defaults to
         # /repo location/.cache/largefiles
         largefiles = Ui()
         largefiles.ui_section = 'largefiles'
         largefiles.ui_key = 'usercache'
         largefiles.ui_value = os.path.join(repo_store_path, '.cache',
                                            'largefiles')
         self.sa.add(largefiles)
         # enable hgsubversion disabled by default
         hgsubversion = Ui()
         hgsubversion.ui_section = 'extensions'
         hgsubversion.ui_key = 'hgsubversion'
         hgsubversion.ui_value = ''
         hgsubversion.ui_active = False
         self.sa.add(hgsubversion)
         # enable hggit disabled by default
         hggit = Ui()
         hggit.ui_section = 'extensions'
         hggit.ui_key = 'hggit'
         hggit.ui_value = ''
         hggit.ui_active = False
         self.sa.add(hggit)
     def create_auth_plugin_options(self, skip_existing=False):
         """
         Create default auth plugin settings, and make it active
         :param skip_existing:
         """
         for k, v, t in [('auth_plugins', 'kallithea.lib.auth_modules.auth_internal', 'list'),
                      ('auth_internal_enabled', 'True', 'bool')]:
             if skip_existing and Setting.get_by_name(k) != None:
                         ('auth_internal_enabled', 'True', 'bool')]:
             if skip_existing and Setting.get_by_name(k) is not None:
                 log.debug('Skipping option %s', k)
                 continue
             setting = Setting(k, v, t)
             self.sa.add(setting)
     def create_default_options(self, skip_existing=False):
         """Creates default settings"""
         for k, v, t in [
             ('default_repo_enable_locking',  False, 'bool'),
             ('default_repo_enable_downloads', False, 'bool'),
             ('default_repo_enable_statistics', False, 'bool'),
             ('default_repo_private', False, 'bool'),
             ('default_repo_type', 'hg', 'unicode')]:
             if skip_existing and Setting.get_by_name(k) is not None:
                 log.debug('Skipping option %s', k)
                 continue
             setting = Setting(k, v, t)
             self.sa.add(setting)
     def fixup_groups(self):
         def_usr = User.get_default_user()
         for g in RepoGroup.query().all():
             g.group_name = g.get_new_name(g.name)
             # get default perm
             default = UserRepoGroupToPerm.query() \
                 .filter(UserRepoGroupToPerm.group == g) \
                 .filter(UserRepoGroupToPerm.user == def_usr) \
                 .scalar()
             if default is None:
                 log.debug('missing default permission for group %s adding', g)
                 RepoGroupModel()._create_default_perms(g)
     def reset_permissions(self, username):
         """
         Resets permissions to default state, useful when old systems had
         bad permissions, we must clean them up
         :param username:
         """
         default_user = User.get_by_username(username)
         if not default_user:
             return
         u2p = UserToPerm.query() \
             .filter(UserToPerm.user == default_user).all()
         fixed = False
         if len(u2p) != len(Permission.DEFAULT_USER_PERMISSIONS):
             for p in u2p:
                 Session().delete(p)
             fixed = True
             self.populate_default_permissions()
         return fixed
     def update_repo_info(self):
         for repo in Repository.query():
             repo.update_changeset_cache()
     def config_prompt(self, test_repo_path='', retries=3):
         _path = self.cli_args.get('repos_location')
         if retries == 3:
             log.info('Setting up repositories config')
         if _path is not None:
             path = _path
         elif not self.tests and not test_repo_path:
             path = raw_input(
                  'Enter a valid absolute path to store repositories. '
                  'All repositories in that path will be added automatically:'
+            )
         else:
             path = test_repo_path
         path_ok = True
         # check proper dir
         if not os.path.isdir(path):
             path_ok = False
             log.error('Given path %s is not a valid directory', path)
         elif not os.path.isabs(path):
             path_ok = False
             log.error('Given path %s is not an absolute path', path)
         # check if path is at least readable.
         if not os.access(path, os.R_OK):
             path_ok = False
             log.error('Given path %s is not readable', path)
         # check write access, warn user about non writeable paths
         elif not os.access(path, os.W_OK) and path_ok:
             log.warning('No write permission to given path %s', path)
             if not self._ask_ok('Given path %s is not writeable, do you want to '
                           'continue with read only mode ? [y/n]' % (path,)):
                 log.error('Canceled by user')
                 sys.exit(-1)
         if retries == 0:
             sys.exit('max retries reached')
         if not path_ok:
             if _path is not None:
                 sys.exit('Invalid repo path: %s' % _path)
             retries -= 1
             return self.config_prompt(test_repo_path, retries) # recursing!!!
         real_path = os.path.normpath(os.path.realpath(path))
         if real_path != os.path.normpath(path):
             log.warning('Using normalized path %s instead of %s', real_path, path)
         return real_path
     def create_settings(self, path):
         self.create_ui_settings(path)
         ui_config = [
             ('web', 'allow_archive', 'gz zip bz2'),
             ('web', 'baseurl', '/'),
             ('paths', '/', path),
             #('phases', 'publish', 'false')
+        ]
         for section, key, value in ui_config:
             ui_conf = Ui()
             setattr(ui_conf, 'ui_section', section)
             setattr(ui_conf, 'ui_key', key)
             setattr(ui_conf, 'ui_value', value)
             self.sa.add(ui_conf)
         settings = [
             ('realm', 'Kallithea', 'unicode'),
             ('title', '', 'unicode'),
             ('ga_code', '', 'unicode'),
             ('show_public_icon', True, 'bool'),
             ('show_private_icon', True, 'bool'),
             ('stylify_metatags', False, 'bool'),
             ('dashboard_items', 100, 'int'), # TODO: call it page_size
             ('admin_grid_items', 25, 'int'),
             ('show_version', True, 'bool'),
             ('use_gravatar', True, 'bool'),
             ('gravatar_url', User.DEFAULT_GRAVATAR_URL, 'unicode'),
             ('clone_uri_tmpl', Repository.DEFAULT_CLONE_URI, 'unicode'),
             ('update_url', Setting.DEFAULT_UPDATE_URL, 'unicode'),
+        ]
         for key, val, type_ in settings:
             sett = Setting(key, val, type_)
             self.sa.add(sett)
         self.create_auth_plugin_options()
         self.create_default_options()
         log.info('created ui config')
     def create_user(self, username, password, email='', admin=False):
         log.info('creating user %s', username)
         UserModel().create_or_update(username, password, email,
                                      firstname=u'Kallithea', lastname=u'Admin',
                                      active=True, admin=admin,
                                      extern_type=User.DEFAULT_AUTH_TYPE)
     def create_default_user(self):
         log.info('creating default user')
         # create default user for handling default permissions.
         user = UserModel().create_or_update(username=User.DEFAULT_USER,
                                             password=str(uuid.uuid1())[:20],
                                             email='anonymous@kallithea-scm.org',
                                             firstname=u'Anonymous',
                                             lastname=u'User')
         # based on configuration options activate/deactivate this user which
         # controls anonymous access
         if self.cli_args.get('public_access') is False:
             log.info('Public access disabled')
             user.active = False
             Session().commit()
     def create_permissions(self):
         """
         Creates all permissions defined in the system
         """
         # module.(access|create|change|delete)_[name]
         # module.(none|read|write|admin)
         log.info('creating permissions')
         PermissionModel().create_permissions()
     def populate_default_permissions(self):
         """
         Populate default permissions. It will create only the default
         permissions that are missing, and not alter already defined ones
         """
         log.info('creating default user permissions')
         PermissionModel().create_default_permissions(user=User.DEFAULT_USER)

kallithea/lib/ipaddr.py

➞

Show inline comments

@@ @@ -4,1548 +4,1547 @@ @@
 # Licensed 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.
 """A fast, lightweight IPv4/IPv6 manipulation library in Python.
 This library is used to create/poke/manipulate IPv4 and IPv6 addresses
 and networks.
 """
 __version__ = 'trunk'
 import struct
 IPV4LENGTH = 32
 IPV6LENGTH = 128
 class AddressValueError(ValueError):
     """A Value Error related to the address."""
 class NetmaskValueError(ValueError):
     """A Value Error related to the netmask."""
 def IPAddress(address, version=None):
     """Take an IP string/int and return an object of the correct type.
     Args:
         address: A string or integer, the IP address.  Either IPv4 or
           IPv6 addresses may be supplied; integers less than 2**32 will
           be considered to be IPv4 by default.
         version: An Integer, 4 or 6. If set, don't try to automatically
           determine what the IP address type is. important for things
           like IPAddress(1), which could be IPv4, '0.0.0.1',  or IPv6,
           '::1'.
     Returns:
         An IPv4Address or IPv6Address object.
     Raises:
         ValueError: if the string passed isn't either a v4 or a v6
           address.
     """
     if version:
         if version == 4:
             return IPv4Address(address)
         elif version == 6:
             return IPv6Address(address)
     try:
         return IPv4Address(address)
     except (AddressValueError, NetmaskValueError):
         pass
     try:
         return IPv6Address(address)
     except (AddressValueError, NetmaskValueError):
         pass
     raise ValueError('%r does not appear to be an IPv4 or IPv6 address' %
                      address)
 def IPNetwork(address, version=None, strict=False):
     """Take an IP string/int and return an object of the correct type.
     Args:
         address: A string or integer, the IP address.  Either IPv4 or
           IPv6 addresses may be supplied; integers less than 2**32 will
           be considered to be IPv4 by default.
         version: An Integer, if set, don't try to automatically
           determine what the IP address type is. important for things
           like IPNetwork(1), which could be IPv4, '0.0.0.1/32', or IPv6,
           '::1/128'.
     Returns:
         An IPv4Network or IPv6Network object.
     Raises:
         ValueError: if the string passed isn't either a v4 or a v6
           address. Or if a strict network was requested and a strict
           network wasn't given.
     """
     if version:
         if version == 4:
             return IPv4Network(address, strict)
         elif version == 6:
             return IPv6Network(address, strict)
     try:
         return IPv4Network(address, strict)
     except (AddressValueError, NetmaskValueError):
         pass
     try:
         return IPv6Network(address, strict)
     except (AddressValueError, NetmaskValueError):
         pass
     raise ValueError('%r does not appear to be an IPv4 or IPv6 network' %
                      address)
 def v4_int_to_packed(address):
     """The binary representation of this address.
     Args:
         address: An integer representation of an IPv4 IP address.
     Returns:
         The binary representation of this address.
     Raises:
         ValueError: If the integer is too large to be an IPv4 IP
           address.
     """
     if address > _BaseV4._ALL_ONES:
         raise ValueError('Address too large for IPv4')
     return Bytes(struct.pack('!I', address))
 def v6_int_to_packed(address):
     """The binary representation of this address.
     Args:
         address: An integer representation of an IPv6 IP address.
     Returns:
         The binary representation of this address.
     """
     return Bytes(struct.pack('!QQ', address >> 64, address & (2 ** 64 - 1)))
 def _find_address_range(addresses):
     """Find a sequence of addresses.
     Args:
         addresses: a list of IPv4 or IPv6 addresses.
     Returns:
         A tuple containing the first and last IP addresses in the sequence.
     """
     first = last = addresses[0]
     for ip in addresses[1:]:
         if ip._ip == last._ip + 1:
             last = ip
         else:
             break
     return (first, last)
 def _get_prefix_length(number1, number2, bits):
     """Get the number of leading bits that are same for two numbers.
     Args:
         number1: an integer.
         number2: another integer.
         bits: the maximum number of bits to compare.
     Returns:
         The number of leading bits that are the same for two numbers.
     """
     for i in range(bits):
         if number1 >> i == number2 >> i:
             return bits - i
     return 0
 def _count_righthand_zero_bits(number, bits):
     """Count the number of zero bits on the right hand side.
     Args:
         number: an integer.
         bits: maximum number of bits to count.
     Returns:
         The number of zero bits on the right hand side of the number.
     """
     if number == 0:
         return bits
     for i in range(bits):
         if (number >> i) % 2:
             return i
 def summarize_address_range(first, last):
     """Summarize a network range given the first and last IP addresses.
     Example:
         >>> summarize_address_range(IPv4Address('1.1.1.0'),
             IPv4Address('1.1.1.130'))
         [IPv4Network('1.1.1.0/25'), IPv4Network('1.1.1.128/31'),
         IPv4Network('1.1.1.130/32')]
     Args:
         first: the first IPv4Address or IPv6Address in the range.
         last: the last IPv4Address or IPv6Address in the range.
     Returns:
         The address range collapsed to a list of IPv4Network's or
         IPv6Network's.
     Raise:
         TypeError:
             If the first and last objects are not IP addresses.
             If the first and last objects are not the same version.
         ValueError:
             If the last object is not greater than the first.
             If the version is not 4 or 6.
     """
     if not (isinstance(first, _BaseIP) and isinstance(last, _BaseIP)):
         raise TypeError('first and last must be IP addresses, not networks')
     if first.version != last.version:
         raise TypeError("%s and %s are not of the same version" % (
                 str(first), str(last)))
     if first > last:
         raise ValueError('last IP address must be greater than first')
     networks = []
     if first.version == 4:
         ip = IPv4Network
     elif first.version == 6:
         ip = IPv6Network
     else:
         raise ValueError('unknown IP version')
     ip_bits = first._max_prefixlen
     first_int = first._ip
     last_int = last._ip
     while first_int <= last_int:
         nbits = _count_righthand_zero_bits(first_int, ip_bits)
         current = None
         while nbits >= 0:
             addend = 2 ** nbits - 1
             current = first_int + addend
             nbits -= 1
             if current <= last_int:
                 break
         prefix = _get_prefix_length(first_int, current, ip_bits)
         net = ip('%s/%d' % (str(first), prefix))
         networks.append(net)
         if current == ip._ALL_ONES:
             break
         first_int = current + 1
         first = IPAddress(first_int, version=first._version)
     return networks
 def _collapse_address_list_recursive(addresses):
     """Loops through the addresses, collapsing concurrent netblocks.
     Example:
         ip1 = IPv4Network('1.1.0.0/24')
         ip2 = IPv4Network('1.1.1.0/24')
         ip3 = IPv4Network('1.1.2.0/24')
         ip4 = IPv4Network('1.1.3.0/24')
         ip5 = IPv4Network('1.1.4.0/24')
         ip6 = IPv4Network('1.1.0.1/22')
         _collapse_address_list_recursive([ip1, ip2, ip3, ip4, ip5, ip6]) ->
           [IPv4Network('1.1.0.0/22'), IPv4Network('1.1.4.0/24')]
         This shouldn't be called directly; it is called via
           collapse_address_list([]).
     Args:
         addresses: A list of IPv4Network's or IPv6Network's
     Returns:
         A list of IPv4Network's or IPv6Network's depending on what we were
         passed.
     """
     ret_array = []
     optimized = False
     for cur_addr in addresses:
         if not ret_array:
             ret_array.append(cur_addr)
             continue
         if cur_addr in ret_array[-1]:
             optimized = True
         elif cur_addr == ret_array[-1].supernet().subnet()[1]:
             ret_array.append(ret_array.pop().supernet())
             optimized = True
         else:
             ret_array.append(cur_addr)
     if optimized:
         return _collapse_address_list_recursive(ret_array)
     return ret_array
 def collapse_address_list(addresses):
     """Collapse a list of IP objects.
     Example:
         collapse_address_list([IPv4('1.1.0.0/24'), IPv4('1.1.1.0/24')]) ->
           [IPv4('1.1.0.0/23')]
     Args:
         addresses: A list of IPv4Network or IPv6Network objects.
     Returns:
         A list of IPv4Network or IPv6Network objects depending on what we
         were passed.
     Raises:
         TypeError: If passed a list of mixed version objects.
     """
     i = 0
     addrs = []
     ips = []
     nets = []
     # split IP addresses and networks
     for ip in addresses:
         if isinstance(ip, _BaseIP):
             if ips and ips[-1]._version != ip._version:
                 raise TypeError("%s and %s are not of the same version" % (
                         str(ip), str(ips[-1])))
             ips.append(ip)
         elif ip._prefixlen == ip._max_prefixlen:
             if ips and ips[-1]._version != ip._version:
                 raise TypeError("%s and %s are not of the same version" % (
                         str(ip), str(ips[-1])))
             ips.append(ip.ip)
         else:
             if nets and nets[-1]._version != ip._version:
                 raise TypeError("%s and %s are not of the same version" % (
                         str(ip), str(nets[-1])))
             nets.append(ip)
     # sort and dedup
     ips = sorted(set(ips))
     nets = sorted(set(nets))
     while i < len(ips):
         (first, last) = _find_address_range(ips[i:])
         i = ips.index(last) + 1
         addrs.extend(summarize_address_range(first, last))
     return _collapse_address_list_recursive(sorted(
         addrs + nets, key=_BaseNet._get_networks_key))
 # backwards compatibility
 CollapseAddrList = collapse_address_list
 # We need to distinguish between the string and packed-bytes representations
 # of an IP address.  For example, b'0::1' is the IPv4 address 48.58.58.49,
 # while '0::1' is an IPv6 address.
+#
 # In Python 3, the native 'bytes' type already provides this functionality,
 # so we use it directly.  For earlier implementations where bytes is not a
 # distinct type, we create a subclass of str to serve as a tag.
+#
 # Usage example (Python 2):
 #   ip = ipaddr.IPAddress(ipaddr.Bytes('xxxx'))
+#
 # Usage example (Python 3):
 #   ip = ipaddr.IPAddress(b'xxxx')
 try:
     if bytes is str:
         raise TypeError("bytes is not a distinct type")
     Bytes = bytes
 except (NameError, TypeError):
     class Bytes(str):
         def __repr__(self):
             return 'Bytes(%s)' % str.__repr__(self)
 def get_mixed_type_key(obj):
     """Return a key suitable for sorting between networks and addresses.
     Address and Network objects are not sortable by default; they're
     fundamentally different so the expression
         IPv4Address('1.1.1.1') <= IPv4Network('1.1.1.1/24')
     doesn't make any sense.  There are some times however, where you may wish
     to have ipaddr sort these for you anyway. If you need to do this, you
     can use this function as the key= argument to sorted().
     Args:
       obj: either a Network or Address object.
     Returns:
       appropriate key.
     """
     if isinstance(obj, _BaseNet):
         return obj._get_networks_key()
     elif isinstance(obj, _BaseIP):
         return obj._get_address_key()
     return NotImplemented
 class _IPAddrBase(object):
     """The mother class."""
     def __index__(self):
         return self._ip
     def __int__(self):
         return self._ip
     def __hex__(self):
         return hex(self._ip)
     @property
     def exploded(self):
         """Return the longhand version of the IP address as a string."""
         return self._explode_shorthand_ip_string()
     @property
     def compressed(self):
         """Return the shorthand version of the IP address as a string."""
         return str(self)
 class _BaseIP(_IPAddrBase):
     """A generic IP object.
     This IP class contains the version independent methods which are
     used by single IP addresses.
     """
     def __eq__(self, other):
         try:
             return (self._ip == other._ip
                     and self._version == other._version)
         except AttributeError:
             return NotImplemented
     def __ne__(self, other):
         eq = self.__eq__(other)
         if eq is NotImplemented:
             return NotImplemented
         return not eq
     def __le__(self, other):
         gt = self.__gt__(other)
         if gt is NotImplemented:
             return NotImplemented
         return not gt
     def __ge__(self, other):
         lt = self.__lt__(other)
         if lt is NotImplemented:
             return NotImplemented
         return not lt
     def __lt__(self, other):
         if self._version != other._version:
             raise TypeError('%s and %s are not of the same version' % (
                     str(self), str(other)))
         if not isinstance(other, _BaseIP):
             raise TypeError('%s and %s are not of the same type' % (
                     str(self), str(other)))
         if self._ip != other._ip:
             return self._ip < other._ip
         return False
     def __gt__(self, other):
         if self._version != other._version:
             raise TypeError('%s and %s are not of the same version' % (
                     str(self), str(other)))
         if not isinstance(other, _BaseIP):
             raise TypeError('%s and %s are not of the same type' % (
                     str(self), str(other)))
         if self._ip != other._ip:
             return self._ip > other._ip
         return False
     # Shorthand for Integer addition and subtraction. This is not
     # meant to ever support addition/subtraction of addresses.
     def __add__(self, other):
         if not isinstance(other, int):
             return NotImplemented
         return IPAddress(int(self) + other, version=self._version)
     def __sub__(self, other):
         if not isinstance(other, int):
             return NotImplemented
         return IPAddress(int(self) - other, version=self._version)
     def __repr__(self):
         return '%s(%r)' % (self.__class__.__name__, str(self))
     def __str__(self):
         return self._string_from_ip_int(self._ip)
     def __hash__(self):
         return hash(hex(long(self._ip)))
     def _get_address_key(self):
         return (self._version, self)
     @property
     def version(self):
         raise NotImplementedError('BaseIP has no version')
 class _BaseNet(_IPAddrBase):
     """A generic IP object.
     This IP class contains the version independent methods which are
     used by networks.
     """
     def __init__(self, address):
         self._cache = {}
     def __repr__(self):
         return '%s(%r)' % (self.__class__.__name__, str(self))
     def iterhosts(self):
         """Generate Iterator over usable hosts in a network.
            This is like __iter__ except it doesn't return the network
            or broadcast addresses.
         """
         cur = int(self.network) + 1
         bcast = int(self.broadcast) - 1
         while cur <= bcast:
             cur += 1
             yield IPAddress(cur - 1, version=self._version)
     def __iter__(self):
         cur = int(self.network)
         bcast = int(self.broadcast)
         while cur <= bcast:
             cur += 1
             yield IPAddress(cur - 1, version=self._version)
     def __getitem__(self, n):
         network = int(self.network)
         broadcast = int(self.broadcast)
         if n >= 0:
             if network + n > broadcast:
                 raise IndexError
             return IPAddress(network + n, version=self._version)
         else:
             n += 1
             if broadcast + n < network:
                 raise IndexError
             return IPAddress(broadcast + n, version=self._version)
     def __lt__(self, other):
         if self._version != other._version:
             raise TypeError('%s and %s are not of the same version' % (
                     str(self), str(other)))
         if not isinstance(other, _BaseNet):
             raise TypeError('%s and %s are not of the same type' % (
                     str(self), str(other)))
         if self.network != other.network:
             return self.network < other.network
         if self.netmask != other.netmask:
             return self.netmask < other.netmask
         return False
     def __gt__(self, other):
         if self._version != other._version:
             raise TypeError('%s and %s are not of the same version' % (
                     str(self), str(other)))
         if not isinstance(other, _BaseNet):
             raise TypeError('%s and %s are not of the same type' % (
                     str(self), str(other)))
         if self.network != other.network:
             return self.network > other.network
         if self.netmask != other.netmask:
             return self.netmask > other.netmask
         return False
     def __le__(self, other):
         gt = self.__gt__(other)
         if gt is NotImplemented:
             return NotImplemented
         return not gt
     def __ge__(self, other):
         lt = self.__lt__(other)
         if lt is NotImplemented:
             return NotImplemented
         return not lt
     def __eq__(self, other):
         try:
             return (self._version == other._version
                     and self.network == other.network
                     and int(self.netmask) == int(other.netmask))
         except AttributeError:
             if isinstance(other, _BaseIP):
                 return (self._version == other._version
                         and self._ip == other._ip)
     def __ne__(self, other):
         eq = self.__eq__(other)
         if eq is NotImplemented:
             return NotImplemented
         return not eq
     def __str__(self):
         return  '%s/%s' % (str(self.ip),
                            str(self._prefixlen))
     def __hash__(self):
         return hash(int(self.network) ^ int(self.netmask))
     def __contains__(self, other):
         # always false if one is v4 and the other is v6.
         if self._version != other._version:
             return False
         # dealing with another network.
         if isinstance(other, _BaseNet):
             return (self.network <= other.network and
                     self.broadcast >= other.broadcast)
         # dealing with another address
         else:
             return (int(self.network) <= int(other._ip) <=
                     int(self.broadcast))
     def overlaps(self, other):
         """Tell if self is partly contained in other."""
         return self.network in other or self.broadcast in other or (
             other.network in self or other.broadcast in self)
     @property
     def network(self):
         x = self._cache.get('network')
         if x is None:
             x = IPAddress(self._ip & int(self.netmask), version=self._version)
             self._cache['network'] = x
         return x
     @property
     def broadcast(self):
         x = self._cache.get('broadcast')
         if x is None:
             x = IPAddress(self._ip | int(self.hostmask), version=self._version)
             self._cache['broadcast'] = x
         return x
     @property
     def hostmask(self):
         x = self._cache.get('hostmask')
         if x is None:
             x = IPAddress(int(self.netmask) ^ self._ALL_ONES,
                           version=self._version)
             self._cache['hostmask'] = x
         return x
     @property
     def with_prefixlen(self):
         return '%s/%d' % (str(self.ip), self._prefixlen)
     @property
     def with_netmask(self):
         return '%s/%s' % (str(self.ip), str(self.netmask))
     @property
     def with_hostmask(self):
         return '%s/%s' % (str(self.ip), str(self.hostmask))
     @property
     def numhosts(self):
         """Number of hosts in the current subnet."""
         return int(self.broadcast) - int(self.network) + 1
     @property
     def version(self):
         raise NotImplementedError('BaseNet has no version')
     @property
     def prefixlen(self):
         return self._prefixlen
     def address_exclude(self, other):
         """Remove an address from a larger block.
         For example:
             addr1 = IPNetwork('10.1.1.0/24')
             addr2 = IPNetwork('10.1.1.0/26')
             addr1.address_exclude(addr2) =
                 [IPNetwork('10.1.1.64/26'), IPNetwork('10.1.1.128/25')]
         or IPv6:
             addr1 = IPNetwork('::1/32')
             addr2 = IPNetwork('::1/128')
             addr1.address_exclude(addr2) = [IPNetwork('::0/128'),
                 IPNetwork('::2/127'),
                 IPNetwork('::4/126'),
                 IPNetwork('::8/125'),
                 ...
                 IPNetwork('0:0:8000::/33')]
         Args:
             other: An IPvXNetwork object of the same type.
         Returns:
             A sorted list of IPvXNetwork objects addresses which is self
             minus other.
         Raises:
             TypeError: If self and other are of differing address
               versions, or if other is not a network object.
             ValueError: If other is not completely contained by self.
         """
         if not self._version == other._version:
             raise TypeError("%s and %s are not of the same version" % (
                 str(self), str(other)))
         if not isinstance(other, _BaseNet):
             raise TypeError("%s is not a network object" % str(other))
         if other not in self:
             raise ValueError('%s not contained in %s' % (str(other),
                                                          str(self)))
         if other == self:
             return []
         ret_addrs = []
         # Make sure we're comparing the network of other.
         other = IPNetwork('%s/%s' % (str(other.network), str(other.prefixlen)),
                    version=other._version)
         s1, s2 = self.subnet()
         while s1 != other and s2 != other:
             if other in s1:
                 ret_addrs.append(s2)
                 s1, s2 = s1.subnet()
             elif other in s2:
                 ret_addrs.append(s1)
                 s1, s2 = s2.subnet()
             else:
                 # If we got here, there's a bug somewhere.
                 assert True == False, ('Error performing exclusion: '
                                        's1: %s s2: %s other: %s' %
                                        (str(s1), str(s2), str(other)))
                 assert False, ('Error performing exclusion: '
                                's1: %s s2: %s other: %s' %
                                (str(s1), str(s2), str(other)))
         if s1 == other:
             ret_addrs.append(s2)
         elif s2 == other:
             ret_addrs.append(s1)
         else:
             # If we got here, there's a bug somewhere.
             assert True == False, ('Error performing exclusion: '
                                    's1: %s s2: %s other: %s' %
                                    (str(s1), str(s2), str(other)))
             assert False, ('Error performing exclusion: s1: %s s2: %s other: %s'
                            % (str(s1), str(s2), str(other)))
         return sorted(ret_addrs, key=_BaseNet._get_networks_key)
     def compare_networks(self, other):
         """Compare two IP objects.
         This is only concerned about the comparison of the integer
         representation of the network addresses.  This means that the
         host bits aren't considered at all in this method.  If you want
         to compare host bits, you can easily enough do a
         'HostA._ip < HostB._ip'
         Args:
             other: An IP object.
         Returns:
             If the IP versions of self and other are the same, returns:
             -1 if self < other:
               eg: IPv4('1.1.1.0/24') < IPv4('1.1.2.0/24')
               IPv6('1080::200C:417A') < IPv6('1080::200B:417B')
 if self == other
               eg: IPv4('1.1.1.1/24') == IPv4('1.1.1.2/24')
               IPv6('1080::200C:417A/96') == IPv6('1080::200C:417B/96')
 if self > other
               eg: IPv4('1.1.1.0/24') > IPv4('1.1.0.0/24')
               IPv6('1080::1:200C:417A/112') >
               IPv6('1080::0:200C:417A/112')
             If the IP versions of self and other are different, returns:
             -1 if self._version < other._version
               eg: IPv4('10.0.0.1/24') < IPv6('::1/128')
 if self._version > other._version
               eg: IPv6('::1/128') > IPv4('255.255.255.0/24')
         """
         if self._version < other._version:
             return -1
         if self._version > other._version:
             return 1
         # self._version == other._version below here:
         if self.network < other.network:
             return -1
         if self.network > other.network:
             return 1
         # self.network == other.network below here:
         if self.netmask < other.netmask:
             return -1
         if self.netmask > other.netmask:
             return 1
         # self.network == other.network and self.netmask == other.netmask
         return 0
     def _get_networks_key(self):
         """Network-only key function.
         Returns an object that identifies this address' network and
         netmask. This function is a suitable "key" argument for sorted()
         and list.sort().
         """
         return (self._version, self.network, self.netmask)
     def _ip_int_from_prefix(self, prefixlen=None):
         """Turn the prefix length netmask into a int for comparison.
         Args:
             prefixlen: An integer, the prefix length.
         Returns:
             An integer.
         """
         if not prefixlen and prefixlen != 0:
             prefixlen = self._prefixlen
         return self._ALL_ONES ^ (self._ALL_ONES >> prefixlen)
     def _prefix_from_ip_int(self, ip_int, mask=32):
         """Return prefix length from the decimal netmask.
         Args:
             ip_int: An integer, the IP address.
             mask: The netmask.  Defaults to 32.
         Returns:
             An integer, the prefix length.
         """
         while mask:
             if ip_int & 1 == 1:
                 break
             ip_int >>= 1
             mask -= 1
         return mask
     def _ip_string_from_prefix(self, prefixlen=None):
         """Turn a prefix length into a dotted decimal string.
         Args:
             prefixlen: An integer, the netmask prefix length.
         Returns:
             A string, the dotted decimal netmask string.
         """
         if not prefixlen:
             prefixlen = self._prefixlen
         return self._string_from_ip_int(self._ip_int_from_prefix(prefixlen))
     def iter_subnets(self, prefixlen_diff=1, new_prefix=None):
         """The subnets which join to make the current subnet.
         In the case that self contains only one IP
         (self._prefixlen == 32 for IPv4 or self._prefixlen == 128
         for IPv6), return a list with just ourself.
         Args:
             prefixlen_diff: An integer, the amount the prefix length
               should be increased by. This should not be set if
               new_prefix is also set.
             new_prefix: The desired new prefix length. This must be a
               larger number (smaller prefix) than the existing prefix.
               This should not be set if prefixlen_diff is also set.
         Returns:
             An iterator of IPv(4|6) objects.
         Raises:
             ValueError: The prefixlen_diff is too small or too large.
                 OR
             prefixlen_diff and new_prefix are both set or new_prefix
               is a smaller number than the current prefix (smaller
               number means a larger network)
         """
         if self._prefixlen == self._max_prefixlen:
             yield self
             return
         if new_prefix is not None:
             if new_prefix < self._prefixlen:
                 raise ValueError('new prefix must be longer')
             if prefixlen_diff != 1:
                 raise ValueError('cannot set prefixlen_diff and new_prefix')
             prefixlen_diff = new_prefix - self._prefixlen
         if prefixlen_diff < 0:
             raise ValueError('prefix length diff must be > 0')
         new_prefixlen = self._prefixlen + prefixlen_diff
         if not self._is_valid_netmask(str(new_prefixlen)):
             raise ValueError(
                 'prefix length diff %d is invalid for netblock %s' % (
                     new_prefixlen, str(self)))
         first = IPNetwork('%s/%s' % (str(self.network),
                                      str(self._prefixlen + prefixlen_diff)),
                          version=self._version)
         yield first
         current = first
         while True:
             broadcast = current.broadcast
             if broadcast == self.broadcast:
                 return
             new_addr = IPAddress(int(broadcast) + 1, version=self._version)
             current = IPNetwork('%s/%s' % (str(new_addr), str(new_prefixlen)),
                                 version=self._version)
             yield current
     def masked(self):
         """Return the network object with the host bits masked out."""
         return IPNetwork('%s/%d' % (self.network, self._prefixlen),
                          version=self._version)
     def subnet(self, prefixlen_diff=1, new_prefix=None):
         """Return a list of subnets, rather than an iterator."""
         return list(self.iter_subnets(prefixlen_diff, new_prefix))
     def supernet(self, prefixlen_diff=1, new_prefix=None):
         """The supernet containing the current network.
         Args:
             prefixlen_diff: An integer, the amount the prefix length of
               the network should be decreased by.  For example, given a
               /24 network and a prefixlen_diff of 3, a supernet with a
               /21 netmask is returned.
         Returns:
             An IPv4 network object.
         Raises:
             ValueError: If self.prefixlen - prefixlen_diff < 0. I.e., you have a
               negative prefix length.
                 OR
             If prefixlen_diff and new_prefix are both set or new_prefix is a
               larger number than the current prefix (larger number means a
               smaller network)
         """
         if self._prefixlen == 0:
             return self
         if new_prefix is not None:
             if new_prefix > self._prefixlen:
                 raise ValueError('new prefix must be shorter')
             if prefixlen_diff != 1:
                 raise ValueError('cannot set prefixlen_diff and new_prefix')
             prefixlen_diff = self._prefixlen - new_prefix
         if self.prefixlen - prefixlen_diff < 0:
             raise ValueError(
                 'current prefixlen is %d, cannot have a prefixlen_diff of %d' %
                 (self.prefixlen, prefixlen_diff))
         return IPNetwork('%s/%s' % (str(self.network),
                                     str(self.prefixlen - prefixlen_diff)),
                          version=self._version)
     # backwards compatibility
     Subnet = subnet
     Supernet = supernet
     AddressExclude = address_exclude
     CompareNetworks = compare_networks
     Contains = __contains__
 class _BaseV4(object):
     """Base IPv4 object.
     The following methods are used by IPv4 objects in both single IP
     addresses and networks.
     """
     # Equivalent to 255.255.255.255 or 32 bits of 1's.
     _ALL_ONES = (2 ** IPV4LENGTH) - 1
     _DECIMAL_DIGITS = frozenset('0123456789')
     def __init__(self, address):
         self._version = 4
         self._max_prefixlen = IPV4LENGTH
     def _explode_shorthand_ip_string(self):
         return str(self)
     def _ip_int_from_string(self, ip_str):
         """Turn the given IP string into an integer for comparison.
         Args:
             ip_str: A string, the IP ip_str.
         Returns:
             The IP ip_str as an integer.
         Raises:
             AddressValueError: if ip_str isn't a valid IPv4 Address.
         """
         octets = ip_str.split('.')
         if len(octets) != 4:
             raise AddressValueError(ip_str)
         packed_ip = 0
         for oc in octets:
             try:
                 packed_ip = (packed_ip << 8) | self._parse_octet(oc)
             except ValueError:
                 raise AddressValueError(ip_str)
         return packed_ip
     def _parse_octet(self, octet_str):
         """Convert a decimal octet into an integer.
         Args:
             octet_str: A string, the number to parse.
         Returns:
             The octet as an integer.
         Raises:
             ValueError: if the octet isn't strictly a decimal from [0..255].
         """
         # Whitelist the characters, since int() allows a lot of bizarre stuff.
         if not self._DECIMAL_DIGITS.issuperset(octet_str):
             raise ValueError
         octet_int = int(octet_str, 10)
         # Disallow leading zeroes, because no clear standard exists on
         # whether these should be interpreted as decimal or octal.
         if octet_int > 255 or (octet_str[0] == '0' and len(octet_str) > 1):
             raise ValueError
         return octet_int
     def _string_from_ip_int(self, ip_int):
         """Turns a 32-bit integer into dotted decimal notation.
         Args:
             ip_int: An integer, the IP address.
         Returns:
             The IP address as a string in dotted decimal notation.
         """
         octets = []
         for _ in xrange(4):
             octets.insert(0, str(ip_int & 0xFF))
             ip_int >>= 8
         return '.'.join(octets)
     @property
     def max_prefixlen(self):
         return self._max_prefixlen
     @property
     def packed(self):
         """The binary representation of this address."""
         return v4_int_to_packed(self._ip)
     @property
     def version(self):
         return self._version
     @property
     def is_reserved(self):
         """Test if the address is otherwise IETF reserved.
          Returns:
              A boolean, True if the address is within the
              reserved IPv4 Network range.
         """
         return self in IPv4Network('240.0.0.0/4')
     @property
     def is_private(self):
         """Test if this address is allocated for private networks.
         Returns:
             A boolean, True if the address is reserved per RFC 1918.
         """
         return (self in IPv4Network('10.0.0.0/8') or
                 self in IPv4Network('172.16.0.0/12') or
                 self in IPv4Network('192.168.0.0/16'))
     @property
     def is_multicast(self):
         """Test if the address is reserved for multicast use.
         Returns:
             A boolean, True if the address is multicast.
             See RFC 3171 for details.
         """
         return self in IPv4Network('224.0.0.0/4')
     @property
     def is_unspecified(self):
         """Test if the address is unspecified.
         Returns:
             A boolean, True if this is the unspecified address as defined in
             RFC 5735 3.
         """
         return self in IPv4Network('0.0.0.0')
     @property
     def is_loopback(self):
         """Test if the address is a loopback address.
         Returns:
             A boolean, True if the address is a loopback per RFC 3330.
         """
         return self in IPv4Network('127.0.0.0/8')
     @property
     def is_link_local(self):
         """Test if the address is reserved for link-local.
         Returns:
             A boolean, True if the address is link-local per RFC 3927.
         """
         return self in IPv4Network('169.254.0.0/16')
 class IPv4Address(_BaseV4, _BaseIP):
     """Represent and manipulate single IPv4 Addresses."""
     def __init__(self, address):
         """
         Args:
             address: A string or integer representing the IP
               '192.168.1.1'
               Additionally, an integer can be passed, so
               IPv4Address('192.168.1.1') == IPv4Address(3232235777).
               or, more generally
               IPv4Address(int(IPv4Address('192.168.1.1'))) ==
                 IPv4Address('192.168.1.1')
         Raises:
             AddressValueError: If ipaddr isn't a valid IPv4 address.
         """
         _BaseV4.__init__(self, address)
         # Efficient constructor from integer.
         if isinstance(address, (int, long)):
             self._ip = address
             if address < 0 or address > self._ALL_ONES:
                 raise AddressValueError(address)
             return
         # Constructing from a packed address
         if isinstance(address, Bytes):
             try:
                 self._ip, = struct.unpack('!I', address)
             except struct.error:
                 raise AddressValueError(address)  # Wrong length.
             return
         # Assume input argument to be string or any object representation
         # which converts into a formatted IP string.
         addr_str = str(address)
         self._ip = self._ip_int_from_string(addr_str)
 class IPv4Network(_BaseV4, _BaseNet):
     """This class represents and manipulates 32-bit IPv4 networks.
     Attributes: [examples for IPv4Network('1.2.3.4/27')]
         ._ip: 16909060
         .ip: IPv4Address('1.2.3.4')
         .network: IPv4Address('1.2.3.0')
         .hostmask: IPv4Address('0.0.0.31')
         .broadcast: IPv4Address('1.2.3.31')
         .netmask: IPv4Address('255.255.255.224')
         .prefixlen: 27
     """
     # the valid octets for host and netmasks. only useful for IPv4.
     _valid_mask_octets = set((255, 254, 252, 248, 240, 224, 192, 128, 0))
     def __init__(self, address, strict=False):
         """Instantiate a new IPv4 network object.
         Args:
             address: A string or integer representing the IP [& network].
               '192.168.1.1/24'
               '192.168.1.1/255.255.255.0'
               '192.168.1.1/0.0.0.255'
               are all functionally the same in IPv4. Similarly,
               '192.168.1.1'
               '192.168.1.1/255.255.255.255'
               '192.168.1.1/32'
               are also functionally equivalent. That is to say, failing to
               provide a subnetmask will create an object with a mask of /32.
               If the mask (portion after the / in the argument) is given in
               dotted quad form, it is treated as a netmask if it starts with a
               non-zero field (e.g. /255.0.0.0 == /8) and as a hostmask if it
               starts with a zero field (e.g. 0.255.255.255 == /8), with the
               single exception of an all-zero mask which is treated as a
               netmask == /0. If no mask is given, a default of /32 is used.
               Additionally, an integer can be passed, so
               IPv4Network('192.168.1.1') == IPv4Network(3232235777).
               or, more generally
               IPv4Network(int(IPv4Network('192.168.1.1'))) ==
                 IPv4Network('192.168.1.1')
             strict: A boolean. If true, ensure that we have been passed
               A true network address, eg, 192.168.1.0/24 and not an
               IP address on a network, eg, 192.168.1.1/24.
         Raises:
             AddressValueError: If ipaddr isn't a valid IPv4 address.
             NetmaskValueError: If the netmask isn't valid for
               an IPv4 address.
             ValueError: If strict was True and a network address was not
               supplied.
         """
         _BaseNet.__init__(self, address)
         _BaseV4.__init__(self, address)
         # Constructing from an integer or packed bytes.
         if isinstance(address, (int, long, Bytes)):
             self.ip = IPv4Address(address)
             self._ip = self.ip._ip
             self._prefixlen = self._max_prefixlen
             self.netmask = IPv4Address(self._ALL_ONES)
             return
         # Assume input argument to be string or any object representation
         # which converts into a formatted IP prefix string.
         addr = str(address).split('/')
         if len(addr) > 2:
             raise AddressValueError(address)
         self._ip = self._ip_int_from_string(addr[0])
         self.ip = IPv4Address(self._ip)
         if len(addr) == 2:
             mask = addr[1].split('.')
             if len(mask) == 4:
                 # We have dotted decimal netmask.
                 if self._is_valid_netmask(addr[1]):
                     self.netmask = IPv4Address(self._ip_int_from_string(
                             addr[1]))
                 elif self._is_hostmask(addr[1]):
                     self.netmask = IPv4Address(
                         self._ip_int_from_string(addr[1]) ^ self._ALL_ONES)
                 else:
                     raise NetmaskValueError('%s is not a valid netmask'
                                                      % addr[1])
                 self._prefixlen = self._prefix_from_ip_int(int(self.netmask))
             else:
                 # We have a netmask in prefix length form.
                 if not self._is_valid_netmask(addr[1]):
                     raise NetmaskValueError(addr[1])
                 self._prefixlen = int(addr[1])
                 self.netmask = IPv4Address(self._ip_int_from_prefix(
                     self._prefixlen))
         else:
             self._prefixlen = self._max_prefixlen
             self.netmask = IPv4Address(self._ip_int_from_prefix(
                 self._prefixlen))
         if strict:
             if self.ip != self.network:
                 raise ValueError('%s has host bits set' %
                                  self.ip)
         if self._prefixlen == (self._max_prefixlen - 1):
             self.iterhosts = self.__iter__
     def _is_hostmask(self, ip_str):
         """Test if the IP string is a hostmask (rather than a netmask).
         Args:
             ip_str: A string, the potential hostmask.
         Returns:
             A boolean, True if the IP string is a hostmask.
         """
         bits = ip_str.split('.')
         try:
             parts = [int(x) for x in bits if int(x) in self._valid_mask_octets]
         except ValueError:
             return False
         if len(parts) != len(bits):
             return False
         if parts[0] < parts[-1]:
             return True
         return False
     def _is_valid_netmask(self, netmask):
         """Verify that the netmask is valid.
         Args:
             netmask: A string, either a prefix or dotted decimal
               netmask.
         Returns:
             A boolean, True if the prefix represents a valid IPv4
             netmask.
         """
         mask = netmask.split('.')
         if len(mask) == 4:
             if [x for x in mask if int(x) not in self._valid_mask_octets]:
                 return False
             if [y for idx, y in enumerate(mask) if idx > 0 and
                 y > mask[idx - 1]]:
                 return False
             return True
         try:
             netmask = int(netmask)
         except ValueError:
             return False
         return 0 <= netmask <= self._max_prefixlen
     # backwards compatibility
     IsRFC1918 = lambda self: self.is_private
     IsMulticast = lambda self: self.is_multicast
     IsLoopback = lambda self: self.is_loopback
     IsLinkLocal = lambda self: self.is_link_local
 class _BaseV6(object):
     """Base IPv6 object.
     The following methods are used by IPv6 objects in both single IP
     addresses and networks.
     """
     _ALL_ONES = (2 ** IPV6LENGTH) - 1
     _HEXTET_COUNT = 8
     _HEX_DIGITS = frozenset('0123456789ABCDEFabcdef')
     def __init__(self, address):
         self._version = 6
         self._max_prefixlen = IPV6LENGTH
     def _ip_int_from_string(self, ip_str):
         """Turn an IPv6 ip_str into an integer.
         Args:
             ip_str: A string, the IPv6 ip_str.
         Returns:
             A long, the IPv6 ip_str.
         Raises:
             AddressValueError: if ip_str isn't a valid IPv6 Address.
         """
         parts = ip_str.split(':')
         # An IPv6 address needs at least 2 colons (3 parts).
         if len(parts) < 3:
             raise AddressValueError(ip_str)
         # If the address has an IPv4-style suffix, convert it to hexadecimal.
         if '.' in parts[-1]:
             ipv4_int = IPv4Address(parts.pop())._ip
             parts.append('%x' % ((ipv4_int >> 16) & 0xFFFF))
             parts.append('%x' % (ipv4_int & 0xFFFF))
         # An IPv6 address can't have more than 8 colons (9 parts).
         if len(parts) > self._HEXTET_COUNT + 1:
             raise AddressValueError(ip_str)
         # Disregarding the endpoints, find '::' with nothing in between.
         # This indicates that a run of zeroes has been skipped.
         try:
             skip_index, = (
                 [i for i in xrange(1, len(parts) - 1) if not parts[i]] or
                 [None])
         except ValueError:
             # Can't have more than one '::'
             raise AddressValueError(ip_str)
         # parts_hi is the number of parts to copy from above/before the '::'
         # parts_lo is the number of parts to copy from below/after the '::'
         if skip_index is not None:
             # If we found a '::', then check if it also covers the endpoints.
             parts_hi = skip_index
             parts_lo = len(parts) - skip_index - 1
             if not parts[0]:
                 parts_hi -= 1
                 if parts_hi:
                     raise AddressValueError(ip_str)  # ^: requires ^::
             if not parts[-1]:
                 parts_lo -= 1
                 if parts_lo:
                     raise AddressValueError(ip_str)  # :$ requires ::$
             parts_skipped = self._HEXTET_COUNT - (parts_hi + parts_lo)
             if parts_skipped < 1:
                 raise AddressValueError(ip_str)
         else:
             # Otherwise, allocate the entire address to parts_hi.  The endpoints
             # could still be empty, but _parse_hextet() will check for that.
             if len(parts) != self._HEXTET_COUNT:
                 raise AddressValueError(ip_str)
             parts_hi = len(parts)
             parts_lo = 0
             parts_skipped = 0
         try:
             # Now, parse the hextets into a 128-bit integer.
             ip_int = 0L
             for i in xrange(parts_hi):
                 ip_int <<= 16
                 ip_int |= self._parse_hextet(parts[i])
             ip_int <<= 16 * parts_skipped
             for i in xrange(-parts_lo, 0):
                 ip_int <<= 16
                 ip_int |= self._parse_hextet(parts[i])
             return ip_int
         except ValueError:
             raise AddressValueError(ip_str)
     def _parse_hextet(self, hextet_str):
         """Convert an IPv6 hextet string into an integer.
         Args:
             hextet_str: A string, the number to parse.
         Returns:
             The hextet as an integer.
         Raises:
             ValueError: if the input isn't strictly a hex number from [0..FFFF].
         """
         # Whitelist the characters, since int() allows a lot of bizarre stuff.
         if not self._HEX_DIGITS.issuperset(hextet_str):
             raise ValueError
         if len(hextet_str) > 4:
             raise ValueError
         hextet_int = int(hextet_str, 16)
         if hextet_int > 0xFFFF:
             raise ValueError
         return hextet_int
     def _compress_hextets(self, hextets):
         """Compresses a list of hextets.
         Compresses a list of strings, replacing the longest continuous
         sequence of "0" in the list with "" and adding empty strings at
         the beginning or at the end of the string such that subsequently
         calling ":".join(hextets) will produce the compressed version of
         the IPv6 address.
         Args:
             hextets: A list of strings, the hextets to compress.
         Returns:
             A list of strings.
         """
         best_doublecolon_start = -1
         best_doublecolon_len = 0
         doublecolon_start = -1
         doublecolon_len = 0
         for index in range(len(hextets)):
             if hextets[index] == '0':
                 doublecolon_len += 1
                 if doublecolon_start == -1:
                     # Start of a sequence of zeros.
                     doublecolon_start = index
                 if doublecolon_len > best_doublecolon_len:
                     # This is the longest sequence of zeros so far.
                     best_doublecolon_len = doublecolon_len
                     best_doublecolon_start = doublecolon_start
             else:
                 doublecolon_len = 0
                 doublecolon_start = -1
         if best_doublecolon_len > 1:
             best_doublecolon_end = (best_doublecolon_start +
                                     best_doublecolon_len)
             # For zeros at the end of the address.
             if best_doublecolon_end == len(hextets):
                 hextets += ['']
             hextets[best_doublecolon_start:best_doublecolon_end] = ['']
             # For zeros at the beginning of the address.
             if best_doublecolon_start == 0:
                 hextets = [''] + hextets
         return hextets

kallithea/tests/api/api_base.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 Tests for the JSON-RPC web api.
 """
 import os
 import random
 import mock
 import re
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.compat import json
 from kallithea.lib.auth import AuthUser
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.meta import Session
 from kallithea.model.scm import ScmModel
 from kallithea.model.gist import GistModel
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.db import Repository, User, Setting, Ui, PullRequest, ChangesetStatus
 from kallithea.lib.utils2 import time_to_datetime
 API_URL = '/_admin/api'
 TEST_USER_GROUP = u'test_user_group'
 TEST_REPO_GROUP = u'test_repo_group'
 fixture = Fixture()
 def _build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     :param random_id:
     """
     random_id = random.randrange(1, 9999)
     return random_id, json.dumps({
         "id": random_id,
         "api_key": apikey,
         "method": method,
         "args": kw
     })
 jsonify = lambda obj: json.loads(json.dumps(obj))
 def crash(*args, **kwargs):
     raise Exception('Total Crash !')
 def api_call(test_obj, params):
     response = test_obj.app.post(API_URL, content_type='application/json',
                                  params=params)
     return response
 ## helpers
 def make_user_group(name=TEST_USER_GROUP):
     gr = fixture.create_user_group(name, cur_user=TEST_USER_ADMIN_LOGIN)
     UserGroupModel().add_user_to_group(user_group=gr,
                                        user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 def make_repo_group(name=TEST_REPO_GROUP):
     gr = fixture.create_repo_group(name, cur_user=TEST_USER_ADMIN_LOGIN)
     Session().commit()
     return gr
 class _BaseTestApi(object):
     REPO = None
     REPO_TYPE = None
     @classmethod
     def setup_class(cls):
         cls.usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         cls.apikey = cls.usr.api_key
         cls.test_user = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='test@example.com',
             firstname=u'first',
             lastname=u'last'
+        )
         Session().commit()
         cls.TEST_USER_LOGIN = cls.test_user.username
         cls.apikey_regular = cls.test_user.api_key
     @classmethod
     def teardown_class(cls):
         pass
     def setup_method(self, method):
         make_user_group()
         make_repo_group()
     def teardown_method(self, method):
         fixture.destroy_user_group(TEST_USER_GROUP)
         fixture.destroy_gists()
         fixture.destroy_repo_group(TEST_REPO_GROUP)
     def _compare_ok(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': None,
             'result': expected
         })
         given = json.loads(given)
         assert expected == given
     def _compare_error(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': expected,
             'result': None
         })
         given = json.loads(given)
         assert expected == given
     def test_Optional_object(self):
         from kallithea.controllers.api.api import Optional
         option1 = Optional(None)
         assert '<Optional:%s>' % None == repr(option1)
-        assert option1() == None
+        assert option1() is None
         assert 1 == Optional.extract(Optional(1))
         assert 'trololo' == Optional.extract('trololo')
     def test_Optional_OAttr(self):
         from kallithea.controllers.api.api import Optional, OAttr
         option1 = Optional(OAttr('apiuser'))
         assert 'apiuser' == Optional.extract(option1)
     def test_OAttr_object(self):
         from kallithea.controllers.api.api import OAttr
         oattr1 = OAttr('apiuser')
         assert '<OptionalAttr:apiuser>' == repr(oattr1)
         assert oattr1() == oattr1
     def test_api_wrong_key(self):
         id_, params = _build_data('trololo', 'get_user')
         response = api_call(self, params)
         expected = 'Invalid API key'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_null(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_missing_non_optional_param_args_bad(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
     def test_api_args_is_null(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         assert response.status == '200 OK'
     def test_api_args_is_bad(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         assert response.status == '200 OK'
     def test_api_args_different_args(self):
         import string
         expected = {
             'ascii_letters': string.ascii_letters,
             'ws': string.whitespace,
             'printables': string.printable
+        }
         id_, params = _build_data(self.apikey, 'test', args=expected)
         response = api_call(self, params)
         assert response.status == '200 OK'
         self._compare_ok(id_, expected, response.body)
     def test_api_get_users(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         response = api_call(self, params)
         ret_all = []
         _users = User.query().filter_by(is_default_user=False) \
             .order_by(User.username).all()
         for usr in _users:
             ret = usr.get_api_data()
             ret_all.append(jsonify(ret))
         expected = ret_all
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_that_does_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid='trololo')
         response = api_call(self, params)
         expected = "user `%s` does not exist" % 'trololo'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid(self):
         id_, params = _build_data(self.apikey, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user')
         response = api_call(self, params)
         usr = User.get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_pull(self):
         repo_name = u'test_pull'
         r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         r.clone_uri = os.path.join(Ui.get_by_key('paths', '/').ui_value, self.REPO)
         Session().commit()
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=repo_name,)
         response = api_call(self, params)
         expected = {'msg': 'Pulled from `%s`' % repo_name,
                     'repository': repo_name}
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_pull_error(self):
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=self.REPO, )
         response = api_call(self, params)
         expected = 'Unable to pull changes from `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_rescan_repos(self):
         id_, params = _build_data(self.apikey, 'rescan_repos')
         response = api_call(self, params)
         expected = {'added': [], 'removed': []}
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'repo_scan', crash)
     def test_api_rescann_error(self):
         id_, params = _build_data(self.apikey, 'rescan_repos', )
         response = api_call(self, params)
         expected = 'Error occurred during rescan repositories action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_invalidate_cache(self):
         repo = RepoModel().get_by_repo_name(self.REPO)
         repo.scm_instance_cached()  # seed cache
         id_, params = _build_data(self.apikey, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = {
             'msg': "Cache for repository `%s` was invalidated" % (self.REPO,),
             'repository': self.REPO
+        }
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(ScmModel, 'mark_for_invalidation', crash)
     def test_api_invalidate_cache_error(self):
         id_, params = _build_data(self.apikey, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = 'Error occurred during cache invalidation action'
         self._compare_error(id_, expected, given=response.body)
     def test_api_invalidate_cache_regular_user_no_permission(self):
         repo = RepoModel().get_by_repo_name(self.REPO)
         repo.scm_instance_cached() # seed cache
         id_, params = _build_data(self.apikey_regular, 'invalidate_cache',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = "repository `%s` does not exist" % (self.REPO,)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_acquire(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO, 'locked': True,
             'locked_since': response.json['result']['locked_since'],
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_acquire_by_non_admin(self):
         repo_name = u'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = {
                 'repo': repo_name,
                 'locked': True,
                 'locked_since': response.json['result']['locked_since'],
                 'locked_by': self.TEST_USER_LOGIN,
                 'lock_state_changed': True,
                 'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                         % (self.TEST_USER_LOGIN, repo_name, True))
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_lock_repo_lock_acquire_non_admin_with_userid(self):
         repo_name = u'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         try:
             id_, params = _build_data(self.apikey_regular, 'lock',
                                       userid=TEST_USER_ADMIN_LOGIN,
                                       repoid=repo_name,
                                       locked=True)
             response = api_call(self, params)
             expected = 'userid is not the same as your user'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_lock_repo_lock_acquire_non_admin_not_his_repo(self):
         id_, params = _build_data(self.apikey_regular, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
     def test_api_lock_repo_lock_release(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=False)
         response = api_call(self, params)
         expected = {
             'repo': self.REPO,
             'locked': False,
             'locked_since': None,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, False))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_acquire_optional_userid(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         time_ = response.json['result']['locked_since']
         expected = {
             'repo': self.REPO,
             'locked': True,
             'locked_since': time_,
             'locked_by': TEST_USER_ADMIN_LOGIN,
             'lock_state_changed': True,
             'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                     % (TEST_USER_ADMIN_LOGIN, self.REPO, True))
+        }
         self._compare_ok(id_, expected, given=response.body)
     def test_api_lock_repo_lock_optional_locked(self):
         try:
             id_, params = _build_data(self.apikey, 'lock',
                                       repoid=self.REPO)
             response = api_call(self, params)
             time_ = response.json['result']['locked_since']
             expected = {
                 'repo': self.REPO,
                 'locked': True,
                 'locked_since': time_,
                 'locked_by': TEST_USER_ADMIN_LOGIN,
                 'lock_state_changed': False,
                 'msg': ('Repo `%s` locked by `%s` on `%s`.'
                         % (self.REPO, TEST_USER_ADMIN_LOGIN,
                            json.dumps(time_to_datetime(time_))))
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             # cleanup
             Repository.unlock(RepoModel().get_by_repo_name(self.REPO))
     def test_api_lock_repo_lock_optional_not_locked(self):
         repo_name = u'api_not_locked'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         assert repo.locked == [None, None]
         try:
             id_, params = _build_data(self.apikey, 'lock',
                                       repoid=repo.repo_id)
             response = api_call(self, params)
             expected = {
                 'repo': repo_name,
                 'locked': False,
                 'locked_since': None,
                 'locked_by': None,
                 'lock_state_changed': False,
                 'msg': ('Repo `%s` not locked.' % (repo_name,))
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     @mock.patch.object(Repository, 'lock', crash)
     def test_api_lock_error(self):
         id_, params = _build_data(self.apikey, 'lock',
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   repoid=self.REPO,
                                   locked=True)
         response = api_call(self, params)
         expected = 'Error occurred locking repository `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_userid_regular_user(self):
         id_, params = _build_data(self.apikey_regular, 'get_locks',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_locks(self):
         id_, params = _build_data(self.apikey, 'get_locks')
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_locks_with_one_locked_repo(self):
         repo_name = u'api_delete_me'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                                    cur_user=self.TEST_USER_LOGIN)
         Repository.lock(repo, User.get_by_username(self.TEST_USER_LOGIN).user_id)
         try:
             id_, params = _build_data(self.apikey, 'get_locks')
             response = api_call(self, params)
             expected = [repo.get_api_data()]
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_get_locks_with_one_locked_repo_for_specific_user(self):
         repo_name = u'api_delete_me'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                                    cur_user=self.TEST_USER_LOGIN)
         Repository.lock(repo, User.get_by_username(self.TEST_USER_LOGIN).user_id)
         try:
             id_, params = _build_data(self.apikey, 'get_locks',
                                       userid=self.TEST_USER_LOGIN)
             response = api_call(self, params)
             expected = [repo.get_api_data()]
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_get_locks_with_userid(self):
         id_, params = _build_data(self.apikey, 'get_locks',
                                   userid=TEST_USER_REGULAR_LOGIN)
         response = api_call(self, params)
         expected = []
         self._compare_ok(id_, expected, given=response.body)
     def test_api_create_existing_user(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN,
                                   email='test@example.com',
                                   password='trololo')
         response = api_call(self, params)
         expected = "user `%s` already exist" % TEST_USER_ADMIN_LOGIN
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user_with_existing_email(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=TEST_USER_ADMIN_LOGIN + 'new',
                                   email=TEST_USER_REGULAR_EMAIL,
                                   password='trololo')
         response = api_call(self, params)
         expected = "email `%s` already exist" % TEST_USER_REGULAR_EMAIL
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         usr = User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email)
         response = api_call(self, params)
         usr = User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     def test_api_create_user_with_extern_name(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email, extern_name='internal')
         response = api_call(self, params)
         usr = User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
+        )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
     @mock.patch.object(UserModel, 'create_or_update', crash)
     def test_api_create_user_when_exception_happened(self):
         username = 'test_new_api_user'
         email = username + "@example.com"
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         expected = 'failed to create user `%s`' % username
         self._compare_error(id_, expected, given=response.body)
     def test_api_delete_user(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'u232@example.com',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         username = usr.username
         email = usr.email
         usr_id = usr.user_id
         ## DELETE THIS USER NOW
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username, )
         response = api_call(self, params)
         ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
                'user': None}
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'delete', crash)
     def test_api_delete_user_when_exception_happened(self):
         usr = UserModel().create_or_update(username=u'test_user',
                                            password=u'qweqwe',
                                            email=u'u232@example.com',
                                            firstname=u'u1', lastname=u'u1')
         Session().commit()
         username = usr.username
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username, )
         response = api_call(self, params)
         ret = 'failed to delete user ID:%s %s' % (usr.user_id,
                                                   usr.username)
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     @parametrize('name,expected', [
         ('firstname', 'new_username'),
         ('lastname', 'new_username'),
         ('email', 'new_username'),
         ('admin', True),
         ('admin', False),
         ('extern_type', 'ldap'),
         ('extern_type', None),
         ('extern_name', 'test'),
         ('extern_name', None),
         ('active', False),
         ('active', True),
         ('password', 'newpass'),
     ])
     def test_api_update_user(self, name, expected):
         usr = User.get_by_username(self.TEST_USER_LOGIN)
         kw = {name: expected,
               'userid': usr.user_id}
         id_, params = _build_data(self.apikey, 'update_user', **kw)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, self.TEST_USER_LOGIN),
             'user': jsonify(User \
                 .get_by_username(self.TEST_USER_LOGIN) \
                 .get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_no_changed_params(self):
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, TEST_USER_ADMIN_LOGIN),
             'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_by_user_id(self):
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, TEST_USER_ADMIN_LOGIN),
             'user': ret
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_update_user_default_user(self):
         usr = User.get_default_user()
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         expected = 'editing default user is forbidden'
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(UserModel, 'update_user', crash)
     def test_api_update_user_when_exception_happens(self):
         usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
         response = api_call(self, params)
         ret = 'failed to update user `%s`' % usr.user_id
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo(self):
         new_group = u'some_new_group'
         make_user_group(new_group)
         RepoModel().grant_user_group_permission(repo=self.REPO,
                                                 group_name=new_group,
                                                 perm='repository.read')
         Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         assert u"tags" not in response.json[u'result']
         assert u'pull_requests' not in response.json[u'result']
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         followers = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = {'name': user.username, 'type': "user",
                          'permission': perm}
             members.append(user_data)
         for user_group in repo.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group = user_group.users_group
             user_group_data = {'name': user_group.users_group_name,
                                'type': "user_group", 'permission': perm}
             members.append(user_group_data)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         ret['members'] = members
         ret['followers'] = followers
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_user_group(new_group)
         id_, params = _build_data(self.apikey, 'get_repo', repoid=self.REPO,
                                   with_revision_names=True,
                                   with_pullrequests=True)
         response = api_call(self, params)
         assert u"v0.2.0" in response.json[u'result'][u'tags']
         assert u'pull_requests' in response.json[u'result']
     @parametrize('grant_perm', [
         ('repository.admin'),
         ('repository.write'),
         ('repository.read'),
     ])
     def test_api_get_repo_by_non_admin(self, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         Session().commit()
         id_, params = _build_data(self.apikey_regular, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
         members = []
         followers = []
         assert 2 == len(repo.repo_to_perm)
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user_obj = user.user
             user_data = {'name': user_obj.username, 'type': "user",
                          'permission': perm}
             members.append(user_data)
         for user_group in repo.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group_obj = user_group.users_group
             user_group_data = {'name': user_group_obj.users_group_name,
                                'type': "user_group", 'permission': perm}
             members.append(user_group_data)
         for user in repo.followers:
             followers.append(user.user.get_api_data())
         ret['members'] = members
         ret['followers'] = followers
         expected = ret
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
     def test_api_get_repo_by_non_admin_no_permission_to_repo(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.none')
         id_, params = _build_data(self.apikey_regular, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_that_doesn_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid='no-such-repo')
         response = api_call(self, params)
         ret = 'repository `%s` does not exist' % 'no-such-repo'
         expected = ret
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repos(self):
         id_, params = _build_data(self.apikey, 'get_repos')
         response = api_call(self, params)
         expected = jsonify([
             repo.get_api_data()
             for repo in Repository.query()
         ])
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repos_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_repos')
         response = api_call(self, params)
         expected = jsonify([
             repo.get_api_data()
             for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN)
         ])
         self._compare_ok(id_, expected, given=response.body)
     @parametrize('name,ret_type', [
         ('all', 'all'),
         ('dirs', 'dirs'),
         ('files', 'files'),
     ])
     def test_api_get_repo_nodes(self, name, ret_type):
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_revisions(self):
         rev = 'i-dont-exist'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_path(self):
         rev = 'tip'
         path = '/idontexits'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_repo_nodes_bad_ret_type(self):
         rev = 'tip'
         path = '/'
         ret_type = 'error'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         expected = ('ret_type must be one of %s'
                     % (','.join(['files', 'dirs', 'all'])))
         self._compare_error(id_, expected, given=response.body)
     @parametrize('name,ret_type,grant_perm', [
         ('all', 'all', 'repository.write'),
         ('dirs', 'dirs', 'repository.admin'),
         ('files', 'files', 'repository.read'),
     ])
     def test_api_get_repo_nodes_by_regular_user(self, name, ret_type, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         Session().commit()
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey_regular, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
     def test_api_create_repo(self):
         repo_name = u'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
-        assert repo != None
+        assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_and_repo_group(self):
         repo_name = u'my_gr/api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         print params
         repo = RepoModel().get_by_repo_name(repo_name)
-        assert repo != None
+        assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(u'my_gr')
     def test_api_create_repo_in_repo_group_without_permission(self):
         repo_group_name = u'%s/api-repo-repo' % TEST_REPO_GROUP
         repo_name = u'%s/api-repo' % repo_group_name
         rg = fixture.create_repo_group(repo_group_name)
         Session().commit()
         RepoGroupModel().grant_user_permission(repo_group_name,
                                                self.TEST_USER_LOGIN,
                                                'group.none')
         Session().commit()
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         # Current result when API access control is different from Web:
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
         # Expected and arguably more correct result:
         #expected = 'failed to create repository `%s`' % repo_name
         #self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo_group(repo_group_name)
     def test_api_create_repo_unknown_owner(self):
         repo_name = u'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=owner,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_create_repo_dont_specify_owner(self):
         repo_name = u'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
-        assert repo != None
+        assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_by_non_admin(self):
         repo_name = u'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
+        )
         response = api_call(self, params)
         repo = RepoModel().get_by_repo_name(repo_name)
-        assert repo != None
+        assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_by_non_admin_specify_owner(self):
         repo_name = u'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
                                   owner=owner)
         response = api_call(self, params)
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
     def test_api_create_repo_exists(self):
         repo_name = self.REPO
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = "repo `%s` already exist" % repo_name
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'create', crash)
     def test_api_create_repo_exception_occurred(self):
         repo_name = u'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = 'failed to create repository `%s`' % repo_name
         self._compare_error(id_, expected, given=response.body)
     @parametrize('changing_attr,updates', [
         ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': u'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}),
         ('clone_uri', {'clone_uri': None}),
         ('landing_rev', {'landing_rev': 'branch:master'}),
         ('enable_statistics', {'enable_statistics': True}),
         ('enable_locking', {'enable_locking': True}),
         ('enable_downloads', {'enable_downloads': True}),
         ('name', {'name': u'new_repo_name'}),
         ('repo_group', {'group': u'test_group_for_update'}),
     ])
     def test_api_update_repo(self, changing_attr, updates):
         repo_name = u'api_update_me'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         if changing_attr == 'repo_group':
             fixture.create_repo_group(updates['group'])
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         if changing_attr == 'name':
             repo_name = updates['name']
         if changing_attr == 'repo_group':
             repo_name = u'/'.join([updates['group'], repo_name])
         try:
             expected = {
                 'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
                 'repository': repo.get_api_data()
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(updates['group'])
     @parametrize('changing_attr,updates', [
         ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': u'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}),
         ('clone_uri', {'clone_uri': None}),
         ('landing_rev', {'landing_rev': 'branch:master'}),
         ('enable_statistics', {'enable_statistics': True}),
         ('enable_locking', {'enable_locking': True}),
         ('enable_downloads', {'enable_downloads': True}),
         ('name', {'name': u'new_repo_name'}),
         ('repo_group', {'group': u'test_group_for_update'}),
     ])
     def test_api_update_group_repo(self, changing_attr, updates):
         group_name = u'lololo'
         fixture.create_repo_group(group_name)
         repo_name = u'%s/api_update_me' % group_name
         repo = fixture.create_repo(repo_name, repo_group=group_name, repo_type=self.REPO_TYPE)
         if changing_attr == 'repo_group':
             fixture.create_repo_group(updates['group'])
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         if changing_attr == 'name':
             repo_name = u'%s/%s' % (group_name, updates['name'])
         if changing_attr == 'repo_group':
             repo_name = u'/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])
         try:
             expected = {
                 'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
                 'repository': repo.get_api_data()
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(updates['group'])
         fixture.destroy_repo_group(group_name)
     def test_api_update_repo_repo_group_does_not_exist(self):
         repo_name = u'admin_owned'
         fixture.create_repo(repo_name)
         updates = {'group': 'test_group_for_update'}
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'repository group `%s` does not exist' % updates['group']
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_update_repo_regular_user_not_allowed(self):
         repo_name = u'admin_owned'
         fixture.create_repo(repo_name)
         updates = {'description': 'something else'}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'repository `%s` does not exist' % repo_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     @mock.patch.object(RepoModel, 'update', crash)
     def test_api_update_repo_exception_occurred(self):
         repo_name = u'api_update_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, owner=TEST_USER_ADMIN_LOGIN,)
         response = api_call(self, params)
         try:
             expected = 'failed to update repo `%s`' % repo_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_update_repo_regular_user_change_repo_name(self):
         repo_name = u'admin_owned'
         new_repo_name = u'new_repo_name'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
         UserModel().revoke_perm('default', 'hg.create.repository')
         UserModel().grant_perm('default', 'hg.create.none')
         updates = {'name': new_repo_name}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'no permission to create (or move) repositories'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             fixture.destroy_repo(new_repo_name)
     def test_api_update_repo_regular_user_change_repo_name_allowed(self):
         repo_name = u'admin_owned'
         new_repo_name = u'new_repo_name'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
         UserModel().revoke_perm('default', 'hg.create.none')
         UserModel().grant_perm('default', 'hg.create.repository')
         updates = {'name': new_repo_name}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = {
                 'msg': 'updated repo ID:%s %s' % (repo.repo_id, new_repo_name),
                 'repository': repo.get_api_data()
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             fixture.destroy_repo(new_repo_name)
     def test_api_update_repo_regular_user_change_owner(self):
         repo_name = u'admin_owned'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
         updates = {'owner': TEST_USER_ADMIN_LOGIN}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'Only Kallithea admin can specify `owner` param'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_delete_repo(self):
         repo_name = u'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         id_, params = _build_data(self.apikey, 'delete_repo',
                                   repoid=repo_name, )
         response = api_call(self, params)
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             'success': True
+        }
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_delete_repo_by_non_admin(self):
         repo_name = u'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                   repoid=repo_name, )
         response = api_call(self, params)
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             'success': True
+        }
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_delete_repo_by_non_admin_no_permission(self):
         repo_name = u'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         try:
             id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                       repoid=repo_name, )
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (repo_name)
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_delete_repo_exception_occurred(self):
         repo_name = u'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         try:
             with mock.patch.object(RepoModel, 'delete', crash):
                 id_, params = _build_data(self.apikey, 'delete_repo',
                                           repoid=repo_name, )
                 response = api_call(self, params)
                 expected = 'failed to delete repository `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
     def test_api_fork_repo(self):
         fork_name = u'api-repo-fork'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     @parametrize('fork_name', [
         u'api-repo-fork',
         u'%s/api-repo-fork' % TEST_REPO_GROUP,
     ])
     def test_api_fork_repo_non_admin(self, fork_name):
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
             'task': None,
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_specify_owner(self):
         fork_name = u'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_non_admin_no_permission_to_fork(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.none')
         fork_name = u'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
         expected = 'repository `%s` does not exist' % (self.REPO)
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     @parametrize('name,perm', [
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin'),
     ])
     def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):
         fork_name = u'api-repo-fork'
         # regardless of base repository permission, forking is disallowed
         # when repository creation is disabled
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=perm)
         UserModel().revoke_perm('default', 'hg.create.repository')
         UserModel().grant_perm('default', 'hg.create.none')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
+        )
         response = api_call(self, params)
         expected = 'no permission to create repositories'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
     def test_api_fork_repo_unknown_owner(self):
         fork_name = u'api-repo-fork'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=owner,
+        )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
     def test_api_fork_repo_fork_exists(self):
         fork_name = u'api-repo-fork'
         fixture.create_fork(self.REPO, fork_name)
         try:
             fork_name = u'api-repo-fork'
             id_, params = _build_data(self.apikey, 'fork_repo',
                                       repoid=self.REPO,
                                       fork_name=fork_name,
                                       owner=TEST_USER_ADMIN_LOGIN,
+            )
             response = api_call(self, params)
             expected = "fork `%s` already exist" % fork_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
     def test_api_fork_repo_repo_exists(self):
         fork_name = self.REPO
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         expected = "repo `%s` already exist" % fork_name
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'create_fork', crash)
     def test_api_fork_repo_exception_occurred(self):
         fork_name = u'api-repo-fork'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=TEST_USER_ADMIN_LOGIN,
+        )
         response = api_call(self, params)
         expected = 'failed to fork repository `%s` as `%s`' % (self.REPO,
                                                                fork_name)
         self._compare_error(id_, expected, given=response.body)
     def test_api_get_user_group(self):
         id_, params = _build_data(self.apikey, 'get_user_group',
                                   usergroupid=TEST_USER_GROUP)
         response = api_call(self, params)
         user_group = UserGroupModel().get_group(TEST_USER_GROUP)
         members = []
         for user in user_group.members:
             user = user.user
             members.append(user.get_api_data())
         ret = user_group.get_api_data()
         ret['members'] = members
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_get_user_groups(self):
         gr_name = u'test_user_group2'
         make_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'get_user_groups', )
             response = api_call(self, params)
             expected = []
             for gr_name in [TEST_USER_GROUP, u'test_user_group2']:
                 user_group = UserGroupModel().get_group(gr_name)
                 ret = user_group.get_api_data()
                 expected.append(ret)
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     def test_api_create_user_group(self):
         group_name = u'some_new_group'
         id_, params = _build_data(self.apikey, 'create_user_group',
                                   group_name=group_name)
         response = api_call(self, params)
         ret = {
             'msg': 'created new user group `%s`' % group_name,
             'user_group': jsonify(UserGroupModel() \
                 .get_by_name(group_name) \
                 .get_api_data())
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_user_group(group_name)
     def test_api_get_user_group_that_exist(self):
         id_, params = _build_data(self.apikey, 'create_user_group',
                                   group_name=TEST_USER_GROUP)
         response = api_call(self, params)
         expected = "user group `%s` already exist" % TEST_USER_GROUP
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(UserGroupModel, 'create', crash)
     def test_api_get_user_group_exception_occurred(self):
         group_name = u'exception_happens'
         id_, params = _build_data(self.apikey, 'create_user_group',
                                   group_name=group_name)
         response = api_call(self, params)
         expected = 'failed to create group `%s`' % group_name
         self._compare_error(id_, expected, given=response.body)
     @parametrize('changing_attr,updates', [
         ('group_name', {'group_name': u'new_group_name'}),
         ('group_name', {'group_name': u'test_group_for_update'}),
         ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
         ('active', {'active': False}),
         ('active', {'active': True}),
     ])
     def test_api_update_user_group(self, changing_attr, updates):
         gr_name = u'test_group_for_update'
         user_group = fixture.create_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'update_user_group',
                                       usergroupid=gr_name, **updates)
             response = api_call(self, params)
             expected = {
                'msg': 'updated user group ID:%s %s' % (user_group.users_group_id,
                                                      user_group.users_group_name),
                'user_group': user_group.get_api_data()
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             if changing_attr == 'group_name':
                 # switch to updated name for proper cleanup
                 gr_name = updates['group_name']
             fixture.destroy_user_group(gr_name)
     @mock.patch.object(UserGroupModel, 'update', crash)
     def test_api_update_user_group_exception_occurred(self):
         gr_name = u'test_group'
         fixture.create_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'update_user_group',
                                       usergroupid=gr_name)
             response = api_call(self, params)
             expected = 'failed to update user group `%s`' % gr_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     def test_api_add_user_to_user_group(self):
         gr_name = u'test_group'
         fixture.create_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'add_user_to_user_group',
                                       usergroupid=gr_name,
                                       userid=TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             expected = {
             'msg': 'added member `%s` to user group `%s`' % (
                     TEST_USER_ADMIN_LOGIN, gr_name),
             'success': True
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     def test_api_add_user_to_user_group_that_doesnt_exist(self):
         id_, params = _build_data(self.apikey, 'add_user_to_user_group',
                                   usergroupid='false-group',
                                   userid=TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         expected = 'user group `%s` does not exist' % 'false-group'
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(UserGroupModel, 'add_user_to_group', crash)
     def test_api_add_user_to_user_group_exception_occurred(self):
         gr_name = u'test_group'
         fixture.create_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'add_user_to_user_group',
                                       usergroupid=gr_name,
                                       userid=TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             expected = 'failed to add member to user group `%s`' % gr_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     def test_api_remove_user_from_user_group(self):
         gr_name = u'test_group_3'
         gr = fixture.create_user_group(gr_name)
         UserGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         try:
             id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
                                       usergroupid=gr_name,
                                       userid=TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             expected = {
                 'msg': 'removed member `%s` from user group `%s`' % (
                     TEST_USER_ADMIN_LOGIN, gr_name
                 ),
                 'success': True}
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     @mock.patch.object(UserGroupModel, 'remove_user_from_group', crash)
     def test_api_remove_user_from_user_group_exception_occurred(self):
         gr_name = u'test_group_3'
         gr = fixture.create_user_group(gr_name)
         UserGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         try:
             id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
                                       usergroupid=gr_name,
                                       userid=TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             expected = 'failed to remove member from user group `%s`' % gr_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     def test_api_delete_user_group(self):
         gr_name = u'test_group'
         ugroup = fixture.create_user_group(gr_name)
         gr_id = ugroup.users_group_id
         try:
             id_, params = _build_data(self.apikey, 'delete_user_group',
                                       usergroupid=gr_name)
             response = api_call(self, params)
             expected = {
                 'user_group': None,
                 'msg': 'deleted user group ID:%s %s' % (gr_id, gr_name)
+            }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             if UserGroupModel().get_by_name(gr_name):
                 fixture.destroy_user_group(gr_name)
     def test_api_delete_user_group_that_is_assigned(self):
         gr_name = u'test_group'
         ugroup = fixture.create_user_group(gr_name)
         gr_id = ugroup.users_group_id
         ugr_to_perm = RepoModel().grant_user_group_permission(self.REPO, gr_name, 'repository.write')
         msg = 'User Group assigned to %s' % ugr_to_perm.repository.repo_name
         try:
             id_, params = _build_data(self.apikey, 'delete_user_group',
                                       usergroupid=gr_name)
             response = api_call(self, params)
             expected = msg
             self._compare_error(id_, expected, given=response.body)
         finally:
             if UserGroupModel().get_by_name(gr_name):
                 fixture.destroy_user_group(gr_name)
     def test_api_delete_user_group_exception_occurred(self):
         gr_name = u'test_group'
         ugroup = fixture.create_user_group(gr_name)
         gr_id = ugroup.users_group_id
         id_, params = _build_data(self.apikey, 'delete_user_group',
                                   usergroupid=gr_name)
         try:
             with mock.patch.object(UserGroupModel, 'delete', crash):
                 response = api_call(self, params)
                 expected = 'failed to delete user group ID:%s %s' % (gr_id, gr_name)
                 self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_user_group(gr_name)
     @parametrize('name,perm', [
         ('none', 'repository.none'),
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin'),
     ])
     def test_api_grant_user_permission(self, name, perm):
         id_, params = _build_data(self.apikey,
                                   'grant_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = api_call(self, params)
         ret = {
             'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
                 perm, TEST_USER_ADMIN_LOGIN, self.REPO
             ),
             'success': True
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_grant_user_permission_wrong_permission(self):
         perm = 'haha.no.permission'
         id_, params = _build_data(self.apikey,
                                   'grant_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = api_call(self, params)
         expected = 'permission `%s` does not exist' % perm
         self._compare_error(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'grant_user_permission', crash)
     def test_api_grant_user_permission_exception_when_adding(self):
         perm = 'repository.read'
         id_, params = _build_data(self.apikey,
                                   'grant_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = api_call(self, params)
         expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
             TEST_USER_ADMIN_LOGIN, self.REPO
+        )
         self._compare_error(id_, expected, given=response.body)
     def test_api_revoke_user_permission(self):
         id_, params = _build_data(self.apikey,
                                   'revoke_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN, )
         response = api_call(self, params)
         expected = {
             'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
                 TEST_USER_ADMIN_LOGIN, self.REPO
             ),
             'success': True
+        }
         self._compare_ok(id_, expected, given=response.body)
     @mock.patch.object(RepoModel, 'revoke_user_permission', crash)
     def test_api_revoke_user_permission_exception_when_adding(self):
         id_, params = _build_data(self.apikey,
                                   'revoke_user_permission',
                                   repoid=self.REPO,
                                   userid=TEST_USER_ADMIN_LOGIN, )
         response = api_call(self, params)
         expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
             TEST_USER_ADMIN_LOGIN, self.REPO
+        )
         self._compare_error(id_, expected, given=response.body)
     @parametrize('name,perm', [
         ('none', 'repository.none'),
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin'),
     ])
     def test_api_grant_user_group_permission(self, name, perm):
         id_, params = _build_data(self.apikey,
                                   'grant_user_group_permission',
                                   repoid=self.REPO,
                                   usergroupid=TEST_USER_GROUP,
                                   perm=perm)
         response = api_call(self, params)
         ret = {
             'msg': 'Granted perm: `%s` for user group: `%s` in repo: `%s`' % (
                 perm, TEST_USER_GROUP, self.REPO
             ),
             'success': True
+        }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
     def test_api_grant_user_group_permission_wrong_permission(self):
         perm = 'haha.no.permission'
         id_, params = _build_data(self.apikey,
                                   'grant_user_group_permission',
                                   repoid=self.REPO,
                                   usergroupid=TEST_USER_GROUP,
                                   perm=perm)
         response = api_call(self, params)
         expected = 'permission `%s` does not exist' % perm

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import os
 import mock
 import urllib
 import pytest
 from sqlalchemy import func
 from kallithea.lib import vcs
 from kallithea.lib.utils2 import safe_str, safe_unicode
 from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \
     Permission, Ui
 from kallithea.model.user import UserModel
 from kallithea.tests.base import *
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session, Base
 from kallithea.tests.fixture import Fixture, error_function
 fixture = Fixture()
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query() \
                 .filter(UserRepoToPerm.repository ==
                         Repository.get_by_repo_name(repo)) \
                 .filter(UserRepoToPerm.user == User.get_by_username(user)) \
                 .all()
     return perm
 class _BaseTestCase(TestController):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     OTHER_TYPE_REPO = None
     OTHER_TYPE = None
     def test_index(self):
         self.log_user()
         response = self.app.get(url('repos'))
     def test_create(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name)
         Session().commit()
     def test_case_insensitivity(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                                  fixture._get_repo_create_params(repo_private=False,
                                                                  repo_name=repo_name,
                                                                  repo_type=self.REPO_TYPE,
                                                                  repo_description=description,
                                                                  _authentication_token=self.authentication_token()))
         # try to create repo with swapped case
         swapped_repo_name = repo_name.swapcase()
         response = self.app.post(url('repos'),
                                  fixture._get_repo_create_params(repo_private=False,
                                                                  repo_name=swapped_repo_name,
                                                                  repo_type=self.REPO_TYPE,
                                                                  repo_description=description,
                                                                  _authentication_token=self.authentication_token()))
         response.mustcontain('already exists')
         RepoModel().delete(repo_name)
         Session().commit()
     def test_create_in_group(self):
         self.log_user()
         ## create GROUP
         group_name = u'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         assert len(inherited_perms) == 1
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_in_group_without_needed_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # avoid spurious RepoGroup DetachedInstanceError ...
         authentication_token = self.authentication_token()
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         ## create GROUP
         group_name = u'reg_sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         group_name_allowed = u'reg_sometest_allowed_%s' % self.REPO_TYPE
         gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,
                                      group_description=u'test',
                                      owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=authentication_token))
         response.mustcontain('Invalid value')
         # user is allowed to create in this group
         repo_name = u'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr_allowed.group_id,
                                                 _authentication_token=authentication_token))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         assert len(inherited_perms) == 1
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         RepoGroupModel().delete(group_name_allowed)
         Session().commit()
     def test_create_in_group_inherit_permissions(self):
         self.log_user()
         ## create GROUP
         group_name = u'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description=u'test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         perm = Permission.get_by_key('repository.write')
         RepoGroupModel().grant_user_permission(gr, TEST_USER_REGULAR_LOGIN, perm)
         ## add repo permissions
         Session().commit()
         repo_name = u'ingroup_inherited_%s' % self.REPO_TYPE
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 repo_copy_permissions=True,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         assert new_repo.repo_name == repo_name_full
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             pytest.fail('no repo %s in filesystem' % repo_name)
         # check if inherited permissiona are applied
         inherited_perms = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         assert len(inherited_perms) == 2
         assert TEST_USER_REGULAR_LOGIN in [x.user.username
                                                     for x in inherited_perms]
         assert 'repository.write' in [x.permission.permission_name
                                                for x in inherited_perms]
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_remote_repo_wrong_clone_uri(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 clone_uri='http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('Invalid repository URL')
     def test_create_remote_repo_wrong_clone_uri_hg_svn(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 clone_uri='svn+http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('Invalid repository URL')
     def test_delete(self):
         self.log_user()
         repo_name = u'vcs_test_new_to_delete_%s' % self.REPO_TYPE
         description = u'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_name=repo_name,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).one()
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
             params={'_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))
         response.follow()
         # check if repo was deleted from db
         deleted_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name).scalar()
-        assert deleted_repo == None
+        assert deleted_repo is None
         assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name)) == False
     def test_delete_non_ascii(self):
         self.log_user()
         non_ascii = "ąęł"
         repo_name = "%s%s" % (safe_str(self.NEW_REPO), non_ascii)
         repo_name_unicode = safe_unicode(repo_name)
         description = 'description for newly created repo' + non_ascii
         description_unicode = safe_unicode(description)
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert response.json == {u'result': True}
         self.checkSessionFlash(response,
                                u'Created repository <a href="/%s">%s</a>'
                                % (urllib.quote(repo_name), repo_name_unicode))
         # test if the repo was created in the database
         new_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).one()
         assert new_repo.repo_name == repo_name_unicode
         assert new_repo.description == description_unicode
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)))
         except vcs.exceptions.VCSError:
             pytest.fail('no repo %s in filesystem' % repo_name)
         response = self.app.post(url('delete_repo', repo_name=repo_name),
             params={'_authentication_token': self.authentication_token()})
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
         response.follow()
         # check if repo was deleted from db
         deleted_repo = Session().query(Repository) \
             .filter(Repository.repo_name == repo_name_unicode).scalar()
-        assert deleted_repo == None
+        assert deleted_repo is None
         assert os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_unicode)) == False
     def test_delete_repo_with_group(self):
         # TODO:
         pass
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('delete_repo', repo_name=self.REPO),
                                  params=dict(_authentication_token=self.authentication_token()))
     def test_show(self):
         self.log_user()
         response = self.app.get(url('summary_home', repo_name=self.REPO))
     def test_edit(self):
         response = self.app.get(url('edit_repo', repo_name=self.REPO))
     def test_set_private_flag_sets_default_to_none(self):
         self.log_user()
         # initially repository perm should be read
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.read'
         assert Repository.get_by_repo_name(self.REPO).private == False
         response = self.app.post(url('update_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=1,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 owner=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         assert Repository.get_by_repo_name(self.REPO).private == True
         # now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.none'
         response = self.app.post(url('update_repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 owner=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         assert Repository.get_by_repo_name(self.REPO).private == False
         # we turn off private now the repo default permission should stay None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         assert len(perm), 1
         assert perm[0].permission.permission_name == 'repository.none'
         # update this permission back
         perm[0].permission = Permission.get_by_key('repository.read')
         Session().commit()
     def test_set_repo_fork_has_no_self_id(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
         opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)
         response.mustcontain(no=[opt])
     def test_set_fork_of_other_repo(self):
         self.log_user()
         other_repo = u'other_%s' % self.REPO_TYPE
         fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         response = self.app.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         self.checkSessionFlash(response,
             'Marked repository %s as fork of %s' % (repo.repo_name, repo2.repo_name))
         assert repo.fork == repo2
         response = response.follow()
         # check if given repo is selected
         opt = """<option value="%s" selected="selected">%s</option>""" % (
                     repo2.repo_id, repo2.repo_name)
         response.mustcontain(opt)
         fixture.destroy_repo(other_repo, forks='detach')
     def test_set_fork_of_other_type_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         response = self.app.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
             'Cannot set repository as fork of repository with other type')
     def test_set_fork_of_none(self):
         self.log_user()
         ## mark it as None
         response = self.app.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=None, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
                                'Marked repository %s as fork of %s'
                                % (repo.repo_name, "Nothing"))
         assert repo.fork is None
     def test_set_fork_of_same_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.post(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo.repo_id, _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'An error occurred during this operation')
     def test_create_on_top_level_without_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         user = User.get(usr['user_id'])
         repo_name = self.NEW_REPO + u'no_perms'
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('<span class="error-message">Invalid value</span>')
         RepoModel().delete(repo_name)
         Session().commit()
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'Error creating repository %s' % repo_name)
         # repo must not be in db
         repo = Repository.get_by_repo_name(repo_name)
-        assert repo == None
+        assert repo is None
         # repo must not be in filesystem !
         assert not os.path.isdir(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name))
 class TestAdminReposControllerGIT(_BaseTestCase):
     REPO = GIT_REPO
     REPO_TYPE = 'git'
     NEW_REPO = NEW_GIT_REPO
     OTHER_TYPE_REPO = HG_REPO
     OTHER_TYPE = 'hg'
 class TestAdminReposControllerHG(_BaseTestCase):
     REPO = HG_REPO
     REPO_TYPE = 'hg'
     NEW_REPO = NEW_HG_REPO
     OTHER_TYPE_REPO = GIT_REPO
     OTHER_TYPE = 'git'
     def test_permanent_url_protocol_access(self):
         repo = Repository.get_by_repo_name(self.REPO)
         permanent_name = '_%d' % repo.repo_id
         # 400 Bad Request - Unable to detect pull/push action
         self.app.get(url('summary_home', repo_name=permanent_name),
             extra_environ={'HTTP_ACCEPT': 'application/mercurial'},
             status=400,
+        )

kallithea/tests/functional/test_admin_user_groups.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 from kallithea.tests.base import *
 from kallithea.model.db import UserGroup, UserGroupToPerm, Permission
 from kallithea.model.meta import Session
 TEST_USER_GROUP = u'admins_test'
 class TestAdminUsersGroupsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url('users_groups'))
         # Test response...
     def test_create(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         self.checkSessionFlash(response,
                                'Created user group <a href="/_admin/user_groups/')
         self.checkSessionFlash(response,
                                '/edit">%s</a>' % TEST_USER_GROUP)
     def test_new(self):
         response = self.app.get(url('new_users_group'))
     def test_update(self):
         response = self.app.post(url('update_users_group', id=1), status=403)
     def test_update_browser_fakeout(self):
         response = self.app.post(url('update_users_group', id=1),
                                  params=dict(_authentication_token=self.authentication_token()))
     def test_delete(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         self.checkSessionFlash(response,
                                'Created user group ')
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).one()
         response = self.app.post(url('delete_users_group', id=gr.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).scalar()
-        assert gr == None
+        assert gr is None
     def test_default_perms_enable_repository_read_on_group(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         self.checkSessionFlash(response,
                                'Created user group ')
         ## ENABLE REPO CREATE ON A GROUP
         response = self.app.post(url('edit_user_group_default_perms_update',
                                      id=ug.users_group_id),
                                  {'create_repo_perm': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.repository')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.post(
             url('edit_user_group_default_perms_update', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]])
         # DELETE !
         ug = UserGroup.get_by_group_name(users_group_name)
         ugid = ug.users_group_id
         response = self.app.post(url('delete_users_group', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response = response.follow()
         gr = Session().query(UserGroup) \
             .filter(UserGroup.users_group_name == users_group_name).scalar()
-        assert gr == None
+        assert gr is None
         p = Permission.get_by_key('hg.create.repository')
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group_id == ugid).all()
         perms = [[x.users_group_id,
                   x.permission_id, ] for x in perms]
         assert perms == []
     def test_default_perms_enable_repository_fork_on_group(self):
         self.log_user()
         users_group_name = TEST_USER_GROUP + 'another2'
         response = self.app.post(url('users_groups'),
                                  {'users_group_name': users_group_name,
                                   'user_group_description': u'DESC',
                                   'active': True,
                                   '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         self.checkSessionFlash(response,
                                'Created user group ')
         ## ENABLE REPO CREATE ON A GROUP
         response = self.app.post(url('edit_user_group_default_perms_update',
                                      id=ug.users_group_id),
                                  {'fork_repo_perm': True, '_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.repository')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]])
         ## DISABLE REPO CREATE ON A GROUP
         response = self.app.post(url('edit_user_group_default_perms_update', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response.follow()
         ug = UserGroup.get_by_group_name(users_group_name)
         p = Permission.get_by_key('hg.create.none')
         p2 = Permission.get_by_key('hg.usergroup.create.false')
         p3 = Permission.get_by_key('hg.fork.none')
         # check if user has this perms, they should be here since
         # defaults are on
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group == ug).all()
         assert sorted([[x.users_group_id, x.permission_id, ] for x in perms]) == sorted([[ug.users_group_id, p.permission_id],
                     [ug.users_group_id, p2.permission_id],
                     [ug.users_group_id, p3.permission_id]])
         # DELETE !
         ug = UserGroup.get_by_group_name(users_group_name)
         ugid = ug.users_group_id
         response = self.app.post(url('delete_users_group', id=ug.users_group_id),
             params={'_authentication_token': self.authentication_token()})
         response = response.follow()
         gr = Session().query(UserGroup) \
                            .filter(UserGroup.users_group_name ==
                                    users_group_name).scalar()
-        assert gr == None
+        assert gr is None
         p = Permission.get_by_key('hg.fork.repository')
         perms = UserGroupToPerm.query() \
             .filter(UserGroupToPerm.users_group_id == ugid).all()
         perms = [[x.users_group_id,
                   x.permission_id, ] for x in perms]
         assert perms == []
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('delete_users_group', id=1),
                                  params=dict(_authentication_token=self.authentication_token()))

kallithea/tests/functional/test_login.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import re
 import time
 import urlparse
 import mock
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.lib.utils2 import generate_api_key
 from kallithea.lib.auth import check_password
 from kallithea.lib import helpers as h
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model import validators
 from kallithea.model.db import User, Notification
 from kallithea.model.meta import Session
 from kallithea.model.user import UserModel
 from tg.util.webtest import test_context
 fixture = Fixture()
 class TestLoginController(TestController):
     def setup_method(self, method):
         self.remove_all_notifications()
         assert Notification.query().all() == []
     def test_index(self):
         response = self.app.get(url(controller='login', action='index'))
         assert response.status == '200 OK'
         # Test response...
     def test_login_admin_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_ADMIN_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_regular_email_ok(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_EMAIL,
                                   'password': TEST_USER_REGULAR_PASS})
         assert response.status == '302 Found'
         self.assert_authenticated_user(response, TEST_USER_REGULAR_LOGIN)
         response = response.follow()
         response.mustcontain('/%s' % HG_REPO)
     def test_login_ok_came_from(self):
         test_came_from = '/_admin/users'
         response = self.app.post(url(controller='login', action='index',
                                      came_from=test_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         assert response.status == '302 Found'
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('Users Administration')
     def test_login_do_not_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': False})
         assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             assert not re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r has expiration date, but should be a session cookie' % cookie
     def test_login_remember(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS,
                                   'remember': True})
         assert 'Set-Cookie' in response.headers
         for cookie in response.headers.getall('Set-Cookie'):
             assert re.search(r';\s+(Max-Age|Expires)=', cookie, re.IGNORECASE), 'Cookie %r should have expiration date, but is a session cookie' % cookie
     def test_logout(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_REGULAR_LOGIN,
                                   'password': TEST_USER_REGULAR_PASS})
         # Verify that a login session has been established.
         response = self.app.get(url(controller='login', action='index'))
         response = response.follow()
         assert 'authuser' in response.session
         response.click('Log Out')
         # Verify that the login session has been terminated.
         response = self.app.get(url(controller='login', action='index'))
         assert 'authuser' not in response.session
     @parametrize('url_came_from', [
           ('data:text/html,<script>window.alert("xss")</script>',),
           ('mailto:test@example.com',),
           ('file:///etc/passwd',),
           ('ftp://ftp.example.com',),
           ('http://other.example.com/bl%C3%A5b%C3%A6rgr%C3%B8d',),
           ('//evil.example.com/',),
           ('/\r\nX-Header-Injection: boo',),
           ('/invälid_url_bytes',),
           ('non-absolute-path',),
     ])
     def test_login_bad_came_froms(self, url_came_from):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url_came_from),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS},
                                  status=400)
     def test_login_short_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': 'as'})
         assert response.status == '200 OK'
         response.mustcontain('Enter 3 characters or more')
     def test_login_wrong_username_password(self):
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('Invalid username or password')
     # verify that get arguments are correctly passed along login redirection
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, (('foo', 'one'), ('bar', 'two'))),
         ({'blue': u'blå'.encode('utf-8'), 'green':u'grøn'},
              (('blue', u'blå'.encode('utf-8')), ('green', u'grøn'.encode('utf-8')))),
     ])
     def test_redirection_to_login_form_preserves_get_args(self, args, args_encoded):
         with fixture.anon_access(False):
             response = self.app.get(url(controller='summary', action='index',
                                         repo_name=HG_REPO,
                                         **args))
             assert response.status == '302 Found'
             came_from = urlparse.parse_qs(urlparse.urlparse(response.location).query)['came_from'][0]
             came_from_qs = urlparse.parse_qsl(urlparse.urlparse(came_from).query)
             for encoded in args_encoded:
                 assert encoded in came_from_qs
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_preserves_get_args(self, args, args_encoded):
         response = self.app.get(url(controller='login', action='index',
                                     came_from=url('/_admin/users', **args)))
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_redirection_after_successful_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': TEST_USER_ADMIN_LOGIN,
                                   'password': TEST_USER_ADMIN_PASS})
         assert response.status == '302 Found'
         for encoded in args_encoded:
             assert encoded in response.location
     @parametrize('args,args_encoded', [
         ({'foo':'one', 'bar':'two'}, ('foo=one', 'bar=two')),
         ({'blue': u'blå', 'green':u'grøn'},
              ('blue=bl%C3%A5', 'green=gr%C3%B8n')),
     ])
     def test_login_form_after_incorrect_login_preserves_get_args(self, args, args_encoded):
         response = self.app.post(url(controller='login', action='index',
                                      came_from=url('/_admin/users', **args)),
                                  {'username': 'error',
                                   'password': 'test12'})
         response.mustcontain('Invalid username or password')
         came_from = urlparse.parse_qs(urlparse.urlparse(response.form.action).query)['came_from'][0]
         for encoded in args_encoded:
             assert encoded in came_from
     #==========================================================================
     # REGISTRATIONS
     #==========================================================================
     def test_register(self):
         response = self.app.get(url(controller='login', action='register'))
         response.mustcontain('Sign Up')
     def test_register_err_same_username(self):
         uname = TEST_USER_ADMIN_LOGIN
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': uname,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmail@example.com',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         with test_context(self.app):
             msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': uname})
         response.mustcontain(msg)
     def test_register_err_same_email(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_0',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL,
                                              'firstname': 'test',
                                              'lastname': 'test'})
         with test_context(self.app):
             msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_same_email_case_sensitive(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'test_admin_1',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': TEST_USER_ADMIN_EMAIL.title(),
                                              'firstname': 'test',
                                              'lastname': 'test'})
         with test_context(self.app):
             msg = validators.UniqSystemEmail()()._messages['email_taken']
         response.mustcontain(msg)
     def test_register_err_wrong_data(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': 'test',
                                              'password_confirmation': 'test',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         assert response.status == '200 OK'
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Enter a value 6 characters long or more')
     def test_register_err_username(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'error user',
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         response.mustcontain('Username may only contain '
                 'alphanumeric characters underscores, '
                 'periods or dashes and must begin with an '
                 'alphanumeric character')
     def test_register_err_case_sensitive(self):
         usr = TEST_USER_ADMIN_LOGIN.title()
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': usr,
                                              'password': 'test12',
                                              'password_confirmation': 'test12',
                                              'email': 'goodmailm',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         response.mustcontain('An email address must contain a single @')
         with test_context(self.app):
             msg = validators.ValidUsername()._messages['username_exists']
         msg = h.html_escape(msg % {'username': usr})
         response.mustcontain(msg)
     def test_register_special_chars(self):
         response = self.app.post(url(controller='login', action='register'),
                                         {'username': 'xxxaxn',
                                          'password': 'ąćźżąśśśś',
                                          'password_confirmation': 'ąćźżąśśśś',
                                          'email': 'goodmailm@test.plx',
                                          'firstname': 'test',
                                          'lastname': 'test'})
         with test_context(self.app):
             msg = validators.ValidPassword()._messages['invalid_password']
         response.mustcontain(msg)
     def test_register_password_mismatch(self):
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': 'xs',
                                              'password': '123qwe',
                                              'password_confirmation': 'qwe123',
                                              'email': 'goodmailm@test.plxa',
                                              'firstname': 'test',
                                              'lastname': 'test'})
         with test_context(self.app):
             msg = validators.ValidPasswordsMatch('password', 'password_confirmation')._messages['password_mismatch']
         response.mustcontain(msg)
     def test_register_ok(self):
         username = 'test_regular4'
         password = 'qweqwe'
         email = 'user4@example.com'
         name = 'testname'
         lastname = 'testlastname'
         response = self.app.post(url(controller='login', action='register'),
                                             {'username': username,
                                              'password': password,
                                              'password_confirmation': password,
                                              'email': email,
                                              'firstname': name,
                                              'lastname': lastname,
                                              'admin': True})  # This should be overridden
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'You have successfully registered with Kallithea')
         ret = Session().query(User).filter(User.username == 'test_regular4').one()
         assert ret.username == username
         assert check_password(password, ret.password) == True
         assert ret.email == email
         assert ret.name == name
         assert ret.lastname == lastname
-        assert ret.api_key != None
+        assert ret.api_key is not None
         assert ret.admin == False
     #==========================================================================
     # PASSWORD RESET
     #==========================================================================
     def test_forgot_password_wrong_mail(self):
         bad_email = 'username%wrongmail.org'
         response = self.app.post(
                         url(controller='login', action='password_reset'),
                             {'email': bad_email, }
+        )
         response.mustcontain('An email address must contain a single @')
     def test_forgot_password(self):
         response = self.app.get(url(controller='login',
                                     action='password_reset'))
         assert response.status == '200 OK'
         username = 'test_password_reset_1'
         password = 'qweqwe'
         email = 'username@example.com'
         name = u'passwd'
         lastname = u'reset'
         timestamp = int(time.time())
         new = User()
         new.username = username
         new.password = password
         new.email = email
         new.name = name
         new.lastname = lastname
         new.api_key = generate_api_key()
         Session().add(new)
         Session().commit()
         response = self.app.post(url(controller='login',
                                      action='password_reset'),
                                  {'email': email, })
         self.checkSessionFlash(response, 'A password reset confirmation code has been sent')
         response = response.follow()
         # BAD TOKEN
         token = "bad"
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                  })
         assert response.status == '200 OK'
         response.mustcontain('Invalid password reset token')
         # GOOD TOKEN
         # TODO: The token should ideally be taken from the mail sent
         # above, instead of being recalculated.
         token = UserModel().get_reset_password_token(
             User.get_by_username(username), timestamp, self.authentication_token())
         response = self.app.get(url(controller='login',
                                     action='password_reset_confirmation',
                                     email=email,
                                     timestamp=timestamp,
                                     token=token))
         assert response.status == '200 OK'
         response.mustcontain("You are about to set a new password for the email address %s" % email)
         response = self.app.post(url(controller='login',
                                      action='password_reset_confirmation'),
                                  {'email': email,
                                   'timestamp': timestamp,
                                   'password': "p@ssw0rd",
                                   'password_confirm': "p@ssw0rd",
                                   'token': token,
                                  })
         assert response.status == '302 Found'
         self.checkSessionFlash(response, 'Successfully updated password')
         response = response.follow()
     #==========================================================================
     # API
     #==========================================================================
     def _get_api_whitelist(self, values=None):
         config = {'api_access_controllers_whitelist': values or []}
         return config
     def _api_key_test(self, api_key, status):
         """Verifies HTTP status code for accessing an auth-requiring page,
         using the given api_key URL parameter as well as using the API key
         with bearer authentication.
         If api_key is None, no api_key is passed at all. If api_key is True,
         a real, working API key is used.
         """
         with fixture.anon_access(False):
             if api_key is None:
                 params = {}
                 headers = {}
             else:
                 if api_key is True:
                     api_key = User.get_first_admin().api_key
                 params = {'api_key': api_key}
                 headers = {'Authorization': 'Bearer ' + str(api_key)}
             self.app.get(url(controller='changeset', action='changeset_raw',
                              repo_name=HG_REPO, revision='tip', **params),
                          status=status)
             self.app.get(url(controller='changeset', action='changeset_raw',
                              repo_name=HG_REPO, revision='tip'),
                          headers=headers,
                          status=status)
     @parametrize('test_name,api_key,code', [
         ('none', None, 302),
         ('empty_string', '', 403),
         ('fake_number', '123456', 403),
         ('proper_api_key', True, 403)
     ])
     def test_access_not_whitelisted_page_via_api_key(self, test_name, api_key, code):
         whitelist = self._get_api_whitelist([])
         with mock.patch('kallithea.CONFIG', whitelist):
             assert [] == whitelist['api_access_controllers_whitelist']
             self._api_key_test(api_key, code)
     @parametrize('test_name,api_key,code', [
         ('none', None, 302),
         ('empty_string', '', 403),
         ('fake_number', '123456', 403),
         ('fake_not_alnum', 'a-z', 403),
         ('fake_api_key', '0123456789abcdef0123456789ABCDEF01234567', 403),
         ('proper_api_key', True, 200)
     ])
     def test_access_whitelisted_page_via_api_key(self, test_name, api_key, code):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == whitelist['api_access_controllers_whitelist']
             self._api_key_test(api_key, code)
     def test_access_page_via_extra_api_key(self):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == whitelist['api_access_controllers_whitelist']
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             Session().commit()
             self._api_key_test(new_api_key.api_key, status=200)
     def test_access_page_via_expired_api_key(self):
         whitelist = self._get_api_whitelist(['ChangesetController:changeset_raw'])
         with mock.patch('kallithea.CONFIG', whitelist):
             assert ['ChangesetController:changeset_raw'] == whitelist['api_access_controllers_whitelist']
             new_api_key = ApiKeyModel().create(TEST_USER_ADMIN_LOGIN, u'test')
             Session().commit()
             # patch the API key and make it expired
             new_api_key.expires = 0
             Session().commit()
             self._api_key_test(new_api_key.api_key, status=403)

kallithea/tests/functional/test_pullrequests.py

➞

Show inline comments

 import re
 import pytest
 from tg.util.webtest import test_context
 from kallithea.tests.base import *
 from kallithea.tests.fixture import Fixture
 from kallithea.model.db import PullRequest, User
 from kallithea.model.meta import Session
 from kallithea.controllers.pullrequests import PullrequestsController
 fixture = Fixture()
 class TestPullrequestsController(TestController):
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='pullrequests', action='index',
                                     repo_name=HG_REPO))
     def test_create_trivial(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('Successfully opened new pull request')
         response.mustcontain('No additional changesets found for iterating on this pull request')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')
     def test_available(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain(no='No additional changesets found for iterating on this pull request')
         response.mustcontain('The following additional changes are available on stable:')
         response.mustcontain('<input id="updaterev_4f7e2131323e0749a740c0a56ab68ae9269c562a" name="updaterev" type="radio" value="4f7e2131323e0749a740c0a56ab68ae9269c562a" />')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"') # as update
     def test_range(self):
         self.log_user()
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         response = response.follow()
         assert response.status == '200 OK'
         response.mustcontain('No additional changesets found for iterating on this pull request')
         response.mustcontain('href="/vcs_test_hg/changeset/4f7e2131323e0749a740c0a56ab68ae9269c562a"')
     def test_update_reviewers(self):
         self.log_user()
         regular_user = User.get_by_username(TEST_USER_REGULAR_LOGIN)
         regular_user2 = User.get_by_username(TEST_USER_REGULAR2_LOGIN)
         admin_user = User.get_by_username(TEST_USER_ADMIN_LOGIN)
         # create initial PR
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
                                  {'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                  },
                                  status=302)
         pull_request1_id = re.search('/pull-request/(\d+)/', response.location).group(1)
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request1_id)
         # create new iteration
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request1_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_authentication_token': self.authentication_token(),
                                   'review_members': [regular_user.user_id],
                                  },
                                  status=302)
         pull_request2_id = re.search('/pull-request/(\d+)/', response.location).group(1)
         assert pull_request2_id != pull_request1_id
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
         response = response.follow()
         # verify reviewer was added
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
         # update without creating new iteration
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request2_id),
+                                 {
                                   'pullrequest_title': 'Title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_authentication_token': self.authentication_token(),
                                   'org_review_members': [admin_user.user_id], # fake - just to get some 'meanwhile' warning ... but it is also added ...
                                   'review_members': [regular_user2.user_id, admin_user.user_id],
                                  },
                                  status=302)
         assert response.location == 'http://localhost/%s/pull-request/%s/_/stable' % (HG_REPO, pull_request2_id)
         response = response.follow()
         # verify reviewers were added / removed
         response.mustcontain('Meanwhile, the following reviewers have been added: test_regular')
         response.mustcontain('Meanwhile, the following reviewers have been removed: test_admin')
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user.user_id)
         response.mustcontain('<input type="hidden" value="%s" name="review_members" />' % regular_user2.user_id)
         response.mustcontain(no='<input type="hidden" value="%s" name="review_members" />' % admin_user.user_id)
     def test_update_with_invalid_reviewer(self):
         invalid_user_id = 99999
         self.log_user()
         # create a valid pull request
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
+                                 {
                                   'org_repo': HG_REPO,
                                   'org_ref': 'rev:94f45ed825a1:94f45ed825a113e61af7e141f44ca578374abef0',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                  },
                                 status=302)
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
-        assert m != None
+        assert m is not None
         pull_request_id = m.group(1)
         # update it
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'updaterev': '4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_authentication_token': self.authentication_token(),
                                   'review_members': [str(invalid_user_id)],
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
     def test_edit_with_invalid_reviewer(self):
         invalid_user_id = 99999
         self.log_user()
         # create a valid pull request
         response = self.app.post(url(controller='pullrequests', action='create',
                                      repo_name=HG_REPO),
+                                 {
                                   'org_repo': HG_REPO,
                                   'org_ref': 'branch:stable:4f7e2131323e0749a740c0a56ab68ae9269c562a',
                                   'other_repo': HG_REPO,
                                   'other_ref': 'branch:default:96507bd11ecc815ebc6270fdf6db110928c09c1e',
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   '_authentication_token': self.authentication_token(),
                                  },
                                 status=302)
         # location is of the form:
         # http://localhost/vcs_test_hg/pull-request/54/_/title
         m = re.search('/pull-request/(\d+)/', response.location)
-        assert m != None
+        assert m is not None
         pull_request_id = m.group(1)
         # edit it
         response = self.app.post(url(controller='pullrequests', action='post',
                                      repo_name=HG_REPO, pull_request_id=pull_request_id),
+                                 {
                                   'pullrequest_title': 'title',
                                   'pullrequest_desc': 'description',
                                   'owner': TEST_USER_ADMIN_LOGIN,
                                   '_authentication_token': self.authentication_token(),
                                   'review_members': [str(invalid_user_id)],
                                  },
                                  status=400)
         response.mustcontain('Invalid reviewer &#34;%s&#34; specified' % invalid_user_id)
     def test_iteration_refs(self):
         # Repo graph excerpt:
         #   o   fb95b340e0d0 webvcs
         #  /:
         # o :   41d2568309a0 default
         # : :
         # : o   5ec21f21aafe webvcs
         # : :
         # : o   9e6119747791 webvcs
         # : :
         # o :   3d1091ee5a53 default
         # :/
         # o     948da46b29c1 default
         self.log_user()
         # create initial PR
         response = self.app.post(
             url(controller='pullrequests', action='create', repo_name=HG_REPO),
+            {
                 'org_repo': HG_REPO,
                 'org_ref': 'rev:9e6119747791:9e6119747791ff886a5abe1193a730b6bf874e1c',
                 'other_repo': HG_REPO,
                 'other_ref': 'branch:default:3d1091ee5a533b1f4577ec7d8a226bb315fb1336',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 '_authentication_token': self.authentication_token(),
             },
             status=302)
         pr1_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         pr1 = PullRequest.get(pr1_id)
         assert pr1.org_ref == 'branch:webvcs:9e6119747791ff886a5abe1193a730b6bf874e1c'
         assert pr1.other_ref == 'branch:default:948da46b29c125838a717f6a8496eb409717078d'
         Session().rollback() # invalidate loaded PR objects before issuing next request.
         # create PR 2 (new iteration with same ancestor)
         response = self.app.post(
             url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr1_id),
+            {
                 'updaterev': '5ec21f21aafe95220f1fc4843a4a57c378498b71',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 'owner': TEST_USER_REGULAR_LOGIN,
                 '_authentication_token': self.authentication_token(),
              },
              status=302)
         pr2_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         pr1 = PullRequest.get(pr1_id)
         pr2 = PullRequest.get(pr2_id)
         assert pr2_id != pr1_id
         assert pr1.status == PullRequest.STATUS_CLOSED
         assert pr2.org_ref == 'branch:webvcs:5ec21f21aafe95220f1fc4843a4a57c378498b71'
         assert pr2.other_ref == pr1.other_ref
         Session().rollback() # invalidate loaded PR objects before issuing next request.
         # create PR 3 (new iteration with new ancestor)
         response = self.app.post(
             url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr2_id),
+            {
                 'updaterev': 'fb95b340e0d03fa51f33c56c991c08077c99303e',
                 'pullrequest_title': 'title',
                 'pullrequest_desc': 'description',
                 'owner': TEST_USER_REGULAR_LOGIN,
                 '_authentication_token': self.authentication_token(),
              },
              status=302)
         pr3_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))
         pr2 = PullRequest.get(pr2_id)
         pr3 = PullRequest.get(pr3_id)
         assert pr3_id != pr2_id
         assert pr2.status == PullRequest.STATUS_CLOSED
         assert pr3.org_ref == 'branch:webvcs:fb95b340e0d03fa51f33c56c991c08077c99303e'
         assert pr3.other_ref == 'branch:default:41d2568309a05f422cffb8008e599d385f8af439'
 @pytest.mark.usefixtures("test_context_fixture") # apply fixture for all test methods
 class TestPullrequestsGetRepoRefs(TestController):
     def setup_method(self, method):
         self.repo_name = u'main'
         repo = fixture.create_repo(self.repo_name, repo_type='hg')
         self.repo_scm_instance = repo.scm_instance
         Session().commit()
         self.c = PullrequestsController()
     def teardown_method(self, method):
         fixture.destroy_repo(u'main')
         Session().commit()
         Session.remove()
     def test_repo_refs_empty_repo(self):
         # empty repo with no commits, no branches, no bookmarks, just one tag
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         assert default == 'tag:null:0000000000000000000000000000000000000000'
     def test_repo_refs_one_commit_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         assert default == 'branch:default:%s' % cs0.raw_id
         assert ([('branch:default:%s' % cs0.raw_id, 'default (current tip)')],
                 'Branches') in refs
     def test_repo_refs_one_commit_rev_hint(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
         expected = 'branch:default:%s' % cs0.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance)
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_rev_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)
         expected = 'rev:%s:%s' % (cs0.raw_id, cs0.raw_id)
         assert default == expected
         assert ([(expected, 'Changeset: %s' % cs0.raw_id[0:12])], 'Special') in refs
         assert ([('branch:default:%s' % cs1.raw_id, 'default (current tip)')], 'Branches') in refs
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs1.raw_id)
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_two_commits_branch_hint(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         cs1 = fixture.commit_change(self.repo_name, filename='file2',
                 content='line2\n', message='commit2', vcs_type='hg',
                 parent=None, newfile=True)
         refs, default = self.c._get_repo_refs(self.repo_scm_instance, branch='default')
         expected = 'branch:default:%s' % cs1.raw_id
         assert default == expected
         assert ([(expected, 'default (current tip)')], 'Branches') in refs
     def test_repo_refs_one_branch_no_hints(self):
         cs0 = fixture.commit_change(self.repo_name, filename='file1',
                 content='line1\n', message='commit1', vcs_type='hg',
                 parent=None, newfile=True)
         # TODO

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)