1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
import pytest
import sqlalchemy as sa
from sqlalchemy_utils import QueryChain
@pytest.fixture
def User(Base):
class User(Base):
__tablename__ = 'user'
id = sa.Column(sa.Integer, primary_key=True)
return User
@pytest.fixture
def Article(Base):
class Article(Base):
__tablename__ = 'article'
id = sa.Column(sa.Integer, primary_key=True)
return Article
@pytest.fixture
def BlogPost(Base):
class BlogPost(Base):
__tablename__ = 'blog_post'
id = sa.Column(sa.Integer, primary_key=True)
return BlogPost
@pytest.fixture
def init_models(User, Article, BlogPost):
pass
@pytest.fixture
def users(session, User):
users = [User(), User()]
session.add_all(users)
session.commit()
return users
@pytest.fixture
def articles(session, Article):
articles = [Article(), Article(), Article(), Article()]
session.add_all(articles)
session.commit()
return articles
@pytest.fixture
def posts(session, BlogPost):
posts = [BlogPost(), BlogPost(), BlogPost()]
session.add_all(posts)
session.commit()
return posts
@pytest.fixture
def chain(session, users, articles, posts, User, Article, BlogPost):
return QueryChain(
[
session.query(User).order_by('id'),
session.query(Article).order_by('id'),
session.query(BlogPost).order_by('id')
]
)
class TestQueryChain:
def test_iter(self, chain):
assert len(list(chain)) == 9
def test_iter_with_limit(self, chain, users, articles):
c = chain.limit(4)
objects = list(c)
assert users == objects[0:2]
assert articles[0:2] == objects[2:]
def test_iter_with_offset(self, chain, articles, posts):
c = chain.offset(3)
objects = list(c)
assert articles[1:] + posts == objects
def test_iter_with_limit_and_offset(self, chain, articles, posts):
c = chain.offset(3).limit(4)
objects = list(c)
assert articles[1:] + posts[0:1] == objects
def test_iter_with_offset_spanning_multiple_queries(self, chain, posts):
c = chain.offset(7)
objects = list(c)
assert posts[1:] == objects
def test_repr(self, chain):
assert repr(chain) == '<QueryChain at 0x%x>' % id(chain)
def test_getitem_with_slice(self, chain):
c = chain[1:]
assert c._offset == 1
assert c._limit is None
def test_getitem_with_single_key(self, chain, articles):
article = chain[2]
assert article == articles[0]
def test_count(self, chain):
assert chain.count() == 9
|
