|
1 from sqlalchemy import Column, Integer, String, ForeignKey, create_engine |
|
2 from sqlalchemy.ext.declarative import declarative_base |
|
3 from sqlalchemy.orm import relationship, backref |
|
4 import unittest #@UnresolvedImport |
|
5 from sqlalchemy.orm import sessionmaker |
|
6 from iri_tweet.utils import ObjectsBuffer |
|
7 |
|
8 Base = declarative_base() |
|
9 |
|
10 class User(Base): |
|
11 __tablename__ = 'users' |
|
12 |
|
13 id = Column(Integer, primary_key=True) |
|
14 name = Column(String) |
|
15 fullname = Column(String) |
|
16 password = Column(String) |
|
17 |
|
18 def __init__(self, name, fullname, password): |
|
19 self.name = name |
|
20 self.fullname = fullname |
|
21 self.password = password |
|
22 |
|
23 def __repr__(self): |
|
24 return "<User('%s','%s', '%s')>" % (self.name, self.fullname, self.password) |
|
25 |
|
26 |
|
27 class Address(Base): |
|
28 __tablename__ = 'addresses' |
|
29 id = Column(Integer, primary_key=True) |
|
30 email_address = Column(String, nullable=False) |
|
31 user_id = Column(Integer, ForeignKey('users.id')) |
|
32 |
|
33 user = relationship("User", backref=backref('addresses', order_by=id)) |
|
34 |
|
35 def __init__(self, user_id, email_address): |
|
36 self.email_address = email_address |
|
37 self.user_id = user_id |
|
38 |
|
39 def __repr__(self): |
|
40 return "<Address('%s')>" % self.email_address |
|
41 |
|
42 |
|
43 |
|
44 class TestObjectBuffer(unittest.TestCase): |
|
45 |
|
46 def setUp(self): |
|
47 self.engine = create_engine('sqlite:///:memory:', echo=False) |
|
48 Base.metadata.create_all(self.engine) |
|
49 sessionMaker = sessionmaker(bind=self.engine) |
|
50 self.session = sessionMaker() |
|
51 |
|
52 def testCreateUser(self): |
|
53 ed_user = User('ed', 'Ed Jones', 'edspassword') |
|
54 self.session.add(ed_user) |
|
55 self.assertTrue(ed_user.id is None) |
|
56 self.session.commit() |
|
57 self.assertTrue(ed_user.id is not None) |
|
58 |
|
59 def testSimpleBuffer(self): |
|
60 obj_buffer = ObjectsBuffer() |
|
61 obj_proxy = obj_buffer.add_object(User, ['ed1', 'Ed1 Jones', 'edspassword'], None, False) |
|
62 self.assertTrue(obj_proxy.id() is None) |
|
63 obj_buffer.persists(self.session) |
|
64 self.assertTrue(obj_proxy.id() is None) |
|
65 self.session.commit() |
|
66 self.assertTrue(obj_proxy.id() is not None) |
|
67 |
|
68 def testSimpleBufferFlush(self): |
|
69 obj_buffer = ObjectsBuffer() |
|
70 obj_proxy = obj_buffer.add_object(User, ['ed2', 'Ed2 Jones', 'edspassword'], None, True) |
|
71 self.assertTrue(obj_proxy.id() is None) |
|
72 obj_buffer.persists(self.session) |
|
73 self.assertTrue(obj_proxy.id() is not None) |
|
74 self.session.commit() |
|
75 self.assertTrue(obj_proxy.id() is not None) |
|
76 |
|
77 def testRelationBuffer(self): |
|
78 obj_buffer = ObjectsBuffer() |
|
79 user1_proxy = obj_buffer.add_object(User, ['ed3', 'Ed3 Jones', 'edspassword'], None, True) |
|
80 obj_buffer.add_object(Address, [user1_proxy.id,'ed3@mail.com'], None, False) |
|
81 obj_buffer.add_object(Address, [user1_proxy.id,'ed3@other.com'], None, False) |
|
82 user2_proxy = obj_buffer.add_object(User, ['ed4', 'Ed3 Jones', 'edspassword'], None, True) |
|
83 obj_buffer.add_object(Address, [user2_proxy.id,'ed4@mail.com'], None, False) |
|
84 obj_buffer.persists(self.session) |
|
85 self.session.commit() |
|
86 ed_user = self.session.query(User).filter_by(name='ed3').first() |
|
87 self.assertEquals(2, len(ed_user.addresses)) |
|
88 ed_user = self.session.query(User).filter_by(name='ed4').first() |
|
89 self.assertEquals(1, len(ed_user.addresses)) |
|
90 |
|
91 |
|
92 def tearDown(self): |
|
93 self.session.close() |
|
94 |
|
95 |
|
96 if __name__ == '__main__': |
|
97 unittest.main() |