ÿØÿà JFIF    ÿÛ „ ( %!1!%*+...983,7(-.- PK}]Vk?k test_cte.pynu[from .. import config from .. import fixtures from ..assertions import eq_ from ..schema import Column from ..schema import Table from ... import ForeignKey from ... import Integer from ... import select from ... import String from ... import testing class CTETest(fixtures.TablesTest): __backend__ = True __requires__ = ("ctes",) run_inserts = "each" run_deletes = "each" @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("data", String(50)), Column("parent_id", ForeignKey("some_table.id")), ) Table( "some_other_table", metadata, Column("id", Integer, primary_key=True), Column("data", String(50)), Column("parent_id", Integer), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.some_table.insert(), [ {"id": 1, "data": "d1", "parent_id": None}, {"id": 2, "data": "d2", "parent_id": 1}, {"id": 3, "data": "d3", "parent_id": 1}, {"id": 4, "data": "d4", "parent_id": 3}, {"id": 5, "data": "d5", "parent_id": 3}, ], ) def test_select_nonrecursive_round_trip(self): some_table = self.tables.some_table with config.db.connect() as conn: cte = ( select([some_table]) .where(some_table.c.data.in_(["d2", "d3", "d4"])) .cte("some_cte") ) result = conn.execute( select([cte.c.data]).where(cte.c.data.in_(["d4", "d5"])) ) eq_(result.fetchall(), [("d4",)]) def test_select_recursive_round_trip(self): some_table = self.tables.some_table with config.db.connect() as conn: cte = ( select([some_table]) .where(some_table.c.data.in_(["d2", "d3", "d4"])) .cte("some_cte", recursive=True) ) cte_alias = cte.alias("c1") st1 = some_table.alias() # note that SQL Server requires this to be UNION ALL, # can't be UNION cte = cte.union_all( select([st1]).where(st1.c.id == cte_alias.c.parent_id) ) result = conn.execute( select([cte.c.data]) .where(cte.c.data != "d2") .order_by(cte.c.data.desc()) ) eq_( result.fetchall(), [("d4",), ("d3",), ("d3",), ("d1",), ("d1",), ("d1",)], ) def test_insert_from_select_round_trip(self): some_table = self.tables.some_table some_other_table = self.tables.some_other_table with config.db.connect() as conn: cte = ( select([some_table]) .where(some_table.c.data.in_(["d2", "d3", "d4"])) .cte("some_cte") ) conn.execute( some_other_table.insert().from_select( ["id", "data", "parent_id"], select([cte]) ) ) eq_( conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), [(2, "d2", 1), (3, "d3", 1), (4, "d4", 3)], ) @testing.requires.ctes_with_update_delete @testing.requires.update_from def test_update_from_round_trip(self): some_table = self.tables.some_table some_other_table = self.tables.some_other_table with config.db.connect() as conn: conn.execute( some_other_table.insert().from_select( ["id", "data", "parent_id"], select([some_table]) ) ) cte = ( select([some_table]) .where(some_table.c.data.in_(["d2", "d3", "d4"])) .cte("some_cte") ) conn.execute( some_other_table.update() .values(parent_id=5) .where(some_other_table.c.data == cte.c.data) ) eq_( conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), [ (1, "d1", None), (2, "d2", 5), (3, "d3", 5), (4, "d4", 5), (5, "d5", 3), ], ) @testing.requires.ctes_with_update_delete @testing.requires.delete_from def test_delete_from_round_trip(self): some_table = self.tables.some_table some_other_table = self.tables.some_other_table with config.db.connect() as conn: conn.execute( some_other_table.insert().from_select( ["id", "data", "parent_id"], select([some_table]) ) ) cte = ( select([some_table]) .where(some_table.c.data.in_(["d2", "d3", "d4"])) .cte("some_cte") ) conn.execute( some_other_table.delete().where( some_other_table.c.data == cte.c.data ) ) eq_( conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), [(1, "d1", None), (5, "d5", 3)], ) @testing.requires.ctes_with_update_delete def test_delete_scalar_subq_round_trip(self): some_table = self.tables.some_table some_other_table = self.tables.some_other_table with config.db.connect() as conn: conn.execute( some_other_table.insert().from_select( ["id", "data", "parent_id"], select([some_table]) ) ) cte = ( select([some_table]) .where(some_table.c.data.in_(["d2", "d3", "d4"])) .cte("some_cte") ) conn.execute( some_other_table.delete().where( some_other_table.c.data == select([cte.c.data]).where( cte.c.id == some_other_table.c.id ) ) ) eq_( conn.execute( select([some_other_table]).order_by(some_other_table.c.id) ).fetchall(), [(1, "d1", None), (5, "d5", 3)], ) PK}]| {{*__pycache__/test_reflection.cpython-37.pycnu[B 4]\@sfddlZddlZddlZddlmZddlmZddlmZddlmZddlm Z ddlm Z dd lm Z dd l m Z dd l mZd d lmZd dlmZd dlmZd dlmZd dlmZd dlmZd dlmZd dlmZd dlmZd dlmZd dl mZd dl mZd dlmZd\Z Z!Gddde j"Z#Gddde j"Z$Gdd d e j"Z%d!Z&dS)"N)assert_raises_message)config)engines)eq_)expect_warnings)fixtures)is_)Column)Table)event)exc) ForeignKey)inspect)Integer)MetaData)String)testing)types) Inspector)DDL)Index) quoted_name)NNc@s$eZdZdZeddZddZdS) HasTableTestTcCs&td|tdtddtdtddS)N test_tableidT) primary_keydata2)r r rr)clsmetadatar"[/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_reflection.py define_tables#s  zHasTableTest.define_tablesc CsBtj.}tjj|ds ttjj|dr4tWdQRXdS)NrZnonexistent_table)rdbbegindialectZ has_tableAssertionError)selfZconnr"r"r#test_has_table,s zHasTableTest.test_has_tableN)__name__ __module__ __qualname__ __backend__ classmethodr$r*r"r"r"r#r s rc@seZdZdZZdZeddZeddZeddZ ed d Z ed d Z ed dZ e jjddZe jjddZe jjddZe jdddZe jjddZe jje jje jjddZe jjddZe jje jjddZe jjd d!Ze jje jj d"d#Z!dd$d%Z"e jje jj d&d'Z#e jj$d(d)Z%e jj$e jj d*d+Z&e jje jj$d,d-Z'dd.d/Z(e jjd0d1Z)e jd2d3Z*e jjd4d5Z+e jjd6d7Z,e jje jd8d9Z-e jje jj d:d;Z.e jj/dd?Z1e jj$d@dAZ2e jj$e jj dBdCZ3e jddDdEZ4e jj5dFdGZ6e jje jj5e jj dHdIZ7e jje jdJdKZ8e jddLdMZ9e jjdNdOZ:e jje jj dPdQZ;e jjdTdUZ?e jj@dVdWZAe jdXdYZBdZd[ZCe jdd\d]ZDe jjEd^d_ZFe jjEe jj d`daZGe jdbdcZHe jjEdddeZIe jjEdfdgZJe jjKe jdhdiZLe jjMdjdkZNe jj/e jjMdldmZOe jj/dndoZPe jjMe jj dpdqZQe jddrdsZRe jjSdtduZTe jjSe jj dvdwZUe jddxdyZVe jddzd{ZWe jjd|d}ZXe jje jj d~dZYe Zdde jdddZ[ddZ\e jj ddZ]e jje jddZ^dS)ComponentReflectionTestNTcCs4tjjjr*ddlm}tjt|j ddStj SdS)Nr)pool)Z poolclass)options) r requirementsZindependent_connectionsenabled sqlalchemyr1rtesting_enginedictZ StaticPoolr%)r r1r"r"r# setup_bind7s   z"ComponentReflectionTest.setup_bindcCs*||dtjjjr&||tjjdS)N)define_reflected_tablesrrequiresschemasr4r test_schema)r r!r"r"r#r$Bs  z%ComponentReflectionTest.define_tablesc Cs|r|d}nd}tjjjrztd|tdtjddtdtdd d td t dd d td tj tj d |dd|dd}nBtd|tdtjddtdtdd d td t dd d |dd}td|tdtj ddtdtj t d|tdt d|ddtd|tdtj tdtj t |j jtdt dtjddd|ddtd|tdtj dddtdt dd d!td"t dd#d!|d$d%tjjjrX|dkrtd&|tdtj ddtdt dtd't d(tjjdtjjjd)nntd*|tdtj ddtd+t d,tjjjtdt d|ddtd-|tdtj ddtdt d|ddtjjjr||||std.|td/t ddd0}td1|tdtj ddtd/t ddd0}td2|j jtd3|j jtjjjr||||stjjjr||dS)4N.usersuser_idT)rtest1F)nullabletest2parent_user_idz%susers.user_id user_id_fk)name)schema test_needs_fk dingalings dingaling_id address_idz%semail_addresses.address_idremail_addressesremote_user_idZ email_address email_ad_pk comment_testrz id comment)rcommentzdata % comment)rSd2z)Comment types type speedily ' " \ '' Fun!zthe test % ' " \ table comment)rHrS local_table remote_idz%s.remote_table_2.id)rIrH remote_tablelocal_idz%s.local_table.idremote_table_2noncol_idx_test_nopkq)rInoncol_idx_test_pknoncol_idx_nopk noncol_idx_pk)rr:self_referential_foreign_keysr4r r saINTZCHARZFloatrrrcr@ZPrimaryKeyConstraintcross_schema_fk_reflectionrr<r%r'default_schema_nameindex_reflection define_indexrr[descview_column_reflection define_viewstemp_table_reflectiondefine_temp_tables)r r!rHZ schema_prefixr?rZr\r"r"r#r9Hs           z/ComponentReflectionTest.define_reflected_tablesc Cstdrdgdd}n ddgi}td|tdtjd d td td td tjtjd ddtdd f|}tj j j rtj j j rt |dtdt |dtddS)NoraclezGLOBAL TEMPORARYz PRESERVE ROWS)prefixesZoracle_on_commitrmZ TEMPORARYuser_tmprT)rrGrfoo user_tmp_uq)rG user_tmp_ix after_createz:create temporary view user_tmp_v as select * from user_tmp before_dropzdrop view user_tmp_v)ragainstr r r`raZVARCHARUniqueConstraintrr:view_reflectionr4temporary_viewsr listenr)r r!kwrnr"r"r#rks*        z*ComponentReflectionTest.define_temp_tablescCs2td|jj|jjtd|jj|jj|jjdS)N users_t_idx users_all_idx)rrbrArDr@)r r!r?r"r"r#rfsz$ComponentReflectionTest.define_indexcCsbx\dD]T}|}|rd||f}|d}d||f}t|dt|t|dtd|qWdS)N)r?rNz%s.%sZ_vz"CREATE VIEW %s AS SELECT * FROM %srrrsz DROP VIEW %s)r rxr)r r!rH table_namefullnameZ view_namequeryr"r"r#ris  z$ComponentReflectionTest.define_viewscCs$ttj}|tjj|kdS)N)rrr%assert_rr<Zget_schema_names)r)inspr"r"r#test_get_schema_namess z-ComponentReflectionTest.test_get_schema_namescCs4t}t|jdrtt|t|jds0tdS)Nrd)rr6hasattrr'r(r)r)Zenginer"r"r#test_dialect_initializesz/ComponentReflectionTest.test_dialect_initializecCs ttj}t|jtjjjdS)N)rrr%rrdr')r)rr"r"r#test_get_default_schema_names z4ComponentReflectionTest.test_get_default_schema_nametablec sddddddg|j}t|j}|dkrR||}|dd g}tt||nl|rld d ||D}n ||}fd d |D}|d krdddg}t||ndddg}tt||dS)NrRr\rZrUrWrYviewemail_addresses_vusers_vcSsg|]}|dr|dqS)rr").0recr"r"r# :szAComponentReflectionTest._test_get_table_names..csg|]}|kr|qSr"r")rt)_ignore_tablesr"r#r@s foreign_keyr?rNrJ) r!rbindZget_view_namessortrsortedZget_sorted_table_and_fkc_namesget_table_names) r)rH table_typeorder_bymetar table_namesanswertablesr")rr#_test_get_table_names"s.      z-ComponentReflectionTest._test_get_table_namescCs&t|j}|}tt|dgdS)Nrn)rrZget_temp_table_namesrr)r)rtemp_table_namesr"r"r#test_get_temp_table_namesIs z1ComponentReflectionTest.test_get_temp_table_namescCs&t|j}|}tt|dgdS)N user_tmp_v)rrZget_temp_view_namesrr)r)rrr"r"r#test_get_temp_view_namesOs z0ComponentReflectionTest.test_get_temp_view_namescCs |dS)N)r)r)r"r"r#test_get_table_namesWsz,ComponentReflectionTest.test_get_table_namescCs|jdddS)Nr)r)r)r)r"r"r#test_get_table_names_fks[sz0ComponentReflectionTest.test_get_table_names_fkscCs |dS)N)_test_get_comments)r)r"r"r#test_get_comments`sz)ComponentReflectionTest.test_get_commentscCs|tjjdS)N)rrrr<)r)r"r"r#test_get_comments_with_schemadsz5ComponentReflectionTest.test_get_comments_with_schemacCstttj}t|jd|dddit|jd|ddditdd|jd|dDdd d d d d d dd gdS)NrR)rHtextzthe test % ' " \ table commentr?cSsg|]}|d|ddqS)rGrS)rGrSr")rrr"r"r#rusz>ComponentReflectionTest._test_get_comments..z id commentr)rSrGzdata % commentrz)Comment types type speedily ' " \ '' Fun!rT)rrr%rZget_table_comment get_columns)r)rHrr"r"r#ris   z*ComponentReflectionTest._test_get_commentscCs|tjjdS)N)rrrr<)r)r"r"r# test_get_table_names_with_schemasz8ComponentReflectionTest.test_get_table_names_with_schemacCs|jdddS)Nr)r)r)r)r"r"r#test_get_view_namessz+ComponentReflectionTest.test_get_view_namescCs|jtjjdddS)Nr)r)rrrr<)r)r"r"r#test_get_view_names_with_schemasz7ComponentReflectionTest.test_get_view_names_with_schemacCs||jdddS)Nr)r)r)r)r"r"r#test_get_tables_and_viewssz1ComponentReflectionTest.test_get_tables_and_viewscCsvttj}|jj|jj}}ddg}|dkr4ddg}t|j}x0t|||fD]\}} |} |j || d} | t | dkt | xt | j D]\} } t| j| | d| | d j}| j}t|tjjr|j}td r|tjtjfkrtj}| t t|j|jtjtjtjtjtjtj tj!gdkd | j| j| | d|f| j"s| | d dkst#qWqPWdS) Nr?rNrrr)rHrrGtyperlz%s(%s), %s(%s)default)$rrr%rr?rNrrziprrlen enumeratecolumnsrrG __class__r isinstancer`rZ TypeEnginert sql_typesZDateZDateTimeset__mro__ intersectionrNumericZTimerZ_Binaryrr()r)rHrrr? addressesrrr|rZ schema_namecolsicolctypeZ ctype_defr"r"r#_test_get_columnssH      z)ComponentReflectionTest._test_get_columnscCs |dS)N)r)r)r"r"r#test_get_columnssz(ComponentReflectionTest.test_get_columnscGsDtd|jfddt|D}|ddt|jjdDS)NrcSsg|]\}}td||qS)zt%d)r )rrtype_r"r"r#rsz.cSsg|] }|dqS)rr")rrbr"r"r#rs)r r!rcreaterrr)r)rrr"r"r#_type_round_trips z(ComponentReflectionTest._type_round_tripcCsHxB|tddD],}t|tjs(tt|jdt|jdqWdS)NrB)rrrrr(rZ precisionZscale)r)typr"r"r#test_numeric_reflections z/ComponentReflectionTest.test_numeric_reflectioncCs4|tdd}t|tjs$tt|jddS)N4r)rrrrr(rlength)r)rr"r"r#test_varchar_reflectionsz/ComponentReflectionTest.test_varchar_reflectionc Cs\td|jtdtddtdtdd}|ttddt|jj dDddd dS) NraT)rCbFcss|]}|d|dfVqdS)rGrCNr")rrr"r"r# szCComponentReflectionTest.test_nullable_reflection..)rr) r r!r rrrr7rrr)r)rr"r"r#test_nullable_reflections z0ComponentReflectionTest.test_nullable_reflectioncCs|jtjjddS)N)rH)rrrr<)r)r"r"r#test_get_columns_with_schemasz4ComponentReflectionTest.test_get_columns_with_schemacCsnt|j}|jj}t|j}|d}|t|dkt|x*t|j D]\}}t |j ||dqJWdS)NrnrrG) rrrrnrrrrrrrrG)r)rrnrrrrr"r"r#test_get_temp_table_columnss   z3ComponentReflectionTest.test_get_temp_table_columnscCs2t|j}|d}tdd|DdddgdS)NrcSsg|] }|dqS)rGr")rrr"r"r#rszFComponentReflectionTest.test_get_temp_view_columns..rrGro)rrrr)r)rrr"r"r#test_get_temp_view_columnss  z2ComponentReflectionTest.test_get_temp_view_columnscCs|jdddS)Nr)r)r)r)r"r"r#test_get_view_columnssz-ComponentReflectionTest.test_get_view_columnscCs|jtjjdddS)Nr)rHr)rrrr<)r)r"r"r#!test_get_view_columns_with_schemasz9ComponentReflectionTest.test_get_view_columns_with_schemac Cs|j}|jj|jj}}t|j}|j|j|d}|d}t|dg|j|j|d}|d} t| dgt j j t|ddWdQRXdS)N)rHconstrained_columnsr@rLrGrQ) r!rr?rNrrZget_pk_constraintrGrrr:Zreflects_pk_namesfail_if) r)rHrr?rrZ users_consZ users_pkeysZ addr_consZ addr_pkeysr"r"r#_test_get_pk_constraint%s   z/ComponentReflectionTest._test_get_pk_constraintcCs |dS)N)r)r)r"r"r#test_get_pk_constraint6sz.ComponentReflectionTest.test_get_pk_constraintcCs|jtjjddS)N)rH)rrrr<)r)r"r"r#"test_get_pk_constraint_with_schema:sz:ComponentReflectionTest.test_get_pk_constraint_with_schemacCs0|j}|jj}t|j}ttjd|j|j dS)Nz+.*get_primary_keys\(\) method is deprecated) r!rr?rrrsa_excZSADeprecationWarningZget_primary_keysrG)r)rr?rr"r"r# test_deprecated_get_primary_keys@s z8ComponentReflectionTest.test_deprecated_get_primary_keysc Cs:|j}|jj|jj}}t|j}|}tjjj r|j |j |d}|d}tjj t|ddWdQRXt|d|t|d|j t|ddgtjjj rt|d d g|j |j |d} | d}tjj ||ddk WdQRXt|d|t|d|j t|ddgt|d d gdS) N)rHrrGrFreferred_schemareferred_tablereferred_columnsr@rrErO)r!rr?rNrrrr:r_r4get_foreign_keysrGZnamed_constraintsrrZimplicitly_named_constraintsr) r)rHrr?rrZexpected_schemaZ users_fkeysfkey1Z addr_fkeysr"r"r#_test_get_foreign_keysMs,   z.ComponentReflectionTest._test_get_foreign_keyscCs |dS)N)r)r)r"r"r#test_get_foreign_keysnsz-ComponentReflectionTest.test_get_foreign_keyscCs|jtjjddS)N)rH)rrrr<)r)r"r"r#!test_get_foreign_keys_with_schemarsz9ComponentReflectionTest.test_get_foreign_keys_with_schemac Cs|dtjjjdtjjdtjj\}}}ttj}||j }t t |d|d}t |dtjjt |d|j t |dd gt |d d g|j|j tjjd }t t |d|d}|ddtjjjfkst t |d|j t |dd gt |d d gdS)Nz%s.local_tablez%s.remote_tablez%s.remote_table_2rrrrrrrV)rHrX) rrr%r'rdrr<rrrGrrr() r)rUrWrYrZ local_fkeysrZ remote_fkeysZfkey2r"r"r#"test_get_inter_schema_foreign_keysws,    z:ComponentReflectionTest.test_get_inter_schema_foreign_keyscCs|jdddS)NZCASCADE)Zondelete)_test_get_foreign_key_options)r)r"r"r#%test_get_foreign_key_options_ondeletesz=ComponentReflectionTest.test_get_foreign_key_options_ondeletecCs|jdddS)NzSET NULL)Zonupdate)r)r)r"r"r#%test_get_foreign_key_options_onupdatesz=ComponentReflectionTest.test_get_foreign_key_options_onupdatec s|j}td|tdtddddtd|tdtddtdttjdd d td td ddtd |tdtddtdtdddtdttjdgdgfddi|dd|t |j }| dddt t fddDi| d ddt t fddD|dS)NxrT)r)rIrZx_idzx.idxid)rGtest userrGrF)rCtidztable.idZmyfkrr2c3s"|]}|r||fVqdS)Nr")rk)optsr"r#rszHComponentReflectionTest._test_get_foreign_key_options..c3s"|]}|r||fVqdS)Nr")rr)rr"r#rs)r!r r rr`rrZForeignKeyConstraint create_allrrrrr7)r)r2rrr")rr#rs8     z5ComponentReflectionTest._test_get_foreign_key_optionscCsbdd|D}xN|D]F}|d|ks(t|||d}x|D]}t||||q@WqWdS)NcSsg|] }|dqS)rGr")rdr"r"r#rsz@ComponentReflectionTest._assert_insp_indexes..rG)r(indexr)r)indexesexpected_indexesZ index_namesZe_indexrkeyr"r"r#_assert_insp_indexess   z,ComponentReflectionTest._assert_insp_indexescCsP|j}t|j}|jd|d}dddgddddddgd dg}|||dS) Nr?)rHFrArDrz)unique column_namesrGr@r{)r!rr get_indexesr)r)rHrrrrr"r"r#_test_get_indexess  z)ComponentReflectionTest._test_get_indexescCs |dS)N)r)r)r"r"r#test_get_indexessz(ComponentReflectionTest.test_get_indexescCs|jtjjddS)N)rH)rrrr<)r)r"r"r#test_get_indexes_with_schemasz4ComponentReflectionTest.test_get_indexes_with_schemacCs|j}t|j}||}d|dg}|||t|||jd}tt|jdt t |jdj |tt |jdj |dS)NF)rrG) autoload_withrr) r!rrrrr rrrr listrrG)r)tnameZixnamerrrrrr"r"r#_test_get_noncol_indexs    z.ComponentReflectionTest._test_get_noncol_indexcCs|dddS)NrZr])r)r)r"r"r#test_get_noncol_index_no_pksz3ComponentReflectionTest.test_get_noncol_index_no_pkcCs|dddS)Nr\r^)r)r)r"r"r#test_get_noncol_index_pk sz0ComponentReflectionTest.test_get_noncol_index_pkc Cstd|jtdtdtdtdt|jdtdt|jdtd|jt|jj }t d"t | dd dgd d gWdQRXdS) NrrrMyrrz+CREATE INDEX t_idx ON t(lower(x), lower(y))zCREATE INDEX t_idx_2 ON t(x)z>Skipped unsupported reflection of expression-based index t_idxZt_idx_2r)rGrr) r r!r rr rxrrrrrrr)r)rr"r"r#%test_reflect_expression_based_indexess$    z=ComponentReflectionTest.test_reflect_expression_based_indexescCs |dS)N)_test_get_unique_constraints)r)r"r"r#test_get_unique_constraints,sz3ComponentReflectionTest.test_get_unique_constraintscCsFt|j}|d}x|D]}|ddqWt|dgddgdS)Nrnduplicates_indexrGrp)rrG)rrget_unique_constraintspopr)r)r reflectedreflr"r"r#&test_get_temp_table_unique_constraints0s    z>ComponentReflectionTest.test_get_temp_table_unique_constraintscCsRt|j}|d}x|D]}|ddqWtdd|DddgddgdS) NrnZdialect_optionscSsg|]}|ddkr|qS)rGrqr")ridxr"r"r#rDszGComponentReflectionTest.test_get_temp_table_indexes..Frorq)rrrG)rrrrr)r)rrindr"r"r#test_get_temp_table_indexes;s    z3ComponentReflectionTest.test_get_temp_table_indexescCs|jtjjddS)N)rH)rrrr<)r)r"r"r#'test_get_unique_constraints_with_schemaNsz?ComponentReflectionTest.test_get_unique_constraints_with_schemac Cstddgdddddgdddddgddd d gdd dgdd dgdgtd d}|j}td|tdtdtdtdtdtjtd tdtd td|d}x*|D]"}| tj |dd |d iqW| t |j }t|jd|dtd d}t}x>t||D]0\} } | dd} | r>|| t| | qWt} td| |j |d}tdd|jD} tdd|jDdg}| |rt|rt|| t|tdS)NZunique_ar)rGrZ unique_a_b_crrbZ unique_c_a_bZunique_asc_keyZascrz i.have.dotsz i have spacesrG)rZtesttblrPrM)rHrr)rrHcSsg|] }|jqSr")rG)rrr"r"r#rszHComponentReflectionTest._test_get_unique_constraints..cSsg|]}t|tjr|jqSr")rr`rurG)rZuqr"r"r#rs)roperator itemgetterr!r r r`rrZappend_constraintrurrrrrrraddrrr constraints differencerr()r)rHuniques orig_metarZuc inspectorrZnames_that_duplicate_indexorigrZdupeZreflected_metadataZ idx_namesZuq_namesr"r"r#rSs^            z4ComponentReflectionTest._test_get_unique_constraintscCs |dS)N)_test_get_check_constraints)r)r"r"r#test_get_check_constraintssz2ComponentReflectionTest.test_get_check_constraintscCs|jtjjddS)N)rH)rrrr<)r)r"r"r#!test_get_check_constraints_schemasz9ComponentReflectionTest.test_get_check_constraints_schemac s|j}td|tdttjdddtjddd|d|t|j}t |j d|dt d d }d d fd d|D}t |ddddddgdS)NZsa_ccrza > 1 AND a < 5Zcc1)rGza = 1 OR (a > 2 AND a < 5)Zcc2)rHrG)rcSsdtd|tjS)N zand|\d|=|a|or|<|>)joinrefindalllowerI)sqltextr"r"r# normalizeszFComponentReflectionTest._test_get_check_constraints..normalizecs"g|]}|d|ddqS)rGr)rGrr")ritem)rr"r#rszGComponentReflectionTest._test_get_check_constraints..za > 1 and a < 5)rGrza = 1 or a > 2 and a < 5)r!r r rr`ZCheckConstraintrrrrZget_check_constraintsr r r)r)rHrrrr")rr#rs(      z3ComponentReflectionTest._test_get_check_constraintscCsL|j}d}d}t|j}|j||d}|||j||d}||dS)Nrr)rH)r!rrZget_view_definitionr)r)rHrZ view_name1Z view_name2rZv1Zv2r"r"r#_test_get_view_definitions  z1ComponentReflectionTest._test_get_view_definitioncCs |dS)N)r!)r)r"r"r#test_get_view_definitionsz0ComponentReflectionTest.test_get_view_definitioncCs|jtjjddS)N)rH)r!rrr<)r)r"r"r#$test_get_view_definition_with_schemaszszBComponentReflectionTest.test_autoincrement_col..Z autoincrementTN)r!rrrgetr()r)rrrZcnamerZid_r"r"r#test_autoincrement_cols   z.ComponentReflectionTest.test_autoincrement_col)NrN)N)Nr)N)N)N)N)N)N)N)_r+r,r-Z run_insertsZ run_deletesr.r/r8r$r9rkrfrirr:Zschema_reflectionrrrZprovide_metadatarrrrvrwrZtable_reflectionrZ!foreign_key_constraint_reflectionrZcomment_reflectionrr;rrrrhrrrrrrrrrrrjrrrrrZ!primary_key_constraint_reflectionrrrrrrrcrZ1foreign_key_constraint_option_reflection_ondeleterZ1foreign_key_constraint_option_reflection_onupdaterrrrrerrrrrZindexes_with_expressionsrZunique_constraint_reflectionrrr r rZcheck_constraint_reflectionrrrr!r"r#Zonly_onr&r'r(r+r"r"r"r#r02s   $   %  8     "-   N '  r0c@s0eZdZdZdZeddZddZddZd S) NormalizedNameTest)Zdenormalized_namesTcCsLttddd|tdtddttddd|tdtddtdtddS) Nt1T)quoter)rt2t1idzt1.id)r rr rr)r r!r"r"r#r$s   z NormalizedNameTest.define_tablescCsttj}ttddd|dd}|jd}|jj|jj s@t ttj}|j ddd|jdjj|jdjj s|t dS) Nr/T)r.)Zautoloadr-cSs |dkS)N)r-r/)r)rGmr"r"r#,zINormalizedNameTest.test_reflect_lowercase_forced_tables..)only) rrr%r rrrbr0Z referencesrr(Zreflect)r)m2Zt2_refZt1_refZm3r"r"r#$test_reflect_lowercase_forced_tables$s   z7NormalizedNameTest.test_reflect_lowercase_forced_tablescCsPddttjD}t|d|dt|d|ddS)NcSsg|]}|dkr|qS))r-r/)r)rrr"r"r#r1sz;NormalizedNameTest.test_get_table_names..rr)rrr%rrupperr)r)Z tablenamesr"r"r#r/sz'NormalizedNameTest.test_get_table_namesN) r+r,r- __requires__r.r/r$r6rr"r"r"r#r,s   r,)r0rr,)'r rr5r`r>rrrrrrr rHr r r rrrrrrrrrrZengine.reflectionrrrZ sql.elementsrr!r?Z TablesTestrr0r,__all__r"r"r"r#sH                      g(PK}]=8y-__pycache__/test_update_delete.cpython-37.pycnu[B 4]@snddlmZddlmZddlmZddlmZddlmZddlmZddlm Z Gd d d ej Z d Z d S) )config)fixtures)eq_)Column)Table)Integer)Stringc@s<eZdZdZdZeddZeddZddZd d Z d S) SimpleUpdateDeleteTestZeachTcCs&td|tdtddtdtddS)Nplain_pkidT)Z primary_keydata2)rrrr )clsmetadatar^/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_update_delete.py define_tabless  z$SimpleUpdateDeleteTest.define_tablescCs2tj|jjdddddddddgdS)Nd1)r r rd2rd3)rdbexecutetablesr insert)rrrr insert_datas  z"SimpleUpdateDeleteTest.insert_datacCsl|jj}tjj||jjdkdd}|j r4t |j r>t t tj| |jjdddgdS)Nrd2_new)r )rr)rr)rr)rr rrrupdatewherecr is_insertAssertionError returns_rowsrselectorder_byfetchall)selftrrrr test_update"s"  z"SimpleUpdateDeleteTest.test_updatecCsf|jj}tj||jjdk}|j r0t |j r:t t tj| |jjddgdS)Nr)rr)rr)rr rrrdeleterr r r!r"r#rr$r%r&)r'r(r)rrr test_delete-s  z"SimpleUpdateDeleteTest.test_deleteN) __name__ __module__ __qualname__Z run_deletesZ __backend__ classmethodrrr*r,rrrrr s   r )r N) rrZ assertionsrZschemarrrr Z TablesTestr __all__rrrrs       .PK}]3dW''#__pycache__/test_cte.cpython-37.pycnu[B 4]@sddlmZddlmZddlmZddlmZddlmZddlmZddlm Z dd lm Z dd lm Z dd lm Z Gd d d ej ZdS))config)fixtures)eq_)Column)Table) ForeignKey)Integer)select)String)testingc@seZdZdZdZdZdZeddZeddZ dd Z d d Z d d Z e jje jjddZe jje jjddZe jjddZdS)CTETestT)ZctesZeachc Cs\td|tdtddtdtdtdtdtd |tdtddtdtdtdtdS) N some_tableidT)Z primary_keydata2 parent_idz some_table.idsome_other_table)rrr r r)clsmetadatarT/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_cte.py define_tabless    zCTETest.define_tablesc CsLtj|jjdddddddddddddd ddd d ddgdS) Nd1)rrrrd2rd3d4d5)rdbexecutetablesrinsert)rrrr insert_data&s     zCTETest.insert_datac Cs|jj}tjd}t|g|jj dddg d}| t|jjg|jj ddg}t | dgWdQRXdS)Nrrrsome_cter )r)r#rrr!connectr wherecrin_cter"rfetchall)selfrconnr+resultrrr#test_select_nonrecursive_round_trip3s  $z+CTETest.test_select_nonrecursive_round_tripc Cs|jj}tj}t|g|jj dddgj ddd}| d}| }| t|g|jj |jjk}|t|jjg|jjdk|jj}t|dd d d d d gWdQRXdS) Nrrrr&T) recursiveZc1)r)r)r)r#rrr!r'r r(r)rr*r+aliasZ union_allrrr"order_bydescrr,)r-rr.r+Z cte_aliasZst1r/rrr test_select_recursive_round_tripAs     z(CTETest.test_select_recursive_round_tripc Cs|jj}|jj}tjx}t|g|jj dddg d}| | dddgt|gt| t|g|jjdd d gWdQRXdS) Nrrrr&rrr)rrr)rrr)rrr)r#rrrr!r'r r(r)rr*r+r"r$ from_selectrr3rr,)r-rrr.r+rrr"test_insert_from_select_round_trip\s  z*CTETest.test_insert_from_select_round_tripc Cs|jj}|jj}tj}||dddgt |gt |g |j j dddgd}||jdd  |j j |j j kt|t |g|j jd d d d dgWdQRXdS)Nrrrrrrr&r)r)rrN)rrr)rrr)rrr)rr r)r#rrrr!r'r"r$r6r r(r)rr*r+updatevaluesrr3rr,)r-rrr.r+rrrtest_update_from_round_triprs*  z#CTETest.test_update_from_round_tripc Cs|jj}|jj}tj}||dddgt |gt |g |j j dddgd}|| |j j |j j kt|t |g|j jdd gWdQRXdS) Nrrrrrrr&)rrN)rr r)r#rrrr!r'r"r$r6r r(r)rr*r+deleterr3rr,)r-rrr.r+rrrtest_delete_from_round_trips   z#CTETest.test_delete_from_round_tripc Cs|jj}|jj}tj}||dddgt |gt |g |j j dddgd}|| |j j t |j j g |j j|j jkkt|t |g|j jdd gWdQRXdS) Nrrrrrrr&)rrN)rr r)r#rrrr!r'r"r$r6r r(r)rr*r+r;rrr3r,)r-rrr.r+rrr"test_delete_scalar_subq_round_trips$  z*CTETest.test_delete_scalar_subq_round_tripN)__name__ __module__ __qualname__Z __backend__ __requires__Z run_insertsZ run_deletes classmethodrr%r0r5r7r requiresZctes_with_update_deleteZ update_fromr:Z delete_fromr<r=rrrrr s  #r N)rrZ assertionsrZschemarrrr r r r Z TablesTestr rrrrs          PK}].[F~sXsX&__pycache__/test_select.cpython-37.pycnu[B 4]N@s`ddlmZddlmZddlmZddlmZddlmZddlmZddlm Z dd lm Z dd lm Z dd lm Z dd lm Z dd lmZddlmZddlmZddlmZddlmZddlmZddlmZddlmZddlmZGdddejZGdddejZGdddejZGdddejZGdddejZGd d!d!ejZd"S)#)config)fixtures)eq_)in_)Column)Table) bindparam)case)false)func)Integer)literal_column)null)select)String)testing)true)tuple_)union)utilc@s@eZdZdZeddZeddZddZej j dd Z d S) CollateTestTcCs&td|tdtddtdtddS)N some_tableidT) primary_keydatad)rrr r)clsmetadatarW/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_select.py define_tabless  zCollateTest.define_tablescCs*tj|jjddddddgdS)Nz collate data1)rrrz collate data2)rdbexecutetablesrinsert)rrrr insert_data#s zCollateTest.insert_datacCsttj||dS)N)rrr#r$fetchall)selfrresultrrr _assert_result-szCollateTest._assert_resultcCsDtjtj}|t|jjg|jjj j | ddgdS)N)r"z collate data1)rz collate data2) rrequiresZget_order_by_collationrr+rr%rorder_bycrcollateZasc)r)Z collationrrr test_collate_order_by0s z!CollateTest.test_collate_order_byN) __name__ __module__ __qualname__ __backend__ classmethodr!r'r+rr,Zorder_by_collationr0rrrr rs  rc@sleZdZdZdZeddZeddZddZd d Z d d Z d dZ ddZ ddZ ejjddZdS)OrderByLabelTestzTest the dialect sends appropriate ORDER BY expressions when labels are used. This essentially exercises the "supports_simple_order_by_label" setting. Tc CsBtd|tdtddtdttdttdtdtd tddS) NrrT)rxyq2p)rrr r)rrrrr r!Gs  zOrderByLabelTest.define_tablesc CsDtj|jjddddddddddddddd d d dgdS) Nr"rZq1Zp3)rr7r8r9r;rZq2p2Zq3p1)rr#r$r%rr&)rrrr r'Ss  zOrderByLabelTest.insert_datacCsttj||dS)N)rrr#r$r()r)rr*rrr r+^szOrderByLabelTest._assert_resultcCs8|jj}|jjd}|t|g|dddgdS)Nlx)r")r)r)r%rr.r7labelr+rr-)r)tabler?rrr test_plainaszOrderByLabelTest.test_plaincCs@|jj}|jj|jjd}|t|g|dddgdS)Nr?)r))) r%rr.r7r8r@r+rr-)r)rAr?rrr test_composed_intfsz"OrderByLabelTest.test_composed_intc Cs|jj}|jj|jjd}t|jj|jj d}| t ||g || dtdfdtdfdtdfgdS) Nr?lyrZq1p3rCZq2p2rDZq3p1)r%rr.r7r8r@r lowerr9r;r+rr-descru)r)rAr?rFrrr test_composed_multipleks z'OrderByLabelTest.test_composed_multiplecCs<|jj}|jjd}|t|g|dddgdS)Nr?)r)r)r") r%rr.r7r@r+rr-rH)r)rAr?rrr test_plain_desctsz OrderByLabelTest.test_plain_desccCsD|jj}|jj|jjd}|t|g| dddgdS)Nr?)rD)rC)r) r%rr.r7r8r@r+rr-rH)r)rAr?rrr test_composed_int_desc{sz'OrderByLabelTest.test_composed_int_desccCsV|jj}|jj|jjd}tt|jj |g | |}| |dddgdS)Nr?)r"r)r"rC)r"rD) r%rr.r7r8r@rr countrZgroup_byr-r+)r)rAexprstmtrrr test_group_by_composeds z'OrderByLabelTest.test_group_by_composedN)r1r2r3__doc__r4r5r!r'r+rBrErJrKrLrr,Zgroup_by_complex_expressionrPrrrr r6<s  r6c@seZdZdZeddZeddZdddZd d Ze j j d d Z e j j d dZ e j j ddZe j jddZe j jddZe j jddZdS)LimitOffsetTestTcCs*td|tdtddtdttdtdS)NrrT)rr7r8)rrr )rrrrr r!s  zLimitOffsetTest.define_tablesc CsBtj|jjddddddddddddddddgdS)Nr"r)rr7r8rr=rC)rr#r$r%rr&)rrrr r's     zLimitOffsetTest.insert_datarcCsttj|||dS)N)rrr#r$r()r)rr*paramsrrr r+szLimitOffsetTest._assert_resultcCs2|jj}|t|g|jjdddgdS)Nr)r"r"r)rrr)r%rr+rr-r.rlimit)r)rArrr test_simple_limitsz!LimitOffsetTest.test_simple_limitcCs2|jj}|t|g|jjdddgdS)Nr)rrr=)r=r=rC)r%rr+rr-r.roffset)r)rArrr test_simple_offsetsz"LimitOffsetTest.test_simple_offsetcCs8|jj}|t|g|jjddddgdS)Nrr")rrr)rrr=) r%rr+rr-r.rrTrV)r)rArrr test_simple_limit_offsetsz(LimitOffsetTest.test_simple_limit_offsetcCsZ|jj}t|g|jjdd}|jt j j ddid}t |}| |ddgdS) z7test that 'literal binds' mode works - no bound params.rr"Z literal_bindsT)dialectZcompile_kwargs)rrr)rrr=N)r%rrr-r.rrTrVcompilerr#rYstrr+)r)rArOZsqlrrr test_limit_offset_nobindss  z)LimitOffsetTest.test_limit_offset_nobindscCs>|jj}|jt|g|jjtdddgddiddS)Nl)r"r"r)rrrr)rS) r%rr+rr-r.rrTr )r)rArrr test_bound_limits z LimitOffsetTest.test_bound_limitcCs>|jj}|jt|g|jjtdddgddiddS)No)rrr=)r=r=rCr)rS) r%rr+rr-r.rrVr )r)rArrr test_bound_offsets z!LimitOffsetTest.test_bound_offsetcCsJ|jj}|jt|g|jjtd tdddgdddddS) Nr]r_)rrr)rrr=rr")r]r_)rS) r%rr+rr-r.rrTr rV)r)rArrr test_bound_limit_offsets   z'LimitOffsetTest.test_bound_limit_offsetN)r)r1r2r3r4r5r!r'r+rUrr,rVrWrXr\Zbound_limit_offsetr^r`rarrrr rRs    rRc@seZdZdZeddZeddZdddZd d Zd d Z e j j e j j d dZe j jddZddZe j j ddZddZdS)CompoundSelectTestTcCs*td|tdtddtdttdtdS)NrrT)rr7r8)rrr )rrrrr r!s  z CompoundSelectTest.define_tablesc CsBtj|jjddddddddddddddddgdS)Nr"r)rr7r8rr=rC)rr#r$r%rr&)rrrr r's     zCompoundSelectTest.insert_datarcCsttj|||dS)N)rrr#r$r()r)rr*rSrrr r+sz!CompoundSelectTest._assert_resultcCs`|jj}t|g|jjdk}t|g|jjdk}t||}|||jjddgdS)Nrr)rrr)rrr=) r%rrwherer.rrr+r-)r)rAs1s2u1rrr test_plain_union s  z#CompoundSelectTest.test_plain_unioncCsh|jj}t|g|jjdk}t|g|jjdk}t||}|| |jjddgdS)Nrr)rrr)rrr=) r%rrrcr.rraliasr+r-)r)rArdrerfrrr test_select_from_plain_unions z/CompoundSelectTest.test_select_from_plain_unioncCs|jj}t|g|jjdkd|jj}t|g|jjdkd|jj}t||d}| ||jjddgdS)Nrr"r)rrr)rrr=) r%rrrcr.rrTr-rr+)r)rArdrerfrrr &test_limit_offset_selectable_in_unionss    z9CompoundSelectTest.test_limit_offset_selectable_in_unionscCsz|jj}t|g|jjdk|jj}t|g|jjdk|jj}t||d}| ||jjddgdS)Nrr)rrr)rrr=) r%rrrcr.rr-rrTr+)r)rArdrerfrrr "test_order_by_selectable_in_unions-s ""z5CompoundSelectTest.test_order_by_selectable_in_unionscCsn|jj}t|g|jjdk}t|g|jjdk}t||d}| | |jjddgdS)Nrr)rrr)rrr=) r%rrrcr.rZdistinctrrTr+r-)r)rArdrerfrrr "test_distinct_selectable_in_unions6s z5CompoundSelectTest.test_distinct_selectable_in_unionscCs|jj}t|g|jjdkd|jj}t|g|jjdkd|jj}t|| }| |d|jjddgdS)Nrr"r)rrr)rrr=) r%rrrcr.rrTr-rrhr+)r)rArdrerfrrr &test_limit_offset_in_unions_from_alias>s    z9CompoundSelectTest.test_limit_offset_in_unions_from_aliascCs|jj}t|g|jjdkd|jj}t|g|jjdkd|jj}t ||d}| ||jjddgdS)Nrr"r)rrr)rrr=) r%rrrcr.rrTr-rhrr+)r)rArdrerfrrr .test_limit_offset_aliased_selectable_in_unionsTs  zACompoundSelectTest.test_limit_offset_aliased_selectable_in_unionsN)r)r1r2r3r4r5r!r'r+rgrirr,Zorder_by_col_from_unionZ/parens_in_union_contained_select_w_limit_offsetrjZ0parens_in_union_contained_select_wo_limit_offsetrkrlrmrnrrrr rbs  rbc@seZdZdZeddZeddZd ddZd d Ze j j d d Z e j j d dZ ddZe j j ddZe j j ddZddZddZddZddZddZdS)!ExpandingBoundInTestTc Cs6td|tdtddtdttdttdtddS) NrrT)rr7r8zr:)rrr r)rrrrr r!ns z"ExpandingBoundInTest.define_tablesc CsJtj|jjdddddddddddddddddd d dgdS) Nr"rZz1)rr7r8rprz2r=z3rCz4)rr#r$r%rr&)rrrr r'ys     z ExpandingBoundInTest.insert_datarcCsttj|||dS)N)rrr#r$r()r)rr*rSrrr r+sz#ExpandingBoundInTest._assert_resultcCsj|jj}t|jjg|jjtddd|jj tddd |jj}|j |gggdddS)Nr9T) expandingr;)r9r;)rS) r%rrr.rrcr7rr r8r-r+)r)rArOrrr test_multiple_empty_setss  z-ExpandingBoundInTest.test_multiple_empty_setscCsZ|jj}t|jjgt|jj|jj t ddd |jj}|j |gdgiddS)Nr9T)rt)rS) r%rrr.rrcrr7rprr r-r+)r)rArOrrr test_empty_heterogeneous_tupless  z4ExpandingBoundInTest.test_empty_heterogeneous_tuplescCsZ|jj}t|jjgt|jj|jj t ddd |jj}|j |gdgiddS)Nr9T)rt)rS) r%rrr.rrcrr7r8rr r-r+)r)rArOrrr test_empty_homogeneous_tupless  z2ExpandingBoundInTest.test_empty_homogeneous_tuplescCs\|jj}t|jjg|jjtddd |jj}|j |dddgdddd gid dS) Nr9T)rt)r)r)r=rrr=)rS) r%rrr.rrcr7rr r-r+)r)rArOrrr test_bound_in_scalars  z)ExpandingBoundInTest.test_bound_in_scalarcCsf|jj}t|jjgt|jj|jj t ddd |jj}|j |dddgdddd gid dS) Nr9T)rt)r)r)r=)rr)rr=)r=rC)rS) r%rrr.rrcrr7r8rr r-r+)r)rArOrrr test_bound_in_two_tuples z,ExpandingBoundInTest.test_bound_in_two_tuplecCsf|jj}t|jjgt|jj|jj t ddd |jj}|j |dddgdddd gid dS) Nr9T)rt)r)r)r=)rrq)rrr)r=rs)rS) r%rrr.rrcrr7rprr r-r+)r)rArOrrr %test_bound_in_heterogeneous_two_tuples z:ExpandingBoundInTest.test_bound_in_heterogeneous_two_tuplecCsP|jj}t|jjg|jjtddd |jj}|j |gdgiddS)Nr9T)rt)rS) r%rrr.rrcr7rr r-r+)r)rArOrrr test_empty_set_against_integers  z3ExpandingBoundInTest.test_empty_set_against_integercCsX|jj}t|jjg|jjtddd |jj}|j |ddddgdgiddS) Nr9T)rt)r")r)r)r=)rS) r%rrr.rrcr7notin_r r-r+)r)rArOrrr 'test_empty_set_against_integer_negations  zKsz*LikeFunctionsTest._test..) r%rrr#connectr$rr.rrcr)r)rNexpectedrZconnZrowsrrr _testGs  $zLikeFunctionsTest._testc Cs6|jjjj}||ddddddddd d d h dS) Nzab%cr"rrr=rCrrDrrr)r%rr.rr startswith)r)colrrr test_startswith_unescapedTs z+LikeFunctionsTest.test_startswith_unescapedcCs(|jjjj}||jddddhdS)Nzab%cT) autoescaper)r%rr.rrr)r)rrrr test_startswith_autoescapeXs z,LikeFunctionsTest.test_startswith_autoescapec Cs:|jjjj}||tddddddddd d d h dS) Nz'ab%c'r"rrr=rCrrDrrr)r%rr.rrrr)r)rrrr test_startswith_sqlexpr\s  z)LikeFunctionsTest.test_startswith_sqlexprcCs(|jjjj}||jddddhdS)Nzab##c#)escaperD)r%rr.rrr)r)rrrr test_startswith_escapecs z(LikeFunctionsTest.test_startswith_escapecCsD|jjjj}||jdddddh||jdddddhdS)Nzab%cTr)rrrzab#crD)r%rr.rrr)r)rrrr !test_startswith_autoescape_escapegs z3LikeFunctionsTest.test_startswith_autoescape_escapec Cs4|jjjj}||ddddddddd d h dS) Nze%fgr"rrr=rCrrDrr)r%rr.rrendswith)r)rrrr test_endswith_unescapedls z)LikeFunctionsTest.test_endswith_unescapedc Cs8|jjjj}||tddddddddd d h dS) Nz'e%fg'r"rrr=rCrrDrr)r%rr.rrrr)r)rrrr test_endswith_sqlexprps z'LikeFunctionsTest.test_endswith_sqlexprcCs(|jjjj}||jddddhdS)Nze%fgT)rr)r%rr.rrr)r)rrrr test_endswith_autoescapevs z*LikeFunctionsTest.test_endswith_autoescapecCs(|jjjj}||jddddhdS)Nze##fgr)rr)r%rr.rrr)r)rrrr test_endswith_escapezs z&LikeFunctionsTest.test_endswith_escapecCsD|jjjj}||jdddddh||jdddddhdS)Nze%fgTr)rrrze#fgr)r%rr.rrr)r)rrrr test_endswith_autoescape_escape~s z1LikeFunctionsTest.test_endswith_autoescape_escapec Cs4|jjjj}||ddddddddd d h dS) Nzb%cder"rrr=rCrrDrr)r%rr.rrcontains)r)rrrr test_contains_unescapeds z)LikeFunctionsTest.test_contains_unescapedcCs(|jjjj}||jddddhdS)Nzb%cdeT)rr)r%rr.rrr)r)rrrr test_contains_autoescapes z*LikeFunctionsTest.test_contains_autoescapecCs(|jjjj}||jddddhdS)Nzb##cder)rrD)r%rr.rrr)r)rrrr test_contains_escapes z&LikeFunctionsTest.test_contains_escapecCsD|jjjj}||jdddddh||jdddddhdS)Nzb%cdTr)rrrzb#cdrD)r%rr.rrr)r)rrrr test_contains_autoescape_escapes z1LikeFunctionsTest.test_contains_autoescape_escape)r1r2r3r4Z run_insertsZ run_deletesr5r!r'rrrrrrrrrrrrrrrrrrr r&s(  rN)rrZ assertionsrrZschemarrr r r r r rrrrrrrrrZ TablesTestrr6rRrbrorrrrr s4                    %R_~d?d?e0ej*Z7Gd@dAdAe0ej*Z8GdBdCdCe0ej*Z9GdDdEdEe(ej.Z:GdFdGdGe(ej.Z;GdHdIdIe(ej*ZdS)MN)config)fixtures)eq_) requirements)Column)Table)and_) BigInteger)Boolean)cast)Date)DateTime)Float)Integer)JSON)literal)MetaData)null)Numeric)select)String)testing)Text)Time) TIMESTAMP) type_coerce)Unicode) UnicodeText)util)declarative_base)Session)uc@s eZdZdZejdddZdS)_LiteralRoundTripFixtureTNc Cstd|jtd|}|tj}x>|D]6}|jt |dj tjj t ddd}| |q.W|jr||jjt |k} n|} | j tjj t ddd} x6| | D](} | d}|dk r||}||kstqWWdQRXdS) ztest literal rendering tx)r&T) literal_binds)dialectcompile_kwargsrN)rmetadatarcreaterdbconnectinsertvaluesrcompiler(dictexecutesupports_whereclauserwherecr&AssertionError) selftype_input_outputfilter_r%connvalueZinsstmtrowr@V/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_types.py_literal_round_trip+s(   z,_LiteralRoundTripFixture._literal_round_trip)N)__name__ __module__ __qualname__r3rprovide_metadatarBr@r@r@rAr$(sr$c@sXeZdZdZedZeddZeddZ ddZ d d Z d d Z d dZ ddZdS)_UnicodeFixture) unicode_datauAlors vous imaginez ma 🐍 surprise, au lever du jour, quand une drôle de petite 🐍 voix m’a réveillé. Elle disait: « S’il vous plaît… dessine-moi 🐍 un mouton! »cCs tjjjS)N)rr"expressions_against_unbounded_textenabled)r7r@r@rAr3[sz$_UnicodeFixture.supports_whereclausec Cs&td|tdtdddtd|jdS)N unicode_tableidT) primary_keytest_needs_autoincrementrH)rrrdatatype)clsr*r@r@rA define_tables_s  z_UnicodeFixture.define_tablescCs`|jj}tj|d|jitjt|jj g }t ||jft |dt js\tdS)NrHr)tablesrKrr,r2r.datarr5rHfirstr isinstancer text_typer6)r7rKr?r@r@rAtest_round_tripjs z_UnicodeFixture.test_round_tripcsjj}tj|fddtdDtjt|jj g }t |fddtdDx |D]}t |dt jshtqhWdS)Ncsg|]}djiqS)rH)rS).0i)r7r@rA ysz?_UnicodeFixture.test_round_trip_executemany..r csg|] }jfqSr@)rS)rXrY)r7r@rArZsr)rRrKrr,r2r.rangerr5rHZfetchallrrUr rVr6)r7rKZrowsr?r@)r7rAtest_round_trip_executemanyts z+_UnicodeFixture.test_round_trip_executemanycCsP|jj}tj|dtditjt|jj g }t |tdfdS)NrH) rRrKrr,r2r.r#rr5rHrTr)r7rKr?r@r@rA_test_empty_stringssz#_UnicodeFixture._test_empty_stringscCs||j|jg|jgdS)N)rBrOrS)r7r@r@rA test_literalsz_UnicodeFixture.test_literalcCs$||jtdgtdgdS)Nuréve🐍 illé)rBrOr r#)r7r@r@rAtest_literal_non_asciisz&_UnicodeFixture.test_literal_non_asciiN)rCrDrE __requires__r#rSpropertyr3 classmethodrQrWr\r^r_r`r@r@r@rArGRs   rGc@s*eZdZdZdZedZejddZ dS)UnicodeVarcharTest)rHTcCs |dS)N)r^)r7r@r@rAtest_empty_strings_varcharsz-UnicodeVarcharTest.test_empty_strings_varcharN) rCrDrEra __backend__rrOrZempty_strings_varcharrfr@r@r@rArdsrdc@s(eZdZdZdZeZejddZ dS)UnicodeTextTest)rHrVTcCs |dS)N)r^)r7r@r@rAtest_empty_strings_textsz'UnicodeTextTest.test_empty_strings_textN) rCrDrErargrrOrZempty_strings_textrir@r@r@rArhsrhc@sdeZdZdZdZeddZeddZddZ d d Z d d Z d dZ ddZ ddZddZdS)TextTest)rVTcCs tjjjS)N)rrrIrJ)r7r@r@rAr3szTextTest.supports_whereclausec Cs$td|tdtdddtdtdS)N text_tablerLT)rMrN text_data)rrrr)rPr*r@r@rArQs  zTextTest.define_tablescCsF|jj}tj|dditjt|jjg }t |ddS)Nrlz some text)z some text) rRrkrr,r2r.rr5rlrTr)r7rkr?r@r@rAtest_text_roundtripszTextTest.test_text_roundtripcCsF|jj}tj|dditjt|jjg }t |ddS)Nrlr])r]) rRrkrr,r2r.rr5rlrTr)r7rkr?r@r@rAtest_text_empty_stringssz TextTest.test_text_empty_stringscCs|tdgdgdS)Nz some text)rBr)r7r@r@rAr_szTextTest.test_literalcCs"|ttdgtdgdS)Nuréve🐍 illé)rBrr r#)r7r@r@rAr`szTextTest.test_literal_non_asciicCsd}|t|g|gdS)Nz&some 'text' hey "hi there" that's text)rBr)r7rSr@r@rAtest_literal_quotingszTextTest.test_literal_quotingcCsd}|t|g|gdS)Nz$backslash one \ backslash two \\ end)rBr)r7rSr@r@rAtest_literal_backslashessz!TextTest.test_literal_backslashescCsd}|t|g|gdS)Nzpercent % signs %% percent)rBr)r7rSr@r@rAtest_literal_percentsignssz"TextTest.test_literal_percentsignsN)rCrDrErargrbr3rcrQrmrnr_r`rorprqr@r@r@rArjs  rjc@s>eZdZdZejddZddZddZdd Z d d Z d S) StringTestTcCs4t}td|tdt}|tj|tjdS)Nfooone)rrrrr+rr,Zdrop)r7r*rsr@r@rAtest_nolength_strings zStringTest.test_nolength_stringcCs|tddgdgdS)N(z some text)rBr)r7r@r@rAr_szStringTest.test_literalcCs&|tdtdgtdgdS)Nrvuréve🐍 illé)rBrr r#)r7r@r@rAr`sz!StringTest.test_literal_non_asciicCsd}|td|g|gdS)Nz&some 'text' hey "hi there" that's textrv)rBr)r7rSr@r@rAroszStringTest.test_literal_quotingcCsd}|td|g|gdS)Nz$backslash one \ backslash two \\ endrv)rBr)r7rSr@r@rArpsz#StringTest.test_literal_backslashesN) rCrDrErgrZunbounded_varcharrur_r`rorpr@r@r@rArrs rrc@s<eZdZdZeddZddZddZej j dd Z dS) _DateFixtureNc Cs&td|tdtdddtd|jdS)N date_tablerLT)rMrN date_data)rrrrO)rPr*r@r@rArQs  z_DateFixture.define_tablescCsl|jj}tj|d|jitjt|jj g }|j pD|j}t ||ft |dt|shtdS)Nryr)rRrxrr,r2r.rSrr5ryrTcomparerrUtyper6)r7rxr?rzr@r@rArW s   z_DateFixture.test_round_tripcCsF|jj}tj|dditjt|jjg }t |ddS)Nry)N) rRrxrr,r2r.rr5ryrTr)r7rxr?r@r@rA test_nullsz_DateFixture.test_nullcCs&|jp |j}||j|jg|gdS)N)rzrSrBrO)r7rzr@r@rAr_s z_DateFixture.test_literal) rCrDrErzrcrQrWr|rrequiresZdatetime_literalsr_r@r@r@rArws   rwc@s,eZdZdZdZeZeddddddZd S) DateTimeTest)datetimeTi  9N) rCrDrErargrrOrrSr@r@r@rAr~$sr~c @s.eZdZdZdZeZeddddddd Zd S) DateTimeMicrosecondsTest)Zdatetime_microsecondsTirrrrriN) rCrDrErargrrOrrSr@r@r@rAr+src @s.eZdZdZdZeZeddddddd Zd S) TimestampMicrosecondsTest)Ztimestamp_microsecondsTirrrrriN) rCrDrErargrrOrrSr@r@r@rAr2src@s&eZdZdZdZeZedddZ dS)TimeTest)timeTrrrN) rCrDrErargrrOrrrSr@r@r@rAr9src@s(eZdZdZdZeZeddddZ dS)TimeMicrosecondsTest)Ztime_microsecondsTrrriN) rCrDrErargrrOrrrSr@r@r@rAr@src@s&eZdZdZdZeZedddZ dS)DateTest)dateTirrN) rCrDrErargrrOrrrSr@r@r@rArGsrc@s:eZdZdZdZeZeddddddZe dddZ d S) DateTimeCoercedToDateTimeTest)rZdate_coerces_from_datetimeTirrrrrN) rCrDrErargrrOrrSrrzr@r@r@rArNs rc@s,eZdZdZdZeZeddddddZdS) DateTimeHistoricTest)Zdatetime_historicTi: r4#N) rCrDrErargrrOrrSr@r@r@rArVsrc@s&eZdZdZdZeZedddZ dS)DateHistoricTest)Z date_historicTiN) rCrDrErargrrOrrrSr@r@r@rAr]src@s.eZdZdZddZddZejddZdS) IntegerTestTcCs|tdgdgdS)N)rBr)r7r@r@rAr_gszIntegerTest.test_literalcCs|tddS)Nl') _round_tripr )r7r@r@rA test_huge_intjszIntegerTest.test_huge_intc Cs|j}td|tdtdddtd|}|tjtj|d|itjt |j j g }t ||ftjrt|dtstnt|dttfstdS)NZ integer_tablerLT)rMrN integer_datar)r*rrrZ create_allrr,r2r.rr5rrTrr Zpy3krUintr6long)r7rOrSr*Z int_tabler?r@r@rArms    zIntegerTest._round_tripN) rCrDrErgr_rrrFrr@r@r@rArdsrc@s2eZdZdZedejd+ddZedddZedd d Z d d Z ej j d dZ ddZddZej jddZej jddZej jddZddZddZej jedddZeddd Zej jd!d"Zej jd#d$Zej jd%d&Zej jd'd(Z ej j!d)d*Z"dS), NumericTestTz/.*does \*not\* support Decimal objects nativelyNFc s|j}td|td|}||dd|Ddd|D}t|}rtfdd|D}tfd d|D}t|||rtd d|Dd d|DdS) Nr%r&cSsg|] }d|iqS)r&r@)rXr&r@r@rArZsz(NumericTest._do_test..cSsh|] }|dqS)rr@)rXr?r@r@rA sz'NumericTest._do_test..c3s|]}|VqdS)Nr@)rXr&)r;r@rA sz'NumericTest._do_test..c3s|]}|VqdS)Nr@)rXr&)r;r@rArscSsg|] }t|qSr@)str)rXr&r@r@rArZs) r*rrr+r.r2rsetr) r7r8r9r:r; check_scaler*r%resultr@)r;rA_do_tests zNumericTest._do_testcCs,|tddddtdgtdgdS)Nr) precisionscaleg9/@z15.7563)rBrdecimalDecimal)r7r@r@rAtest_render_literal_numerics  z'NumericTest.test_render_literal_numericcCs(|tdddddtdgdgdS)NrrF)rr asdecimalg9/@z15.7563)rBrrr)r7r@r@rA#test_render_literal_numeric_asfloats  z/NumericTest.test_render_literal_numeric_asfloatcCs*|jtddtdgdgddddS)Nrg9/@z15.7563cSs|dk rt|dpdS)Nr)round)nr@r@rAz7NumericTest.test_render_literal_float..)r;)rBrrr)r7r@r@rAtest_render_literal_floats  z%NumericTest.test_render_literal_floatcCs2|jtdddddtdgtdgdddS)NT)Zdecimal_return_scalerg6ߗD/@z 15.7563827)r)rrrr)r7r@r@rAtest_float_custom_scales    z#NumericTest.test_float_custom_scalecCs,|tddddtdgtdgdS)Nrr)rrg9/@z15.7563)rrrr)r7r@r@rAtest_numeric_as_decimals  z#NumericTest.test_numeric_as_decimalcCs(|tdddddtdgdgdS)NrrF)rrrg9/@z15.7563)rrrr)r7r@r@rAtest_numeric_as_floats  z!NumericTest.test_numeric_as_floatcCs|tddddgdgdS)Nrr)rr)rr)r7r@r@rAtest_numeric_null_as_decimalsz(NumericTest.test_numeric_null_as_decimalcCs |tdddddgdgdS)NrrF)rrr)rr)r7r@r@rAtest_numeric_null_as_floatsz&NumericTest.test_numeric_null_as_floatcCs0|tddddtddgtddgdS)NrT)rrg9/@z15.7563)rrrr)r7r@r@rAtest_float_as_decimals z!NumericTest.test_float_as_decimalcCs,|jtdddtdgdgddddS)Nr)rg9/@z15.7563cSs|dk rt|dpdS)Nr)r)rr@r@rArrz1NumericTest.test_float_as_float..)r;)rrrr)r7r@r@rAtest_float_as_floats  zNumericTest.test_float_as_floatcCs(d}tjtt|g}t||dS)Ng9/@)rr,scalarrrr)r7exprvalr@r@rAtest_float_coerce_round_tripsz(NumericTest.test_float_coerce_round_tripcCs.td}tjtt|g}t||dS)Nz15.7563)rrrr,rrrr)r7rrr@r@rAtest_decimal_coerce_round_trips z*NumericTest.test_decimal_coerce_round_tripcCs6td}tjtt|tddg}t||dS)Nz15.7563rr) rrrr,rrr rr)r7rrr@r@rA%test_decimal_coerce_round_trip_w_casts z1NumericTest.test_decimal_coerce_round_trip_w_castcCs:ttdtdtdg}|tddd||dS)Nz54.234246451650z0.004354z900.0rr)rr)rrrrr)r7numbersr@r@rAtest_precision_decimals z"NumericTest.test_precision_decimalcCsttdtdtdtdtdtdtdtdtd td td td g }|td dd||dS)ztest exceedingly small decimals. Decimal reports values with E notation when the exponent is greater than 6. z1E-2z1E-3z1E-4z1E-5z1E-6z1E-7z1E-8z0.01000005940696z0.00000005940696z0.00000000000696z0.70000000000696z696E-12r)rrN)rrrrr)r7rr@r@rAtest_enotation_decimals z"NumericTest.test_enotation_decimalcCsBttdtdtdtdg}|tddd||dS) z*test exceedingly large decimals. z4E+8z5748E+15z 1.521E+15z00000000000000.1E+12r)rrN)rrrrr)r7rr@r@rAtest_enotation_decimal_large#s z(NumericTest.test_enotation_decimal_largecCs:ttdtdtdg}|tddd||dS)Nz31943874831932418390.01z319438950232418390.273596z87673.594069654243&r)rr)rrrrr)r7rr@r@rAtest_many_significant_digits3s z(NumericTest.test_many_significant_digitscCs.ttdg}|jtddd||dddS)Nz1.000rr )rrT)r)rrrrr)r7rr@r@rAtest_numeric_no_decimal>sz#NumericTest.test_numeric_no_decimal)NF)#rCrDrErgrZ emits_warningrFrrrrr}Zprecision_generic_float_typerrrZfetch_null_from_numericrrZfloats_to_four_decimalsrrrZimplicit_decimal_bindsrrZprecision_numerics_generalrZ"precision_numerics_enotation_largerrZ*precision_numerics_many_significant_digitsrZ-precision_numerics_retains_significant_digitsrr@r@r@rArs.    rc@s<eZdZdZeddZddZddZdd Zd d Z d S) BooleanTestTc Cs2td|tdtdddtdttdtdddS) N boolean_tablerLTF)rMZ autoincrementr=unconstrained_value)Zcreate_constraint)rrrr )rPr*r@r@rArQIs zBooleanTest.define_tablescCs|tddgddgdS)NTF)rBr )r7r@r@rAtest_render_literal_boolSsz$BooleanTest.test_render_literal_boolcCsb|jj}tj|ddddtjt|jj|jj g }t |dt |dt s^tdS)NrTF)rLr=r)TFr)rRrrr,r2r.rr5r=rrTrrUboolr6)r7rr?r@r@rArWVs zBooleanTest.test_round_tripcCsP|jj}tj|ddddtjt|jj|jj g }t |ddS)Nr)rLr=r)NN) rRrrr,r2r.rr5r=rrTr)r7rr?r@r@rAr|gszBooleanTest.test_nullc Cs|jj}tj}||ddddddddgt|t |j j g |j j dt|t |j j g |j jdt|t |j j g |j j dt|t |j j g |j jdWdQRXdS)NrT)rLr=rrF)rRrrr,r-r2r.rrrr5rLr4r=r)r7rr<r@r@rAtest_whereclausews0    zBooleanTest.test_whereclauseN) rCrDrErgrcrQrrWr|rr@r@r@rArFs  rc@s0eZdZdZdZeZdddZddddZdd d gd d d gddiddigdZ d d d gZ dddddddgdddiidiZ ddd d!id"Z e d#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-d.Zd/d0ZdHd1d2Zd3d4Zejjd5d6Zd7d8Zd9d:Zd;d<Zd=d>Zd?d@ZdAdBZ dCdDZ!dEdFZ"dGS)IJSONTest)Z json_typeTZvalue1Zvalue2)key1key2zvalue ' three ')z Key 'One'zkey twoz key threerrr rtZtwoZthreeZfourZfivesixZseven)rrkey3nestedbd)ar5fh)egelem3elem4Zelem5)Zelem1elem2rz some valuersZbar)rrr5c CsDtd|tdtddtdtdddtd |jtd |jdd dS) N data_tablerLT)rMnameF)ZnullablerSnulldata)Z none_as_null)rrrrrO)rPr*r@r@rArQs  zJSONTest.define_tablescCs||jdS)N)_test_round_tripdata1)r7r@r@rAtest_round_trip_data1szJSONTest.test_round_trip_data1cCsJ|jj}tj|d|dtjt|jjg }t ||fdS)NZrow1)rrS) rRrrr,r2r.rr5rSrTr)r7Z data_elementrr?r@r@rArs zJSONTest._test_round_tripc Cs|jjjd}tjb}||jjdddt| t |jjjj g | tdt| t |gdWdQRXdS)Nrr1)rrS)rRrr5rr,r-r2r.rrrrr4is_r)r7colr<r@r@rA test_round_trip_none_as_sql_nulls z)JSONTest.test_round_trip_none_as_sql_nullc Cs|jjjd}tjf}||jjdtj dt | t |jjjj gt|tdkdt | t |gdWdQRXdS)NrSr)rrSr)rRrr5rr,r-r2r.rNULLrrrrr4r r)r7rr<r@r@rA&test_round_trip_json_null_as_json_nulls  z/JSONTest.test_round_trip_json_null_as_json_nullc Cs|jjjd}tjd}||jjdddt| t |jjjj g t |tdkdt| t |gdWdQRXdS)NrSr)rrSr)rRrr5rr,r-r2r.rrrrr4r r)r7rr<r@r@rA!test_round_trip_none_as_json_nulls z*JSONTest.test_round_trip_none_as_json_nullc CsVtj|jjd|jdd|jdd|jdd|j dd|j dd|j dgdS)Nr)rrSr2r3r4r5r6) rr,r2rRrr.rdata2data3data4data5data6)r7r@r@rA_criteria_fixtures      zJSONTest._criteria_fixturec Csv|tjZ}t|jjjjg |}t | |||rht |j tjddid}t | ||WdQRXdS)Nr'T)r))rrr,r-rrRrr5rr4rrrr0)r7Zcritexpectedr_r<r>Z literal_sqlr@r@rA_test_index_criteria!s zJSONTest._test_index_criteriacCsJ|jjjj}|jjjd}|t|dddgt|dtdkddS)NrSrrrzkey twoz"value2") rRrr5rrr Zin_r r)r7rrr@r@rAtest_crit_spaces_in_key1s z JSONTest.test_crit_spaces_in_keycCsB|jjjj}|jjjd}|t|dkt|dtdkddS)NrSrrz"two")rRrr5rrr r r)r7rrr@r@rAtest_crit_simple_int@s zJSONTest.test_crit_simple_intcCs,|jjjd}|t|dtdkddS)NrS)rrrz"seven"r)rRrr5rr r)r7rr@r@rAtest_crit_mixed_pathLszJSONTest.test_crit_mixed_pathcCs,|jjjd}|t|dtdkddS)NrS)rrrrz"elem5"r)rRrr5rr r)r7rr@r@rAtest_crit_string_pathRs  zJSONTest.test_crit_string_pathcCsB|jjjj}|jjjd}|t|dkt|dtdkddS)NrSrrz "some value")rRrr5rrr r r)r7rrr@r@rAtest_crit_against_string_basicZs z'JSONTest.test_crit_against_string_basiccCsL|jjjj}|jjjd}|jt|dkt|dttdt kddddS)NrSrrz some valueF)r_) rRrr5rrr r rrr)r7rrr@r@rA$test_crit_against_string_coerce_typebs z-JSONTest.test_crit_against_string_coerce_typecCsB|jjjj}|jjjd}|t|dkt|dtdkddS)NrSrr5)rRrr5rrr r r)r7rrr@r@rAtest_crit_against_int_basicos z$JSONTest.test_crit_against_int_basiccCsL|jjjj}|jjjd}|jt|dkt|dttdt kddddS)NrSrrrF)r_) rRrr5rrr r rrr)r7rrr@r@rA!test_crit_against_int_coerce_typews   z*JSONTest.test_crit_against_int_coerce_typec Cstjz}||jjdtdtdddtdiidt | t |jjj j gtdtdddtdiiWdQRXdS)Nruréve🐍 illérSk1u drôl🐍e)rrS)rr,r-r2rRrr.r r#rrrr5rS)r7r<r@r@rAtest_unicode_round_trips  z JSONTest.test_unicode_round_tripcst}Gfddd|}ttj}|dddd}|||||ddddgt|t j j j j tt j j j jtj j j jdkdt|t j j j j tt j j j jtj j j jdkddS)NcseZdZjjZdS)z.JSONTest.test_eval_none_flag_orm..DataN)rCrDrErRrZ __table__r@)r7r@rADatasrd1)rrSrd2)rN)r!r"rr,addZcommitZbulk_insert_mappingsrqueryr rRrr5rSrrfilterrrT)r7ZBasersrr@)r7rAtest_eval_none_flag_orms(  z JSONTest.test_eval_none_flag_ormN)T)#rCrDrErargrrOrrrrrrrcrQrrrrrrrrrrZjson_array_indexesrrrrrrrrr r@r@r@rArsB        r)rdrhrrr~rjrrrrrrrrrrrr)?rrr]rrZ assertionsrrZschemarrr r r r rrrrrrrrrrrrrrrrrrr Zext.declarativer!Zormr"r#objectr$rGZ TablesTestrdrhrjZTestBaserrrwr~rrrrrrrrrrrr__all__r@r@r@rAsr                                *A  6'#@\PK}]r"2#__pycache__/__init__.cpython-37.pycnu[B 4]f@sTddlTddlTddlTddlTddlTddlTddlTddlTddlTddl TdS))*N) Ztest_cteZtest_ddlZ test_dialectZ test_insertZtest_reflectionZ test_resultsZ test_selectZ test_sequenceZ test_typesZtest_update_deleterrT/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/__init__.pysPK}]죔!!'__pycache__/test_dialect.cpython-37.pycnu[B 4]@sddlmZddlmZddlmZddlmZddlmZddlmZddlmZddlm Z d d lm Z d d lm Z d d lm Z d d lm Z d dlmZd dlmZGdddejZGdddejZGdddejZdS)) assert_raises)config)eq_)fixtures)provide_metadata) requirements)Column)Table)exc)Integer)literal_column)select)String)compatc@s:eZdZdZdZdZeddZej ddZ dd Z d S) ExceptionTestzTest basic exception wrapping. DBAPIs vary a lot in exception behavior so to actually anticipate specific exceptions from real round trips, we need to be conservative. eachTc Cs(td|tdtdddtdtddS)N manual_pkidTF) primary_key autoincrementdata2)r rr r)clsmetadatarX/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_dialect.py define_tabless zExceptionTest.define_tablesc CsdtjP}|}||jjdddtt j |j|jjddd| WdQRXdS)Nd1)rr) rdbconnectbeginexecutetablesrinsertrr ZIntegrityErrorrollback)selfconntransrrrtest_integrity_error(s   z"ExceptionTest.test_integrity_errorc Cstj}y |ttdgds*tWnBtjk rn}z"t |}t |j t |ks^tWdd}~XYnXt j rt |t stnt |t stWdQRXdS)NuméilF)rr r!r#rr AssertionErrorr Z DBAPIErrorstrorigrZpy2k isinstance)r'r(errerr_strrrrtest_exception_with_non_ascii;s  (z+ExceptionTest.test_exception_with_non_asciiN) __name__ __module__ __qualname____doc__ run_deletes __backend__ classmethodrrZ$duplicate_key_raises_integrity_errorr*r1rrrrrs  rc@s<eZdZdZdZdZeddZddZdd Z d d Z d S) AutocommitTestr) autocommitTc Cs,td|tdtdddtdtddddS) N some_tablerTF)rrrr)Ztest_needs_acid)r rr r)rrrrrrZs  zAutocommitTest.define_tablescCsf|}||jjddd|t|t|jjj j g|rJdnd||jj dS)Nrz some data)rr) r"r#r$r;r%r&rscalarrcrdelete)r'r(r:r)rrr_test_conn_autocommitsdsz%AutocommitTest._test_conn_autocommitscCs:tj}|jdd}||d|||ddS)NZ AUTOCOMMIT)Zisolation_levelTF)rr r!Zexecution_optionsr?Z invalidate)r'r(Zc2rrrtest_autocommit_onrs    z!AutocommitTest.test_autocommit_oncCstj}||ddS)NF)rr r!r?)r'r(rrrtest_autocommit_offys z"AutocommitTest.test_autocommit_offN) r2r3r4r6 __requires__r7r8rr?r@rArrrrr9Rs r9c@seZdZeddZdS) EscapingTestc Cs|j}td|tdtd}|tjtj}|| t dd|| t ddt | t |jjg|jjtdkdt | t |jjg|jjtdkdWd QRXd S) z|test that the DBAPI accommodates for escaped / nonescaped percent signs in a way that matches the compiler trrz some % value)rzsome %% other valuez'some % value'z'some %% other value'N)rr rrcreaterr r"r#r%dictrr<rr=rwherer )r'mrDr(rrrtest_percent_sign_round_trips   z)EscapingTest.test_percent_sign_round_tripN)r2r3r4rrIrrrrrC~srCN)rrrrrrZschemarr r r r rrutilrZ TablesTestrr9ZTestBaserCrrrrs              ?,PK}].ZZ(__pycache__/test_sequence.cpython-37.pycnu[B 4]5@sddlmZddlmZddlmZddlmZddlmZddlmZddlm Z dd lm Z dd lmZdd lm Z dd lm Z dd lm Z GdddejZGddde jejZGdddejZdS))config)fixtures)eq_) requirements)Column)Table)Integer)MetaData)schema)Sequence)String)testingc@sReZdZdZdZdZeddZddZdd Z d d Z e j d d Z ddZdS) SequenceTest) sequencesTZeachc CsXtd|tdttdddtdtdtd|tdttddd ddtdtddS) Nseq_pkidZ tab_id_seqT)Z primary_keydata2 seq_opt_pk)optional)rrr r r )clsmetadatarY/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_sequence.py define_tabless zSequenceTest.define_tablescCs.tjj|jjdd||jjtjdS)Nz some data)r)rdbexecutetablesrinsert_assert_round_trip)selfrrrtest_insert_roundtrip*sz"SequenceTest.test_insert_roundtripcCs*tjj|jjdd}t|jdgdS)Nz some data)r)rrrrrrrinserted_primary_key)r!rrrrtest_insert_lastrowid.sz"SequenceTest.test_insert_lastrowidcCs$tj|jjjjj}t|ddS)Nr#) rrrrrcrdefaultr)r!r%rrrtest_nextval_direct2sz SequenceTest.test_nextval_directcCs*tjj|jjdd}t|jdgdS)Nz some data)rr#)rrrrrrrr$)r!r%rrrtest_optional_seq6szSequenceTest.test_optional_seqcCs ||}t|ddS)N)r#z some data)rselectfirstr)r!tableZconnrowrrrr =szSequenceTest._assert_round_tripN)__name__ __module__ __qualname__ __requires__ __backend__Zrun_create_tables classmethodrr"r&r)rZsequences_optionalr*r rrrrrs rc@seZdZdZdZddZdS)SequenceCompilerTest)rTcCsrtdttdttdtdt}|jdd}tjj j dtjj d td}|j |d|fd tjj d dS) NxyZy_seqq)r8) statementdialectz#INSERT INTO x (y, q) VALUES (%s, 5)T)Z literal_bindsr;) rr rr r rvaluesrrr;Zstatement_compilerZvisit_sequenceZassert_compile)r!r-stmtZ seq_nextvalrrr!test_literal_binds_inline_compileFs  z6SequenceCompilerTest.test_literal_binds_inline_compileN)r/r0r1r2r3r>rrrrr5Bsr5c@sdeZdZdZdZddZejjddZ ddZ ejjd d Z ejjd d Z ejjd dZ dS)HasSequenceTest)rTc CsPtd}tjt|zttjjtjddWdtjt |XdS)N user_id_seqT) r rrrr CreateSequencerr; has_sequence DropSequence)r!s1rrrtest_has_sequence_s z!HasSequenceTest.test_has_sequencec Cs\tdtjd}tjt|z"ttjj j tjdtjddWdtjt |XdS)Nr@)r T) r r test_schemarrrr rArr;rBrC)r!rDrrrtest_has_sequence_schemajs z(HasSequenceTest.test_has_sequence_schemacCsttjjtjdddS)Nr@F)rrrr;rB)r!rrrtest_has_sequence_negxsz%HasSequenceTest.test_has_sequence_negcCs"ttjjjtjdtjdddS)Nr@)r F)rrrr;rBrrF)r!rrrtest_has_sequence_schemas_neg{sz-HasSequenceTest.test_has_sequence_schemas_negc CsVtd}tjt|z"ttjjjtjdt j ddWdtjt |XdS)Nr@)r F) r rrrr rArr;rBrrFrC)r!rDrrr'test_has_sequence_default_not_in_remotes z7HasSequenceTest.test_has_sequence_default_not_in_remotec CsVtdtjd}tjt|zttjj tjddWdtjt |XdS)Nr@)r F) r rrFrrrr rArr;rBrC)r!rDrrr'test_has_sequence_remote_not_in_defaults z7HasSequenceTest.test_has_sequence_remote_not_in_defaultN)r/r0r1r2r3rErrequiresZschemasrGrHrIrJrKrrrrr?[s  r?N)rrZ assertionsrrr rrr r r r rZ TablesTestrZAssertsCompiledSQLZTestBaser5r?rrrrs            3PK}]&<"j  #__pycache__/test_ddl.cpython-37.pycnu[B 4]6 @sddlmZddlmZddlmZddlmZddlmZddlmZddlmZdd lm Z dd lm Z dd lm Z dd lm Z Gd ddej ZdZdS))config)fixtures)util)eq_) requirements)Column)inspect)Integer)schema)String)Tablec@seZdZdZdddZddZddZeje j d d Z eje j d d Z ej e j d dZeje j ddZeje j ddZeje j ddZdS) TableDDLTestTNc Cs*td|jtdtdddtdtd|dS) N test_tableidTF) primary_key autoincrementdata2)r )r metadatarr r )selfr rT/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_ddl.py_simple_fixtures  zTableDDLTest._simple_fixturec Cs&td|jtdtdddtdtdS)NZ _test_tablerTF)rr_datar)r rrr r )rrrr_underscore_fixtures z TableDDLTest._underscore_fixturec CsJtj6}||d||}t|dWdQRXdS)N)z some data) rdbbeginexecuteinsertvaluesselectrfirst)rtableZconnresultrrr_simple_roundtrip"s zTableDDLTest._simple_roundtripcCs&|}|jtjdd||dS)NF) checkfirst)rcreaterrr&)rr$rrrtest_create_table(szTableDDLTest.test_create_tablecCs,|jtjd}|jtjdd||dS)N)r F)r')rrZ test_schemar(rr&)rr$rrrtest_create_table_schema/sz%TableDDLTest.test_create_table_schemacCs,|}|jtjdd|jtjdddS)NF)r')rr(rrZdrop)rr$rrrtest_drop_table6szTableDDLTest.test_drop_tablecCs&|}|jtjdd||dS)NF)r')rr(rrr&)rr$rrrtest_underscore_names=sz"TableDDLTest.test_underscore_namescCsN|}|jtjddd|_tjt|tt tj dddidS)NF)r'z a commentrtext) rr(rrcommentrr SetTableCommentrr get_table_comment)rr$rrrtest_add_table_commentDsz#TableDDLTest.test_add_table_commentcCs`|}|jtjddd|_tjt|tjt|t t tj dddidS)NF)r'z a commentrr-) rr(rrr.rr r/ZDropTableCommentrr r0)rr$rrrtest_drop_table_commentPs z$TableDDLTest.test_drop_table_comment)N)__name__ __module__ __qualname__Z __backend__rrr&rZ create_tablerZprovide_metadatar)r*Z drop_tabler+r,Zcomment_reflectionr1r2rrrrrs   r)rN)rrrZ assertionsrrrr r r r r ZTestBaser__all__rrrrs           MPK}]9[S""&__pycache__/test_insert.cpython-37.pycnu[B 4]r%@sddlmZddlmZddlmZddlmZddlmZddlmZddlm Z dd lm Z dd lm Z dd lm Z dd lm Z dd lmZGdddejZGdddejZGdddejZdZdS))config)engines)fixtures)eq_) requirements)Column)Table)Integer)literal)literal_column)select)Stringc@sReZdZdZdZdZddiZeddZdd Z d d Z d d Z e j ddZdS) LastrowidTesteachT)Zimplements_get_lastrowidautoincrement_insertimplicit_returningFc CsLtd|tdtdddtdtdtd|tdtddd tdtddS) N autoinc_pkidT) primary_keytest_needs_autoincrementdata2 manual_pkF)r autoincrement)rrr r)clsmetadatarW/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_insert.py define_tabless zLastrowidTest.define_tablescCs*||}t|tjjjdfdS)Nz some data)executer firstrrdbdialectdefault_sequence_base)selftableconnrowrrr_assert_round_trip*sz LastrowidTest._assert_round_tripcCs.tjj|jjdd||jjtjdS)Nz some data)r)rr"r tablesrinsertr))r%rrrtest_autoincrement_on_insert.sz*LastrowidTest.test_autoincrement_on_insertcCsDtjj|jjdd}tjt|jjjj g}t |j |gdS)Nz some data)r) rr"r r*rr+scalarr crrinserted_primary_key)r%rpkrrrtest_last_inserted_id3sz#LastrowidTest.test_last_inserted_idcCsFtjj|jjdd}|j}tjt|jjj j g}t ||dS)Nz some data)r) rr"r r*rr+ lastrowidr-r r.rr)r%r0r3r1rrrtest_native_lastrowid_autoinc?s z+LastrowidTest.test_native_lastrowid_autoincN)__name__ __module__ __qualname__ run_deletes __backend__ __requires____engine_options__ classmethodrr)r,r2rZdbapi_lastrowidr4rrrrrs  rc@s|eZdZdZdZeddZddZej ddZ ej d d Z ej d d Zej d dZej ddZej ddZdS)InsertBehaviorTestrTc Cstd|tdtdddtdtdtd|tdtddd tdtdtd |tdtdddtdtdtd td d tdttdtdtdd dS)NrrT)rrrrrF)rrincludes_defaultsx)defaulty2)type_r)rrr rr r )rrrrrrMs,    z InsertBehaviorTest.define_tablescCsbtjjrtjddid}ntj}|j|jj dd}|j s@t |j rJt |jsTt |jr^t dS)NrF)optionsz some data)r)r returningenabledrZtesting_enginerr"r r*rr+ _soft_closedAssertionErrorclosed is_insert returns_rows)r%enginer0rrrtest_autoclose_on_insertls   z+InsertBehaviorTest.test_autoclose_on_insertcCsDtjj|jjdd}|js"t|jr,t|j s6t|j r@tdS)Nz some data)r) rr"r r*rr+rHrIrJrKrL)r%r0rrr+test_autoclose_on_insert_implicit_returningzs    z>InsertBehaviorTest.test_autoclose_on_insert_implicit_returningcCsbtj|jj}|jst|jr(ttj|jj |jjj j dk}t |s^tdS)N)rr"r r*rr+rHrIrJr wherer.rlenfetchall)r%r0rrrtest_empty_inserts   z$InsertBehaviorTest.test_empty_insertc Cs|jj}|jj}tj|tdddtdddtdddgtj|dt |j j g |j j ddg}t|jdgtjt |j j g|j j }t|d d gdS) Ndata1)rrrdata2r data3)r)rV)rW)r*rrrr"r r+dict from_selectr r.rrPin_rr/order_byrR)r% src_table dest_tableresultrrrtest_insert_from_select_autoincs   z2InsertBehaviorTest.test_insert_from_select_autoincc Cs|jj}|jj}tj|dt|j j g |j j ddg}t |jdgtjt|j j g|j j }t |gdS)N)rrVrW)r*rrrr"r r+rYr r.rrPrZrr/r[rR)r%r\r]r^rrr'test_insert_from_select_autoinc_no_rowssz:InsertBehaviorTest.test_insert_from_select_autoinc_no_rowsc Cs|jj}tj|tdddtdddtdddgtj|jdd dt|j j d |j j g |j j ddgttjt|j j g|j j d d d d d gdS)NrTrU)rrrrVr rWT)inliner@)rU)rV)rW)r*rrr"r r+rXrYr r.rrrPrZrr[rR)r%r&rrrtest_insert_from_selects   z*InsertBehaviorTest.test_insert_from_selectc Cs|jj}tj|tdddtdddtdddgtj|jdd dt|j j d |j j g |j j ddgttjt|g|j j |j j d d d ddgdS)NrTrU)rrrrVr rWT)rar@)rTrUr@)rrVr@rc)rVr@rc)r rWr@rc)rWr@rc)r*r>rr"r r+rXrYr r.rrrPrZrr[rR)r%r&rrr%test_insert_from_select_with_defaultss&   z8InsertBehaviorTest.test_insert_from_select_with_defaultsN)r5r6r7r8r9r<rrNrrFrOZ empty_insertsrSZinsert_from_selectr_r`rbrfrrrrr=Is  r=c@sZeZdZdZdZdZddiZddZeddZ e j d d Z d d Z d dZddZdS) ReturningTestr)rFrTrcCs*||}t|tjjjdfdS)Nz some data)r r r!rrr"r#r$)r%r&r'r(rrrr) sz ReturningTest._assert_round_tripc Cs(td|tdtdddtdtddS)NrrT)rrrr)rrr r)rrrrrrs  zReturningTest.define_tablescCsZtj}|jj}|j||jjdd}| d}tj t |jjg}t ||dS)Nz some data)r) rr"r*rr r+rFr.rr!r-r r)r%rMr&r0r1 fetched_pkrrr%test_explicit_returning_pk_autocommits z3ReturningTest.test_explicit_returning_pk_autocommitc Csntj}|jj}|.}|j||jj dd}| d}WdQRXtj t |jj g}t ||dS)Nz some data)rrh)rr"r*rbeginr r+rFr.rr!r-r r)r%rMr&r'r0r1rirrr(test_explicit_returning_pk_no_autocommit$s z6ReturningTest.test_explicit_returning_pk_no_autocommitcCs.tjj|jjdd||jjtjdS)Nz some data)r)rr"r r*rr+r))r%rrr/test_autoincrement_on_insert_implicit_returning/sz=ReturningTest.test_autoincrement_on_insert_implicit_returningcCsDtjj|jjdd}tjt|jjjj g}t |j |gdS)Nz some data)r) rr"r r*rr+r-r r.rrr/)r%r0r1rrr(test_last_inserted_id_implicit_returning4sz6ReturningTest.test_last_inserted_id_implicit_returningN)r5r6r7Zrun_create_tablesr:r9r;r)r<rrZfetch_rows_post_commitrjrlrmrnrrrrrgs   rg)rr=rgN)rrrZ assertionsrrZschemarrr r r r rZ TablesTestrr=rg__all__rrrrs            :;:PK}]]e?$++'__pycache__/test_results.cpython-37.pycnu[B 4],@sddlZddlmZddlmZddlmZddlmZddlmZddlm Z dd lm Z d d lm Z d d lm Z d d lm Z d dlmZd dlmZd dlmZd dlmZd dlmZGdddejZGdddejZGdddejejZdS)N)config)engines)fixtures)eq_) requirements)Column)Table)DateTime)func)Integer)select)sql)String)testing)textc@sVeZdZdZeddZeddZddZdd Zd d Z e j d d Z ddZ dS) RowFetchTestTcCsDtd|tdtddtdtdtd|tdtddtdtdS) Nplain_pkidT) primary_keydata2 has_datestoday)r rr rr )clsmetadatarX/opt/alt/python37/lib64/python3.7/site-packages/sqlalchemy/testing/suite/test_results.py define_tabless  zRowFetchTest.define_tablesc Cs`tj|jjdddddddddgtj|jjdtdd d d d d d gdS) Nd1)rrrd2r Zd3i r)rr)rdbexecutetablesrinsertrdatetime)rrrr insert_data&s  zRowFetchTest.insert_datacCsFtj|jj|jjjj }t |ddt |dddS)Nrr rr!) rr%r&r'rrorder_bycrfirstr)selfrowrrrtest_via_string6s zRowFetchTest.test_via_stringcCsFtj|jj|jjjj }t |ddt |dddS)Nrr r!) rr%r&r'rrr+r,rr-r)r.r/rrr test_via_int>s zRowFetchTest.test_via_intcCsVtj|jj|jjjj }t ||jjjjdt ||jjjj ddS)Nr r!) rr%r&r'rrr+r,rr-rr)r.r/rrrtest_via_col_objectFs z RowFetchTest.test_via_col_objectcCs`tjt|jjjj|jjjjdg |jjjj }| }t | ddgt |ddS)Nr)r!r!)rr%r&rr'rr,rlabelr+rr-rkeys)r.resultr/rrrtest_row_with_dupe_namesNs z%RowFetchTest.test_row_with_dupe_namesc Csh|jj}t|djjg}t|jj|dg}t j | }t |dtdddddddS)ztest that a scalar select as a column is returned as such and that type conversion works OK. (this is half a SQLAlchemy Core test and half to catch database backends that may have unusual behavior with scalar selects.) xZ somelabelir#r$rN)r'rraliasr,rZ as_scalarrr3rr%r&r-rr))r.Z datetabless2r/rrrtest_row_w_scalar_select\s z%RowFetchTest.test_row_w_scalar_selectN)__name__ __module__ __qualname__ __backend__ classmethodrr*r0r1r2rZ%duplicate_names_in_cursor_descriptionr6r;rrrrrs  rc@s<eZdZdZdZdZeddZddZdd Z d d Z d S) PercentSchemaNamesTestztests using percent signs, spaces in table and column names. This is a very fringe use case, doesn't work for MySQL or PostgreSQL. the requirement, "percent_schema_names", is marked "skip" by default. )Zpercent_schema_namesTcCs@td|tdttdt|j_tdtdtd|j_dS)Nz percent%tablezpercent%zspaces % more spaces) r rr r' percent_tablertablecolumnlightweight_percent_table)rrrrrrysz$PercentSchemaNamesTest.define_tablescCsT|jj}x>ddddddddddddgD]}tj||q.W|dS)Nr#r$)zpercent%zspaces % more spaces )r'rBrr%r&r( _assert_table)r.rBparamsrrrtest_single_roundtripsz,PercentSchemaNamesTest.test_single_roundtripcCsV|jj}tj|dddtj|dddddddddg|dS)Nr#r$)zpercent%zspaces % more spacesrFrGrHrI)r'rBrr%r&r(rJ)r.rBrrrtest_executemany_roundtripsz1PercentSchemaNamesTest.test_executemany_roundtripc CsR|jj}|jj}x||||fD]}tttj| |j dddddgtttj| |j d ddg |j dddgtj| |j d}t|dd t|dd t||j dd t||j dd q&Wtj||j dd itttj| |j dd d ddgdS)Nzpercent%)r#r$)rFrG)rHrI)rGrHzspaces % more spacesrHrIr#r$)r#rN)rFrN)rHrN)rGrN)r'rBrEr8rlistrr%r&rr+r,whereZin_r-updatevalues)r.rBrErCr/rrrrJsD z$PercentSchemaNamesTest._assert_tableN) r<r=r>__doc__ __requires__r?r@rrLrMrJrrrrrAls  rAc@seZdZdZdZddZddZddZd d Zd d Z d dZ ddZ ddZ ddZ ddZddZddZddZddZdd Zd!d"Zejd#d$Zd%S)&ServerSideCursorsTest)server_side_cursorsTcCsd|jjjdkr|jS|jjjdkr8tdjj}t||S|jjjdkr\tdjj}t||SdSdS)Npsycopg2pymysqlzpymysql.cursorsZmysqldbzMySQLdb.cursorsF)engineZdialectZdrivername __import__ZcursorsZSSCursor isinstance)r.cursorZsscursorrrr_is_server_sides    z%ServerSideCursorsTest._is_server_sidecCstjd|id|_|jS)NrV)options)rZtesting_enginerY)r.rVrrr_fixtureszServerSideCursorsTest._fixturecCstj|jdS)N)rZtesting_reaperZ close_allrYZdispose)r.rrrtearDowns zServerSideCursorsTest.tearDowncCs(|d}|d}||js$tdS)NTzselect 1)r`r&r^r]AssertionError)r.rYr5rrrtest_global_strings  z(ServerSideCursorsTest.test_global_stringcCs,|d}|td}||js(tdS)NTzselect 1)r`r&rr^r]rb)r.rYr5rrrtest_global_texts z&ServerSideCursorsTest.test_global_textcCs.|d}|tdg}||js*tdS)NTr )r`r&rr^r]rb)r.rYr5rrrtest_global_exprs z&ServerSideCursorsTest.test_global_exprcCs,|d}|td}||jr(tdS)NFzselect 1)r`r&rr^r]rb)r.rYr5rrrtest_global_off_explicits z.ServerSideCursorsTest.test_global_off_explicitcCs:|d}tdgjdd}||}||js6tdS)NFr T)stream_results)r`rexecution_optionsr&r^r]rb)r.rYr9r5rrrtest_stmt_options  z&ServerSideCursorsTest.test_stmt_optioncCs4|d}|jddd}||js0tdS)NFT)rgzselect 1)r`connectrhr&r^r]rb)r.rYr5rrrtest_conn_options  z&ServerSideCursorsTest.test_conn_optioncCsF|d}tdgjdd}|jdd|}||jrBtdS)NFr T)rg)r`rrhrjr&r^r]rb)r.rYr9r5rrr&test_stmt_enabled_conn_option_disabled$s zrTr?r^r`rarcrdrerfrirkrlrmrorprqrrrsrZprovide_metadatar{rrrrrUs&     rU)r)rrrZ assertionsrrZschemarr r r r rrrrrZ TablesTestrrAZTestBaseZAssertsExecutionResultsrUrrrrs&               Xn PK}] ,,test_results.pynu[import datetime from .. import config from .. import engines from .. import fixtures from ..assertions import eq_ from ..config import requirements from ..schema import Column from ..schema import Table from ... import DateTime from ... import func from ... import Integer from ... import select from ... import sql from ... import String from ... import testing from ... import text class RowFetchTest(fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "plain_pk", metadata, Column("id", Integer, primary_key=True), Column("data", String(50)), ) Table( "has_dates", metadata, Column("id", Integer, primary_key=True), Column("today", DateTime), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.plain_pk.insert(), [ {"id": 1, "data": "d1"}, {"id": 2, "data": "d2"}, {"id": 3, "data": "d3"}, ], ) config.db.execute( cls.tables.has_dates.insert(), [{"id": 1, "today": datetime.datetime(2006, 5, 12, 12, 0, 0)}], ) def test_via_string(self): row = config.db.execute( self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id) ).first() eq_(row["id"], 1) eq_(row["data"], "d1") def test_via_int(self): row = config.db.execute( self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id) ).first() eq_(row[0], 1) eq_(row[1], "d1") def test_via_col_object(self): row = config.db.execute( self.tables.plain_pk.select().order_by(self.tables.plain_pk.c.id) ).first() eq_(row[self.tables.plain_pk.c.id], 1) eq_(row[self.tables.plain_pk.c.data], "d1") @requirements.duplicate_names_in_cursor_description def test_row_with_dupe_names(self): result = config.db.execute( select( [ self.tables.plain_pk.c.data, self.tables.plain_pk.c.data.label("data"), ] ).order_by(self.tables.plain_pk.c.id) ) row = result.first() eq_(result.keys(), ["data", "data"]) eq_(row, ("d1", "d1")) def test_row_w_scalar_select(self): """test that a scalar select as a column is returned as such and that type conversion works OK. (this is half a SQLAlchemy Core test and half to catch database backends that may have unusual behavior with scalar selects.) """ datetable = self.tables.has_dates s = select([datetable.alias("x").c.today]).as_scalar() s2 = select([datetable.c.id, s.label("somelabel")]) row = config.db.execute(s2).first() eq_(row["somelabel"], datetime.datetime(2006, 5, 12, 12, 0, 0)) class PercentSchemaNamesTest(fixtures.TablesTest): """tests using percent signs, spaces in table and column names. This is a very fringe use case, doesn't work for MySQL or PostgreSQL. the requirement, "percent_schema_names", is marked "skip" by default. """ __requires__ = ("percent_schema_names",) __backend__ = True @classmethod def define_tables(cls, metadata): cls.tables.percent_table = Table( "percent%table", metadata, Column("percent%", Integer), Column("spaces % more spaces", Integer), ) cls.tables.lightweight_percent_table = sql.table( "percent%table", sql.column("percent%"), sql.column("spaces % more spaces"), ) def test_single_roundtrip(self): percent_table = self.tables.percent_table for params in [ {"percent%": 5, "spaces % more spaces": 12}, {"percent%": 7, "spaces % more spaces": 11}, {"percent%": 9, "spaces % more spaces": 10}, {"percent%": 11, "spaces % more spaces": 9}, ]: config.db.execute(percent_table.insert(), params) self._assert_table() def test_executemany_roundtrip(self): percent_table = self.tables.percent_table config.db.execute( percent_table.insert(), {"percent%": 5, "spaces % more spaces": 12} ) config.db.execute( percent_table.insert(), [ {"percent%": 7, "spaces % more spaces": 11}, {"percent%": 9, "spaces % more spaces": 10}, {"percent%": 11, "spaces % more spaces": 9}, ], ) self._assert_table() def _assert_table(self): percent_table = self.tables.percent_table lightweight_percent_table = self.tables.lightweight_percent_table for table in ( percent_table, percent_table.alias(), lightweight_percent_table, lightweight_percent_table.alias(), ): eq_( list( config.db.execute( table.select().order_by(table.c["percent%"]) ) ), [(5, 12), (7, 11), (9, 10), (11, 9)], ) eq_( list( config.db.execute( table.select() .where(table.c["spaces % more spaces"].in_([9, 10])) .order_by(table.c["percent%"]) ) ), [(9, 10), (11, 9)], ) row = config.db.execute( table.select().order_by(table.c["percent%"]) ).first() eq_(row["percent%"], 5) eq_(row["spaces % more spaces"], 12) eq_(row[table.c["percent%"]], 5) eq_(row[table.c["spaces % more spaces"]], 12) config.db.execute( percent_table.update().values( {percent_table.c["spaces % more spaces"]: 15} ) ) eq_( list( config.db.execute( percent_table.select().order_by( percent_table.c["percent%"] ) ) ), [(5, 15), (7, 15), (9, 15), (11, 15)], ) class ServerSideCursorsTest( fixtures.TestBase, testing.AssertsExecutionResults ): __requires__ = ("server_side_cursors",) __backend__ = True def _is_server_side(self, cursor): if self.engine.dialect.driver == "psycopg2": return cursor.name elif self.engine.dialect.driver == "pymysql": sscursor = __import__("pymysql.cursors").cursors.SSCursor return isinstance(cursor, sscursor) elif self.engine.dialect.driver == "mysqldb": sscursor = __import__("MySQLdb.cursors").cursors.SSCursor return isinstance(cursor, sscursor) else: return False def _fixture(self, server_side_cursors): self.engine = engines.testing_engine( options={"server_side_cursors": server_side_cursors} ) return self.engine def tearDown(self): engines.testing_reaper.close_all() self.engine.dispose() def test_global_string(self): engine = self._fixture(True) result = engine.execute("select 1") assert self._is_server_side(result.cursor) def test_global_text(self): engine = self._fixture(True) result = engine.execute(text("select 1")) assert self._is_server_side(result.cursor) def test_global_expr(self): engine = self._fixture(True) result = engine.execute(select([1])) assert self._is_server_side(result.cursor) def test_global_off_explicit(self): engine = self._fixture(False) result = engine.execute(text("select 1")) # It should be off globally ... assert not self._is_server_side(result.cursor) def test_stmt_option(self): engine = self._fixture(False) s = select([1]).execution_options(stream_results=True) result = engine.execute(s) # ... but enabled for this one. assert self._is_server_side(result.cursor) def test_conn_option(self): engine = self._fixture(False) # and this one result = ( engine.connect() .execution_options(stream_results=True) .execute("select 1") ) assert self._is_server_side(result.cursor) def test_stmt_enabled_conn_option_disabled(self): engine = self._fixture(False) s = select([1]).execution_options(stream_results=True) # not this one result = ( engine.connect().execution_options(stream_results=False).execute(s) ) assert not self._is_server_side(result.cursor) def test_stmt_option_disabled(self): engine = self._fixture(True) s = select([1]).execution_options(stream_results=False) result = engine.execute(s) assert not self._is_server_side(result.cursor) def test_aliases_and_ss(self): engine = self._fixture(False) s1 = select([1]).execution_options(stream_results=True).alias() result = engine.execute(s1) assert self._is_server_side(result.cursor) # s1's options shouldn't affect s2 when s2 is used as a # from_obj. s2 = select([1], from_obj=s1) result = engine.execute(s2) assert not self._is_server_side(result.cursor) def test_for_update_expr(self): engine = self._fixture(True) s1 = select([1]).with_for_update() result = engine.execute(s1) assert self._is_server_side(result.cursor) def test_for_update_string(self): engine = self._fixture(True) result = engine.execute("SELECT 1 FOR UPDATE") assert self._is_server_side(result.cursor) def test_text_no_ss(self): engine = self._fixture(False) s = text("select 42") result = engine.execute(s) assert not self._is_server_side(result.cursor) def test_text_ss_option(self): engine = self._fixture(False) s = text("select 42").execution_options(stream_results=True) result = engine.execute(s) assert self._is_server_side(result.cursor) @testing.provide_metadata def test_roundtrip(self): md = self.metadata self._fixture(True) test_table = Table( "test_table", md, Column("id", Integer, primary_key=True), Column("data", String(50)), ) test_table.create(checkfirst=True) test_table.insert().execute(data="data1") test_table.insert().execute(data="data2") eq_( test_table.select().order_by(test_table.c.id).execute().fetchall(), [(1, "data1"), (2, "data2")], ) test_table.update().where(test_table.c.id == 2).values( data=test_table.c.data + " updated" ).execute() eq_( test_table.select().order_by(test_table.c.id).execute().fetchall(), [(1, "data1"), (2, "data2 updated")], ) test_table.delete().execute() eq_(select([func.count("*")]).select_from(test_table).scalar(), 0) PK}]Atest_dialect.pynu[#! coding: utf-8 from .. import assert_raises from .. import config from .. import eq_ from .. import fixtures from .. import provide_metadata from ..config import requirements from ..schema import Column from ..schema import Table from ... import exc from ... import Integer from ... import literal_column from ... import select from ... import String from ...util import compat class ExceptionTest(fixtures.TablesTest): """Test basic exception wrapping. DBAPIs vary a lot in exception behavior so to actually anticipate specific exceptions from real round trips, we need to be conservative. """ run_deletes = "each" __backend__ = True @classmethod def define_tables(cls, metadata): Table( "manual_pk", metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("data", String(50)), ) @requirements.duplicate_key_raises_integrity_error def test_integrity_error(self): with config.db.connect() as conn: trans = conn.begin() conn.execute( self.tables.manual_pk.insert(), {"id": 1, "data": "d1"} ) assert_raises( exc.IntegrityError, conn.execute, self.tables.manual_pk.insert(), {"id": 1, "data": "d1"}, ) trans.rollback() def test_exception_with_non_ascii(self): with config.db.connect() as conn: try: # try to create an error message that likely has non-ascii # characters in the DBAPI's message string. unfortunately # there's no way to make this happen with some drivers like # mysqlclient, pymysql. this at least does produce a non- # ascii error message for cx_oracle, psycopg2 conn.execute(select([literal_column(u"méil")])) assert False except exc.DBAPIError as err: err_str = str(err) assert str(err.orig) in str(err) # test that we are actually getting string on Py2k, unicode # on Py3k. if compat.py2k: assert isinstance(err_str, str) else: assert isinstance(err_str, str) class AutocommitTest(fixtures.TablesTest): run_deletes = "each" __requires__ = ("autocommit",) __backend__ = True @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("data", String(50)), test_needs_acid=True, ) def _test_conn_autocommits(self, conn, autocommit): trans = conn.begin() conn.execute( self.tables.some_table.insert(), {"id": 1, "data": "some data"} ) trans.rollback() eq_( conn.scalar(select([self.tables.some_table.c.id])), 1 if autocommit else None, ) conn.execute(self.tables.some_table.delete()) def test_autocommit_on(self): conn = config.db.connect() c2 = conn.execution_options(isolation_level="AUTOCOMMIT") self._test_conn_autocommits(c2, True) conn.invalidate() self._test_conn_autocommits(conn, False) def test_autocommit_off(self): conn = config.db.connect() self._test_conn_autocommits(conn, False) class EscapingTest(fixtures.TestBase): @provide_metadata def test_percent_sign_round_trip(self): """test that the DBAPI accommodates for escaped / nonescaped percent signs in a way that matches the compiler """ m = self.metadata t = Table("t", m, Column("data", String(50))) t.create(config.db) with config.db.begin() as conn: conn.execute(t.insert(), dict(data="some % value")) conn.execute(t.insert(), dict(data="some %% other value")) eq_( conn.scalar( select([t.c.data]).where( t.c.data == literal_column("'some % value'") ) ), "some % value", ) eq_( conn.scalar( select([t.c.data]).where( t.c.data == literal_column("'some %% other value'") ) ), "some %% other value", ) PK}][!Jss test_types.pynu[# coding: utf-8 import datetime import decimal from .. import config from .. import fixtures from ..assertions import eq_ from ..config import requirements from ..schema import Column from ..schema import Table from ... import and_ from ... import BigInteger from ... import Boolean from ... import cast from ... import Date from ... import DateTime from ... import Float from ... import Integer from ... import JSON from ... import literal from ... import MetaData from ... import null from ... import Numeric from ... import select from ... import String from ... import testing from ... import Text from ... import Time from ... import TIMESTAMP from ... import type_coerce from ... import Unicode from ... import UnicodeText from ... import util from ...ext.declarative import declarative_base from ...orm import Session from ...util import u class _LiteralRoundTripFixture(object): supports_whereclause = True @testing.provide_metadata def _literal_round_trip(self, type_, input_, output, filter_=None): """test literal rendering """ # for literal, we test the literal render in an INSERT # into a typed column. we can then SELECT it back as its # official type; ideally we'd be able to use CAST here # but MySQL in particular can't CAST fully t = Table("t", self.metadata, Column("x", type_)) t.create() with testing.db.connect() as conn: for value in input_: ins = ( t.insert() .values(x=literal(value)) .compile( dialect=testing.db.dialect, compile_kwargs=dict(literal_binds=True), ) ) conn.execute(ins) if self.supports_whereclause: stmt = t.select().where(t.c.x == literal(value)) else: stmt = t.select() stmt = stmt.compile( dialect=testing.db.dialect, compile_kwargs=dict(literal_binds=True), ) for row in conn.execute(stmt): value = row[0] if filter_ is not None: value = filter_(value) assert value in output class _UnicodeFixture(_LiteralRoundTripFixture): __requires__ = ("unicode_data",) data = u( "Alors vous imaginez ma 🐍 surprise, au lever du jour, " "quand une drôle de petite 🐍 voix m’a réveillé. Elle " "disait: « S’il vous plaît… dessine-moi 🐍 un mouton! »" ) @property def supports_whereclause(self): return config.requirements.expressions_against_unbounded_text.enabled @classmethod def define_tables(cls, metadata): Table( "unicode_table", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("unicode_data", cls.datatype), ) def test_round_trip(self): unicode_table = self.tables.unicode_table config.db.execute(unicode_table.insert(), {"unicode_data": self.data}) row = config.db.execute(select([unicode_table.c.unicode_data])).first() eq_(row, (self.data,)) assert isinstance(row[0], util.text_type) def test_round_trip_executemany(self): unicode_table = self.tables.unicode_table config.db.execute( unicode_table.insert(), [{"unicode_data": self.data} for i in range(3)], ) rows = config.db.execute( select([unicode_table.c.unicode_data]) ).fetchall() eq_(rows, [(self.data,) for i in range(3)]) for row in rows: assert isinstance(row[0], util.text_type) def _test_empty_strings(self): unicode_table = self.tables.unicode_table config.db.execute(unicode_table.insert(), {"unicode_data": u("")}) row = config.db.execute(select([unicode_table.c.unicode_data])).first() eq_(row, (u(""),)) def test_literal(self): self._literal_round_trip(self.datatype, [self.data], [self.data]) def test_literal_non_ascii(self): self._literal_round_trip( self.datatype, [util.u("réve🐍 illé")], [util.u("réve🐍 illé")] ) class UnicodeVarcharTest(_UnicodeFixture, fixtures.TablesTest): __requires__ = ("unicode_data",) __backend__ = True datatype = Unicode(255) @requirements.empty_strings_varchar def test_empty_strings_varchar(self): self._test_empty_strings() class UnicodeTextTest(_UnicodeFixture, fixtures.TablesTest): __requires__ = "unicode_data", "text_type" __backend__ = True datatype = UnicodeText() @requirements.empty_strings_text def test_empty_strings_text(self): self._test_empty_strings() class TextTest(_LiteralRoundTripFixture, fixtures.TablesTest): __requires__ = ("text_type",) __backend__ = True @property def supports_whereclause(self): return config.requirements.expressions_against_unbounded_text.enabled @classmethod def define_tables(cls, metadata): Table( "text_table", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("text_data", Text), ) def test_text_roundtrip(self): text_table = self.tables.text_table config.db.execute(text_table.insert(), {"text_data": "some text"}) row = config.db.execute(select([text_table.c.text_data])).first() eq_(row, ("some text",)) def test_text_empty_strings(self): text_table = self.tables.text_table config.db.execute(text_table.insert(), {"text_data": ""}) row = config.db.execute(select([text_table.c.text_data])).first() eq_(row, ("",)) def test_literal(self): self._literal_round_trip(Text, ["some text"], ["some text"]) def test_literal_non_ascii(self): self._literal_round_trip( Text, [util.u("réve🐍 illé")], [util.u("réve🐍 illé")] ) def test_literal_quoting(self): data = """some 'text' hey "hi there" that's text""" self._literal_round_trip(Text, [data], [data]) def test_literal_backslashes(self): data = r"backslash one \ backslash two \\ end" self._literal_round_trip(Text, [data], [data]) def test_literal_percentsigns(self): data = r"percent % signs %% percent" self._literal_round_trip(Text, [data], [data]) class StringTest(_LiteralRoundTripFixture, fixtures.TestBase): __backend__ = True @requirements.unbounded_varchar def test_nolength_string(self): metadata = MetaData() foo = Table("foo", metadata, Column("one", String)) foo.create(config.db) foo.drop(config.db) def test_literal(self): # note that in Python 3, this invokes the Unicode # datatype for the literal part because all strings are unicode self._literal_round_trip(String(40), ["some text"], ["some text"]) def test_literal_non_ascii(self): self._literal_round_trip( String(40), [util.u("réve🐍 illé")], [util.u("réve🐍 illé")] ) def test_literal_quoting(self): data = """some 'text' hey "hi there" that's text""" self._literal_round_trip(String(40), [data], [data]) def test_literal_backslashes(self): data = r"backslash one \ backslash two \\ end" self._literal_round_trip(String(40), [data], [data]) class _DateFixture(_LiteralRoundTripFixture): compare = None @classmethod def define_tables(cls, metadata): Table( "date_table", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("date_data", cls.datatype), ) def test_round_trip(self): date_table = self.tables.date_table config.db.execute(date_table.insert(), {"date_data": self.data}) row = config.db.execute(select([date_table.c.date_data])).first() compare = self.compare or self.data eq_(row, (compare,)) assert isinstance(row[0], type(compare)) def test_null(self): date_table = self.tables.date_table config.db.execute(date_table.insert(), {"date_data": None}) row = config.db.execute(select([date_table.c.date_data])).first() eq_(row, (None,)) @testing.requires.datetime_literals def test_literal(self): compare = self.compare or self.data self._literal_round_trip(self.datatype, [self.data], [compare]) class DateTimeTest(_DateFixture, fixtures.TablesTest): __requires__ = ("datetime",) __backend__ = True datatype = DateTime data = datetime.datetime(2012, 10, 15, 12, 57, 18) class DateTimeMicrosecondsTest(_DateFixture, fixtures.TablesTest): __requires__ = ("datetime_microseconds",) __backend__ = True datatype = DateTime data = datetime.datetime(2012, 10, 15, 12, 57, 18, 396) class TimestampMicrosecondsTest(_DateFixture, fixtures.TablesTest): __requires__ = ("timestamp_microseconds",) __backend__ = True datatype = TIMESTAMP data = datetime.datetime(2012, 10, 15, 12, 57, 18, 396) class TimeTest(_DateFixture, fixtures.TablesTest): __requires__ = ("time",) __backend__ = True datatype = Time data = datetime.time(12, 57, 18) class TimeMicrosecondsTest(_DateFixture, fixtures.TablesTest): __requires__ = ("time_microseconds",) __backend__ = True datatype = Time data = datetime.time(12, 57, 18, 396) class DateTest(_DateFixture, fixtures.TablesTest): __requires__ = ("date",) __backend__ = True datatype = Date data = datetime.date(2012, 10, 15) class DateTimeCoercedToDateTimeTest(_DateFixture, fixtures.TablesTest): __requires__ = "date", "date_coerces_from_datetime" __backend__ = True datatype = Date data = datetime.datetime(2012, 10, 15, 12, 57, 18) compare = datetime.date(2012, 10, 15) class DateTimeHistoricTest(_DateFixture, fixtures.TablesTest): __requires__ = ("datetime_historic",) __backend__ = True datatype = DateTime data = datetime.datetime(1850, 11, 10, 11, 52, 35) class DateHistoricTest(_DateFixture, fixtures.TablesTest): __requires__ = ("date_historic",) __backend__ = True datatype = Date data = datetime.date(1727, 4, 1) class IntegerTest(_LiteralRoundTripFixture, fixtures.TestBase): __backend__ = True def test_literal(self): self._literal_round_trip(Integer, [5], [5]) def test_huge_int(self): self._round_trip(BigInteger, 1376537018368127) @testing.provide_metadata def _round_trip(self, datatype, data): metadata = self.metadata int_table = Table( "integer_table", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("integer_data", datatype), ) metadata.create_all(config.db) config.db.execute(int_table.insert(), {"integer_data": data}) row = config.db.execute(select([int_table.c.integer_data])).first() eq_(row, (data,)) if util.py3k: assert isinstance(row[0], int) else: assert isinstance(row[0], (long, int)) # noqa class NumericTest(_LiteralRoundTripFixture, fixtures.TestBase): __backend__ = True @testing.emits_warning(r".*does \*not\* support Decimal objects natively") @testing.provide_metadata def _do_test(self, type_, input_, output, filter_=None, check_scale=False): metadata = self.metadata t = Table("t", metadata, Column("x", type_)) t.create() t.insert().execute([{"x": x} for x in input_]) result = {row[0] for row in t.select().execute()} output = set(output) if filter_: result = set(filter_(x) for x in result) output = set(filter_(x) for x in output) eq_(result, output) if check_scale: eq_([str(x) for x in result], [str(x) for x in output]) @testing.emits_warning(r".*does \*not\* support Decimal objects natively") def test_render_literal_numeric(self): self._literal_round_trip( Numeric(precision=8, scale=4), [15.7563, decimal.Decimal("15.7563")], [decimal.Decimal("15.7563")], ) @testing.emits_warning(r".*does \*not\* support Decimal objects natively") def test_render_literal_numeric_asfloat(self): self._literal_round_trip( Numeric(precision=8, scale=4, asdecimal=False), [15.7563, decimal.Decimal("15.7563")], [15.7563], ) def test_render_literal_float(self): self._literal_round_trip( Float(4), [15.7563, decimal.Decimal("15.7563")], [15.7563], filter_=lambda n: n is not None and round(n, 5) or None, ) @testing.requires.precision_generic_float_type def test_float_custom_scale(self): self._do_test( Float(None, decimal_return_scale=7, asdecimal=True), [15.7563827, decimal.Decimal("15.7563827")], [decimal.Decimal("15.7563827")], check_scale=True, ) def test_numeric_as_decimal(self): self._do_test( Numeric(precision=8, scale=4), [15.7563, decimal.Decimal("15.7563")], [decimal.Decimal("15.7563")], ) def test_numeric_as_float(self): self._do_test( Numeric(precision=8, scale=4, asdecimal=False), [15.7563, decimal.Decimal("15.7563")], [15.7563], ) @testing.requires.fetch_null_from_numeric def test_numeric_null_as_decimal(self): self._do_test(Numeric(precision=8, scale=4), [None], [None]) @testing.requires.fetch_null_from_numeric def test_numeric_null_as_float(self): self._do_test( Numeric(precision=8, scale=4, asdecimal=False), [None], [None] ) @testing.requires.floats_to_four_decimals def test_float_as_decimal(self): self._do_test( Float(precision=8, asdecimal=True), [15.7563, decimal.Decimal("15.7563"), None], [decimal.Decimal("15.7563"), None], ) def test_float_as_float(self): self._do_test( Float(precision=8), [15.7563, decimal.Decimal("15.7563")], [15.7563], filter_=lambda n: n is not None and round(n, 5) or None, ) def test_float_coerce_round_trip(self): expr = 15.7563 val = testing.db.scalar(select([literal(expr)])) eq_(val, expr) # this does not work in MySQL, see #4036, however we choose not # to render CAST unconditionally since this is kind of an edge case. @testing.requires.implicit_decimal_binds @testing.emits_warning(r".*does \*not\* support Decimal objects natively") def test_decimal_coerce_round_trip(self): expr = decimal.Decimal("15.7563") val = testing.db.scalar(select([literal(expr)])) eq_(val, expr) @testing.emits_warning(r".*does \*not\* support Decimal objects natively") def test_decimal_coerce_round_trip_w_cast(self): expr = decimal.Decimal("15.7563") val = testing.db.scalar(select([cast(expr, Numeric(10, 4))])) eq_(val, expr) @testing.requires.precision_numerics_general def test_precision_decimal(self): numbers = set( [ decimal.Decimal("54.234246451650"), decimal.Decimal("0.004354"), decimal.Decimal("900.0"), ] ) self._do_test(Numeric(precision=18, scale=12), numbers, numbers) @testing.requires.precision_numerics_enotation_large def test_enotation_decimal(self): """test exceedingly small decimals. Decimal reports values with E notation when the exponent is greater than 6. """ numbers = set( [ decimal.Decimal("1E-2"), decimal.Decimal("1E-3"), decimal.Decimal("1E-4"), decimal.Decimal("1E-5"), decimal.Decimal("1E-6"), decimal.Decimal("1E-7"), decimal.Decimal("1E-8"), decimal.Decimal("0.01000005940696"), decimal.Decimal("0.00000005940696"), decimal.Decimal("0.00000000000696"), decimal.Decimal("0.70000000000696"), decimal.Decimal("696E-12"), ] ) self._do_test(Numeric(precision=18, scale=14), numbers, numbers) @testing.requires.precision_numerics_enotation_large def test_enotation_decimal_large(self): """test exceedingly large decimals. """ numbers = set( [ decimal.Decimal("4E+8"), decimal.Decimal("5748E+15"), decimal.Decimal("1.521E+15"), decimal.Decimal("00000000000000.1E+12"), ] ) self._do_test(Numeric(precision=25, scale=2), numbers, numbers) @testing.requires.precision_numerics_many_significant_digits def test_many_significant_digits(self): numbers = set( [ decimal.Decimal("31943874831932418390.01"), decimal.Decimal("319438950232418390.273596"), decimal.Decimal("87673.594069654243"), ] ) self._do_test(Numeric(precision=38, scale=12), numbers, numbers) @testing.requires.precision_numerics_retains_significant_digits def test_numeric_no_decimal(self): numbers = set([decimal.Decimal("1.000")]) self._do_test( Numeric(precision=5, scale=3), numbers, numbers, check_scale=True ) class BooleanTest(_LiteralRoundTripFixture, fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "boolean_table", metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("value", Boolean), Column("unconstrained_value", Boolean(create_constraint=False)), ) def test_render_literal_bool(self): self._literal_round_trip(Boolean(), [True, False], [True, False]) def test_round_trip(self): boolean_table = self.tables.boolean_table config.db.execute( boolean_table.insert(), {"id": 1, "value": True, "unconstrained_value": False}, ) row = config.db.execute( select( [boolean_table.c.value, boolean_table.c.unconstrained_value] ) ).first() eq_(row, (True, False)) assert isinstance(row[0], bool) def test_null(self): boolean_table = self.tables.boolean_table config.db.execute( boolean_table.insert(), {"id": 1, "value": None, "unconstrained_value": None}, ) row = config.db.execute( select( [boolean_table.c.value, boolean_table.c.unconstrained_value] ) ).first() eq_(row, (None, None)) def test_whereclause(self): # testing "WHERE " renders a compatible expression boolean_table = self.tables.boolean_table with config.db.connect() as conn: conn.execute( boolean_table.insert(), [ {"id": 1, "value": True, "unconstrained_value": True}, {"id": 2, "value": False, "unconstrained_value": False}, ], ) eq_( conn.scalar( select([boolean_table.c.id]).where(boolean_table.c.value) ), 1, ) eq_( conn.scalar( select([boolean_table.c.id]).where( boolean_table.c.unconstrained_value ) ), 1, ) eq_( conn.scalar( select([boolean_table.c.id]).where(~boolean_table.c.value) ), 2, ) eq_( conn.scalar( select([boolean_table.c.id]).where( ~boolean_table.c.unconstrained_value ) ), 2, ) class JSONTest(_LiteralRoundTripFixture, fixtures.TablesTest): __requires__ = ("json_type",) __backend__ = True datatype = JSON data1 = {"key1": "value1", "key2": "value2"} data2 = { "Key 'One'": "value1", "key two": "value2", "key three": "value ' three '", } data3 = { "key1": [1, 2, 3], "key2": ["one", "two", "three"], "key3": [{"four": "five"}, {"six": "seven"}], } data4 = ["one", "two", "three"] data5 = { "nested": { "elem1": [{"a": "b", "c": "d"}, {"e": "f", "g": "h"}], "elem2": {"elem3": {"elem4": "elem5"}}, } } data6 = {"a": 5, "b": "some value", "c": {"foo": "bar"}} @classmethod def define_tables(cls, metadata): Table( "data_table", metadata, Column("id", Integer, primary_key=True), Column("name", String(30), nullable=False), Column("data", cls.datatype), Column("nulldata", cls.datatype(none_as_null=True)), ) def test_round_trip_data1(self): self._test_round_trip(self.data1) def _test_round_trip(self, data_element): data_table = self.tables.data_table config.db.execute( data_table.insert(), {"name": "row1", "data": data_element} ) row = config.db.execute(select([data_table.c.data])).first() eq_(row, (data_element,)) def test_round_trip_none_as_sql_null(self): col = self.tables.data_table.c["nulldata"] with config.db.connect() as conn: conn.execute( self.tables.data_table.insert(), {"name": "r1", "data": None} ) eq_( conn.scalar( select([self.tables.data_table.c.name]).where( col.is_(null()) ) ), "r1", ) eq_(conn.scalar(select([col])), None) def test_round_trip_json_null_as_json_null(self): col = self.tables.data_table.c["data"] with config.db.connect() as conn: conn.execute( self.tables.data_table.insert(), {"name": "r1", "data": JSON.NULL}, ) eq_( conn.scalar( select([self.tables.data_table.c.name]).where( cast(col, String) == "null" ) ), "r1", ) eq_(conn.scalar(select([col])), None) def test_round_trip_none_as_json_null(self): col = self.tables.data_table.c["data"] with config.db.connect() as conn: conn.execute( self.tables.data_table.insert(), {"name": "r1", "data": None} ) eq_( conn.scalar( select([self.tables.data_table.c.name]).where( cast(col, String) == "null" ) ), "r1", ) eq_(conn.scalar(select([col])), None) def _criteria_fixture(self): config.db.execute( self.tables.data_table.insert(), [ {"name": "r1", "data": self.data1}, {"name": "r2", "data": self.data2}, {"name": "r3", "data": self.data3}, {"name": "r4", "data": self.data4}, {"name": "r5", "data": self.data5}, {"name": "r6", "data": self.data6}, ], ) def _test_index_criteria(self, crit, expected, test_literal=True): self._criteria_fixture() with config.db.connect() as conn: stmt = select([self.tables.data_table.c.name]).where(crit) eq_(conn.scalar(stmt), expected) if test_literal: literal_sql = str( stmt.compile( config.db, compile_kwargs={"literal_binds": True} ) ) eq_(conn.scalar(literal_sql), expected) def test_crit_spaces_in_key(self): name = self.tables.data_table.c.name col = self.tables.data_table.c["data"] # limit the rows here to avoid PG error # "cannot extract field from a non-object", which is # fixed in 9.4 but may exist in 9.3 self._test_index_criteria( and_( name.in_(["r1", "r2", "r3"]), cast(col["key two"], String) == '"value2"', ), "r2", ) @config.requirements.json_array_indexes def test_crit_simple_int(self): name = self.tables.data_table.c.name col = self.tables.data_table.c["data"] # limit the rows here to avoid PG error # "cannot extract array element from a non-array", which is # fixed in 9.4 but may exist in 9.3 self._test_index_criteria( and_(name == "r4", cast(col[1], String) == '"two"'), "r4" ) def test_crit_mixed_path(self): col = self.tables.data_table.c["data"] self._test_index_criteria( cast(col[("key3", 1, "six")], String) == '"seven"', "r3" ) def test_crit_string_path(self): col = self.tables.data_table.c["data"] self._test_index_criteria( cast(col[("nested", "elem2", "elem3", "elem4")], String) == '"elem5"', "r5", ) def test_crit_against_string_basic(self): name = self.tables.data_table.c.name col = self.tables.data_table.c["data"] self._test_index_criteria( and_(name == "r6", cast(col["b"], String) == '"some value"'), "r6" ) def test_crit_against_string_coerce_type(self): name = self.tables.data_table.c.name col = self.tables.data_table.c["data"] self._test_index_criteria( and_( name == "r6", cast(col["b"], String) == type_coerce("some value", JSON), ), "r6", test_literal=False, ) def test_crit_against_int_basic(self): name = self.tables.data_table.c.name col = self.tables.data_table.c["data"] self._test_index_criteria( and_(name == "r6", cast(col["a"], String) == "5"), "r6" ) def test_crit_against_int_coerce_type(self): name = self.tables.data_table.c.name col = self.tables.data_table.c["data"] self._test_index_criteria( and_(name == "r6", cast(col["a"], String) == type_coerce(5, JSON)), "r6", test_literal=False, ) def test_unicode_round_trip(self): with config.db.connect() as conn: conn.execute( self.tables.data_table.insert(), { "name": "r1", "data": { util.u("réve🐍 illé"): util.u("réve🐍 illé"), "data": {"k1": util.u("drôl🐍e")}, }, }, ) eq_( conn.scalar(select([self.tables.data_table.c.data])), { util.u("réve🐍 illé"): util.u("réve🐍 illé"), "data": {"k1": util.u("drôl🐍e")}, }, ) def test_eval_none_flag_orm(self): Base = declarative_base() class Data(Base): __table__ = self.tables.data_table s = Session(testing.db) d1 = Data(name="d1", data=None, nulldata=None) s.add(d1) s.commit() s.bulk_insert_mappings( Data, [{"name": "d2", "data": None, "nulldata": None}] ) eq_( s.query( cast(self.tables.data_table.c.data, String()), cast(self.tables.data_table.c.nulldata, String), ) .filter(self.tables.data_table.c.name == "d1") .first(), ("null", None), ) eq_( s.query( cast(self.tables.data_table.c.data, String()), cast(self.tables.data_table.c.nulldata, String), ) .filter(self.tables.data_table.c.name == "d2") .first(), ("null", None), ) __all__ = ( "UnicodeVarcharTest", "UnicodeTextTest", "JSONTest", "DateTest", "DateTimeTest", "TextTest", "NumericTest", "IntegerTest", "DateTimeHistoricTest", "DateTimeCoercedToDateTimeTest", "TimeMicrosecondsTest", "TimestampMicrosecondsTest", "TimeTest", "DateTimeMicrosecondsTest", "DateHistoricTest", "StringTest", "BooleanTest", ) PK}]䗙\\test_reflection.pynu[import operator import re import sqlalchemy as sa from .. import assert_raises_message from .. import config from .. import engines from .. import eq_ from .. import expect_warnings from .. import fixtures from .. import is_ from ..schema import Column from ..schema import Table from ... import event from ... import exc as sa_exc from ... import ForeignKey from ... import inspect from ... import Integer from ... import MetaData from ... import String from ... import testing from ... import types as sql_types from ...engine.reflection import Inspector from ...schema import DDL from ...schema import Index from ...sql.elements import quoted_name metadata, users = None, None class HasTableTest(fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "test_table", metadata, Column("id", Integer, primary_key=True), Column("data", String(50)), ) def test_has_table(self): with config.db.begin() as conn: assert config.db.dialect.has_table(conn, "test_table") assert not config.db.dialect.has_table(conn, "nonexistent_table") class ComponentReflectionTest(fixtures.TablesTest): run_inserts = run_deletes = None __backend__ = True @classmethod def setup_bind(cls): if config.requirements.independent_connections.enabled: from sqlalchemy import pool return engines.testing_engine( options=dict(poolclass=pool.StaticPool) ) else: return config.db @classmethod def define_tables(cls, metadata): cls.define_reflected_tables(metadata, None) if testing.requires.schemas.enabled: cls.define_reflected_tables(metadata, testing.config.test_schema) @classmethod def define_reflected_tables(cls, metadata, schema): if schema: schema_prefix = schema + "." else: schema_prefix = "" if testing.requires.self_referential_foreign_keys.enabled: users = Table( "users", metadata, Column("user_id", sa.INT, primary_key=True), Column("test1", sa.CHAR(5), nullable=False), Column("test2", sa.Float(5), nullable=False), Column( "parent_user_id", sa.Integer, sa.ForeignKey( "%susers.user_id" % schema_prefix, name="user_id_fk" ), ), schema=schema, test_needs_fk=True, ) else: users = Table( "users", metadata, Column("user_id", sa.INT, primary_key=True), Column("test1", sa.CHAR(5), nullable=False), Column("test2", sa.Float(5), nullable=False), schema=schema, test_needs_fk=True, ) Table( "dingalings", metadata, Column("dingaling_id", sa.Integer, primary_key=True), Column( "address_id", sa.Integer, sa.ForeignKey("%semail_addresses.address_id" % schema_prefix), ), Column("data", sa.String(30)), schema=schema, test_needs_fk=True, ) Table( "email_addresses", metadata, Column("address_id", sa.Integer), Column( "remote_user_id", sa.Integer, sa.ForeignKey(users.c.user_id) ), Column("email_address", sa.String(20)), sa.PrimaryKeyConstraint("address_id", name="email_ad_pk"), schema=schema, test_needs_fk=True, ) Table( "comment_test", metadata, Column("id", sa.Integer, primary_key=True, comment="id comment"), Column("data", sa.String(20), comment="data % comment"), Column( "d2", sa.String(20), comment=r"""Comment types type speedily ' " \ '' Fun!""", ), schema=schema, comment=r"""the test % ' " \ table comment""", ) if testing.requires.cross_schema_fk_reflection.enabled: if schema is None: Table( "local_table", metadata, Column("id", sa.Integer, primary_key=True), Column("data", sa.String(20)), Column( "remote_id", ForeignKey( "%s.remote_table_2.id" % testing.config.test_schema ), ), test_needs_fk=True, schema=config.db.dialect.default_schema_name, ) else: Table( "remote_table", metadata, Column("id", sa.Integer, primary_key=True), Column( "local_id", ForeignKey( "%s.local_table.id" % config.db.dialect.default_schema_name ), ), Column("data", sa.String(20)), schema=schema, test_needs_fk=True, ) Table( "remote_table_2", metadata, Column("id", sa.Integer, primary_key=True), Column("data", sa.String(20)), schema=schema, test_needs_fk=True, ) if testing.requires.index_reflection.enabled: cls.define_index(metadata, users) if not schema: # test_needs_fk is at the moment to force MySQL InnoDB noncol_idx_test_nopk = Table( "noncol_idx_test_nopk", metadata, Column("q", sa.String(5)), test_needs_fk=True, ) noncol_idx_test_pk = Table( "noncol_idx_test_pk", metadata, Column("id", sa.Integer, primary_key=True), Column("q", sa.String(5)), test_needs_fk=True, ) Index("noncol_idx_nopk", noncol_idx_test_nopk.c.q.desc()) Index("noncol_idx_pk", noncol_idx_test_pk.c.q.desc()) if testing.requires.view_column_reflection.enabled: cls.define_views(metadata, schema) if not schema and testing.requires.temp_table_reflection.enabled: cls.define_temp_tables(metadata) @classmethod def define_temp_tables(cls, metadata): # cheat a bit, we should fix this with some dialect-level # temp table fixture if testing.against("oracle"): kw = { "prefixes": ["GLOBAL TEMPORARY"], "oracle_on_commit": "PRESERVE ROWS", } else: kw = {"prefixes": ["TEMPORARY"]} user_tmp = Table( "user_tmp", metadata, Column("id", sa.INT, primary_key=True), Column("name", sa.VARCHAR(50)), Column("foo", sa.INT), sa.UniqueConstraint("name", name="user_tmp_uq"), sa.Index("user_tmp_ix", "foo"), **kw ) if ( testing.requires.view_reflection.enabled and testing.requires.temporary_views.enabled ): event.listen( user_tmp, "after_create", DDL( "create temporary view user_tmp_v as " "select * from user_tmp" ), ) event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v")) @classmethod def define_index(cls, metadata, users): Index("users_t_idx", users.c.test1, users.c.test2) Index("users_all_idx", users.c.user_id, users.c.test2, users.c.test1) @classmethod def define_views(cls, metadata, schema): for table_name in ("users", "email_addresses"): fullname = table_name if schema: fullname = "%s.%s" % (schema, table_name) view_name = fullname + "_v" query = "CREATE VIEW %s AS SELECT * FROM %s" % ( view_name, fullname, ) event.listen(metadata, "after_create", DDL(query)) event.listen( metadata, "before_drop", DDL("DROP VIEW %s" % view_name) ) @testing.requires.schema_reflection def test_get_schema_names(self): insp = inspect(testing.db) self.assert_(testing.config.test_schema in insp.get_schema_names()) @testing.requires.schema_reflection def test_dialect_initialize(self): engine = engines.testing_engine() assert not hasattr(engine.dialect, "default_schema_name") inspect(engine) assert hasattr(engine.dialect, "default_schema_name") @testing.requires.schema_reflection def test_get_default_schema_name(self): insp = inspect(testing.db) eq_(insp.default_schema_name, testing.db.dialect.default_schema_name) @testing.provide_metadata def _test_get_table_names( self, schema=None, table_type="table", order_by=None ): _ignore_tables = [ "comment_test", "noncol_idx_test_pk", "noncol_idx_test_nopk", "local_table", "remote_table", "remote_table_2", ] meta = self.metadata insp = inspect(meta.bind) if table_type == "view": table_names = insp.get_view_names(schema) table_names.sort() answer = ["email_addresses_v", "users_v"] eq_(sorted(table_names), answer) else: if order_by: tables = [ rec[0] for rec in insp.get_sorted_table_and_fkc_names(schema) if rec[0] ] else: tables = insp.get_table_names(schema) table_names = [t for t in tables if t not in _ignore_tables] if order_by == "foreign_key": answer = ["users", "email_addresses", "dingalings"] eq_(table_names, answer) else: answer = ["dingalings", "email_addresses", "users"] eq_(sorted(table_names), answer) @testing.requires.temp_table_names def test_get_temp_table_names(self): insp = inspect(self.bind) temp_table_names = insp.get_temp_table_names() eq_(sorted(temp_table_names), ["user_tmp"]) @testing.requires.view_reflection @testing.requires.temp_table_names @testing.requires.temporary_views def test_get_temp_view_names(self): insp = inspect(self.bind) temp_table_names = insp.get_temp_view_names() eq_(sorted(temp_table_names), ["user_tmp_v"]) @testing.requires.table_reflection def test_get_table_names(self): self._test_get_table_names() @testing.requires.table_reflection @testing.requires.foreign_key_constraint_reflection def test_get_table_names_fks(self): self._test_get_table_names(order_by="foreign_key") @testing.requires.comment_reflection def test_get_comments(self): self._test_get_comments() @testing.requires.comment_reflection @testing.requires.schemas def test_get_comments_with_schema(self): self._test_get_comments(testing.config.test_schema) def _test_get_comments(self, schema=None): insp = inspect(testing.db) eq_( insp.get_table_comment("comment_test", schema=schema), {"text": r"""the test % ' " \ table comment"""}, ) eq_(insp.get_table_comment("users", schema=schema), {"text": None}) eq_( [ {"name": rec["name"], "comment": rec["comment"]} for rec in insp.get_columns("comment_test", schema=schema) ], [ {"comment": "id comment", "name": "id"}, {"comment": "data % comment", "name": "data"}, { "comment": ( r"""Comment types type speedily ' " \ '' Fun!""" ), "name": "d2", }, ], ) @testing.requires.table_reflection @testing.requires.schemas def test_get_table_names_with_schema(self): self._test_get_table_names(testing.config.test_schema) @testing.requires.view_column_reflection def test_get_view_names(self): self._test_get_table_names(table_type="view") @testing.requires.view_column_reflection @testing.requires.schemas def test_get_view_names_with_schema(self): self._test_get_table_names( testing.config.test_schema, table_type="view" ) @testing.requires.table_reflection @testing.requires.view_column_reflection def test_get_tables_and_views(self): self._test_get_table_names() self._test_get_table_names(table_type="view") def _test_get_columns(self, schema=None, table_type="table"): meta = MetaData(testing.db) users, addresses = (self.tables.users, self.tables.email_addresses) table_names = ["users", "email_addresses"] if table_type == "view": table_names = ["users_v", "email_addresses_v"] insp = inspect(meta.bind) for table_name, table in zip(table_names, (users, addresses)): schema_name = schema cols = insp.get_columns(table_name, schema=schema_name) self.assert_(len(cols) > 0, len(cols)) # should be in order for i, col in enumerate(table.columns): eq_(col.name, cols[i]["name"]) ctype = cols[i]["type"].__class__ ctype_def = col.type if isinstance(ctype_def, sa.types.TypeEngine): ctype_def = ctype_def.__class__ # Oracle returns Date for DateTime. if testing.against("oracle") and ctype_def in ( sql_types.Date, sql_types.DateTime, ): ctype_def = sql_types.Date # assert that the desired type and return type share # a base within one of the generic types. self.assert_( len( set(ctype.__mro__) .intersection(ctype_def.__mro__) .intersection( [ sql_types.Integer, sql_types.Numeric, sql_types.DateTime, sql_types.Date, sql_types.Time, sql_types.String, sql_types._Binary, ] ) ) > 0, "%s(%s), %s(%s)" % (col.name, col.type, cols[i]["name"], ctype), ) if not col.primary_key: assert cols[i]["default"] is None @testing.requires.table_reflection def test_get_columns(self): self._test_get_columns() @testing.provide_metadata def _type_round_trip(self, *types): t = Table( "t", self.metadata, *[Column("t%d" % i, type_) for i, type_ in enumerate(types)] ) t.create() return [ c["type"] for c in inspect(self.metadata.bind).get_columns("t") ] @testing.requires.table_reflection def test_numeric_reflection(self): for typ in self._type_round_trip(sql_types.Numeric(18, 5)): assert isinstance(typ, sql_types.Numeric) eq_(typ.precision, 18) eq_(typ.scale, 5) @testing.requires.table_reflection def test_varchar_reflection(self): typ = self._type_round_trip(sql_types.String(52))[0] assert isinstance(typ, sql_types.String) eq_(typ.length, 52) @testing.requires.table_reflection @testing.provide_metadata def test_nullable_reflection(self): t = Table( "t", self.metadata, Column("a", Integer, nullable=True), Column("b", Integer, nullable=False), ) t.create() eq_( dict( (col["name"], col["nullable"]) for col in inspect(self.metadata.bind).get_columns("t") ), {"a": True, "b": False}, ) @testing.requires.table_reflection @testing.requires.schemas def test_get_columns_with_schema(self): self._test_get_columns(schema=testing.config.test_schema) @testing.requires.temp_table_reflection def test_get_temp_table_columns(self): meta = MetaData(self.bind) user_tmp = self.tables.user_tmp insp = inspect(meta.bind) cols = insp.get_columns("user_tmp") self.assert_(len(cols) > 0, len(cols)) for i, col in enumerate(user_tmp.columns): eq_(col.name, cols[i]["name"]) @testing.requires.temp_table_reflection @testing.requires.view_column_reflection @testing.requires.temporary_views def test_get_temp_view_columns(self): insp = inspect(self.bind) cols = insp.get_columns("user_tmp_v") eq_([col["name"] for col in cols], ["id", "name", "foo"]) @testing.requires.view_column_reflection def test_get_view_columns(self): self._test_get_columns(table_type="view") @testing.requires.view_column_reflection @testing.requires.schemas def test_get_view_columns_with_schema(self): self._test_get_columns( schema=testing.config.test_schema, table_type="view" ) @testing.provide_metadata def _test_get_pk_constraint(self, schema=None): meta = self.metadata users, addresses = self.tables.users, self.tables.email_addresses insp = inspect(meta.bind) users_cons = insp.get_pk_constraint(users.name, schema=schema) users_pkeys = users_cons["constrained_columns"] eq_(users_pkeys, ["user_id"]) addr_cons = insp.get_pk_constraint(addresses.name, schema=schema) addr_pkeys = addr_cons["constrained_columns"] eq_(addr_pkeys, ["address_id"]) with testing.requires.reflects_pk_names.fail_if(): eq_(addr_cons["name"], "email_ad_pk") @testing.requires.primary_key_constraint_reflection def test_get_pk_constraint(self): self._test_get_pk_constraint() @testing.requires.table_reflection @testing.requires.primary_key_constraint_reflection @testing.requires.schemas def test_get_pk_constraint_with_schema(self): self._test_get_pk_constraint(schema=testing.config.test_schema) @testing.requires.table_reflection @testing.provide_metadata def test_deprecated_get_primary_keys(self): meta = self.metadata users = self.tables.users insp = Inspector(meta.bind) assert_raises_message( sa_exc.SADeprecationWarning, r".*get_primary_keys\(\) method is deprecated", insp.get_primary_keys, users.name, ) @testing.provide_metadata def _test_get_foreign_keys(self, schema=None): meta = self.metadata users, addresses = (self.tables.users, self.tables.email_addresses) insp = inspect(meta.bind) expected_schema = schema # users if testing.requires.self_referential_foreign_keys.enabled: users_fkeys = insp.get_foreign_keys(users.name, schema=schema) fkey1 = users_fkeys[0] with testing.requires.named_constraints.fail_if(): eq_(fkey1["name"], "user_id_fk") eq_(fkey1["referred_schema"], expected_schema) eq_(fkey1["referred_table"], users.name) eq_(fkey1["referred_columns"], ["user_id"]) if testing.requires.self_referential_foreign_keys.enabled: eq_(fkey1["constrained_columns"], ["parent_user_id"]) # addresses addr_fkeys = insp.get_foreign_keys(addresses.name, schema=schema) fkey1 = addr_fkeys[0] with testing.requires.implicitly_named_constraints.fail_if(): self.assert_(fkey1["name"] is not None) eq_(fkey1["referred_schema"], expected_schema) eq_(fkey1["referred_table"], users.name) eq_(fkey1["referred_columns"], ["user_id"]) eq_(fkey1["constrained_columns"], ["remote_user_id"]) @testing.requires.foreign_key_constraint_reflection def test_get_foreign_keys(self): self._test_get_foreign_keys() @testing.requires.foreign_key_constraint_reflection @testing.requires.schemas def test_get_foreign_keys_with_schema(self): self._test_get_foreign_keys(schema=testing.config.test_schema) @testing.requires.cross_schema_fk_reflection @testing.requires.schemas def test_get_inter_schema_foreign_keys(self): local_table, remote_table, remote_table_2 = self.tables( "%s.local_table" % testing.db.dialect.default_schema_name, "%s.remote_table" % testing.config.test_schema, "%s.remote_table_2" % testing.config.test_schema, ) insp = inspect(config.db) local_fkeys = insp.get_foreign_keys(local_table.name) eq_(len(local_fkeys), 1) fkey1 = local_fkeys[0] eq_(fkey1["referred_schema"], testing.config.test_schema) eq_(fkey1["referred_table"], remote_table_2.name) eq_(fkey1["referred_columns"], ["id"]) eq_(fkey1["constrained_columns"], ["remote_id"]) remote_fkeys = insp.get_foreign_keys( remote_table.name, schema=testing.config.test_schema ) eq_(len(remote_fkeys), 1) fkey2 = remote_fkeys[0] assert fkey2["referred_schema"] in ( None, testing.db.dialect.default_schema_name, ) eq_(fkey2["referred_table"], local_table.name) eq_(fkey2["referred_columns"], ["id"]) eq_(fkey2["constrained_columns"], ["local_id"]) @testing.requires.foreign_key_constraint_option_reflection_ondelete def test_get_foreign_key_options_ondelete(self): self._test_get_foreign_key_options(ondelete="CASCADE") @testing.requires.foreign_key_constraint_option_reflection_onupdate def test_get_foreign_key_options_onupdate(self): self._test_get_foreign_key_options(onupdate="SET NULL") @testing.provide_metadata def _test_get_foreign_key_options(self, **options): meta = self.metadata Table( "x", meta, Column("id", Integer, primary_key=True), test_needs_fk=True, ) Table( "table", meta, Column("id", Integer, primary_key=True), Column("x_id", Integer, sa.ForeignKey("x.id", name="xid")), Column("test", String(10)), test_needs_fk=True, ) Table( "user", meta, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("tid", Integer), sa.ForeignKeyConstraint( ["tid"], ["table.id"], name="myfk", **options ), test_needs_fk=True, ) meta.create_all() insp = inspect(meta.bind) # test 'options' is always present for a backend # that can reflect these, since alembic looks for this opts = insp.get_foreign_keys("table")[0]["options"] eq_(dict((k, opts[k]) for k in opts if opts[k]), {}) opts = insp.get_foreign_keys("user")[0]["options"] eq_(dict((k, opts[k]) for k in opts if opts[k]), options) def _assert_insp_indexes(self, indexes, expected_indexes): index_names = [d["name"] for d in indexes] for e_index in expected_indexes: assert e_index["name"] in index_names index = indexes[index_names.index(e_index["name"])] for key in e_index: eq_(e_index[key], index[key]) @testing.provide_metadata def _test_get_indexes(self, schema=None): meta = self.metadata # The database may decide to create indexes for foreign keys, etc. # so there may be more indexes than expected. insp = inspect(meta.bind) indexes = insp.get_indexes("users", schema=schema) expected_indexes = [ { "unique": False, "column_names": ["test1", "test2"], "name": "users_t_idx", }, { "unique": False, "column_names": ["user_id", "test2", "test1"], "name": "users_all_idx", }, ] self._assert_insp_indexes(indexes, expected_indexes) @testing.requires.index_reflection def test_get_indexes(self): self._test_get_indexes() @testing.requires.index_reflection @testing.requires.schemas def test_get_indexes_with_schema(self): self._test_get_indexes(schema=testing.config.test_schema) @testing.provide_metadata def _test_get_noncol_index(self, tname, ixname): meta = self.metadata insp = inspect(meta.bind) indexes = insp.get_indexes(tname) # reflecting an index that has "x DESC" in it as the column. # the DB may or may not give us "x", but make sure we get the index # back, it has a name, it's connected to the table. expected_indexes = [{"unique": False, "name": ixname}] self._assert_insp_indexes(indexes, expected_indexes) t = Table(tname, meta, autoload_with=meta.bind) eq_(len(t.indexes), 1) is_(list(t.indexes)[0].table, t) eq_(list(t.indexes)[0].name, ixname) @testing.requires.index_reflection def test_get_noncol_index_no_pk(self): self._test_get_noncol_index("noncol_idx_test_nopk", "noncol_idx_nopk") @testing.requires.index_reflection def test_get_noncol_index_pk(self): self._test_get_noncol_index("noncol_idx_test_pk", "noncol_idx_pk") @testing.requires.indexes_with_expressions @testing.provide_metadata def test_reflect_expression_based_indexes(self): Table( "t", self.metadata, Column("x", String(30)), Column("y", String(30)), ) event.listen( self.metadata, "after_create", DDL("CREATE INDEX t_idx ON t(lower(x), lower(y))"), ) event.listen( self.metadata, "after_create", DDL("CREATE INDEX t_idx_2 ON t(x)") ) self.metadata.create_all() insp = inspect(self.metadata.bind) with expect_warnings( "Skipped unsupported reflection of expression-based index t_idx" ): eq_( insp.get_indexes("t"), [{"name": "t_idx_2", "column_names": ["x"], "unique": 0}], ) @testing.requires.unique_constraint_reflection def test_get_unique_constraints(self): self._test_get_unique_constraints() @testing.requires.temp_table_reflection @testing.requires.unique_constraint_reflection def test_get_temp_table_unique_constraints(self): insp = inspect(self.bind) reflected = insp.get_unique_constraints("user_tmp") for refl in reflected: # Different dialects handle duplicate index and constraints # differently, so ignore this flag refl.pop("duplicates_index", None) eq_(reflected, [{"column_names": ["name"], "name": "user_tmp_uq"}]) @testing.requires.temp_table_reflection def test_get_temp_table_indexes(self): insp = inspect(self.bind) indexes = insp.get_indexes("user_tmp") for ind in indexes: ind.pop("dialect_options", None) eq_( # TODO: we need to add better filtering for indexes/uq constraints # that are doubled up [idx for idx in indexes if idx["name"] == "user_tmp_ix"], [ { "unique": False, "column_names": ["foo"], "name": "user_tmp_ix", } ], ) @testing.requires.unique_constraint_reflection @testing.requires.schemas def test_get_unique_constraints_with_schema(self): self._test_get_unique_constraints(schema=testing.config.test_schema) @testing.provide_metadata def _test_get_unique_constraints(self, schema=None): # SQLite dialect needs to parse the names of the constraints # separately from what it gets from PRAGMA index_list(), and # then matches them up. so same set of column_names in two # constraints will confuse it. Perhaps we should no longer # bother with index_list() here since we have the whole # CREATE TABLE? uniques = sorted( [ {"name": "unique_a", "column_names": ["a"]}, {"name": "unique_a_b_c", "column_names": ["a", "b", "c"]}, {"name": "unique_c_a_b", "column_names": ["c", "a", "b"]}, {"name": "unique_asc_key", "column_names": ["asc", "key"]}, {"name": "i.have.dots", "column_names": ["b"]}, {"name": "i have spaces", "column_names": ["c"]}, ], key=operator.itemgetter("name"), ) orig_meta = self.metadata table = Table( "testtbl", orig_meta, Column("a", sa.String(20)), Column("b", sa.String(30)), Column("c", sa.Integer), # reserved identifiers Column("asc", sa.String(30)), Column("key", sa.String(30)), schema=schema, ) for uc in uniques: table.append_constraint( sa.UniqueConstraint(*uc["column_names"], name=uc["name"]) ) orig_meta.create_all() inspector = inspect(orig_meta.bind) reflected = sorted( inspector.get_unique_constraints("testtbl", schema=schema), key=operator.itemgetter("name"), ) names_that_duplicate_index = set() for orig, refl in zip(uniques, reflected): # Different dialects handle duplicate index and constraints # differently, so ignore this flag dupe = refl.pop("duplicates_index", None) if dupe: names_that_duplicate_index.add(dupe) eq_(orig, refl) reflected_metadata = MetaData() reflected = Table( "testtbl", reflected_metadata, autoload_with=orig_meta.bind, schema=schema, ) # test "deduplicates for index" logic. MySQL and Oracle # "unique constraints" are actually unique indexes (with possible # exception of a unique that is a dupe of another one in the case # of Oracle). make sure # they aren't duplicated. idx_names = set([idx.name for idx in reflected.indexes]) uq_names = set( [ uq.name for uq in reflected.constraints if isinstance(uq, sa.UniqueConstraint) ] ).difference(["unique_c_a_b"]) assert not idx_names.intersection(uq_names) if names_that_duplicate_index: eq_(names_that_duplicate_index, idx_names) eq_(uq_names, set()) @testing.requires.check_constraint_reflection def test_get_check_constraints(self): self._test_get_check_constraints() @testing.requires.check_constraint_reflection @testing.requires.schemas def test_get_check_constraints_schema(self): self._test_get_check_constraints(schema=testing.config.test_schema) @testing.provide_metadata def _test_get_check_constraints(self, schema=None): orig_meta = self.metadata Table( "sa_cc", orig_meta, Column("a", Integer()), sa.CheckConstraint("a > 1 AND a < 5", name="cc1"), sa.CheckConstraint("a = 1 OR (a > 2 AND a < 5)", name="cc2"), schema=schema, ) orig_meta.create_all() inspector = inspect(orig_meta.bind) reflected = sorted( inspector.get_check_constraints("sa_cc", schema=schema), key=operator.itemgetter("name"), ) # trying to minimize effect of quoting, parenthesis, etc. # may need to add more to this as new dialects get CHECK # constraint reflection support def normalize(sqltext): return " ".join( re.findall(r"and|\d|=|a|or|<|>", sqltext.lower(), re.I) ) reflected = [ {"name": item["name"], "sqltext": normalize(item["sqltext"])} for item in reflected ] eq_( reflected, [ {"name": "cc1", "sqltext": "a > 1 and a < 5"}, {"name": "cc2", "sqltext": "a = 1 or a > 2 and a < 5"}, ], ) @testing.provide_metadata def _test_get_view_definition(self, schema=None): meta = self.metadata view_name1 = "users_v" view_name2 = "email_addresses_v" insp = inspect(meta.bind) v1 = insp.get_view_definition(view_name1, schema=schema) self.assert_(v1) v2 = insp.get_view_definition(view_name2, schema=schema) self.assert_(v2) @testing.requires.view_reflection def test_get_view_definition(self): self._test_get_view_definition() @testing.requires.view_reflection @testing.requires.schemas def test_get_view_definition_with_schema(self): self._test_get_view_definition(schema=testing.config.test_schema) @testing.only_on("postgresql", "PG specific feature") @testing.provide_metadata def _test_get_table_oid(self, table_name, schema=None): meta = self.metadata insp = inspect(meta.bind) oid = insp.get_table_oid(table_name, schema) self.assert_(isinstance(oid, int)) def test_get_table_oid(self): self._test_get_table_oid("users") @testing.requires.schemas def test_get_table_oid_with_schema(self): self._test_get_table_oid("users", schema=testing.config.test_schema) @testing.requires.table_reflection @testing.provide_metadata def test_autoincrement_col(self): """test that 'autoincrement' is reflected according to sqla's policy. Don't mark this test as unsupported for any backend ! (technically it fails with MySQL InnoDB since "id" comes before "id2") A backend is better off not returning "autoincrement" at all, instead of potentially returning "False" for an auto-incrementing primary key column. """ meta = self.metadata insp = inspect(meta.bind) for tname, cname in [ ("users", "user_id"), ("email_addresses", "address_id"), ("dingalings", "dingaling_id"), ]: cols = insp.get_columns(tname) id_ = {c["name"]: c for c in cols}[cname] assert id_.get("autoincrement", True) class NormalizedNameTest(fixtures.TablesTest): __requires__ = ("denormalized_names",) __backend__ = True @classmethod def define_tables(cls, metadata): Table( quoted_name("t1", quote=True), metadata, Column("id", Integer, primary_key=True), ) Table( quoted_name("t2", quote=True), metadata, Column("id", Integer, primary_key=True), Column("t1id", ForeignKey("t1.id")), ) def test_reflect_lowercase_forced_tables(self): m2 = MetaData(testing.db) t2_ref = Table(quoted_name("t2", quote=True), m2, autoload=True) t1_ref = m2.tables["t1"] assert t2_ref.c.t1id.references(t1_ref.c.id) m3 = MetaData(testing.db) m3.reflect(only=lambda name, m: name.lower() in ("t1", "t2")) assert m3.tables["t2"].c.t1id.references(m3.tables["t1"].c.id) def test_get_table_names(self): tablenames = [ t for t in inspect(testing.db).get_table_names() if t.lower() in ("t1", "t2") ] eq_(tablenames[0].upper(), tablenames[0].lower()) eq_(tablenames[1].upper(), tablenames[1].lower()) __all__ = ("ComponentReflectionTest", "HasTableTest", "NormalizedNameTest") PK}]sh*NNtest_select.pynu[from .. import config from .. import fixtures from ..assertions import eq_ from ..assertions import in_ from ..schema import Column from ..schema import Table from ... import bindparam from ... import case from ... import false from ... import func from ... import Integer from ... import literal_column from ... import null from ... import select from ... import String from ... import testing from ... import true from ... import tuple_ from ... import union from ... import util class CollateTest(fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("data", String(100)), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.some_table.insert(), [ {"id": 1, "data": "collate data1"}, {"id": 2, "data": "collate data2"}, ], ) def _assert_result(self, select, result): eq_(config.db.execute(select).fetchall(), result) @testing.requires.order_by_collation def test_collate_order_by(self): collation = testing.requires.get_order_by_collation(testing.config) self._assert_result( select([self.tables.some_table]).order_by( self.tables.some_table.c.data.collate(collation).asc() ), [(1, "collate data1"), (2, "collate data2")], ) class OrderByLabelTest(fixtures.TablesTest): """Test the dialect sends appropriate ORDER BY expressions when labels are used. This essentially exercises the "supports_simple_order_by_label" setting. """ __backend__ = True @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("x", Integer), Column("y", Integer), Column("q", String(50)), Column("p", String(50)), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.some_table.insert(), [ {"id": 1, "x": 1, "y": 2, "q": "q1", "p": "p3"}, {"id": 2, "x": 2, "y": 3, "q": "q2", "p": "p2"}, {"id": 3, "x": 3, "y": 4, "q": "q3", "p": "p1"}, ], ) def _assert_result(self, select, result): eq_(config.db.execute(select).fetchall(), result) def test_plain(self): table = self.tables.some_table lx = table.c.x.label("lx") self._assert_result(select([lx]).order_by(lx), [(1,), (2,), (3,)]) def test_composed_int(self): table = self.tables.some_table lx = (table.c.x + table.c.y).label("lx") self._assert_result(select([lx]).order_by(lx), [(3,), (5,), (7,)]) def test_composed_multiple(self): table = self.tables.some_table lx = (table.c.x + table.c.y).label("lx") ly = (func.lower(table.c.q) + table.c.p).label("ly") self._assert_result( select([lx, ly]).order_by(lx, ly.desc()), [(3, util.u("q1p3")), (5, util.u("q2p2")), (7, util.u("q3p1"))], ) def test_plain_desc(self): table = self.tables.some_table lx = table.c.x.label("lx") self._assert_result( select([lx]).order_by(lx.desc()), [(3,), (2,), (1,)] ) def test_composed_int_desc(self): table = self.tables.some_table lx = (table.c.x + table.c.y).label("lx") self._assert_result( select([lx]).order_by(lx.desc()), [(7,), (5,), (3,)] ) @testing.requires.group_by_complex_expression def test_group_by_composed(self): table = self.tables.some_table expr = (table.c.x + table.c.y).label("lx") stmt = ( select([func.count(table.c.id), expr]) .group_by(expr) .order_by(expr) ) self._assert_result(stmt, [(1, 3), (1, 5), (1, 7)]) class LimitOffsetTest(fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("x", Integer), Column("y", Integer), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.some_table.insert(), [ {"id": 1, "x": 1, "y": 2}, {"id": 2, "x": 2, "y": 3}, {"id": 3, "x": 3, "y": 4}, {"id": 4, "x": 4, "y": 5}, ], ) def _assert_result(self, select, result, params=()): eq_(config.db.execute(select, params).fetchall(), result) def test_simple_limit(self): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).limit(2), [(1, 1, 2), (2, 2, 3)], ) @testing.requires.offset def test_simple_offset(self): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).offset(2), [(3, 3, 4), (4, 4, 5)], ) @testing.requires.offset def test_simple_limit_offset(self): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).limit(2).offset(1), [(2, 2, 3), (3, 3, 4)], ) @testing.requires.offset def test_limit_offset_nobinds(self): """test that 'literal binds' mode works - no bound params.""" table = self.tables.some_table stmt = select([table]).order_by(table.c.id).limit(2).offset(1) sql = stmt.compile( dialect=config.db.dialect, compile_kwargs={"literal_binds": True} ) sql = str(sql) self._assert_result(sql, [(2, 2, 3), (3, 3, 4)]) @testing.requires.bound_limit_offset def test_bound_limit(self): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).limit(bindparam("l")), [(1, 1, 2), (2, 2, 3)], params={"l": 2}, ) @testing.requires.bound_limit_offset def test_bound_offset(self): table = self.tables.some_table self._assert_result( select([table]).order_by(table.c.id).offset(bindparam("o")), [(3, 3, 4), (4, 4, 5)], params={"o": 2}, ) @testing.requires.bound_limit_offset def test_bound_limit_offset(self): table = self.tables.some_table self._assert_result( select([table]) .order_by(table.c.id) .limit(bindparam("l")) .offset(bindparam("o")), [(2, 2, 3), (3, 3, 4)], params={"l": 2, "o": 1}, ) class CompoundSelectTest(fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("x", Integer), Column("y", Integer), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.some_table.insert(), [ {"id": 1, "x": 1, "y": 2}, {"id": 2, "x": 2, "y": 3}, {"id": 3, "x": 3, "y": 4}, {"id": 4, "x": 4, "y": 5}, ], ) def _assert_result(self, select, result, params=()): eq_(config.db.execute(select, params).fetchall(), result) def test_plain_union(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2) s2 = select([table]).where(table.c.id == 3) u1 = union(s1, s2) self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) def test_select_from_plain_union(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2) s2 = select([table]).where(table.c.id == 3) u1 = union(s1, s2).alias().select() self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) @testing.requires.order_by_col_from_union @testing.requires.parens_in_union_contained_select_w_limit_offset def test_limit_offset_selectable_in_unions(self): table = self.tables.some_table s1 = ( select([table]) .where(table.c.id == 2) .limit(1) .order_by(table.c.id) ) s2 = ( select([table]) .where(table.c.id == 3) .limit(1) .order_by(table.c.id) ) u1 = union(s1, s2).limit(2) self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) @testing.requires.parens_in_union_contained_select_wo_limit_offset def test_order_by_selectable_in_unions(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2).order_by(table.c.id) s2 = select([table]).where(table.c.id == 3).order_by(table.c.id) u1 = union(s1, s2).limit(2) self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) def test_distinct_selectable_in_unions(self): table = self.tables.some_table s1 = select([table]).where(table.c.id == 2).distinct() s2 = select([table]).where(table.c.id == 3).distinct() u1 = union(s1, s2).limit(2) self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) @testing.requires.parens_in_union_contained_select_w_limit_offset def test_limit_offset_in_unions_from_alias(self): table = self.tables.some_table s1 = ( select([table]) .where(table.c.id == 2) .limit(1) .order_by(table.c.id) ) s2 = ( select([table]) .where(table.c.id == 3) .limit(1) .order_by(table.c.id) ) # this necessarily has double parens u1 = union(s1, s2).alias() self._assert_result( u1.select().limit(2).order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)] ) def test_limit_offset_aliased_selectable_in_unions(self): table = self.tables.some_table s1 = ( select([table]) .where(table.c.id == 2) .limit(1) .order_by(table.c.id) .alias() .select() ) s2 = ( select([table]) .where(table.c.id == 3) .limit(1) .order_by(table.c.id) .alias() .select() ) u1 = union(s1, s2).limit(2) self._assert_result(u1.order_by(u1.c.id), [(2, 2, 3), (3, 3, 4)]) class ExpandingBoundInTest(fixtures.TablesTest): __backend__ = True @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("x", Integer), Column("y", Integer), Column("z", String(50)), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.some_table.insert(), [ {"id": 1, "x": 1, "y": 2, "z": "z1"}, {"id": 2, "x": 2, "y": 3, "z": "z2"}, {"id": 3, "x": 3, "y": 4, "z": "z3"}, {"id": 4, "x": 4, "y": 5, "z": "z4"}, ], ) def _assert_result(self, select, result, params=()): eq_(config.db.execute(select, params).fetchall(), result) def test_multiple_empty_sets(self): # test that any anonymous aliasing used by the dialect # is fine with duplicates table = self.tables.some_table stmt = ( select([table.c.id]) .where(table.c.x.in_(bindparam("q", expanding=True))) .where(table.c.y.in_(bindparam("p", expanding=True))) .order_by(table.c.id) ) self._assert_result(stmt, [], params={"q": [], "p": []}) @testing.requires.tuple_in def test_empty_heterogeneous_tuples(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where( tuple_(table.c.x, table.c.z).in_( bindparam("q", expanding=True) ) ) .order_by(table.c.id) ) self._assert_result(stmt, [], params={"q": []}) @testing.requires.tuple_in def test_empty_homogeneous_tuples(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where( tuple_(table.c.x, table.c.y).in_( bindparam("q", expanding=True) ) ) .order_by(table.c.id) ) self._assert_result(stmt, [], params={"q": []}) def test_bound_in_scalar(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where(table.c.x.in_(bindparam("q", expanding=True))) .order_by(table.c.id) ) self._assert_result(stmt, [(2,), (3,), (4,)], params={"q": [2, 3, 4]}) @testing.requires.tuple_in def test_bound_in_two_tuple(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where( tuple_(table.c.x, table.c.y).in_( bindparam("q", expanding=True) ) ) .order_by(table.c.id) ) self._assert_result( stmt, [(2,), (3,), (4,)], params={"q": [(2, 3), (3, 4), (4, 5)]} ) @testing.requires.tuple_in def test_bound_in_heterogeneous_two_tuple(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where( tuple_(table.c.x, table.c.z).in_( bindparam("q", expanding=True) ) ) .order_by(table.c.id) ) self._assert_result( stmt, [(2,), (3,), (4,)], params={"q": [(2, "z2"), (3, "z3"), (4, "z4")]}, ) def test_empty_set_against_integer(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where(table.c.x.in_(bindparam("q", expanding=True))) .order_by(table.c.id) ) self._assert_result(stmt, [], params={"q": []}) def test_empty_set_against_integer_negation(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where(table.c.x.notin_(bindparam("q", expanding=True))) .order_by(table.c.id) ) self._assert_result(stmt, [(1,), (2,), (3,), (4,)], params={"q": []}) def test_empty_set_against_string(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where(table.c.z.in_(bindparam("q", expanding=True))) .order_by(table.c.id) ) self._assert_result(stmt, [], params={"q": []}) def test_empty_set_against_string_negation(self): table = self.tables.some_table stmt = ( select([table.c.id]) .where(table.c.z.notin_(bindparam("q", expanding=True))) .order_by(table.c.id) ) self._assert_result(stmt, [(1,), (2,), (3,), (4,)], params={"q": []}) def test_null_in_empty_set_is_false(self): stmt = select( [ case( [ ( null().in_( bindparam("foo", value=(), expanding=True) ), true(), ) ], else_=false(), ) ] ) in_(config.db.execute(stmt).fetchone()[0], (False, 0)) class LikeFunctionsTest(fixtures.TablesTest): __backend__ = True run_inserts = "once" run_deletes = None @classmethod def define_tables(cls, metadata): Table( "some_table", metadata, Column("id", Integer, primary_key=True), Column("data", String(50)), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.some_table.insert(), [ {"id": 1, "data": "abcdefg"}, {"id": 2, "data": "ab/cdefg"}, {"id": 3, "data": "ab%cdefg"}, {"id": 4, "data": "ab_cdefg"}, {"id": 5, "data": "abcde/fg"}, {"id": 6, "data": "abcde%fg"}, {"id": 7, "data": "ab#cdefg"}, {"id": 8, "data": "ab9cdefg"}, {"id": 9, "data": "abcde#fg"}, {"id": 10, "data": "abcd9fg"}, ], ) def _test(self, expr, expected): some_table = self.tables.some_table with config.db.connect() as conn: rows = { value for value, in conn.execute( select([some_table.c.id]).where(expr) ) } eq_(rows, expected) def test_startswith_unescaped(self): col = self.tables.some_table.c.data self._test(col.startswith("ab%c"), {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}) def test_startswith_autoescape(self): col = self.tables.some_table.c.data self._test(col.startswith("ab%c", autoescape=True), {3}) def test_startswith_sqlexpr(self): col = self.tables.some_table.c.data self._test( col.startswith(literal_column("'ab%c'")), {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, ) def test_startswith_escape(self): col = self.tables.some_table.c.data self._test(col.startswith("ab##c", escape="#"), {7}) def test_startswith_autoescape_escape(self): col = self.tables.some_table.c.data self._test(col.startswith("ab%c", autoescape=True, escape="#"), {3}) self._test(col.startswith("ab#c", autoescape=True, escape="#"), {7}) def test_endswith_unescaped(self): col = self.tables.some_table.c.data self._test(col.endswith("e%fg"), {1, 2, 3, 4, 5, 6, 7, 8, 9}) def test_endswith_sqlexpr(self): col = self.tables.some_table.c.data self._test( col.endswith(literal_column("'e%fg'")), {1, 2, 3, 4, 5, 6, 7, 8, 9} ) def test_endswith_autoescape(self): col = self.tables.some_table.c.data self._test(col.endswith("e%fg", autoescape=True), {6}) def test_endswith_escape(self): col = self.tables.some_table.c.data self._test(col.endswith("e##fg", escape="#"), {9}) def test_endswith_autoescape_escape(self): col = self.tables.some_table.c.data self._test(col.endswith("e%fg", autoescape=True, escape="#"), {6}) self._test(col.endswith("e#fg", autoescape=True, escape="#"), {9}) def test_contains_unescaped(self): col = self.tables.some_table.c.data self._test(col.contains("b%cde"), {1, 2, 3, 4, 5, 6, 7, 8, 9}) def test_contains_autoescape(self): col = self.tables.some_table.c.data self._test(col.contains("b%cde", autoescape=True), {3}) def test_contains_escape(self): col = self.tables.some_table.c.data self._test(col.contains("b##cde", escape="#"), {7}) def test_contains_autoescape_escape(self): col = self.tables.some_table.c.data self._test(col.contains("b%cd", autoescape=True, escape="#"), {3}) self._test(col.contains("b#cd", autoescape=True, escape="#"), {7}) PK}]Bc0r%r%test_insert.pynu[from .. import config from .. import engines from .. import fixtures from ..assertions import eq_ from ..config import requirements from ..schema import Column from ..schema import Table from ... import Integer from ... import literal from ... import literal_column from ... import select from ... import String class LastrowidTest(fixtures.TablesTest): run_deletes = "each" __backend__ = True __requires__ = "implements_get_lastrowid", "autoincrement_insert" __engine_options__ = {"implicit_returning": False} @classmethod def define_tables(cls, metadata): Table( "autoinc_pk", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("data", String(50)), ) Table( "manual_pk", metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("data", String(50)), ) def _assert_round_trip(self, table, conn): row = conn.execute(table.select()).first() eq_(row, (config.db.dialect.default_sequence_base, "some data")) def test_autoincrement_on_insert(self): config.db.execute(self.tables.autoinc_pk.insert(), data="some data") self._assert_round_trip(self.tables.autoinc_pk, config.db) def test_last_inserted_id(self): r = config.db.execute( self.tables.autoinc_pk.insert(), data="some data" ) pk = config.db.scalar(select([self.tables.autoinc_pk.c.id])) eq_(r.inserted_primary_key, [pk]) # failed on pypy1.9 but seems to be OK on pypy 2.1 # @exclusions.fails_if(lambda: util.pypy, # "lastrowid not maintained after " # "connection close") @requirements.dbapi_lastrowid def test_native_lastrowid_autoinc(self): r = config.db.execute( self.tables.autoinc_pk.insert(), data="some data" ) lastrowid = r.lastrowid pk = config.db.scalar(select([self.tables.autoinc_pk.c.id])) eq_(lastrowid, pk) class InsertBehaviorTest(fixtures.TablesTest): run_deletes = "each" __backend__ = True @classmethod def define_tables(cls, metadata): Table( "autoinc_pk", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("data", String(50)), ) Table( "manual_pk", metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("data", String(50)), ) Table( "includes_defaults", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("data", String(50)), Column("x", Integer, default=5), Column( "y", Integer, default=literal_column("2", type_=Integer) + literal(2), ), ) def test_autoclose_on_insert(self): if requirements.returning.enabled: engine = engines.testing_engine( options={"implicit_returning": False} ) else: engine = config.db r = engine.execute(self.tables.autoinc_pk.insert(), data="some data") assert r._soft_closed assert not r.closed assert r.is_insert assert not r.returns_rows @requirements.returning def test_autoclose_on_insert_implicit_returning(self): r = config.db.execute( self.tables.autoinc_pk.insert(), data="some data" ) assert r._soft_closed assert not r.closed assert r.is_insert assert not r.returns_rows @requirements.empty_inserts def test_empty_insert(self): r = config.db.execute(self.tables.autoinc_pk.insert()) assert r._soft_closed assert not r.closed r = config.db.execute( self.tables.autoinc_pk.select().where( self.tables.autoinc_pk.c.id != None ) ) assert len(r.fetchall()) @requirements.insert_from_select def test_insert_from_select_autoinc(self): src_table = self.tables.manual_pk dest_table = self.tables.autoinc_pk config.db.execute( src_table.insert(), [ dict(id=1, data="data1"), dict(id=2, data="data2"), dict(id=3, data="data3"), ], ) result = config.db.execute( dest_table.insert().from_select( ("data",), select([src_table.c.data]).where( src_table.c.data.in_(["data2", "data3"]) ), ) ) eq_(result.inserted_primary_key, [None]) result = config.db.execute( select([dest_table.c.data]).order_by(dest_table.c.data) ) eq_(result.fetchall(), [("data2",), ("data3",)]) @requirements.insert_from_select def test_insert_from_select_autoinc_no_rows(self): src_table = self.tables.manual_pk dest_table = self.tables.autoinc_pk result = config.db.execute( dest_table.insert().from_select( ("data",), select([src_table.c.data]).where( src_table.c.data.in_(["data2", "data3"]) ), ) ) eq_(result.inserted_primary_key, [None]) result = config.db.execute( select([dest_table.c.data]).order_by(dest_table.c.data) ) eq_(result.fetchall(), []) @requirements.insert_from_select def test_insert_from_select(self): table = self.tables.manual_pk config.db.execute( table.insert(), [ dict(id=1, data="data1"), dict(id=2, data="data2"), dict(id=3, data="data3"), ], ) config.db.execute( table.insert(inline=True).from_select( ("id", "data"), select([table.c.id + 5, table.c.data]).where( table.c.data.in_(["data2", "data3"]) ), ) ) eq_( config.db.execute( select([table.c.data]).order_by(table.c.data) ).fetchall(), [("data1",), ("data2",), ("data2",), ("data3",), ("data3",)], ) @requirements.insert_from_select def test_insert_from_select_with_defaults(self): table = self.tables.includes_defaults config.db.execute( table.insert(), [ dict(id=1, data="data1"), dict(id=2, data="data2"), dict(id=3, data="data3"), ], ) config.db.execute( table.insert(inline=True).from_select( ("id", "data"), select([table.c.id + 5, table.c.data]).where( table.c.data.in_(["data2", "data3"]) ), ) ) eq_( config.db.execute( select([table]).order_by(table.c.data, table.c.id) ).fetchall(), [ (1, "data1", 5, 4), (2, "data2", 5, 4), (7, "data2", 5, 4), (3, "data3", 5, 4), (8, "data3", 5, 4), ], ) class ReturningTest(fixtures.TablesTest): run_create_tables = "each" __requires__ = "returning", "autoincrement_insert" __backend__ = True __engine_options__ = {"implicit_returning": True} def _assert_round_trip(self, table, conn): row = conn.execute(table.select()).first() eq_(row, (config.db.dialect.default_sequence_base, "some data")) @classmethod def define_tables(cls, metadata): Table( "autoinc_pk", metadata, Column( "id", Integer, primary_key=True, test_needs_autoincrement=True ), Column("data", String(50)), ) @requirements.fetch_rows_post_commit def test_explicit_returning_pk_autocommit(self): engine = config.db table = self.tables.autoinc_pk r = engine.execute( table.insert().returning(table.c.id), data="some data" ) pk = r.first()[0] fetched_pk = config.db.scalar(select([table.c.id])) eq_(fetched_pk, pk) def test_explicit_returning_pk_no_autocommit(self): engine = config.db table = self.tables.autoinc_pk with engine.begin() as conn: r = conn.execute( table.insert().returning(table.c.id), data="some data" ) pk = r.first()[0] fetched_pk = config.db.scalar(select([table.c.id])) eq_(fetched_pk, pk) def test_autoincrement_on_insert_implicit_returning(self): config.db.execute(self.tables.autoinc_pk.insert(), data="some data") self._assert_round_trip(self.tables.autoinc_pk, config.db) def test_last_inserted_id_implicit_returning(self): r = config.db.execute( self.tables.autoinc_pk.insert(), data="some data" ) pk = config.db.scalar(select([self.tables.autoinc_pk.c.id])) eq_(r.inserted_primary_key, [pk]) __all__ = ("LastrowidTest", "InsertBehaviorTest", "ReturningTest") PK}]"!test_update_delete.pynu[from .. import config from .. import fixtures from ..assertions import eq_ from ..schema import Column from ..schema import Table from ... import Integer from ... import String class SimpleUpdateDeleteTest(fixtures.TablesTest): run_deletes = "each" __backend__ = True @classmethod def define_tables(cls, metadata): Table( "plain_pk", metadata, Column("id", Integer, primary_key=True), Column("data", String(50)), ) @classmethod def insert_data(cls): config.db.execute( cls.tables.plain_pk.insert(), [ {"id": 1, "data": "d1"}, {"id": 2, "data": "d2"}, {"id": 3, "data": "d3"}, ], ) def test_update(self): t = self.tables.plain_pk r = config.db.execute(t.update().where(t.c.id == 2), data="d2_new") assert not r.is_insert assert not r.returns_rows eq_( config.db.execute(t.select().order_by(t.c.id)).fetchall(), [(1, "d1"), (2, "d2_new"), (3, "d3")], ) def test_delete(self): t = self.tables.plain_pk r = config.db.execute(t.delete().where(t.c.id == 2)) assert not r.is_insert assert not r.returns_rows eq_( config.db.execute(t.select().order_by(t.c.id)).fetchall(), [(1, "d1"), (3, "d3")], ) __all__ = ("SimpleUpdateDeleteTest",) PK}]Dǝ6 6 test_ddl.pynu[from .. import config from .. import fixtures from .. import util from ..assertions import eq_ from ..config import requirements from ... import Column from ... import inspect from ... import Integer from ... import schema from ... import String from ... import Table class TableDDLTest(fixtures.TestBase): __backend__ = True def _simple_fixture(self, schema=None): return Table( "test_table", self.metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("data", String(50)), schema=schema, ) def _underscore_fixture(self): return Table( "_test_table", self.metadata, Column("id", Integer, primary_key=True, autoincrement=False), Column("_data", String(50)), ) def _simple_roundtrip(self, table): with config.db.begin() as conn: conn.execute(table.insert().values((1, "some data"))) result = conn.execute(table.select()) eq_(result.first(), (1, "some data")) @requirements.create_table @util.provide_metadata def test_create_table(self): table = self._simple_fixture() table.create(config.db, checkfirst=False) self._simple_roundtrip(table) @requirements.create_table @util.provide_metadata def test_create_table_schema(self): table = self._simple_fixture(schema=config.test_schema) table.create(config.db, checkfirst=False) self._simple_roundtrip(table) @requirements.drop_table @util.provide_metadata def test_drop_table(self): table = self._simple_fixture() table.create(config.db, checkfirst=False) table.drop(config.db, checkfirst=False) @requirements.create_table @util.provide_metadata def test_underscore_names(self): table = self._underscore_fixture() table.create(config.db, checkfirst=False) self._simple_roundtrip(table) @requirements.comment_reflection @util.provide_metadata def test_add_table_comment(self): table = self._simple_fixture() table.create(config.db, checkfirst=False) table.comment = "a comment" config.db.execute(schema.SetTableComment(table)) eq_( inspect(config.db).get_table_comment("test_table"), {"text": "a comment"}, ) @requirements.comment_reflection @util.provide_metadata def test_drop_table_comment(self): table = self._simple_fixture() table.create(config.db, checkfirst=False) table.comment = "a comment" config.db.execute(schema.SetTableComment(table)) config.db.execute(schema.DropTableComment(table)) eq_(inspect(config.db).get_table_comment("test_table"), {"text": None}) __all__ = ("TableDDLTest",) PK}]Ys"ff __init__.pynu[from .test_cte import * # noqa from .test_ddl import * # noqa from .test_dialect import * # noqa from .test_insert import * # noqa from .test_reflection import * # noqa from .test_results import * # noqa from .test_select import * # noqa from .test_sequence import * # noqa from .test_types import * # noqa from .test_update_delete import * # noqa PK}]d 55test_sequence.pynu[from .. import config from .. import fixtures from ..assertions import eq_ from ..config import requirements from ..schema import Column from ..schema import Table from ... import Integer from ... import MetaData from ... import schema from ... import Sequence from ... import String from ... import testing class SequenceTest(fixtures.TablesTest): __requires__ = ("sequences",) __backend__ = True run_create_tables = "each" @classmethod def define_tables(cls, metadata): Table( "seq_pk", metadata, Column("id", Integer, Sequence("tab_id_seq"), primary_key=True), Column("data", String(50)), ) Table( "seq_opt_pk", metadata, Column( "id", Integer, Sequence("tab_id_seq", optional=True), primary_key=True, ), Column("data", String(50)), ) def test_insert_roundtrip(self): config.db.execute(self.tables.seq_pk.insert(), data="some data") self._assert_round_trip(self.tables.seq_pk, config.db) def test_insert_lastrowid(self): r = config.db.execute(self.tables.seq_pk.insert(), data="some data") eq_(r.inserted_primary_key, [1]) def test_nextval_direct(self): r = config.db.execute(self.tables.seq_pk.c.id.default) eq_(r, 1) @requirements.sequences_optional def test_optional_seq(self): r = config.db.execute( self.tables.seq_opt_pk.insert(), data="some data" ) eq_(r.inserted_primary_key, [1]) def _assert_round_trip(self, table, conn): row = conn.execute(table.select()).first() eq_(row, (1, "some data")) class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase): __requires__ = ("sequences",) __backend__ = True def test_literal_binds_inline_compile(self): table = Table( "x", MetaData(), Column("y", Integer, Sequence("y_seq")), Column("q", Integer), ) stmt = table.insert().values(q=5) seq_nextval = testing.db.dialect.statement_compiler( statement=None, dialect=testing.db.dialect ).visit_sequence(Sequence("y_seq")) self.assert_compile( stmt, "INSERT INTO x (y, q) VALUES (%s, 5)" % (seq_nextval,), literal_binds=True, dialect=testing.db.dialect, ) class HasSequenceTest(fixtures.TestBase): __requires__ = ("sequences",) __backend__ = True def test_has_sequence(self): s1 = Sequence("user_id_seq") testing.db.execute(schema.CreateSequence(s1)) try: eq_( testing.db.dialect.has_sequence(testing.db, "user_id_seq"), True, ) finally: testing.db.execute(schema.DropSequence(s1)) @testing.requires.schemas def test_has_sequence_schema(self): s1 = Sequence("user_id_seq", schema=config.test_schema) testing.db.execute(schema.CreateSequence(s1)) try: eq_( testing.db.dialect.has_sequence( testing.db, "user_id_seq", schema=config.test_schema ), True, ) finally: testing.db.execute(schema.DropSequence(s1)) def test_has_sequence_neg(self): eq_(testing.db.dialect.has_sequence(testing.db, "user_id_seq"), False) @testing.requires.schemas def test_has_sequence_schemas_neg(self): eq_( testing.db.dialect.has_sequence( testing.db, "user_id_seq", schema=config.test_schema ), False, ) @testing.requires.schemas def test_has_sequence_default_not_in_remote(self): s1 = Sequence("user_id_seq") testing.db.execute(schema.CreateSequence(s1)) try: eq_( testing.db.dialect.has_sequence( testing.db, "user_id_seq", schema=config.test_schema ), False, ) finally: testing.db.execute(schema.DropSequence(s1)) @testing.requires.schemas def test_has_sequence_remote_not_in_default(self): s1 = Sequence("user_id_seq", schema=config.test_schema) testing.db.execute(schema.CreateSequence(s1)) try: eq_( testing.db.dialect.has_sequence(testing.db, "user_id_seq"), False, ) finally: testing.db.execute(schema.DropSequence(s1)) PK}]Vk?k test_cte.pynu[PK}]| {{*__pycache__/test_reflection.cpython-37.pycnu[PK}]=8y-*__pycache__/test_update_delete.cpython-37.pycnu[PK}]3dW''#J__pycache__/test_cte.cpython-37.pycnu[PK}].[F~sXsX&ı__pycache__/test_select.cpython-37.pycnu[PK}]NE=ss% __pycache__/test_types.cpython-37.pycnu[PK}]r"2#t~__pycache__/__init__.cpython-37.pycnu[PK}]죔!!'e__pycache__/test_dialect.cpython-37.pycnu[PK}].ZZ(ݑ__pycache__/test_sequence.cpython-37.pycnu[PK}]&<"j  #__pycache__/test_ddl.cpython-37.pycnu[PK}]9[S""&__pycache__/test_insert.cpython-37.pycnu[PK}]]e?$++'U__pycache__/test_results.cpython-37.pycnu[PK}] ,,test_results.pynu[PK}]A.test_dialect.pynu[PK}][!Jss Atest_types.pynu[PK}]䗙\\test_reflection.pynu[PK}]sh*NNGtest_select.pynu[PK}]Bc0r%r%test_insert.pynu[PK}]"!test_update_delete.pynu[PK}]Dǝ6 6 test_ddl.pynu[PK}]Ys"ff *__init__.pynu[PK}]d 55test_sequence.pynu[PK@