Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use create_session method in AutoItDriverServer

Best Python code snippet using AutoItDriverServer_python

test_polymorphic_rel.py

Source:test_polymorphic_rel.py

...57        """58        Test that all objects load from the full query, when59        with_polymorphic is used.60        """61        sess = create_session()62        def go():63            eq_(64                sess.query(Person).order_by(Person.person_id).all(),65                all_employees,66            )67        count = {"": 14, "Polymorphic": 9}.get(self.select_type, 10)68        self.assert_sql_count(testing.db, go, count)69    def test_primary_eager_aliasing_one(self):70        # For both joinedload() and subqueryload(), if the original q is71        # not loading the subclass table, the joinedload doesn't happen.72        sess = create_session()73        def go():74            eq_(75                sess.query(Person)76                .order_by(Person.person_id)77                .options(joinedload(Engineer.machines))[1:3],78                all_employees[1:3],79            )80        count = {"": 6, "Polymorphic": 3}.get(self.select_type, 4)81        self.assert_sql_count(testing.db, go, count)82    def test_primary_eager_aliasing_two(self):83        sess = create_session()84        def go():85            eq_(86                sess.query(Person)87                .order_by(Person.person_id)88                .options(subqueryload(Engineer.machines))89                .all(),90                all_employees,91            )92        count = {"": 14, "Polymorphic": 7}.get(self.select_type, 8)93        self.assert_sql_count(testing.db, go, count)94    def test_primary_eager_aliasing_three(self):95        # assert the JOINs don't over JOIN96        sess = create_session()97        def go():98            eq_(99                sess.query(Person)100                .with_polymorphic("*")101                .order_by(Person.person_id)102                .options(joinedload(Engineer.machines))[1:3],103                all_employees[1:3],104            )105        self.assert_sql_count(testing.db, go, 3)106        eq_(107            select([func.count("*")])108            .select_from(109                sess.query(Person)110                .with_polymorphic("*")111                .options(joinedload(Engineer.machines))112                .order_by(Person.person_id)113                .limit(2)114                .offset(1)115                .with_labels()116                .subquery()117            )118            .scalar(),119            2,120        )121    def test_get_one(self):122        """123        For all mappers, ensure the primary key has been calculated as124        just the "person_id" column.125        """126        sess = create_session()127        eq_(128            sess.query(Person).get(e1.person_id),129            Engineer(name="dilbert", primary_language="java"),130        )131    def test_get_two(self):132        sess = create_session()133        eq_(134            sess.query(Engineer).get(e1.person_id),135            Engineer(name="dilbert", primary_language="java"),136        )137    def test_get_three(self):138        sess = create_session()139        eq_(140            sess.query(Manager).get(b1.person_id),141            Boss(name="pointy haired boss", golf_swing="fore"),142        )143    def test_multi_join(self):144        sess = create_session()145        e = aliased(Person)146        c = aliased(Company)147        q = (148            sess.query(Company, Person, c, e)149            .join(Person, Company.employees)150            .join(e, c.employees)151            .filter(Person.name == "dilbert")152            .filter(e.name == "wally")153        )154        eq_(q.count(), 1)155        eq_(156            q.all(),157            [158                (159                    Company(company_id=1, name="MegaCorp, Inc."),160                    Engineer(161                        status="regular engineer",162                        engineer_name="dilbert",163                        name="dilbert",164                        company_id=1,165                        primary_language="java",166                        person_id=1,167                        type="engineer",168                    ),169                    Company(company_id=1, name="MegaCorp, Inc."),170                    Engineer(171                        status="regular engineer",172                        engineer_name="wally",173                        name="wally",174                        company_id=1,175                        primary_language="c++",176                        person_id=2,177                        type="engineer",178                    ),179                )180            ],181        )182    def test_filter_on_subclass_one(self):183        sess = create_session()184        eq_(sess.query(Engineer).all()[0], Engineer(name="dilbert"))185    def test_filter_on_subclass_two(self):186        sess = create_session()187        eq_(sess.query(Engineer).first(), Engineer(name="dilbert"))188    def test_filter_on_subclass_three(self):189        sess = create_session()190        eq_(191            sess.query(Engineer)192            .filter(Engineer.person_id == e1.person_id)193            .first(),194            Engineer(name="dilbert"),195        )196    def test_filter_on_subclass_four(self):197        sess = create_session()198        eq_(199            sess.query(Manager)200            .filter(Manager.person_id == m1.person_id)201            .one(),202            Manager(name="dogbert"),203        )204    def test_filter_on_subclass_five(self):205        sess = create_session()206        eq_(207            sess.query(Manager)208            .filter(Manager.person_id == b1.person_id)209            .one(),210            Boss(name="pointy haired boss"),211        )212    def test_filter_on_subclass_six(self):213        sess = create_session()214        eq_(215            sess.query(Boss).filter(Boss.person_id == b1.person_id).one(),216            Boss(name="pointy haired boss"),217        )218    def test_join_from_polymorphic_nonaliased_one(self):219        sess = create_session()220        eq_(221            sess.query(Person)222            .join("paperwork", aliased=False)223            .filter(Paperwork.description.like("%review%"))224            .all(),225            [b1, m1],226        )227    def test_join_from_polymorphic_nonaliased_two(self):228        sess = create_session()229        eq_(230            sess.query(Person)231            .order_by(Person.person_id)232            .join("paperwork", aliased=False)233            .filter(Paperwork.description.like("%#2%"))234            .all(),235            [e1, m1],236        )237    def test_join_from_polymorphic_nonaliased_three(self):238        sess = create_session()239        eq_(240            sess.query(Engineer)241            .order_by(Person.person_id)242            .join("paperwork", aliased=False)243            .filter(Paperwork.description.like("%#2%"))244            .all(),245            [e1],246        )247    def test_join_from_polymorphic_nonaliased_four(self):248        sess = create_session()249        eq_(250            sess.query(Person)251            .order_by(Person.person_id)252            .join("paperwork", aliased=False)253            .filter(Person.name.like("%dog%"))254            .filter(Paperwork.description.like("%#2%"))255            .all(),256            [m1],257        )258    def test_join_from_polymorphic_aliased_one(self):259        sess = create_session()260        eq_(261            sess.query(Person)262            .order_by(Person.person_id)263            .join("paperwork", aliased=True)264            .filter(Paperwork.description.like("%review%"))265            .all(),266            [b1, m1],267        )268    def test_join_from_polymorphic_aliased_two(self):269        sess = create_session()270        eq_(271            sess.query(Person)272            .order_by(Person.person_id)273            .join("paperwork", aliased=True)274            .filter(Paperwork.description.like("%#2%"))275            .all(),276            [e1, m1],277        )278    def test_join_from_polymorphic_aliased_three(self):279        sess = create_session()280        eq_(281            sess.query(Engineer)282            .order_by(Person.person_id)283            .join("paperwork", aliased=True)284            .filter(Paperwork.description.like("%#2%"))285            .all(),286            [e1],287        )288    def test_join_from_polymorphic_aliased_four(self):289        sess = create_session()290        eq_(291            sess.query(Person)292            .order_by(Person.person_id)293            .join("paperwork", aliased=True)294            .filter(Person.name.like("%dog%"))295            .filter(Paperwork.description.like("%#2%"))296            .all(),297            [m1],298        )299    def test_join_from_with_polymorphic_nonaliased_one(self):300        sess = create_session()301        eq_(302            sess.query(Person)303            .with_polymorphic(Manager)304            .order_by(Person.person_id)305            .join("paperwork")306            .filter(Paperwork.description.like("%review%"))307            .all(),308            [b1, m1],309        )310    def test_join_from_with_polymorphic_nonaliased_two(self):311        sess = create_session()312        eq_(313            sess.query(Person)314            .with_polymorphic([Manager, Engineer])315            .order_by(Person.person_id)316            .join("paperwork")317            .filter(Paperwork.description.like("%#2%"))318            .all(),319            [e1, m1],320        )321    def test_join_from_with_polymorphic_nonaliased_three(self):322        sess = create_session()323        eq_(324            sess.query(Person)325            .with_polymorphic([Manager, Engineer])326            .order_by(Person.person_id)327            .join("paperwork")328            .filter(Person.name.like("%dog%"))329            .filter(Paperwork.description.like("%#2%"))330            .all(),331            [m1],332        )333    def test_join_from_with_polymorphic_aliased_one(self):334        sess = create_session()335        eq_(336            sess.query(Person)337            .with_polymorphic(Manager)338            .join("paperwork", aliased=True)339            .filter(Paperwork.description.like("%review%"))340            .all(),341            [b1, m1],342        )343    def test_join_from_with_polymorphic_aliased_two(self):344        sess = create_session()345        eq_(346            sess.query(Person)347            .with_polymorphic([Manager, Engineer])348            .order_by(Person.person_id)349            .join("paperwork", aliased=True)350            .filter(Paperwork.description.like("%#2%"))351            .all(),352            [e1, m1],353        )354    def test_join_from_with_polymorphic_aliased_three(self):355        sess = create_session()356        eq_(357            sess.query(Person)358            .with_polymorphic([Manager, Engineer])359            .order_by(Person.person_id)360            .join("paperwork", aliased=True)361            .filter(Person.name.like("%dog%"))362            .filter(Paperwork.description.like("%#2%"))363            .all(),364            [m1],365        )366    def test_join_to_polymorphic_nonaliased(self):367        sess = create_session()368        eq_(369            sess.query(Company)370            .join("employees")371            .filter(Person.name == "vlad")372            .one(),373            c2,374        )375    def test_join_to_polymorphic_aliased(self):376        sess = create_session()377        eq_(378            sess.query(Company)379            .join("employees", aliased=True)380            .filter(Person.name == "vlad")381            .one(),382            c2,383        )384    def test_polymorphic_any_one(self):385        sess = create_session()386        any_ = Company.employees.any(Person.name == "vlad")387        eq_(sess.query(Company).filter(any_).all(), [c2])388    def test_polymorphic_any_two(self):389        sess = create_session()390        # test that the aliasing on "Person" does not bleed into the391        # EXISTS clause generated by any()392        any_ = Company.employees.any(Person.name == "wally")393        eq_(394            sess.query(Company)395            .join(Company.employees, aliased=True)396            .filter(Person.name == "dilbert")397            .filter(any_)398            .all(),399            [c1],400        )401    def test_polymorphic_any_three(self):402        sess = create_session()403        any_ = Company.employees.any(Person.name == "vlad")404        eq_(405            sess.query(Company)406            .join(Company.employees, aliased=True)407            .filter(Person.name == "dilbert")408            .filter(any_)409            .all(),410            [],411        )412    def test_polymorphic_any_eight(self):413        sess = create_session()414        any_ = Engineer.machines.any(Machine.name == "Commodore 64")415        eq_(416            sess.query(Person).order_by(Person.person_id).filter(any_).all(),417            [e2, e3],418        )419    def test_polymorphic_any_nine(self):420        sess = create_session()421        any_ = Person.paperwork.any(Paperwork.description == "review #2")422        eq_(423            sess.query(Person).order_by(Person.person_id).filter(any_).all(),424            [m1],425        )426    def test_join_from_columns_or_subclass_one(self):427        sess = create_session()428        expected = [("dogbert",), ("pointy haired boss",)]429        eq_(sess.query(Manager.name).order_by(Manager.name).all(), expected)430    def test_join_from_columns_or_subclass_two(self):431        sess = create_session()432        expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)]433        eq_(434            sess.query(Manager.name)435            .join(Paperwork, Manager.paperwork)436            .order_by(Manager.name)437            .all(),438            expected,439        )440    def test_join_from_columns_or_subclass_three(self):441        sess = create_session()442        expected = [443            ("dilbert",),444            ("dilbert",),445            ("dogbert",),446            ("dogbert",),447            ("pointy haired boss",),448            ("vlad",),449            ("wally",),450            ("wally",),451        ]452        eq_(453            sess.query(Person.name)454            .join(Paperwork, Person.paperwork)455            .order_by(Person.name)456            .all(),457            expected,458        )459    def test_join_from_columns_or_subclass_four(self):460        sess = create_session()461        # Load Person.name, joining from Person -> paperwork, get all462        # the people.463        expected = [464            ("dilbert",),465            ("dilbert",),466            ("dogbert",),467            ("dogbert",),468            ("pointy haired boss",),469            ("vlad",),470            ("wally",),471            ("wally",),472        ]473        eq_(474            sess.query(Person.name)475            .join(paperwork, Person.person_id == paperwork.c.person_id)476            .order_by(Person.name)477            .all(),478            expected,479        )480    def test_join_from_columns_or_subclass_five(self):481        sess = create_session()482        # same, on manager.  get only managers.483        expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)]484        eq_(485            sess.query(Manager.name)486            .join(paperwork, Manager.person_id == paperwork.c.person_id)487            .order_by(Person.name)488            .all(),489            expected,490        )491    def test_join_from_columns_or_subclass_six(self):492        sess = create_session()493        if self.select_type == "":494            # this now raises, due to [ticket:1892].  Manager.person_id495            # is now the "person_id" column on Manager. SQL is incorrect.496            assert_raises(497                sa_exc.DBAPIError,498                sess.query(Person.name)499                .join(paperwork, Manager.person_id == paperwork.c.person_id)500                .order_by(Person.name)501                .all,502            )503        elif self.select_type == "Unions":504            # with the union, not something anyone would really be using505            # here, it joins to the full result set.  This is 0.6's506            # behavior and is more or less wrong.507            expected = [508                ("dilbert",),509                ("dilbert",),510                ("dogbert",),511                ("dogbert",),512                ("pointy haired boss",),513                ("vlad",),514                ("wally",),515                ("wally",),516            ]517            eq_(518                sess.query(Person.name)519                .join(paperwork, Manager.person_id == paperwork.c.person_id)520                .order_by(Person.name)521                .all(),522                expected,523            )524        else:525            # when a join is present and managers.person_id is available,526            # you get the managers.527            expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)]528            eq_(529                sess.query(Person.name)530                .join(paperwork, Manager.person_id == paperwork.c.person_id)531                .order_by(Person.name)532                .all(),533                expected,534            )535    def test_join_from_columns_or_subclass_seven(self):536        sess = create_session()537        eq_(538            sess.query(Manager)539            .join(Paperwork, Manager.paperwork)540            .order_by(Manager.name)541            .all(),542            [m1, b1],543        )544    def test_join_from_columns_or_subclass_eight(self):545        sess = create_session()546        expected = [("dogbert",), ("dogbert",), ("pointy haired boss",)]547        eq_(548            sess.query(Manager.name)549            .join(paperwork, Manager.person_id == paperwork.c.person_id)550            .order_by(Manager.name)551            .all(),552            expected,553        )554    def test_join_from_columns_or_subclass_nine(self):555        sess = create_session()556        eq_(557            sess.query(Manager.person_id)558            .join(paperwork, Manager.person_id == paperwork.c.person_id)559            .order_by(Manager.name)560            .all(),561            [(4,), (4,), (3,)],562        )563    def test_join_from_columns_or_subclass_ten(self):564        sess = create_session()565        expected = [566            ("pointy haired boss", "review #1"),567            ("dogbert", "review #2"),568            ("dogbert", "review #3"),569        ]570        eq_(571            sess.query(Manager.name, Paperwork.description)572            .join(Paperwork, Manager.person_id == Paperwork.person_id)573            .order_by(Paperwork.paperwork_id)574            .all(),575            expected,576        )577    def test_join_from_columns_or_subclass_eleven(self):578        sess = create_session()579        expected = [("pointy haired boss",), ("dogbert",), ("dogbert",)]580        malias = aliased(Manager)581        eq_(582            sess.query(malias.name)583            .join(paperwork, malias.person_id == paperwork.c.person_id)584            .all(),585            expected,586        )587    def test_subclass_option_pathing(self):588        sess = create_session()589        dilbert = (590            sess.query(Person)591            .options(defaultload(Engineer.machines).defer(Machine.name))592            .filter(Person.name == "dilbert")593            .first()594        )595        m = dilbert.machines[0]596        assert "name" not in m.__dict__597        eq_(m.name, "IBM ThinkPad")598    def test_expire(self):599        """600        Test that individual column refresh doesn't get tripped up by601        the select_table mapper.602        """603        sess = create_session()604        name = "dogbert"605        m1 = sess.query(Manager).filter(Manager.name == name).one()606        sess.expire(m1)607        assert m1.status == "regular manager"608        name = "pointy haired boss"609        m2 = sess.query(Manager).filter(Manager.name == name).one()610        sess.expire(m2, ["manager_name", "golf_swing"])611        assert m2.golf_swing == "fore"612    def test_with_polymorphic_one(self):613        sess = create_session()614        def go():615            eq_(616                sess.query(Person)617                .with_polymorphic(Engineer)618                .filter(Engineer.primary_language == "java")619                .all(),620                self._emps_wo_relationships_fixture()[0:1],621            )622        self.assert_sql_count(testing.db, go, 1)623    def test_with_polymorphic_two(self):624        sess = create_session()625        def go():626            eq_(627                sess.query(Person)628                .with_polymorphic("*")629                .order_by(Person.person_id)630                .all(),631                self._emps_wo_relationships_fixture(),632            )633        self.assert_sql_count(testing.db, go, 1)634    def test_with_polymorphic_three(self):635        sess = create_session()636        def go():637            eq_(638                sess.query(Person)639                .with_polymorphic(Engineer)640                .order_by(Person.person_id)641                .all(),642                self._emps_wo_relationships_fixture(),643            )644        self.assert_sql_count(testing.db, go, 3)645    def test_with_polymorphic_four(self):646        sess = create_session()647        def go():648            eq_(649                sess.query(Person)650                .with_polymorphic(Engineer, people.outerjoin(engineers))651                .order_by(Person.person_id)652                .all(),653                self._emps_wo_relationships_fixture(),654            )655        self.assert_sql_count(testing.db, go, 3)656    def test_with_polymorphic_five(self):657        sess = create_session()658        def go():659            # limit the polymorphic join down to just "Person",660            # overriding select_table661            eq_(662                sess.query(Person).with_polymorphic(Person).all(),663                self._emps_wo_relationships_fixture(),664            )665        self.assert_sql_count(testing.db, go, 6)666    def test_with_polymorphic_six(self):667        sess = create_session()668        assert_raises(669            sa_exc.InvalidRequestError,670            sess.query(Person).with_polymorphic,671            Paperwork,672        )673        assert_raises(674            sa_exc.InvalidRequestError,675            sess.query(Engineer).with_polymorphic,676            Boss,677        )678        assert_raises(679            sa_exc.InvalidRequestError,680            sess.query(Engineer).with_polymorphic,681            Person,682        )683    def test_with_polymorphic_seven(self):684        sess = create_session()685        # compare to entities without related collections to prevent686        # additional lazy SQL from firing on loaded entities687        eq_(688            sess.query(Person)689            .with_polymorphic("*")690            .order_by(Person.person_id)691            .all(),692            self._emps_wo_relationships_fixture(),693        )694    def test_relationship_to_polymorphic_one(self):695        expected = self._company_with_emps_machines_fixture()696        sess = create_session()697        def go():698            # test load Companies with lazy load to 'employees'699            eq_(sess.query(Company).all(), expected)700        count = {"": 10, "Polymorphic": 5}.get(self.select_type, 6)701        self.assert_sql_count(testing.db, go, count)702    def test_relationship_to_polymorphic_two(self):703        expected = self._company_with_emps_machines_fixture()704        sess = create_session()705        def go():706            # with #2438, of_type() is recognized.  This707            # overrides the with_polymorphic of the mapper708            # and we get a consistent 3 queries now.709            eq_(710                sess.query(Company)711                .options(712                    joinedload(Company.employees.of_type(Engineer)).joinedload(713                        Engineer.machines714                    )715                )716                .all(),717                expected,718            )719        # in the old case, we would get this720        # count = {'':7, 'Polymorphic':1}.get(self.select_type, 2)721        # query one is company->Person/Engineer->Machines722        # query two is managers + boss for row #3723        # query three is managers for row #4724        count = 3725        self.assert_sql_count(testing.db, go, count)726    def test_relationship_to_polymorphic_three(self):727        expected = self._company_with_emps_machines_fixture()728        sess = create_session()729        sess = create_session()730        def go():731            eq_(732                sess.query(Company)733                .options(734                    subqueryload(735                        Company.employees.of_type(Engineer)736                    ).subqueryload(Engineer.machines)737                )738                .all(),739                expected,740            )741        # the old case where subqueryload_all742        # didn't work with of_tyoe743        # count = { '':8, 'Joins':4, 'Unions':4, 'Polymorphic':3,744        #    'AliasedJoins':4}[self.select_type]745        # query one is company->Person/Engineer->Machines746        # query two is Person/Engineer subq747        # query three is Machines subq748        # (however this test can't tell if the Q was a749        # lazyload or subqload ...)750        # query four is managers + boss for row #3751        # query five is managers for row #4752        count = 5753        self.assert_sql_count(testing.db, go, count)754    def test_joinedload_on_subclass(self):755        sess = create_session()756        expected = [757            Engineer(758                name="dilbert",759                engineer_name="dilbert",760                primary_language="java",761                status="regular engineer",762                machines=[763                    Machine(name="IBM ThinkPad"),764                    Machine(name="IPhone"),765                ],766            )767        ]768        def go():769            # test load People with joinedload to engineers + machines770            eq_(771                sess.query(Person)772                .with_polymorphic("*")773                .options(joinedload(Engineer.machines))774                .filter(Person.name == "dilbert")775                .all(),776                expected,777            )778        self.assert_sql_count(testing.db, go, 1)779    def test_subqueryload_on_subclass(self):780        sess = create_session()781        expected = [782            Engineer(783                name="dilbert",784                engineer_name="dilbert",785                primary_language="java",786                status="regular engineer",787                machines=[788                    Machine(name="IBM ThinkPad"),789                    Machine(name="IPhone"),790                ],791            )792        ]793        def go():794            eq_(795                sess.query(Person)796                .with_polymorphic("*")797                .options(subqueryload(Engineer.machines))798                .filter(Person.name == "dilbert")799                .all(),800                expected,801            )802        self.assert_sql_count(testing.db, go, 2)803    def test_query_subclass_join_to_base_relationship(self):804        sess = create_session()805        # non-polymorphic806        eq_(sess.query(Engineer).join(Person.paperwork).all(), [e1, e2, e3])807    def test_join_to_subclass(self):808        sess = create_session()809        eq_(810            sess.query(Company)811            .join(people.join(engineers), "employees")812            .filter(Engineer.primary_language == "java")813            .all(),814            [c1],815        )816    def test_join_to_subclass_one(self):817        sess = create_session()818        eq_(819            sess.query(Company)820            .select_from(companies.join(people).join(engineers))821            .filter(Engineer.primary_language == "java")822            .all(),823            [c1],824        )825    def test_join_to_subclass_two(self):826        sess = create_session()827        eq_(828            sess.query(Company)829            .join(people.join(engineers), "employees")830            .filter(Engineer.primary_language == "java")831            .all(),832            [c1],833        )834    def test_join_to_subclass_three(self):835        sess = create_session()836        ealias = aliased(Engineer)837        eq_(838            sess.query(Company)839            .join(ealias, "employees")840            .filter(ealias.primary_language == "java")841            .all(),842            [c1],843        )844    def test_join_to_subclass_six(self):845        sess = create_session()846        eq_(847            sess.query(Company)848            .join(people.join(engineers), "employees")849            .join(Engineer.machines)850            .all(),851            [c1, c2],852        )853    def test_join_to_subclass_six_point_five(self):854        sess = create_session()855        eq_(856            sess.query(Company)857            .join(people.join(engineers), "employees")858            .join(Engineer.machines)859            .filter(Engineer.name == "dilbert")860            .all(),861            [c1],862        )863    def test_join_to_subclass_seven(self):864        sess = create_session()865        eq_(866            sess.query(Company)867            .join(people.join(engineers), "employees")868            .join(Engineer.machines)869            .filter(Machine.name.ilike("%thinkpad%"))870            .all(),871            [c1],872        )873    def test_join_to_subclass_eight(self):874        sess = create_session()875        eq_(sess.query(Person).join(Engineer.machines).all(), [e1, e2, e3])876    def test_join_to_subclass_nine(self):877        sess = create_session()878        eq_(879            sess.query(Company)880            .select_from(companies.join(people).join(engineers))881            .filter(Engineer.primary_language == "java")882            .all(),883            [c1],884        )885    def test_join_to_subclass_ten(self):886        sess = create_session()887        eq_(888            sess.query(Company)889            .join("employees")890            .filter(Engineer.primary_language == "java")891            .all(),892            [c1],893        )894    def test_join_to_subclass_eleven(self):895        sess = create_session()896        eq_(897            sess.query(Company)898            .select_from(companies.join(people).join(engineers))899            .filter(Engineer.primary_language == "java")900            .all(),901            [c1],902        )903    def test_join_to_subclass_twelve(self):904        sess = create_session()905        eq_(sess.query(Person).join(Engineer.machines).all(), [e1, e2, e3])906    def test_join_to_subclass_thirteen(self):907        sess = create_session()908        eq_(909            sess.query(Person)910            .join(Engineer.machines)911            .filter(Machine.name.ilike("%ibm%"))912            .all(),913            [e1, e3],914        )915    def test_join_to_subclass_fourteen(self):916        sess = create_session()917        eq_(918            sess.query(Company).join("employees", Engineer.machines).all(),919            [c1, c2],920        )921    def test_join_to_subclass_fifteen(self):922        sess = create_session()923        eq_(924            sess.query(Company)925            .join("employees", Engineer.machines)926            .filter(Machine.name.ilike("%thinkpad%"))927            .all(),928            [c1],929        )930    def test_join_to_subclass_sixteen(self):931        sess = create_session()932        # non-polymorphic933        eq_(sess.query(Engineer).join(Engineer.machines).all(), [e1, e2, e3])934    def test_join_to_subclass_seventeen(self):935        sess = create_session()936        eq_(937            sess.query(Engineer)938            .join(Engineer.machines)939            .filter(Machine.name.ilike("%ibm%"))940            .all(),941            [e1, e3],942        )943    def test_join_through_polymorphic_nonaliased_one(self):944        sess = create_session()945        eq_(946            sess.query(Company)947            .join("employees", "paperwork", aliased=False)948            .filter(Paperwork.description.like("%#2%"))949            .all(),950            [c1],951        )952    def test_join_through_polymorphic_nonaliased_two(self):953        sess = create_session()954        eq_(955            sess.query(Company)956            .join("employees", "paperwork", aliased=False)957            .filter(Paperwork.description.like("%#%"))958            .all(),959            [c1, c2],960        )961    def test_join_through_polymorphic_nonaliased_three(self):962        sess = create_session()963        eq_(964            sess.query(Company)965            .join("employees", "paperwork", aliased=False)966            .filter(Person.name.in_(["dilbert", "vlad"]))967            .filter(Paperwork.description.like("%#2%"))968            .all(),969            [c1],970        )971    def test_join_through_polymorphic_nonaliased_four(self):972        sess = create_session()973        eq_(974            sess.query(Company)975            .join("employees", "paperwork", aliased=False)976            .filter(Person.name.in_(["dilbert", "vlad"]))977            .filter(Paperwork.description.like("%#%"))978            .all(),979            [c1, c2],980        )981    def test_join_through_polymorphic_nonaliased_five(self):982        sess = create_session()983        eq_(984            sess.query(Company)985            .join("employees", aliased=aliased)986            .filter(Person.name.in_(["dilbert", "vlad"]))987            .join("paperwork", from_joinpoint=True, aliased=False)988            .filter(Paperwork.description.like("%#2%"))989            .all(),990            [c1],991        )992    def test_join_through_polymorphic_nonaliased_six(self):993        sess = create_session()994        eq_(995            sess.query(Company)996            .join("employees", aliased=aliased)997            .filter(Person.name.in_(["dilbert", "vlad"]))998            .join("paperwork", from_joinpoint=True, aliased=False)999            .filter(Paperwork.description.like("%#%"))1000            .all(),1001            [c1, c2],1002        )1003    def test_join_through_polymorphic_aliased_one(self):1004        sess = create_session()1005        eq_(1006            sess.query(Company)1007            .join("employees", "paperwork", aliased=True)1008            .filter(Paperwork.description.like("%#2%"))1009            .all(),1010            [c1],1011        )1012    def test_join_through_polymorphic_aliased_two(self):1013        sess = create_session()1014        eq_(1015            sess.query(Company)1016            .join("employees", "paperwork", aliased=True)1017            .filter(Paperwork.description.like("%#%"))1018            .all(),1019            [c1, c2],1020        )1021    def test_join_through_polymorphic_aliased_three(self):1022        sess = create_session()1023        eq_(1024            sess.query(Company)1025            .join("employees", "paperwork", aliased=True)1026            .filter(Person.name.in_(["dilbert", "vlad"]))1027            .filter(Paperwork.description.like("%#2%"))1028            .all(),1029            [c1],1030        )1031    def test_join_through_polymorphic_aliased_four(self):1032        sess = create_session()1033        eq_(1034            sess.query(Company)1035            .join("employees", "paperwork", aliased=True)1036            .filter(Person.name.in_(["dilbert", "vlad"]))1037            .filter(Paperwork.description.like("%#%"))1038            .all(),1039            [c1, c2],1040        )1041    def test_join_through_polymorphic_aliased_five(self):1042        sess = create_session()1043        eq_(1044            sess.query(Company)1045            .join("employees", aliased=aliased)1046            .filter(Person.name.in_(["dilbert", "vlad"]))1047            .join("paperwork", from_joinpoint=True, aliased=True)1048            .filter(Paperwork.description.like("%#2%"))1049            .all(),1050            [c1],1051        )1052    def test_join_through_polymorphic_aliased_six(self):1053        sess = create_session()1054        eq_(1055            sess.query(Company)1056            .join("employees", aliased=aliased)1057            .filter(Person.name.in_(["dilbert", "vlad"]))1058            .join("paperwork", from_joinpoint=True, aliased=True)1059            .filter(Paperwork.description.like("%#%"))1060            .all(),1061            [c1, c2],1062        )1063    def test_explicit_polymorphic_join_one(self):1064        sess = create_session()1065        # join from Company to Engineer; join condition formulated by1066        # ORMJoin using regular table foreign key connections.  Engineer1067        # is expressed as "(select * people join engineers) as anon_1"1068        # so the join is contained.1069        eq_(1070            sess.query(Company)1071            .join(Engineer)1072            .filter(Engineer.engineer_name == "vlad")1073            .one(),1074            c2,1075        )1076    def test_explicit_polymorphic_join_two(self):1077        sess = create_session()1078        # same, using explicit join condition.  Query.join() must1079        # adapt the on clause here to match the subquery wrapped around1080        # "people join engineers".1081        eq_(1082            sess.query(Company)1083            .join(Engineer, Company.company_id == Engineer.company_id)1084            .filter(Engineer.engineer_name == "vlad")1085            .one(),1086            c2,1087        )1088    def test_filter_on_baseclass(self):1089        sess = create_session()1090        eq_(sess.query(Person).order_by(Person.person_id).all(), all_employees)1091        eq_(1092            sess.query(Person).order_by(Person.person_id).first(),1093            all_employees[0],1094        )1095        eq_(1096            sess.query(Person)1097            .order_by(Person.person_id)1098            .filter(Person.person_id == e2.person_id)1099            .one(),1100            e2,1101        )1102    def test_from_alias(self):1103        sess = create_session()1104        palias = aliased(Person)1105        eq_(1106            sess.query(palias)1107            .order_by(palias.person_id)1108            .filter(palias.name.in_(["dilbert", "wally"]))1109            .all(),1110            [e1, e2],1111        )1112    def test_self_referential_one(self):1113        sess = create_session()1114        palias = aliased(Person)1115        expected = [(m1, e1), (m1, e2), (m1, b1)]1116        eq_(1117            sess.query(Person, palias)1118            .filter(Person.company_id == palias.company_id)1119            .filter(Person.name == "dogbert")1120            .filter(Person.person_id > palias.person_id)1121            .order_by(Person.person_id, palias.person_id)1122            .all(),1123            expected,1124        )1125    def test_self_referential_two(self):1126        sess = create_session()1127        palias = aliased(Person)1128        expected = [(m1, e1), (m1, e2), (m1, b1)]1129        eq_(1130            sess.query(Person, palias)1131            .filter(Person.company_id == palias.company_id)1132            .filter(Person.name == "dogbert")1133            .filter(Person.person_id > palias.person_id)1134            .from_self()1135            .order_by(Person.person_id, palias.person_id)1136            .all(),1137            expected,1138        )1139    def test_nesting_queries(self):1140        # query.statement places a flag "no_adapt" on the returned1141        # statement.  This prevents the polymorphic adaptation in the1142        # second "filter" from hitting it, which would pollute the1143        # subquery and usually results in recursion overflow errors1144        # within the adaption.1145        sess = create_session()1146        subq = (1147            sess.query(engineers.c.person_id)1148            .filter(Engineer.primary_language == "java")1149            .statement.as_scalar()1150        )1151        eq_(sess.query(Person).filter(Person.person_id.in_(subq)).one(), e1)1152    def test_mixed_entities_one(self):1153        sess = create_session()1154        expected = [1155            (1156                Engineer(1157                    status="regular engineer",1158                    engineer_name="dilbert",1159                    name="dilbert",1160                    company_id=1,1161                    primary_language="java",1162                    person_id=1,1163                    type="engineer",1164                ),1165                "MegaCorp, Inc.",1166            ),1167            (1168                Engineer(1169                    status="regular engineer",1170                    engineer_name="wally",1171                    name="wally",1172                    company_id=1,1173                    primary_language="c++",1174                    person_id=2,1175                    type="engineer",1176                ),1177                "MegaCorp, Inc.",1178            ),1179            (1180                Engineer(1181                    status="elbonian engineer",1182                    engineer_name="vlad",1183                    name="vlad",1184                    company_id=2,1185                    primary_language="cobol",1186                    person_id=5,1187                    type="engineer",1188                ),1189                "Elbonia, Inc.",1190            ),1191        ]1192        eq_(1193            sess.query(Engineer, Company.name)1194            .join(Company.employees)1195            .order_by(Person.person_id)1196            .filter(Person.type == "engineer")1197            .all(),1198            expected,1199        )1200    def _join_to_poly_wp_one(self, sess):1201        wp = with_polymorphic(self.classes.Person, "*")1202        return (1203            sess.query(wp.name, self.classes.Company.name)1204            .join(self.classes.Company.employees.of_type(wp))1205            .order_by(wp.person_id)1206        )1207    def _join_to_poly_wp_two(self, sess):1208        wp = with_polymorphic(self.classes.Person, "*", aliased=True)1209        return (1210            sess.query(wp.name, self.classes.Company.name)1211            .join(self.classes.Company.employees.of_type(wp))1212            .order_by(wp.person_id)1213        )1214    def _join_to_poly_wp_three(self, sess):1215        wp = with_polymorphic(1216            self.classes.Person, "*", aliased=True, flat=True1217        )1218        return (1219            sess.query(wp.name, self.classes.Company.name)1220            .join(self.classes.Company.employees.of_type(wp))1221            .order_by(wp.person_id)1222        )1223    @testing.combinations(1224        lambda self, sess: (1225            sess.query(self.classes.Person.name, self.classes.Company.name)1226            .join(self.classes.Company.employees)1227            .order_by(self.classes.Person.person_id)1228        ),1229        _join_to_poly_wp_one,1230        _join_to_poly_wp_two,1231        _join_to_poly_wp_three,1232    )1233    def test_mixed_entities_join_to_poly(self, q):1234        sess = create_session()1235        expected = [1236            ("dilbert", "MegaCorp, Inc."),1237            ("wally", "MegaCorp, Inc."),1238            ("pointy haired boss", "MegaCorp, Inc."),1239            ("dogbert", "MegaCorp, Inc."),1240            ("vlad", "Elbonia, Inc."),1241        ]1242        eq_(1243            q(self, sess).all(), expected,1244        )1245    def test_mixed_entities_two(self):1246        sess = create_session()1247        expected = [1248            ("java", "MegaCorp, Inc."),1249            ("cobol", "Elbonia, Inc."),1250            ("c++", "MegaCorp, Inc."),1251        ]1252        eq_(1253            sess.query(Engineer.primary_language, Company.name)1254            .join(Company.employees)1255            .filter(Person.type == "engineer")1256            .order_by(desc(Engineer.primary_language))1257            .all(),1258            expected,1259        )1260    def test_mixed_entities_three(self):1261        sess = create_session()1262        palias = aliased(Person)1263        expected = [1264            (1265                Engineer(1266                    status="elbonian engineer",1267                    engineer_name="vlad",1268                    name="vlad",1269                    primary_language="cobol",1270                ),1271                "Elbonia, Inc.",1272                Engineer(1273                    status="regular engineer",1274                    engineer_name="dilbert",1275                    name="dilbert",1276                    company_id=1,1277                    primary_language="java",1278                    person_id=1,1279                    type="engineer",1280                ),1281            )1282        ]1283        eq_(1284            sess.query(Person, Company.name, palias)1285            .join(Company.employees)1286            .filter(Company.name == "Elbonia, Inc.")1287            .filter(palias.name == "dilbert")1288            .all(),1289            expected,1290        )1291    def test_mixed_entities_four(self):1292        sess = create_session()1293        palias = aliased(Person)1294        expected = [1295            (1296                Engineer(1297                    status="regular engineer",1298                    engineer_name="dilbert",1299                    name="dilbert",1300                    company_id=1,1301                    primary_language="java",1302                    person_id=1,1303                    type="engineer",1304                ),1305                "Elbonia, Inc.",1306                Engineer(1307                    status="elbonian engineer",1308                    engineer_name="vlad",1309                    name="vlad",1310                    primary_language="cobol",1311                ),1312            )1313        ]1314        eq_(1315            sess.query(palias, Company.name, Person)1316            .join(Company.employees)1317            .filter(Company.name == "Elbonia, Inc.")1318            .filter(palias.name == "dilbert")1319            .all(),1320            expected,1321        )1322    def test_mixed_entities_five(self):1323        sess = create_session()1324        palias = aliased(Person)1325        expected = [("vlad", "Elbonia, Inc.", "dilbert")]1326        eq_(1327            sess.query(Person.name, Company.name, palias.name)1328            .join(Company.employees)1329            .filter(Company.name == "Elbonia, Inc.")1330            .filter(palias.name == "dilbert")1331            .all(),1332            expected,1333        )1334    def test_mixed_entities_six(self):1335        sess = create_session()1336        palias = aliased(Person)1337        expected = [1338            ("manager", "dogbert", "engineer", "dilbert"),1339            ("manager", "dogbert", "engineer", "wally"),1340            ("manager", "dogbert", "boss", "pointy haired boss"),1341        ]1342        eq_(1343            sess.query(Person.type, Person.name, palias.type, palias.name)1344            .filter(Person.company_id == palias.company_id)1345            .filter(Person.name == "dogbert")1346            .filter(Person.person_id > palias.person_id)1347            .order_by(Person.person_id, palias.person_id)1348            .all(),1349            expected,1350        )1351    def test_mixed_entities_seven(self):1352        sess = create_session()1353        expected = [1354            ("dilbert", "tps report #1"),1355            ("dilbert", "tps report #2"),1356            ("dogbert", "review #2"),1357            ("dogbert", "review #3"),1358            ("pointy haired boss", "review #1"),1359            ("vlad", "elbonian missive #3"),1360            ("wally", "tps report #3"),1361            ("wally", "tps report #4"),1362        ]1363        eq_(1364            sess.query(Person.name, Paperwork.description)1365            .filter(Person.person_id == Paperwork.person_id)1366            .order_by(Person.name, Paperwork.description)1367            .all(),1368            expected,1369        )1370    def test_mixed_entities_eight(self):1371        sess = create_session()1372        eq_(1373            sess.query(func.count(Person.person_id))1374            .filter(Engineer.primary_language == "java")1375            .all(),1376            [(1,)],1377        )1378    def test_mixed_entities_nine(self):1379        sess = create_session()1380        expected = [("Elbonia, Inc.", 1), ("MegaCorp, Inc.", 4)]1381        eq_(1382            sess.query(Company.name, func.count(Person.person_id))1383            .filter(Company.company_id == Person.company_id)1384            .group_by(Company.name)1385            .order_by(Company.name)1386            .all(),1387            expected,1388        )1389    def test_mixed_entities_ten(self):1390        sess = create_session()1391        expected = [("Elbonia, Inc.", 1), ("MegaCorp, Inc.", 4)]1392        eq_(1393            sess.query(Company.name, func.count(Person.person_id))1394            .join(Company.employees)1395            .group_by(Company.name)1396            .order_by(Company.name)1397            .all(),1398            expected,1399        )1400    # def test_mixed_entities(self):1401    #    sess = create_session()1402    # TODO: I think raise error on these for now.  different1403    # inheritance/loading schemes have different results here,1404    # all incorrect1405    #1406    # eq_(1407    #    sess.query(Person.name, Engineer.primary_language).all(),1408    #    [])1409    # def test_mixed_entities(self):1410    #    sess = create_session()1411    # eq_(sess.query(1412    #             Person.name,1413    #             Engineer.primary_language,1414    #             Manager.manager_name)1415    #          .all(),1416    #     [])1417    def test_mixed_entities_eleven(self):1418        sess = create_session()1419        expected = [("java",), ("c++",), ("cobol",)]1420        eq_(1421            sess.query(Engineer.primary_language)1422            .filter(Person.type == "engineer")1423            .all(),1424            expected,1425        )1426    def test_mixed_entities_twelve(self):1427        sess = create_session()1428        expected = [("vlad", "Elbonia, Inc.")]1429        eq_(1430            sess.query(Person.name, Company.name)1431            .join(Company.employees)1432            .filter(Company.name == "Elbonia, Inc.")1433            .all(),1434            expected,1435        )1436    def test_mixed_entities_thirteen(self):1437        sess = create_session()1438        expected = [("pointy haired boss", "fore")]1439        eq_(sess.query(Boss.name, Boss.golf_swing).all(), expected)1440    def test_mixed_entities_fourteen(self):1441        sess = create_session()1442        expected = [("dilbert", "java"), ("wally", "c++"), ("vlad", "cobol")]1443        eq_(1444            sess.query(Engineer.name, Engineer.primary_language).all(),1445            expected,1446        )1447    def test_mixed_entities_fifteen(self):1448        sess = create_session()1449        expected = [1450            (1451                "Elbonia, Inc.",1452                Engineer(1453                    status="elbonian engineer",1454                    engineer_name="vlad",1455                    name="vlad",1456                    primary_language="cobol",1457                ),1458            )1459        ]1460        eq_(1461            sess.query(Company.name, Person)1462            .join(Company.employees)1463            .filter(Company.name == "Elbonia, Inc.")1464            .all(),1465            expected,1466        )1467    def test_mixed_entities_sixteen(self):1468        sess = create_session()1469        expected = [1470            (1471                Engineer(1472                    status="elbonian engineer",1473                    engineer_name="vlad",1474                    name="vlad",1475                    primary_language="cobol",1476                ),1477                "Elbonia, Inc.",1478            )1479        ]1480        eq_(1481            sess.query(Person, Company.name)1482            .join(Company.employees)1483            .filter(Company.name == "Elbonia, Inc.")1484            .all(),1485            expected,1486        )1487    def test_mixed_entities_seventeen(self):1488        sess = create_session()1489        expected = [("pointy haired boss",), ("dogbert",)]1490        eq_(sess.query(Manager.name).all(), expected)1491    def test_mixed_entities_eighteen(self):1492        sess = create_session()1493        expected = [("pointy haired boss foo",), ("dogbert foo",)]1494        eq_(sess.query(Manager.name + " foo").all(), expected)1495    def test_mixed_entities_nineteen(self):1496        sess = create_session()1497        row = (1498            sess.query(Engineer.name, Engineer.primary_language)1499            .filter(Engineer.name == "dilbert")1500            .first()1501        )1502        assert row.name == "dilbert"1503        assert row.primary_language == "java"1504    def test_correlation_one(self):1505        sess = create_session()1506        # this for a long time did not work with PolymorphicAliased and1507        # PolymorphicUnions, which was due to the no_replacement_traverse1508        # annotation added to query.statement which then went into as_scalar().1509        # this is removed as of :ticket:`4304` so now works.1510        eq_(1511            sess.query(Person.name)1512            .filter(1513                sess.query(Company.name)1514                .filter(Company.company_id == Person.company_id)1515                .correlate(Person)1516                .as_scalar()1517                == "Elbonia, Inc."1518            )1519            .all(),1520            [(e3.name,)],1521        )1522    def test_correlation_two(self):1523        sess = create_session()1524        paliased = aliased(Person)1525        eq_(1526            sess.query(paliased.name)1527            .filter(1528                sess.query(Company.name)1529                .filter(Company.company_id == paliased.company_id)1530                .correlate(paliased)1531                .as_scalar()1532                == "Elbonia, Inc."1533            )1534            .all(),1535            [(e3.name,)],1536        )1537    def test_correlation_three(self):1538        sess = create_session()1539        paliased = aliased(Person, flat=True)1540        eq_(1541            sess.query(paliased.name)1542            .filter(1543                sess.query(Company.name)1544                .filter(Company.company_id == paliased.company_id)1545                .correlate(paliased)1546                .as_scalar()1547                == "Elbonia, Inc."1548            )1549            .all(),1550            [(e3.name,)],1551        )1552class PolymorphicTest(_PolymorphicTestBase, _Polymorphic):1553    def test_join_to_subclass_four(self):1554        sess = create_session()1555        eq_(1556            sess.query(Person)1557            .select_from(people.join(engineers))1558            .join(Engineer.machines)1559            .all(),1560            [e1, e2, e3],1561        )1562    def test_join_to_subclass_five(self):1563        sess = create_session()1564        eq_(1565            sess.query(Person)1566            .select_from(people.join(engineers))1567            .join(Engineer.machines)1568            .filter(Machine.name.ilike("%ibm%"))1569            .all(),1570            [e1, e3],1571        )1572    def test_correlation_w_polymorphic(self):1573        sess = create_session()1574        p_poly = with_polymorphic(Person, "*")1575        eq_(1576            sess.query(p_poly.name)1577            .filter(1578                sess.query(Company.name)1579                .filter(Company.company_id == p_poly.company_id)1580                .correlate(p_poly)1581                .as_scalar()1582                == "Elbonia, Inc."1583            )1584            .all(),1585            [(e3.name,)],1586        )1587    def test_correlation_w_polymorphic_flat(self):1588        sess = create_session()1589        p_poly = with_polymorphic(Person, "*", flat=True)1590        eq_(1591            sess.query(p_poly.name)1592            .filter(1593                sess.query(Company.name)1594                .filter(Company.company_id == p_poly.company_id)1595                .correlate(p_poly)1596                .as_scalar()1597                == "Elbonia, Inc."1598            )1599            .all(),1600            [(e3.name,)],1601        )1602    def test_join_to_subclass_ten(self):1603        pass1604    def test_mixed_entities_one(self):1605        pass1606    def test_mixed_entities_two(self):1607        pass1608    def test_mixed_entities_eight(self):1609        pass1610    def test_polymorphic_any_eight(self):1611        pass1612class PolymorphicPolymorphicTest(1613    _PolymorphicTestBase, _PolymorphicPolymorphic1614):1615    __dialect__ = "default"1616    def test_aliased_not_polluted_by_join(self):1617        # aliased(polymorphic) will normally do the old-school1618        # "(SELECT * FROM a JOIN b ...) AS anon_1" thing.1619        # this is the safest1620        sess = create_session()1621        palias = aliased(Person)1622        self.assert_compile(1623            sess.query(palias, Company.name)1624            .order_by(palias.person_id)1625            .join(Person, Company.employees)1626            .filter(palias.name == "dilbert"),1627            "SELECT anon_1.people_person_id AS anon_1_people_person_id, "1628            "anon_1.people_company_id AS anon_1_people_company_id, "1629            "anon_1.people_name AS anon_1_people_name, "1630            "anon_1.people_type AS anon_1_people_type, "1631            "anon_1.engineers_person_id AS anon_1_engineers_person_id, "1632            "anon_1.engineers_status AS anon_1_engineers_status, "1633            "anon_1.engineers_engineer_name AS anon_1_engineers_engineer_name, "  # noqa1634            "anon_1.engineers_primary_language AS "1635            "anon_1_engineers_primary_language, "1636            "anon_1.managers_person_id AS anon_1_managers_person_id, "1637            "anon_1.managers_status AS anon_1_managers_status, "1638            "anon_1.managers_manager_name AS anon_1_managers_manager_name, "1639            "anon_1.boss_boss_id AS anon_1_boss_boss_id, "1640            "anon_1.boss_golf_swing AS anon_1_boss_golf_swing, "1641            "companies.name AS companies_name "1642            "FROM (SELECT people.person_id AS people_person_id, "1643            "people.company_id AS people_company_id, "1644            "people.name AS people_name, people.type AS people_type, "1645            "engineers.person_id AS engineers_person_id, "1646            "engineers.status AS engineers_status, "1647            "engineers.engineer_name AS engineers_engineer_name, "1648            "engineers.primary_language AS engineers_primary_language, "1649            "managers.person_id AS managers_person_id, "1650            "managers.status AS managers_status, "1651            "managers.manager_name AS managers_manager_name, "1652            "boss.boss_id AS boss_boss_id, "1653            "boss.golf_swing AS boss_golf_swing "1654            "FROM people LEFT OUTER JOIN engineers "1655            "ON people.person_id = engineers.person_id "1656            "LEFT OUTER JOIN managers "1657            "ON people.person_id = managers.person_id LEFT OUTER JOIN boss "1658            "ON managers.person_id = boss.boss_id) AS anon_1, "1659            "companies JOIN "1660            "(people LEFT OUTER JOIN engineers "1661            "ON people.person_id = engineers.person_id "1662            "LEFT OUTER JOIN managers "1663            "ON people.person_id = managers.person_id "1664            "LEFT OUTER JOIN boss ON managers.person_id = boss.boss_id) "1665            "ON companies.company_id = people.company_id "1666            "WHERE anon_1.people_name = :people_name_1 "1667            "ORDER BY anon_1.people_person_id",1668        )1669    def test_flat_aliased_w_select_from(self):1670        sess = create_session()1671        palias = aliased(Person, flat=True)1672        self.assert_compile(1673            sess.query(palias, Company.name)1674            .select_from(palias)1675            .order_by(palias.person_id)1676            .join(Person, Company.employees)1677            .filter(palias.name == "dilbert"),1678            "SELECT people_1.person_id AS people_1_person_id, "1679            "people_1.company_id AS people_1_company_id, "1680            "people_1.name AS people_1_name, people_1.type AS people_1_type, "1681            "engineers_1.person_id AS engineers_1_person_id, "1682            "engineers_1.status AS engineers_1_status, "1683            "engineers_1.engineer_name AS engineers_1_engineer_name, "1684            "engineers_1.primary_language AS engineers_1_primary_language, "1685            "managers_1.person_id AS managers_1_person_id, "1686            "managers_1.status AS managers_1_status, "1687            "managers_1.manager_name AS managers_1_manager_name, "1688            "boss_1.boss_id AS boss_1_boss_id, "1689            "boss_1.golf_swing AS boss_1_golf_swing, "1690            "companies.name AS companies_name "1691            "FROM people AS people_1 "1692            "LEFT OUTER JOIN engineers AS engineers_1 "1693            "ON people_1.person_id = engineers_1.person_id "1694            "LEFT OUTER JOIN managers AS managers_1 "1695            "ON people_1.person_id = managers_1.person_id "1696            "LEFT OUTER JOIN boss AS boss_1 "1697            "ON managers_1.person_id = boss_1.boss_id, "1698            "companies JOIN (people LEFT OUTER JOIN engineers "1699            "ON people.person_id = engineers.person_id "1700            "LEFT OUTER JOIN managers "1701            "ON people.person_id = managers.person_id "1702            "LEFT OUTER JOIN boss ON managers.person_id = boss.boss_id) "1703            "ON companies.company_id = people.company_id "1704            "WHERE people_1.name = :name_1 ORDER BY people_1.person_id",1705        )1706class PolymorphicUnionsTest(_PolymorphicTestBase, _PolymorphicUnions):1707    def test_subqueryload_on_subclass_uses_path_correctly(self):1708        sess = create_session()1709        expected = [1710            Engineer(1711                name="dilbert",1712                engineer_name="dilbert",1713                primary_language="java",1714                status="regular engineer",1715                machines=[1716                    Machine(name="IBM ThinkPad"),1717                    Machine(name="IPhone"),1718                ],1719            )1720        ]1721        with self.sql_execution_asserter(testing.db) as asserter:1722            wp = with_polymorphic(Person, "*")...

st_create_session.py

Source:st_create_session.py

...5import random6import nfs4lib7import threading8from rpc import RPCAcceptError, GARBAGE_ARGS9def create_session(c, clientid, sequenceid, cred=None, flags=0):10    """Send a simple CREATE_SESSION"""11    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[])12    res = c.compound([op.create_session(clientid, sequenceid, flags,13                                        chan_attrs, chan_attrs,14                                        123, [])], cred)15    return res16###############################################17def testSupported1(t, env):18    """Do a simple CREATE_SESSION19    FLAGS: create_session all20    CODE: CSESS121    """22    c = env.c1.new_client(env.testname(t))23    sess = c.create_session()24def testSupported2(t, env):25    """Do a CREATE_SESSION after a SEQUENCE (for same client)26    FLAGS: create_session all27    CODE: CSESS228    """29    c1 = env.c1.new_client(env.testname(t))30    sess1 = c1.create_session()31    # Create second session32    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[])33    sec = [callback_sec_parms4(0)]34    cs_op = op.create_session(c1.clientid, c1.seqid, 0,35                              chan_attrs, chan_attrs, c1.c.prog, sec)36    res = sess1.compound([cs_op])37    check(res)38    sess2 = c1._add_session(res.resarray[-1])39    # Now make sure sess2 works40    res = sess2.compound([op.putrootfh()])41    check(res)42    43def testSupported2b(t, env):44    """Do a CREATE_SESSION after a SEQUENCE (for different client)45    FLAGS: create_session all46    CODE: CSESS2b47    """48    c1 = env.c1.new_client("%s_1" % env.testname(t))49    c2 = env.c1.new_client("%s_2" % env.testname(t))50    sess1 = c1.create_session()51    # Create second session52    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[])53    sec = [callback_sec_parms4(0)]54    cs_op = op.create_session(c2.clientid, c2.seqid, 0,55                              chan_attrs, chan_attrs, c2.c.prog, sec)56    res = sess1.compound([cs_op])57    check(res)58    sess2 = c2._add_session(res.resarray[-1])59    # Now make sure sess2 works60    res = sess2.compound([op.putrootfh()])61    check(res)62    63def testNoExchange(t, env):64    """Send CREATE_SESSION when server has no record of clientid65    FLAGS: create_session all66    CODE: CSESS367    """68    c = env.c169    # NOTE no real way to guarantee this will not collide with previously70    # seen clientid, but odds are pretty low.71    cid = random.randint(0, 2**64 - 1)72    res = create_session(c, cid, 256)73    check(res, NFS4ERR_STALE_CLIENTID)74def testContrivedReplay(t, env):75    """Server is supposed to create a contrived replay result76    FLAGS: create_session all77    CODE: CSESS478    """79    c = env.c1.new_client(env.testname(t))80    # CREATE_SESSION81    res = create_session(c.c, c.clientid,82                         nfs4lib.dec_u32(c.seqid))83    check(res, NFS4ERR_SEQ_MISORDERED)84def testReplay1(t, env):85    """Replay a successful CREATE_SESSION86    87    FLAGS: create_session all88    CODE: CSESS589    """90    c = env.c1.new_client(env.testname(t))91    # CREATE_SESSION92    res1 = create_session(c.c, c.clientid, c.seqid)93    check(res1)94    # REPLAY95    res2 = create_session(c.c, c.clientid, c.seqid)96    check(res2)97    # Test results are equal (ignoring tags)98    res1.tag = res2.tag = ""99    if not nfs4lib.test_equal(res1, res2):100        fail("Replay results not equal")101def testReplay1a(t, env):102    """Replay a successful CREATE_SESSION with a SEQUENCE from a different session103    FLAGS: create_session all104    CODE: CSESS5a105    """106    c = env.c1.new_client(env.testname(t))107    # CREATE_SESSION108    sess1 = c.create_session()109    # another CREATE_SESSION110    c.seqid = 2111    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[])112    sec = [callback_sec_parms4(0)]113    res1 = create_session(c.c, c.clientid, c.seqid)114    check(res1)115    # REPLAY first CREATE_SESSION with SEQUENCE from 2nd session116    cs_op = op.create_session(c.clientid, c.seqid, 0,117                              chan_attrs, chan_attrs, c.c.prog, sec)118    res2 = sess1.compound([cs_op])119    check(res2)120    # Test results are equal (ignoring tags)121    res1.tag = res2.tag = ""122    if not nfs4lib.test_equal(res1, res2):123        fail("Replay results not equal")124def testReplay1b(t, env):125    """Replay a successful SEQUENCE:CREATE_SESSION without a preceeding SEQUENCE126    FLAGS: create_session all127    CODE: CSESS5b128    """129    c = env.c1.new_client(env.testname(t))130    # CREATE_SESSION131    sess1 = c.create_session()132    # another CREATE_SESSION with SEQUENCE from first session133    c.seqid = 2134    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[])135    sec = [callback_sec_parms4(0)]136    cs_op = op.create_session(c.clientid, c.seqid, 0,137                              chan_attrs, chan_attrs, c.c.prog, sec)138    res1 = sess1.compound([cs_op])139    check(res1)140    # REPLAY second CREATE_SESSION without SEQUENCE141    res2 = create_session(c.c, c.clientid, c.seqid)142    check(res2)143    # Test results are equal (ignoring tags)144    res1.tag = res2.tag = ""145    if not nfs4lib.test_equal(res1, res2):146        fail("Replay results not equal")147def testReplay2(t, env):148    """Replay a unsuccessful CREATE_SESSION149    150    FLAGS: create_session all151    DEPEND: CSESS9152    CODE: CSESS6153    """154    c = env.c1.new_client(env.testname(t), cred=env.cred1)155    res1 = create_session(c.c, c.clientid, c.seqid, cred=env.cred2)156    check(res1, NFS4ERR_CLID_INUSE)157    # REPLAY158    res2 = create_session(c.c, c.clientid, c.seqid, cred=env.cred2)159    check(res2, NFS4ERR_CLID_INUSE)160    # Test results are equal (ignoring tags)161    res1.tag = res2.tag = ""162    if not nfs4lib.test_equal(res1, res2):163        fail("Replay results not equal")164def testBadSeqnum1(t, env):165    """Send too large seqnum166    167    FLAGS: create_session all168    CODE: CSESS7169    """170    c = env.c1.new_client(env.testname(t))171    res1 = create_session(c.c, c.clientid, c.seqid)172    check(res1)173    # REPLAY174    badseqid = nfs4lib.inc_u32(nfs4lib.inc_u32(c.seqid))175    res2 = create_session(c.c, c.clientid, badseqid)176    check(res2, NFS4ERR_SEQ_MISORDERED)177def testBadSeqnum2(t, env):178    """Send too small seqnum179    180    FLAGS: create_session all181    CODE: CSESS8182    """183    c = env.c1.new_client(env.testname(t))184    res1 = create_session(c.c, c.clientid, c.seqid)185    check(res1)186    # REPLAY187    badseqid = nfs4lib.dec_u32(c.seqid)188    res2 = create_session(c.c, c.clientid, badseqid)189    check(res2, NFS4ERR_SEQ_MISORDERED)190def testPrincipalCollision1(t, env):191    """Change of principal before confirmation is bad192    193    FLAGS: create_session all194    CODE: CSESS9195    """196    c = env.c1.new_client(env.testname(t), cred=env.cred1)197    res = create_session(c.c, c.clientid, c.seqid, cred=env.cred2)198    check(res, NFS4ERR_CLID_INUSE)199def testPrincipalCollision2(t, env):200    """Change of principal after confirmation is good201    202    FLAGS: create_session all203    CODE: CSESS10204    """205    c = env.c1.new_client(env.testname(t), cred=env.cred1)206    res1 = create_session(c.c, c.clientid, c.seqid, cred=env.cred1)207    check(res1)208    csr = res1.resarray[0]209    res2 = create_session(c.c, c.clientid, csr.csr_sequence,210                          cred=env.cred2)211    check(res2)212def testBadFlag(t, env):213    """Use invalid flag bits214    FLAGS: create_session all215    CODE: CSESS15216    """217    c = env.c1.new_client(env.testname(t))218    res = create_session(c.c, c.clientid, c.seqid, flags = 0xf)219    # XXX Where is this required?  What about ignoring bits?220    check(res, NFS4ERR_INVAL)221def testCbSecParms(t, env):222    """Send each type of security parameter223    NOTE this is a bit strange, no one would really send mixture224    FLAGS: create_session all225    CODE: CSESS16226    """227    sec = [callback_sec_parms4(AUTH_NONE),228           callback_sec_parms4(AUTH_SYS, cbsp_sys_cred=authsys_parms(5, "Random machine name", 7, 11, [13, 17, 19, 23, 29])),229           callback_sec_parms4(RPCSEC_GSS, cbsp_gss_handles=gss_cb_handles4(RPC_GSS_SVC_PRIVACY, "Handle from server", "Client handle")),230           ]231                               232    c1 = env.c1.new_client(env.testname(t))233    sess1 = c1.create_session(sec=sec)234def testCbSecParmsDec(t, env):235    """A decode problem was found at NFS server236       (wrong index used in inner loop).237       http://marc.info/?l=linux-kernel&m=129961996327640&w=2238    FLAGS: create_session all239    CODE: CSESS16a240    """241    sec = [callback_sec_parms4(AUTH_NONE),242           callback_sec_parms4(RPCSEC_GSS, cbsp_gss_handles=gss_cb_handles4(RPC_GSS_SVC_PRIVACY, "Handle from server", "Client handle")),243           callback_sec_parms4(AUTH_SYS, cbsp_sys_cred=authsys_parms(5, "Random machine name", 7, 11, [])),244           ]245    c1 = env.c1.new_client(env.testname(t))246    sess1 = c1.create_session(sec=sec)247def testRdmaArray0(t, env):248    """Test 0 length rdma arrays249    FLAGS: create_session all250    CODE: CSESS17251    """252    c1 = env.c1.new_client(env.testname(t))253    chan_attrs = channel_attrs4(0, 8192,8192,8192,128,8,[])254    sess1 = c1.create_session(fore_attrs=chan_attrs,255                              back_attrs=chan_attrs)256def testRdmaArray1(t, env):257    """Test length 1 rdma arrays258    FLAGS: create_session all259    CODE: CSESS18260    """261    c1 = env.c1.new_client(env.testname(t))262    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[57])263    sess1 = c1.create_session(fore_attrs=chan_attrs,264                              back_attrs=chan_attrs)265def testRdmaArray2(t, env):266    """Test length 2 rdma arrays267    FLAGS: create_session all268    CODE: CSESS19269    """270    c = env.c1.new_client(env.testname(t))271    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[13, 57])272    ops = [op.create_session(c.clientid, c.seqid, 0,273                             chan_attrs, chan_attrs,274                             c.c.prog, [])]275    c = c.c276    xid = c.compound_async(ops, checks=False)277    try:278        res = c.listen(xid)279        print res280    except RPCAcceptError, e:281        if e.stat == GARBAGE_ARGS:282            # Legitimate return283            return284        else:285            raise286    check(res, NFS4ERR_BADXDR)287def testManyClients(t, env):288    """Create and confirm many clients289    FLAGS: create_session all290    CODE: CSESS200291    """292    N = 10 # number of clients to create293    for i in range(N):294        c = env.c1.new_client("%s_Client_%i" % (env.testname(t), i))295        sess = c.create_session()296        297def testCallbackProgram(t, env):298    """Check server can handle random transient program number299    FLAGS: create_session all300    CODE: CSESS20301    """302    cb_occurred = threading.Event()303    transient = 0x40000004304    def mycheck(prog):305        print "Got call using prog=0x%x" % prog306        cb_occurred.prog = prog307        cb_occurred.set()308    orig = env.c1._check_program309    try:310        env.c1._check_program = mycheck311        c = env.c1.new_client(env.testname(t))312        sess = c.create_session(prog=transient)313        cb_occurred.wait(10)314        if not cb_occurred.isSet():315            fail("No CB_NULL sent")316        if cb_occurred.prog != transient:317            fail("Expected cb progam 0x%x, got 0x%x" %318                 (transient, cb_occurred.prog))319    finally:320        env.c1._check_program = orig321def testCallbackVersion(t, env):322    """Check server sends callback program with a version listed in nfs4client.py323    FLAGS: create_session all324    CODE: CSESS21325    """326    cb_occurred = threading.Event()327    transient = 0x40000000328    def mycheck(low, hi, vers):329        print "Got call using version=%i" % vers330        cb_occurred.low = low331        cb_occurred.hi = hi332        cb_occurred.vers = vers333        cb_occurred.set()334        return (low <= vers <= hi)335    orig = env.c1._check_version336    try:337        env.c1._check_version = mycheck338        c = env.c1.new_client(env.testname(t))339        sess = c.create_session(prog=transient)340        cb_occurred.wait(10)341        if not cb_occurred.isSet():342            fail("No CB_NULL sent")343        if not (cb_occurred.low <= cb_occurred.vers <= cb_occurred.hi):344            fail("Expected cb version between %i and %i, got %i" %345                 (cb_occurred.low, cb_occurred.hi, cb_occurred.vers))346    finally:347        env.c1._check_version = orig348def testMaxreqs(t, env):349    """A CREATE_SESSION with maxreqs too large should return350       a modified value351    FLAGS: create_session all352    CODE: CSESS22353    """354    # Assuming this is too large for any server; increase if necessary:355    # but too huge will eat many memory for replay_cache, be careful!356    TOO_MANY_SLOTS = 500357    c = env.c1.new_client(env.testname(t))358    # CREATE_SESSION with fore_channel = TOO_MANY_SLOTS359    chan_attrs = channel_attrs4(0,8192,8192,8192,128, TOO_MANY_SLOTS, [])360    sess1 = c.create_session(fore_attrs=chan_attrs)361    if nfs4lib.test_equal(sess1.fore_channel.maxrequests,362                          chan_attrs.ca_maxrequests, "count4"):363        fail("Server allows surprisingly large fore_channel maxreqs")364def testNotOnlyOp(t, env):365    """Check for NFS4ERR_NOT_ONLY_OP366    FLAGS: create_session all367    CODE: CSESS23368    """369    c = env.c1.new_client(env.testname(t))370    # CREATE_SESSION with PUT_ROOTFH371    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[])372    res = c.c.compound([op.create_session(c.clientid, c.seqid, 0,373                                        chan_attrs, chan_attrs,374                                        123, []), op.putrootfh()], None)375    check(res, NFS4ERR_NOT_ONLY_OP)376def testCsr_sequence(t, env):377    """The corresponding result of csa_sequence is csr_sequence,378       which MUST be equal to csa_sequence.379    FLAGS: create_session all380    CODE: CSESS24381    """382    c = env.c1.new_client(env.testname(t))383    # CREATE_SESSION384    chan_attrs = channel_attrs4(0,8192,8192,8192,128,8,[])385    csa_sequence = c.seqid386    sess1 = c.create_session(fore_attrs=chan_attrs)387    if not nfs4lib.test_equal(sess1.seqid, csa_sequence, "int"):388        fail("Server returns bad csr_sequence which not equal to csa_sequence")389def testTooSmallMaxRS(t, env):390    """If client selects a value for ca_maxresponsesize such that391       a replier on a channel could never send a response,392       server SHOULD return NFS4ERR_TOOSMALL393    FLAGS: create_session all394    CODE: CSESS25395    """396    c = env.c1.new_client(env.testname(t))397    # CREATE_SESSION with too small ca_maxresponsesize398    chan_attrs = channel_attrs4(0,8192,0,8192,128,8,[])399    res = c.c.compound([op.create_session(c.clientid, c.seqid, 0,400                                        chan_attrs, chan_attrs,401                                        123, [])], None)402    check(res, NFS4ERR_TOOSMALL)403def testRepTooBig(t, env):404    """If requester sends a request for which the size of the reply405       would exceed ca_maxresponsesize, the replier will return406       NFS4ERR_REP_TOO_BIG407    FLAGS: create_session all408    CODE: CSESS26409    """410    name = env.testname(t)411    c1 = env.c1.new_client(name)412    # create session with a small ca_maxresponsesize413    chan_attrs = channel_attrs4(0,8192,500,8192,128,8,[])414    sess1 = c1.create_session(fore_attrs=chan_attrs)415    sess1.compound([op.reclaim_complete(FALSE)])416    owner = "owner_%s" % name417    path = sess1.c.homedir + [name]418    res = create_file(sess1, owner, path, access=OPEN4_SHARE_ACCESS_BOTH)419    check(res)420    # write some data to file421    fh = res.resarray[-1].object422    stateid = res.resarray[-2].stateid423    res = sess1.compound([op.putfh(fh), op.write(stateid, 5, FILE_SYNC4, "write test data " * 10)])424    check(res)425    # read data rather than ca_maxresponsesize426    res = sess1.compound([op.putfh(fh), op.read(stateid, 0, 500)])427    check(res, NFS4ERR_REP_TOO_BIG)428def testRepTooBigToCache(t, env):429    """If requester sends a request for which the size of the reply430       would exceed ca_maxresponsesize_cached, the replier will return431       NFS4ERR_REP_TOO_BIG_TO_CACHE432    FLAGS: create_session all433    CODE: CSESS27434    """435    c = env.c1.new_client(env.testname(t))436    # CREATE_SESSION with a small ca_maxresponsesize_cached437    chan_attrs = channel_attrs4(0,8192,8192,10,128,8,[])438    res = c.c.compound([op.create_session(c.clientid, c.seqid, 0,439                                        chan_attrs, chan_attrs,440                                        123, [])], None)441    check(res)442    # SEQUENCE with cache this443    sid = res.resarray[0].csr_sessionid444    res = c.c.compound([op.sequence(sid, 1, 0, 0, True)])...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.