How to use create_session method in AutoItDriverServer

Best Python code snippet using AutoItDriverServer_python

test_polymorphic_rel.py

Source:test_polymorphic_rel.py Github

copy

Full Screen

...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, "*")...

Full Screen

Full Screen

st_create_session.py

Source:st_create_session.py Github

copy

Full Screen

...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)])...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run AutoItDriverServer automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful