Python models.User 模块，create() 实例源码

def user_creator(num_user):
    """Create users from an Italian-like context. Due to param in factory create 'it_iT'."""
    for i in range(num_user):
        user_uuid = fake.uuid4()
        first_name = fake.first_name()
        last_name = fake.last_name()
        email_provider = fake.free_email_domain()
        email_user = '{}.{}@{}'.format(
            first_name.lower(), last_name.lower(), email_provider)
        password = fake.password(length=3, special_chars=False, digits=True,
                                 upper_case=True, lower_case=False)
        User.create(
            uuid=user_uuid,
            first_name=first_name,
            last_name=last_name,
            email=email_user,
            password=User.hash_password(password)
        )

def item_creator(num_item):
    LIST_CATEGORIES = ['scarpe', 'accessori', 'abbigliamento uomo', 'abbigliamento donna']
    for i in range(num_item):
        item_id = fake.uuid4()
        item_name = fake.sentence(nb_words=3, variable_nb_words=True)
        item_price = fake.pyfloat(left_digits=2, right_digits=2, positive=True)
        item_category = random.choice(LIST_CATEGORIES)
        item = Item.create(
            uuid=item_id,
            name=item_name,
            price=item_price,
            description=fake.paragraph(
                nb_sentences=3, variable_nb_sentences=True),
            availability=random.randint(35, 60),
            category=item_category,
        )
        picture_creator(num_item, i, item)

def test_authenticated(self, request):
        func = authenticated(lambda *a, **k: 'success')
        request.json.get.return_value = '1234'
        # test decorator, token is not in cache
        self.assertEqual(func(), 'success')
        self.assertIsNone(request.user)
        self.assertIsNone(request.auth)
        # test decorator, token is in cache, no user
        set_cache('1234', 1)
        self.assertEqual(func(), 'success')
        self.assertIsNone(request.user)
        self.assertIsNone(request.auth)
        # test decorator success
        user = User.create(username='user1', password='passwd')
        set_cache('1234', user.pk)
        self.assertEqual(func(), 'success')
        self.assertEqual(request.user.pk, user.pk)
        self.assertEqual(request.auth, '1234')

def test_use_cache(self):
        def _func(a, b):
            invert_color(a)
            return 'ok'

        func = use_cache(10, name='test_func')(_func)
        a, b = Game.create(white='1234', black='qwer'), delete_cache
        # run first time, no cache
        with patch('tests.decorators_t.invert_color') as mock:
            self.assertEqual(func(a, b=b), 'ok')
            mock.assert_called_once_with(a)
        # run second time, results from cached
        with patch('tests.decorators_t.invert_color') as mock:
            self.assertEqual(func(a, b=b), 'ok')
            self.assertFalse(mock.called)
        # delete cache and run again, no cache
        delete_cache(get_cache_func_name('test_func', a, b=b))
        with patch('tests.decorators_t.invert_color') as mock:
            self.assertEqual(func(a, b=b), 'ok')
            mock.assert_called_once_with(a)

def test_atomic_with_delete(flushdb):
    for i in range(3):
        await User.create(username=f'u{i}')

    async with db.atomic():
        user = await User.get(User.username == 'u1')
        await user.delete_instance()

    usernames = [u.username async for u in User.select()]
    assert sorted(usernames) == ['u0', 'u2']

    async with db.atomic():
        async with db.atomic():
            user = await User.get(User.username == 'u2')
            await user.delete_instance()

    usernames = [u.username async for u in User.select()]
    assert usernames == ['u0']

def get_databases():
    """create a list with the name of each .db file from main folder."""
    list_of_db = glob.glob('*.db')
    return list_of_db

def picture_creator(num_picture, index, item):
    ALLOWED_EXTENSION = ['jpg', 'jpeg', 'png', 'gif']
    pictures_path = get_random_pictures(num_picture)
    picture_id = fake.uuid4()
    extension = random.choice(ALLOWED_EXTENSION)
    Picture.create(
        uuid=picture_id,
        extension=extension,
        item=item
    )
    image_folder = utils.get_image_folder()
    if not os.path.exists(image_folder):
        os.makedirs(image_folder)
    shutil.copy2(pictures_path[index], '/{}/{}.{}'.format(image_folder, picture_id, extension))

def order_creator(num_order):
    for i in range(num_order):
        order_id = fake.uuid4()
        Order.create(
            uuid=order_id,
            user=User.select().order_by(fn.Random()).get(),
            total_price=0,
            delivery_address=Address.select().order_by(fn.Random()).get(),
            items=[]
        )

def favorite_creator(num_favorites):
    for i in range(num_favorites):
        Favorite.create(
            uuid=fake.uuid4(),
            item=Item.select().order_by(fn.Random()).get(),
            user=User.select().order_by(fn.Random()).get(),
            )

def __init__(self, cls):
        self.original = cls.create
        self.uuid_generator = mock_uuid_generator()

def get_all_models_names():
    """
    Returns the names of all the classes defined inside the 'models' module.
    """
    return [name for
            (name, cls) in inspect.getmembers(
                sys.modules['models'], inspect.isclass)
            if cls.__module__ is 'models']


# ###########################################################
# Peewee models helpers
# Functions to create new instances with overridable defaults

def add_user(email, password, id=None, first_name='John', last_name='Doe'):
    """
    Create a single user in the test database.
    If an email is provided it will be used, otherwise it will be generated
    by the function before adding the User to the database.
    """
    email = email or 'johndoe{}@email.com'.format(int(random.random() * 100))

    return User.create(
        first_name=first_name,
        last_name=last_name,
        email=email,
        password=User.hash_password(password),
        uuid=id or uuid.uuid4(),
    )

def add_address(user, country='Italy', city='Pistoia', post_code='51100',
                address='Via Verdi 12', phone='3294882773', id=None):

    return Address.create(
        uuid=id or uuid.uuid4(),
        user=user,
        country=country,
        city=city,
        post_code=post_code,
        address=address,
        phone=phone,
    )

def add_favorite(user, item, id=None):
    """Link the favorite item to user."""
    return Favorite.create(
        uuid=id or uuid.uuid4(),
        item=item,
        user=user,
    )

def add_item(name='Item Test', price='15.99',
             description='test test test', id=None, category='scarpe'):
    return Item.create(
        uuid=id or uuid.uuid4(),
        name=name,
        price=price,
        description=description,
        availability=random.randint(35, 60),
        category=category,
    )


# ###########################################################
# Flask helpers
# Common operations for flask functionalities

def post(self):
        """ Add an user to the database."""
        data = request.get_json(force=True)

        errors = User.validate_input(data)
        if errors:
            return errors, BAD_REQUEST

        # Extract the user attributes to check and generate the User row
        data = data['data']['attributes']

        # If email is present in the database return a BAD_REQUEST response.
        if User.exists(data['email']):
            msg = {'message': 'email already present.'}
            return msg, CONFLICT

        new_user = User.create(
            uuid=uuid.uuid4(),
            first_name=data['first_name'],
            last_name=data['last_name'],
            email=data['email'],
            password=User.hash_password(data['password'])
        )
        notify_new_user(first_name=new_user.first_name,
                        last_name=new_user.last_name)

        # If everything went OK return the newly created user and CREATED code
        # TODO: Handle json() return value (data, errors) and handle errors not
        # empty
        return generate_response(new_user.json(), CREATED)

def test_base(self):
        # init fields
        BaseValidator.fields = {
            'f1': dict(type=str, required=True),
            'f2': dict(type=int),
            'f3': dict(type=str, default='ok'),
        }
        # try to create validator without reqired field
        with self.assertRaises(errors.ValidationRequiredError, msg='f1'):
            BaseValidator(MockRequest(form={'f2': '42'}))
        # try to create validator with wrong type
        with self.assertRaises(errors.ValidationError, msg='f2'):
            BaseValidator(MockRequest(form={'f1': 'v1', 'f2': 'v2'}))
        # create validator without not required field and check it
        val = BaseValidator(MockRequest(form={'f1': 'v1'}))
        self.assertEqual(val.form, {'f1': 'v1', 'f2': None, 'f3': 'ok'})
        # create validator with all fields and check it
        val = BaseValidator(MockRequest(form={'f1': 'v1', 'f2': '42'}))
        self.assertEqual(val.form, {'f1': 'v1', 'f2': 42, 'f3': 'ok'})
        # test is_valid and get_error
        self.assertIsNone(val.get_error())
        self.assertFalse(val.error('test'))
        self.assertEqual(val.get_error(), 'test')
        self.assertTrue(val.is_valid())
        self.assertIsNone(val.get_error())
        self.assertEqual(val.form, val.cleaned_data)

def test_login_required(self, request):
        request.user = None
        func = login_required(lambda *a, **k: 'success')
        with patch('decorators.send_error') as mock:
            mock.return_value = '_error'
            self.assertEqual(func(), '_error')
            mock.assert_called_once_with('not authorized')
        request.user = User.create(username='user1', password='passwd')
        self.assertEqual(func(), 'success')

def test_with_game_2(self, request):
        request.json.get.return_value = 'user_token'
        # create game with white user and try to get it anonymously
        func = authenticated(with_game(lambda *a, **k: (a, k)))
        user = User.create(username='user1', password='passwd')
        game = Game.create(white='1234', black='qwer', player_white=user)
        with patch('decorators.send_error') as mock:
            func('1234')
            mock.assert_called_once_with('wrong user')
        self.assertEqual(func('qwer')[0][0].model.pk, game.pk)
        # set auth and check again
        set_cache('user_token', user.pk)
        self.assertEqual(func('1234')[0][0].model.pk, game.pk)

def createUser(self, screen_name, is_bot=False):
        api_user = self.twitter_client.user_shows(screen_name=screen_name)

        user = User.create(
            screen_name=screen_name,
            is_bot=is_bot,
            followers=api_user.followers_count,
            following=api_user.friends_count
        )

        return user

def fromUser(self, screen_name, tweets_number=10, is_bot=False):
        user = self.createUser(screen_name, is_bot)

        tweets = self.twitter_client.user_timeline(screen_name=screen_name, count=tweets_number)
        for i, status in enumerate(tweets):
            tweet = status._json
            text = tweet['text']
            date = tweet['created_at']
            entities = tweet['entities']
            user_mentions = entities['user_mentions']
            mentions_list = []

            if len(user_mentions) > 0:
                for mention in user_mentions:
                    mentions_list.append(mention['screen_name'])

            text_string = unicodedata.normalize('NFKD', text).encode('ascii','ignore')
            date_string = unicodedata.normalize('NFKD', date).encode('ascii','ignore')
            name_mentions_string = ",".join(mentions_list)

            Tweet.create(
                    user = user,
                    text = text_string,
                    date = date_string,
                    source = status.source,
                    mentions = name_mentions_string
            )

def main(first_name, last_name, email, password):
    click.echo('####################\n####################\nADMIN USER SCRIPT\n')
    click.echo('####################\n####################\n')
    click.echo('Here you can create an admin user. For eachone you have to insert:\n')
    click.echo('first name\n-last name\n-email\n-password')

    if not first_name:
        first_name = click.prompt('Please enter your first name')
    if not last_name:
        last_name = click.prompt('Please enter your last name')
    if not email:
        email = click.prompt('Please enter a valid email')
    if not password:
        password = click.prompt('Please enter your password')

    request_data = {
        'first_name': first_name,
        'last_name': last_name,
        'email': email,
        'password': password
    }

    for field in request_data:
        try:
            value = request_data[field]
            non_empty_str(value, field)
        except ValueError:
            print('ERROR! Some fields are empty or required')
            sys.exit(-1)

    # If email is present in the database return a ERROR and close the program.
    if User.exists(request_data['email']):
        print('ERROR! email already exists')
        sys.exit(-1)

    User.create(
        uuid=uuid.uuid4(),
        first_name=first_name,
        last_name=last_name,
        email=email,
        password=User.hash_password(password),
        admin=True
    )
    print("Great! Insert successfully")