<commit_before>from testbase import PersistTest, AssertMixin
import unittest, sys, os
from sqlalchemy.mapper import *
import sqlalchemy.objectstore as objectstore

#ECHO = True
ECHO = False
DATA = False
execfile("test/tables.py")

keywords.insert().execute(
    dict(keyword_id=1, name='blue'),
    dict(keyword_id=2, name='red'),
    dict(keyword_id=3, name='green'),
    dict(keyword_id=4, name='big'),
    dict(keyword_id=5, name='small'),
    dict(keyword_id=6, name='round'),
    dict(keyword_id=7, name='square')
)

db.connection().commit()

db.echo = True

class HistoryTest(AssertMixin):
    def testattr(self):
        m = mapper(User, users, properties = dict(addresses = relation(Address, addresses)))
        u = User()
        u.user_id = 7
        u.user_name = 'afdas'
        u.addresses.append(Address())
        u.addresses[0].email_address = 'hi'
        u.addresses.append(Address())
        u.addresses[1].email_address = 'there'
        print repr(u.__dict__)
        print repr(u.addresses)
        objectstore.uow().rollback_object(u)
        print repr(u.__dict__)
        
class SaveTest(AssertMixin):

    def setUp(self):
        objectstore.clear()
        clear_mappers()
        
    def testbasic(self):
        # save two users
        u = User()
        u.user_name = 'savetester'

        m = mapper(User, users)
        u2 = User()
        u2.user_name = 'savetester2'

        objectstore.uow().register_new(u)
        
        objectstore.uow().commit(u)
        objectstore.uow().commit()

        # assert the first one retreives the same from the identity map
        nu = m.get(u.user_id)
        self.assert_(u is nu)
        
        # clear out the identity map, so next get forces a SELECT
        objectstore.clear()

        # check it again, identity should be different but ids the same
        nu = m.get(u.user_id)
        self.assert_(u is not nu and u.user_id == nu.user_id and nu.user_name == 'savetester')

        # change first users name and save
        u.user_name = 'modifiedname'
        objectstore.uow().commit()

        # select both
        #objectstore.clear()
        userlist = m.select(users.c.user_id.in_(u.user_id, u2.user_id), order_by=[users.c.user_name])
        self.assert_(u.user_id == userlist[0].user_id and userlist[0].user_name == 'modifiedname')
        self.assert_(u2.user_id == userlist[1].user_id and userlist[1].user_name == 'savetester2')

    def testmultitable(self):
        """tests a save of an object where each instance spans two tables. also tests
        redefinition of the keynames for the column properties."""
        usersaddresses = sql.join(users, addresses, users.c.user_id == addresses.c.user_id)
        m = mapper(User, usersaddresses, table = users,  
            properties = dict(
                email = ColumnProperty(addresses.c.email_address), 
                foo_id = ColumnProperty(users.c.user_id, addresses.c.user_id)
                )
            )
            
        u = User()
        u.user_name = 'multitester'
        u.email = '<EMAIL>'

        objectstore.uow().commit()

        usertable = engine.ResultProxy(users.select(users.c.user_id.in_(u.foo_id)).execute()).fetchall()
        self.assert_(usertable[0].row == (u.foo_id, 'multitester'))
        addresstable = engine.ResultProxy(addresses.select(addresses.c.address_id.in_(u.address_id)).execute()).fetchall()
        self.assert_(addresstable[0].row == (u.address_id, u.foo_id, '<EMAIL>'))

        u.email = '<EMAIL>'
        u.user_name = 'imnew'
        objectstore.uow().commit()

        usertable = engine.ResultProxy(users.select(users.c.user_id.in_(u.foo_id)).execute()).fetchall()
        self.assert_(usertable[0].row == (u.foo_id, 'imnew'))
        addresstable = engine.ResultProxy(addresses.select(addresses.c.address_id.in_(u.address_id)).execute()).fetchall()
        self.assert_(addresstable[0].row == (u.address_id, u.foo_id, '<EMAIL>'))

        u = m.select(users.c.user_id==u.foo_id)[0]
        print repr(u.__dict__)

    def testonetoone(self):
        m = mapper(User, users, properties = dict(
            address = relation(Address, addresses, lazy = True, uselist = False)
        ))
        u = User()
        u.user_name = 'one2onetester'
        u.address = Address()
        u.address.email_address = '<EMAIL>'
        objectstore.uow().commit()
        u.user_name = 'imnew'
        objectstore.uow().commit()
        u.address.email_address = '<EMAIL>'
        objectstore.uow().commit()

    def testdelete(self):
        m = mapper(User, users, properties = dict(
            address = relation(Address, addresses, lazy = True, uselist = False, private = False)
        ))
        u = User()
        a = Address()
        u.user_name = 'one2onetester'
        u.address = a
        u.address.email_address = '<EMAIL>'
        objectstore.uow().commit()

        print "OK"
        objectstore.uow().register_deleted(u)
        objectstore.uow().commit()
        self.assert_(a.address_id is not None and a.user_id is None)
        
    def testbackwardsonetoone(self):
        # test 'backwards'
#        m = mapper(Address, addresses, properties = dict(
#            user = relation(User, users, foreignkey = addresses.c.user_id, primaryjoin = users.c.user_id == addresses.c.user_id, lazy = True, uselist = False)
#        ))
        # TODO: put assertion in here !!!
        m = mapper(Address, addresses, properties = dict(
            user = relation(User, users, lazy = True, uselist = False)
        ))
        data = [
            {'user_name' : 'thesub' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'assdkfj' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'n4knd' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'v88f4' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'asdf8d' , 'email_address' : '<EMAIL>'}
        ]
        objects = []
        for elem in data:
            a = Address()
            a.email_address = elem['email_address']
            a.user = User()
            a.user.user_name = elem['user_name']
            objects.append(a)
            
        objectstore.uow().commit()<commit_msg>dependency sort thing getting out of hand<commit_after>from testbase import PersistTest, AssertMixin
import unittest, sys, os
from sqlalchemy.mapper import *
import sqlalchemy.objectstore as objectstore

#ECHO = True
ECHO = False
DATA = False
execfile("test/tables.py")

keywords.insert().execute(
    dict(keyword_id=1, name='blue'),
    dict(keyword_id=2, name='red'),
    dict(keyword_id=3, name='green'),
    dict(keyword_id=4, name='big'),
    dict(keyword_id=5, name='small'),
    dict(keyword_id=6, name='round'),
    dict(keyword_id=7, name='square')
)

db.connection().commit()

db.echo = True

class HistoryTest(AssertMixin):
    def testattr(self):
        m = mapper(User, users, properties = dict(addresses = relation(Address, addresses)))
        u = User()
        u.user_id = 7
        u.user_name = 'afdas'
        u.addresses.append(Address())
        u.addresses[0].email_address = 'hi'
        u.addresses.append(Address())
        u.addresses[1].email_address = 'there'
        print repr(u.__dict__)
        print repr(u.addresses)
        objectstore.uow().rollback_object(u)
        print repr(u.__dict__)
        
class SaveTest(AssertMixin):

    def setUp(self):
        objectstore.clear()
        clear_mappers()
        
    def testbasic(self):
        # save two users
        u = User()
        u.user_name = 'savetester'

        m = mapper(User, users)
        u2 = User()
        u2.user_name = 'savetester2'

        objectstore.uow().register_new(u)
        
        objectstore.uow().commit(u)
        objectstore.uow().commit()

        # assert the first one retreives the same from the identity map
        nu = m.get(u.user_id)
        self.assert_(u is nu)
        
        # clear out the identity map, so next get forces a SELECT
        objectstore.clear()

        # check it again, identity should be different but ids the same
        nu = m.get(u.user_id)
        self.assert_(u is not nu and u.user_id == nu.user_id and nu.user_name == 'savetester')

        # change first users name and save
        u.user_name = 'modifiedname'
        objectstore.uow().commit()

        # select both
        #objectstore.clear()
        userlist = m.select(users.c.user_id.in_(u.user_id, u2.user_id), order_by=[users.c.user_name])
        self.assert_(u.user_id == userlist[0].user_id and userlist[0].user_name == 'modifiedname')
        self.assert_(u2.user_id == userlist[1].user_id and userlist[1].user_name == 'savetester2')

    def testmultitable(self):
        """tests a save of an object where each instance spans two tables. also tests
        redefinition of the keynames for the column properties."""
        usersaddresses = sql.join(users, addresses, users.c.user_id == addresses.c.user_id)
        m = mapper(User, usersaddresses, table = users,  
            properties = dict(
                email = ColumnProperty(addresses.c.email_address), 
                foo_id = ColumnProperty(users.c.user_id, addresses.c.user_id)
                )
            )
            
        u = User()
        u.user_name = 'multitester'
        u.email = '<EMAIL>'

        objectstore.uow().commit()

        usertable = engine.ResultProxy(users.select(users.c.user_id.in_(u.foo_id)).execute()).fetchall()
        self.assert_(usertable[0].row == (u.foo_id, 'multitester'))
        addresstable = engine.ResultProxy(addresses.select(addresses.c.address_id.in_(u.address_id)).execute()).fetchall()
        self.assert_(addresstable[0].row == (u.address_id, u.foo_id, '<EMAIL>'))

        u.email = '<EMAIL>'
        u.user_name = 'imnew'
        objectstore.uow().commit()

        usertable = engine.ResultProxy(users.select(users.c.user_id.in_(u.foo_id)).execute()).fetchall()
        self.assert_(usertable[0].row == (u.foo_id, 'imnew'))
        addresstable = engine.ResultProxy(addresses.select(addresses.c.address_id.in_(u.address_id)).execute()).fetchall()
        self.assert_(addresstable[0].row == (u.address_id, u.foo_id, '<EMAIL>'))

        u = m.select(users.c.user_id==u.foo_id)[0]
        print repr(u.__dict__)

    def testonetoone(self):
        m = mapper(User, users, properties = dict(
            address = relation(Address, addresses, lazy = True, uselist = False)
        ))
        u = User()
        u.user_name = 'one2onetester'
        u.address = Address()
        u.address.email_address = '<EMAIL>'
        objectstore.uow().commit()
        u.user_name = 'imnew'
        objectstore.uow().commit()
        u.address.email_address = '<EMAIL>'
        objectstore.uow().commit()

    def testdelete(self):
        m = mapper(User, users, properties = dict(
            address = relation(Address, addresses, lazy = True, uselist = False, private = False)
        ))
        u = User()
        a = Address()
        u.user_name = 'one2onetester'
        u.address = a
        u.address.email_address = '<EMAIL>'
        objectstore.uow().commit()

        objectstore.uow().register_deleted(u)
        objectstore.uow().commit()
        self.assert_(a.address_id is not None and a.user_id is None and not objectstore.uow().identity_map.has_key(u._instance_key) and objectstore.uow().identity_map.has_key(a._instance_key))

    def testcascadingdelete(self):
        m = mapper(User, users, properties = dict(
            address = relation(Address, addresses, lazy = False, uselist = False, private = True),
            orders = relation(
                mapper(Order, orders, properties = dict (
                    items = relation(Item, orderitems, lazy = False, uselist =True, private = True)
                )), 
                lazy = True, uselist = True, private = True)
        ))
        
        data = [User,
            {'user_name' : 'ed', 
                'address' : (Address, {'email_address' : '<EMAIL>'}),
                'orders' : (Order, [
                    {'description' : 'eds 1st order', 'items' : (Item, [{'item_name' : 'eds o1 item'}, {'item_name' : 'eds other o1 item'}])}, 
                    {'description' : 'eds 2nd order', 'items' : (Item, [{'item_name' : 'eds o2 item'}, {'item_name' : 'eds other o2 item'}])}
                 ])
            },
            {'user_name' : 'jack', 
                'address' : (Address, {'email_address' : '<EMAIL>'}),
                'orders' : (Order, [
                    {'description' : 'jacks 1st order', 'items' : (Item, [{'item_name' : 'im a lumberjack'}, {'item_name' : 'and im ok'}])}
                 ])
            },
            {'user_name' : 'foo', 
                'address' : (Address, {'email_address': '<EMAIL>'}),
                'orders' : (Order, [
                    {'description' : 'foo order', 'items' : (Item, [])}, 
                    {'description' : 'foo order 2', 'items' : (Item, [{'item_name' : 'hi'}])}, 
                    {'description' : 'foo order three', 'items' : (Item, [{'item_name' : 'there'}])}
                ])
            }        
        ]
        
        for elem in data[1:]:
            u = User()
            u.user_name = elem['user_name']
            u.address = Address()
            u.address.email_address = elem['address'][1]['email_address']
            u.orders = []
            for order in elem['orders'][1]:
                o = Order()
                o.isopen = None
                o.description = order['description']
                u.orders.append(o)
                o.items = []
                for item in order['items'][1]:
                    i = Item()
                    i.item_name = item['item_name']
                    o.items.append(i)
                
        objectstore.uow().commit()
        objectstore.clear()
        
        l = m.select()
        for u in l:
            print repr(u.orders)
        #self.assert_result(l, data[0], *data[1])
        
        objectstore.uow().register_deleted(l[0])
        objectstore.uow().register_deleted(l[2])
        objectstore.uow().commit()
        
    def testbackwardsonetoone(self):
        # test 'backwards'
#        m = mapper(Address, addresses, properties = dict(
#            user = relation(User, users, foreignkey = addresses.c.user_id, primaryjoin = users.c.user_id == addresses.c.user_id, lazy = True, uselist = False)
#        ))
        # TODO: put assertion in here !!!
        m = mapper(Address, addresses, properties = dict(
            user = relation(User, users, lazy = True, uselist = False)
        ))
        data = [
            {'user_name' : 'thesub' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'assdkfj' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'n4knd' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'v88f4' , 'email_address' : '<EMAIL>'},
            {'user_name' : 'asdf8d' , 'email_address' : '<EMAIL>'}
        ]
        objects = []
        for elem in data:
            a = Address()
            a.email_address = elem['email_address']
            a.user = User()
            a.user.user_name = elem['user_name']
            objects.append(a)
            
        objectstore.uow().commit()
{'fraction_non_alphanumeric': 0.1406178264576638, 'fraction_numerical': 0.005246933823047157, 'mean_word_length': 2.1753436068304874, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3471620', 'n_tokens_mistral': 4842, 'n_tokens_neox': 4724, 'n_words': 974}