<commit_before>import testenv; testenv.configure_for_tests()

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base, declared_synonym
from testlib.fixtures import Base as Fixture
from testlib import *


class DeclarativeTest(TestBase):
    def setUp(self):
        global Base
        Base = declarative_base(testing.db)
        
    def tearDown(self):
        Base.metadata.drop_all()
        
    def test_basic(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            addresses = relation("Address", backref="user")

        class Address(Base, Fixture):
            __tablename__ = 'addresses'
            
            id = Column('id', Integer, primary_key=True)
            email = Column('email', String(50))
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
            
        Base.metadata.create_all()
        
        u1 = User(name='u1', addresses=[
            Address(email='one'),
            Address(email='two'),
        ])
        sess = create_session()
        sess.save(u1)
        sess.flush()
        sess.clear()
        
        self.assertEquals(sess.query(User).all(), [User(name='u1', addresses=[
            Address(email='one'),
            Address(email='two'),
        ])])
        
        a1 = sess.query(Address).filter(Address.email=='two').one()
        self.assertEquals(a1, Address(email='two'))
        self.assertEquals(a1.user, User(name='u1'))

    def test_expression(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            addresses = relation("Address", backref="user")
            
        class Address(Base, Fixture):
            __tablename__ = 'addresses'

            id = Column('id', Integer, primary_key=True)
            email = Column('email', String(50))
            user_id = Column('user_id', Integer, ForeignKey('users.id'))

        User.address_count = column_property(select([func.count(Address.id)]).where(Address.user_id==User.id).as_scalar())

        Base.metadata.create_all()

        u1 = User(name='u1', addresses=[
            Address(email='one'),
            Address(email='two'),
        ])
        sess = create_session()
        sess.save(u1)
        sess.flush()
        sess.clear()

        self.assertEquals(sess.query(User).all(), [User(name='u1', address_count=2, addresses=[
            Address(email='one'),
            Address(email='two'),
        ])])

    def test_synonym_inline(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            _name = Column('name', String(50))
            def _set_name(self, name):
                self._name = "SOMENAME " + name
            def _get_name(self):
                return self._name
            name = declared_synonym(property(_get_name, _set_name), '_name')
            
        Base.metadata.create_all()
        
        sess = create_session()
        u1 = User(name='someuser')
        assert u1.name == "SOMENAME someuser", u1.name
        sess.save(u1)
        sess.flush()
        self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)

    def test_synonym_added(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            _name = Column('name', String(50))
            def _set_name(self, name):
                self._name = "SOMENAME " + name
            def _get_name(self):
                return self._name
            name = property(_get_name, _set_name)
        User.name = declared_synonym(User.name, '_name')

        Base.metadata.create_all()

        sess = create_session()
        u1 = User(name='someuser')
        assert u1.name == "SOMENAME someuser", u1.name
        sess.save(u1)
        sess.flush()
        self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)
    
    def test_joined_inheritance(self):
        class Company(Base, Fixture):
            __tablename__ = 'companies'
            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            employees = relation("Person")
            
        class Person(Base, Fixture):
            __tablename__ = 'people'
            id = Column('id', Integer, primary_key=True)
            company_id = Column('company_id', Integer, ForeignKey('companies.id'))
            name = Column('name', String(50))
            discriminator = Column('type', String(50))
            __mapper_args__ = {'polymorphic_on':discriminator}
            
        class Engineer(Person):
            __tablename__ = 'engineers'
            __mapper_args__ = {'polymorphic_identity':'engineer'}
            id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
            primary_language = Column('primary_language', String(50))

        class Manager(Person):
            __tablename__ = 'managers'
            __mapper_args__ = {'polymorphic_identity':'manager'}
            id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
            golf_swing = Column('golf_swing', String(50))
        
        Base.metadata.create_all()

        sess = create_session()
        c1 = Company(name="MegaCorp, Inc.", employees=[
            Engineer(name="dilbert", primary_language="java"),
            Engineer(name="wally", primary_language="c++"),
            Manager(name="dogbert", golf_swing="fore!")
        ])

        c2 = Company(name="Elbonia, Inc.", employees=[
            Engineer(name="vlad", primary_language="cobol")
        ])

        sess.save(c1)
        sess.save(c2)
        sess.flush()
        sess.clear()
        
        self.assertEquals(sess.query(Company).filter(Company.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).first(), c2)

    def test_single_inheritance(self):
        class Company(Base, Fixture):
            __tablename__ = 'companies'
            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            employees = relation("Person")
            
        class Person(Base, Fixture):
            __tablename__ = 'people'
            id = Column('id', Integer, primary_key=True)
            company_id = Column('company_id', Integer, ForeignKey('companies.id'))
            name = Column('name', String(50))
            discriminator = Column('type', String(50))
            primary_language = Column('primary_language', String(50))
            golf_swing = Column('golf_swing', String(50))
            __mapper_args__ = {'polymorphic_on':discriminator}
            
        class Engineer(Person):
            __mapper_args__ = {'polymorphic_identity':'engineer'}
<commit_msg>- Added a primaryjoin= test<commit_after>import testenv; testenv.configure_for_tests()

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base, declared_synonym
from testlib.fixtures import Base as Fixture
from testlib import *


class DeclarativeTest(TestBase):
    def setUp(self):
        global Base
        Base = declarative_base(testing.db)
        
    def tearDown(self):
        Base.metadata.drop_all()
        
    def test_basic(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            addresses = relation("Address", backref="user")

        class Address(Base, Fixture):
            __tablename__ = 'addresses'
            
            id = Column('id', Integer, primary_key=True)
            email = Column('email', String(50))
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
            
        Base.metadata.create_all()
        
        u1 = User(name='u1', addresses=[
            Address(email='one'),
            Address(email='two'),
        ])
        sess = create_session()
        sess.save(u1)
        sess.flush()
        sess.clear()
        
        self.assertEquals(sess.query(User).all(), [User(name='u1', addresses=[
            Address(email='one'),
            Address(email='two'),
        ])])
        
        a1 = sess.query(Address).filter(Address.email=='two').one()
        self.assertEquals(a1, Address(email='two'))
        self.assertEquals(a1.user, User(name='u1'))

    def test_expression(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            addresses = relation("Address", backref="user")
            
        class Address(Base, Fixture):
            __tablename__ = 'addresses'

            id = Column('id', Integer, primary_key=True)
            email = Column('email', String(50))
            user_id = Column('user_id', Integer, ForeignKey('users.id'))

        User.address_count = column_property(select([func.count(Address.id)]).where(Address.user_id==User.id).as_scalar())

        Base.metadata.create_all()

        u1 = User(name='u1', addresses=[
            Address(email='one'),
            Address(email='two'),
        ])
        sess = create_session()
        sess.save(u1)
        sess.flush()
        sess.clear()

        self.assertEquals(sess.query(User).all(), [User(name='u1', address_count=2, addresses=[
            Address(email='one'),
            Address(email='two'),
        ])])

    def test_synonym_inline(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            _name = Column('name', String(50))
            def _set_name(self, name):
                self._name = "SOMENAME " + name
            def _get_name(self):
                return self._name
            name = declared_synonym(property(_get_name, _set_name), '_name')
            
        Base.metadata.create_all()
        
        sess = create_session()
        u1 = User(name='someuser')
        assert u1.name == "SOMENAME someuser", u1.name
        sess.save(u1)
        sess.flush()
        self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)

    def test_synonym_added(self):
        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            _name = Column('name', String(50))
            def _set_name(self, name):
                self._name = "SOMENAME " + name
            def _get_name(self):
                return self._name
            name = property(_get_name, _set_name)
        User.name = declared_synonym(User.name, '_name')

        Base.metadata.create_all()

        sess = create_session()
        u1 = User(name='someuser')
        assert u1.name == "SOMENAME someuser", u1.name
        sess.save(u1)
        sess.flush()
        self.assertEquals(sess.query(User).filter(User.name=="SOMENAME someuser").one(), u1)
    
    def test_joined_inheritance(self):
        class Company(Base, Fixture):
            __tablename__ = 'companies'
            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            employees = relation("Person")
            
        class Person(Base, Fixture):
            __tablename__ = 'people'
            id = Column('id', Integer, primary_key=True)
            company_id = Column('company_id', Integer, ForeignKey('companies.id'))
            name = Column('name', String(50))
            discriminator = Column('type', String(50))
            __mapper_args__ = {'polymorphic_on':discriminator}
            
        class Engineer(Person):
            __tablename__ = 'engineers'
            __mapper_args__ = {'polymorphic_identity':'engineer'}
            id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
            primary_language = Column('primary_language', String(50))

        class Manager(Person):
            __tablename__ = 'managers'
            __mapper_args__ = {'polymorphic_identity':'manager'}
            id = Column('id', Integer, ForeignKey('people.id'), primary_key=True)
            golf_swing = Column('golf_swing', String(50))
        
        Base.metadata.create_all()

        sess = create_session()
        c1 = Company(name="MegaCorp, Inc.", employees=[
            Engineer(name="dilbert", primary_language="java"),
            Engineer(name="wally", primary_language="c++"),
            Manager(name="dogbert", golf_swing="fore!")
        ])

        c2 = Company(name="Elbonia, Inc.", employees=[
            Engineer(name="vlad", primary_language="cobol")
        ])

        sess.save(c1)
        sess.save(c2)
        sess.flush()
        sess.clear()
        
        self.assertEquals(sess.query(Company).filter(Company.employees.of_type(Engineer).any(Engineer.primary_language=='cobol')).first(), c2)

    def test_relation_reference(self):
        class Address(Base, Fixture):
            __tablename__ = 'addresses'

            id = Column('id', Integer, primary_key=True)
            email = Column('email', String(50))
            user_id = Column('user_id', Integer, ForeignKey('users.id'))

        class User(Base, Fixture):
            __tablename__ = 'users'

            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            addresses = relation("Address", backref="user",
                                 primaryjoin=id==Address.user_id)

        User.address_count = column_property(select([func.count(Address.id)]).where(Address.user_id==User.id).as_scalar())

        Base.metadata.create_all()

        u1 = User(name='u1', addresses=[
            Address(email='one'),
            Address(email='two'),
        ])
        sess = create_session()
        sess.save(u1)
        sess.flush()
        sess.clear()

        self.assertEquals(sess.query(User).all(), [User(name='u1', address_count=2, addresses=[
            Address(email='one'),
            Address(email='two'),
        ])])

    def test_single_inheritance(self):
        class Company(Base, Fixture):
            __tablename__ = 'companies'
            id = Column('id', Integer, primary_key=True)
            name = Column('name', String(50))
            employees = relation("Person")
            
        class Person(Base, Fixture):
            __tablename__ = 'people'
            id = Column('id', Integer, primary_key=True)
            company_id = Column('company_id', Integer, ForeignKey('companies.id'))
            name = Column('name', String(50))
            discriminator = Column('type', String(50))
            primary_language = Column('primary_language', String(50))
            golf_swing = Column('golf_swing', String(50))
            __mapper_args__ = {'polymorphic_on':discriminator}
            
        class Engineer(Person):
            __mapper_args__ = {'polymorphic_identity':'engineer'}

{'fraction_non_alphanumeric': 0.1325916230366492, 'fraction_numerical': 0.008442408376963351, 'mean_word_length': 2.1941889632107023, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5770386', 'n_tokens_mistral': 4514, 'n_tokens_neox': 4380, 'n_words': 789}