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

def test_post_new_user_no_email__fail(self):
        user = format_jsonapi_request('user', {
            'first_name': 'Mario',
            'last_name': 'Rossi',
            'password': 'aksdg',
        })
        resp = self.app.post(API_ENDPOINT.format('users/'),
                             data=json.dumps(user),
                             content_type='application/json')

        # TODO: Refactor resource to use the Schema to validate and return
        # errors. response.data should contain the error missing required field
        # email
        assert resp.status_code == BAD_REQUEST
        expected_result = EXPECTED_RESULTS['post_new_user_no_email__fail']
        assert_valid_response(resp.data, expected_result)

        assert User.select().count() == 0

def test_move(self):
        # add game and check it
        game = Game.create(white='123', black='456', state='Ke1,ke8')
        self.assertTrue(abs((game.date_created - game.date_state).total_seconds()) < 1)
        self.assertEqual(game.next_color, WHITE)
        self.assertEqual(Game.select().count(), 1)
        self.assertEqual(Move.select().count(), 0)
        self.assertFalse(game.ended)
        # wait a second
        time.sleep(1)
        # add move and check
        game.add_move('K', 'e1-e2', 'Ke2,ke8')
        self.assertEqual(Move.select().count(), 1)
        # reload game
        game = Game.get(pk=game.pk)
        self.assertEqual(game.next_color, BLACK)
        self.assertEqual(game.state, 'Ke2,ke8')
        self.assertTrue((game.date_state - game.date_created).total_seconds() > 1)
        self.assertAlmostEqual(game.moves.get().time_move, 1, places=1)
        self.assertFalse(game.ended)
        # add move with ending game
        game.add_move('k', 'e8-e7', 'Ke2,ke7', True)
        self.assertTrue(game.ended)
        self.assertEqual(game.winner, BLACK)

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 address_creator(num_addr):
    LIST_COUNTRIES = ['Belgium', 'France', 'Germany',
                      'Greece', 'Italy', 'Portugal', 'Spain']
    for i in range(num_addr):
        country = random.choice(LIST_COUNTRIES)
        Address.create(
            uuid=fake.uuid4(),
            user=User.select().order_by(fn.Random()).get(),
            country=country,
            city=fake.city(),
            post_code=fake.postcode(),
            address=fake.street_name(),
            phone=fake.phone_number(),
        )

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 order_item_creator(num_items):
    orders = Order.select()
    for order in orders:
        for e in range(num_items):
            an_item = Item.select().order_by(fn.Random()).get()
            quantity = random.randint(1, 5)
            order.add_item(an_item, quantity)

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 name_exists(form, field):
    if User.select().where(User.username == field.data).exists():
        raise ValidationError('User with this name already exists.')

def email_exists(form,field):
    if User.select().where(User.email == field.data).exists():
        raise ValidationError('User with this email already exists.')

def is_valid(self):
        if len(self.form['username']) < 4:
            return self.error('username must be at least 4 characters')
        if self.form['username'] == 'anonymous':
            return self.error('username cannot be anonymous')
        if len(self.form['password']) < 8:
            return self.error('password must be at least 8 characters')
        if User.select().where(User.username == self.form['username']).count():
            return self.error('username is already in use')
        if self.form['email']:
            if not validate_email(self.form['email']):
                return self.error('email is not valid')
            if User.select().where(User.email == self.form['email']).count():
                return self.error('email is already in use')
        return super(RegistrationValidator, self).is_valid()

def name_exists(form, field):
    if User.select().where(User.username == field.data).exists():
        raise ValidationError('User with that name already exists.')

def email_exists(form, field):
    if User.select().where(User.email == field.data).exists():
        raise ValidationError('User with that email already exists.')

def name_exists(form, field):
    if User.select().where(User.username == field.data).exists():
        raise ValidationError('User with that name already exists.')

def email_exists(form, field):
    if User.select().where(User.email == field.data).exists():
        raise ValidationError('User with that email already exists.')