Source code for examples.versioned_history.test_versioning
"""Unit tests illustrating usage of the ``history_meta.py``module functions."""importunittestimportwarningsfromsqlalchemyimportBooleanfromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimportForeignKeyfromsqlalchemyimportForeignKeyConstraintfromsqlalchemyimportIndexfromsqlalchemyimportinspectfromsqlalchemyimportIntegerfromsqlalchemyimportjoinfromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemyimporttestingfromsqlalchemyimportUniqueConstraintfromsqlalchemy.ormimportclear_mappersfromsqlalchemy.ormimportcolumn_propertyfromsqlalchemy.ormimportdeclarative_basefromsqlalchemy.ormimportDeclarativeBasefromsqlalchemy.ormimportdeferredfromsqlalchemy.ormimportexcasorm_excfromsqlalchemy.ormimportrelationshipfromsqlalchemy.ormimportSessionfromsqlalchemy.testingimportassert_raisesfromsqlalchemy.testingimportAssertsCompiledSQLfromsqlalchemy.testingimporteq_fromsqlalchemy.testingimporteq_ignore_whitespacefromsqlalchemy.testingimportis_fromsqlalchemy.testingimportne_fromsqlalchemy.testing.entitiesimportComparableEntityfrom.history_metaimportVersionedfrom.history_metaimportversioned_sessionwarnings.simplefilter("error")classTestVersioning(AssertsCompiledSQL):__dialect__="default"defsetUp(self):self.engine=engine=create_engine("sqlite://")self.session=Session(engine)self.make_base()versioned_session(self.session)deftearDown(self):self.session.close()clear_mappers()self.Base.metadata.drop_all(self.engine)defmake_base(self):self.Base=declarative_base()defcreate_tables(self):self.Base.metadata.create_all(self.engine)deftest_plain(self):classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)name=Column(String(50))self.create_tables()sess=self.sessionsc=SomeClass(name="sc1")sess.add(sc)sess.commit()sc.name="sc1modified"sess.commit()assertsc.version==2SomeClassHistory=SomeClass.__history_mapper__.class_eq_(sess.query(SomeClassHistory).filter(SomeClassHistory.version==1).all(),[SomeClassHistory(version=1,name="sc1")],)sc.name="sc1modified2"eq_(sess.query(SomeClassHistory).order_by(SomeClassHistory.version).all(),[SomeClassHistory(version=1,name="sc1"),SomeClassHistory(version=2,name="sc1modified"),],)assertsc.version==3sess.commit()sc.name="temp"sc.name="sc1modified2"sess.commit()eq_(sess.query(SomeClassHistory).order_by(SomeClassHistory.version).all(),[SomeClassHistory(version=1,name="sc1"),SomeClassHistory(version=2,name="sc1modified"),],)sess.delete(sc)sess.commit()eq_(sess.query(SomeClassHistory).order_by(SomeClassHistory.version).all(),[SomeClassHistory(version=1,name="sc1"),SomeClassHistory(version=2,name="sc1modified"),SomeClassHistory(version=3,name="sc1modified2"),],)@testing.variation("constraint_type",["index_single_col","composite_index","explicit_name_index","unique_constraint","unique_constraint_naming_conv","unique_constraint_explicit_name","fk_constraint","fk_constraint_naming_conv","fk_constraint_explicit_name",],)deftest_index_naming(self,constraint_type):"""test #10920"""if(constraint_type.unique_constraint_naming_convorconstraint_type.fk_constraint_naming_conv):self.Base.metadata.naming_convention={"ix":"ix_%(column_0_label)s","uq":"uq_%(table_name)s_%(column_0_name)s","fk":("fk_%(table_name)s_%(column_0_name)s""_%(referred_table_name)s"),}if(constraint_type.fk_constraintorconstraint_type.fk_constraint_naming_convorconstraint_type.fk_constraint_explicit_name):classRelated(self.Base):__tablename__="related"id=Column(Integer,primary_key=True)classSomeClass(Versioned,self.Base):__tablename__="sometable"id=Column(Integer,primary_key=True)x=Column(Integer)y=Column(Integer)# Index objects are copied and these have to have a new nameifconstraint_type.index_single_col:__table_args__=(Index(None,x,),)elifconstraint_type.composite_index:__table_args__=(Index(None,x,y),)elifconstraint_type.explicit_name_index:__table_args__=(Index("my_index",x,y),)# unique constraint objects are discarded.elif(constraint_type.unique_constraintorconstraint_type.unique_constraint_naming_conv):__table_args__=(UniqueConstraint(x,y),)elifconstraint_type.unique_constraint_explicit_name:__table_args__=(UniqueConstraint(x,y,name="my_uq"),)# foreign key constraint objects are copied and have the same# name, but no database in Core has any problem with this as the# names are local to the parent table.elif(constraint_type.fk_constraintorconstraint_type.fk_constraint_naming_conv):__table_args__=(ForeignKeyConstraint([x],[Related.id]),)elifconstraint_type.fk_constraint_explicit_name:__table_args__=(ForeignKeyConstraint([x],[Related.id],name="my_fk"),)else:constraint_type.fail()eq_(set(idx.name+"_history"foridxinSomeClass.__table__.indexes),set(idx.nameforidxinSomeClass.__history_mapper__.local_table.indexes),)self.create_tables()deftest_discussion_9546(self):classThingExternal(Versioned,self.Base):__tablename__="things_external"id=Column(Integer,primary_key=True)external_attribute=Column(String)classThingLocal(Versioned,self.Base):__tablename__="things_local"id=Column(Integer,ForeignKey(ThingExternal.id),primary_key=True)internal_attribute=Column(String)is_(ThingExternal.__table__,inspect(ThingExternal).local_table)classThing(self.Base):__table__=join(ThingExternal,ThingLocal)id=column_property(ThingExternal.id,ThingLocal.id)version=column_property(ThingExternal.version,ThingLocal.version)eq_ignore_whitespace(str(select(Thing)),"SELECT things_external.id, things_local.id AS id_1, ""things_external.external_attribute, things_external.version, ""things_local.version AS version_1, ""things_local.internal_attribute FROM things_external ""JOIN things_local ON things_external.id = things_local.id",)deftest_w_mapper_versioning(self):classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"use_mapper_versioning=Trueid=Column(Integer,primary_key=True)name=Column(String(50))self.create_tables()sess=self.sessionsc=SomeClass(name="sc1")sess.add(sc)sess.commit()s2=Session(sess.bind)sc2=s2.query(SomeClass).first()sc2.name="sc1modified"sc.name="sc1modified_again"sess.commit()eq_(sc.version,2)assert_raises(orm_exc.StaleDataError,s2.flush)deftest_from_null(self):classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)name=Column(String(50))self.create_tables()sess=self.sessionsc=SomeClass()sess.add(sc)sess.commit()sc.name="sc1"sess.commit()assertsc.version==2deftest_insert_null(self):classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)boole=Column(Boolean,default=False)self.create_tables()sess=self.sessionsc=SomeClass(boole=True)sess.add(sc)sess.commit()sc.boole=Nonesess.commit()sc.boole=Falsesess.commit()SomeClassHistory=SomeClass.__history_mapper__.class_eq_(sess.query(SomeClassHistory.boole).order_by(SomeClassHistory.id).all(),[(True,),(None,)],)eq_(sc.version,3)deftest_deferred(self):"""test versioning of unloaded, deferred columns."""classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)name=Column(String(50))data=deferred(Column(String(25)))self.create_tables()sess=self.sessionsc=SomeClass(name="sc1",data="somedata")sess.add(sc)sess.commit()sess.close()sc=sess.query(SomeClass).first()assert"data"notinsc.__dict__sc.name="sc1modified"sess.commit()assertsc.version==2SomeClassHistory=SomeClass.__history_mapper__.class_eq_(sess.query(SomeClassHistory).filter(SomeClassHistory.version==1).all(),[SomeClassHistory(version=1,name="sc1",data="somedata")],)deftest_joined_inheritance(self):classBaseClass(Versioned,self.Base,ComparableEntity):__tablename__="basetable"id=Column(Integer,primary_key=True)name=Column(String(50))type=Column(String(20))__mapper_args__={"polymorphic_on":type,"polymorphic_identity":"base",}classSubClassSeparatePk(BaseClass):__tablename__="subtable1"id=column_property(Column(Integer,primary_key=True),BaseClass.id)base_id=Column(Integer,ForeignKey("basetable.id"))subdata1=Column(String(50))__mapper_args__={"polymorphic_identity":"sep"}classSubClassSamePk(BaseClass):__tablename__="subtable2"id=Column(Integer,ForeignKey("basetable.id"),primary_key=True)subdata2=Column(String(50))__mapper_args__={"polymorphic_identity":"same"}self.create_tables()sess=self.sessionsep1=SubClassSeparatePk(name="sep1",subdata1="sep1subdata")base1=BaseClass(name="base1")same1=SubClassSamePk(name="same1",subdata2="same1subdata")sess.add_all([sep1,base1,same1])sess.commit()base1.name="base1mod"same1.subdata2="same1subdatamod"sep1.name="sep1mod"sess.commit()BaseClassHistory=BaseClass.__history_mapper__.class_SubClassSeparatePkHistory=(SubClassSeparatePk.__history_mapper__.class_)SubClassSamePkHistory=SubClassSamePk.__history_mapper__.class_eq_(sess.query(BaseClassHistory).order_by(BaseClassHistory.id).all(),[SubClassSeparatePkHistory(id=1,name="sep1",type="sep",version=1),BaseClassHistory(id=2,name="base1",type="base",version=1),SubClassSamePkHistory(id=3,name="same1",type="same",version=1),],)same1.subdata2="same1subdatamod2"eq_(sess.query(BaseClassHistory).order_by(BaseClassHistory.id,BaseClassHistory.version).all(),[SubClassSeparatePkHistory(id=1,name="sep1",type="sep",version=1),BaseClassHistory(id=2,name="base1",type="base",version=1),SubClassSamePkHistory(id=3,name="same1",type="same",version=1),SubClassSamePkHistory(id=3,name="same1",type="same",version=2),],)base1.name="base1mod2"eq_(sess.query(BaseClassHistory).order_by(BaseClassHistory.id,BaseClassHistory.version).all(),[SubClassSeparatePkHistory(id=1,name="sep1",type="sep",version=1),BaseClassHistory(id=2,name="base1",type="base",version=1),BaseClassHistory(id=2,name="base1mod",type="base",version=2),SubClassSamePkHistory(id=3,name="same1",type="same",version=1),SubClassSamePkHistory(id=3,name="same1",type="same",version=2),],)deftest_joined_inheritance_multilevel(self):classBaseClass(Versioned,self.Base,ComparableEntity):__tablename__="basetable"id=Column(Integer,primary_key=True)name=Column(String(50))type=Column(String(20))__mapper_args__={"polymorphic_on":type,"polymorphic_identity":"base",}classSubClass(BaseClass):__tablename__="subtable"id=column_property(Column(Integer,primary_key=True),BaseClass.id)base_id=Column(Integer,ForeignKey("basetable.id"))subdata1=Column(String(50))__mapper_args__={"polymorphic_identity":"sub"}classSubSubClass(SubClass):__tablename__="subsubtable"id=Column(Integer,ForeignKey("subtable.id"),primary_key=True)subdata2=Column(String(50))__mapper_args__={"polymorphic_identity":"subsub"}self.create_tables()SubSubHistory=SubSubClass.__history_mapper__.class_sess=self.sessionq=sess.query(SubSubHistory)self.assert_compile(q,"SELECT ""basetable_history.name AS basetable_history_name, ""basetable_history.type AS basetable_history_type, ""subsubtable_history.version AS subsubtable_history_version, ""subtable_history.version AS subtable_history_version, ""basetable_history.version AS basetable_history_version, ""subtable_history.base_id AS subtable_history_base_id, ""subtable_history.subdata1 AS subtable_history_subdata1, ""subsubtable_history.id AS subsubtable_history_id, ""subtable_history.id AS subtable_history_id, ""basetable_history.id AS basetable_history_id, ""subsubtable_history.changed AS subsubtable_history_changed, ""subtable_history.changed AS subtable_history_changed, ""basetable_history.changed AS basetable_history_changed, ""subsubtable_history.subdata2 AS subsubtable_history_subdata2 ""FROM basetable_history ""JOIN subtable_history ""ON basetable_history.id = subtable_history.base_id ""AND basetable_history.version = subtable_history.version ""JOIN subsubtable_history ON subtable_history.id = ""subsubtable_history.id AND subtable_history.version = ""subsubtable_history.version",)ssc=SubSubClass(name="ss1",subdata1="sd1",subdata2="sd2")sess.add(ssc)sess.commit()eq_(sess.query(SubSubHistory).all(),[])ssc.subdata1="sd11"ssc.subdata2="sd22"sess.commit()eq_(sess.query(SubSubHistory).all(),[SubSubHistory(name="ss1",subdata1="sd1",subdata2="sd2",type="subsub",version=1,)],)eq_(ssc,SubSubClass(name="ss1",subdata1="sd11",subdata2="sd22",version=2),)deftest_joined_inheritance_changed(self):classBaseClass(Versioned,self.Base,ComparableEntity):__tablename__="basetable"id=Column(Integer,primary_key=True)name=Column(String(50))type=Column(String(20))__mapper_args__={"polymorphic_on":type,"polymorphic_identity":"base",}classSubClass(BaseClass):__tablename__="subtable"id=Column(Integer,ForeignKey("basetable.id"),primary_key=True)__mapper_args__={"polymorphic_identity":"sep"}self.create_tables()BaseClassHistory=BaseClass.__history_mapper__.class_SubClassHistory=SubClass.__history_mapper__.class_sess=self.sessions1=SubClass(name="s1")sess.add(s1)sess.commit()s1.name="s2"sess.commit()actual_changed_base=sess.scalar(select(BaseClass.__history_mapper__.local_table.c.changed))actual_changed_sub=sess.scalar(select(SubClass.__history_mapper__.local_table.c.changed))h1=sess.query(BaseClassHistory).first()eq_(h1.changed,actual_changed_base)eq_(h1.changed,actual_changed_sub)h1=sess.query(SubClassHistory).first()eq_(h1.changed,actual_changed_base)eq_(h1.changed,actual_changed_sub)deftest_single_inheritance(self):classBaseClass(Versioned,self.Base,ComparableEntity):__tablename__="basetable"id=Column(Integer,primary_key=True)name=Column(String(50))type=Column(String(50))__mapper_args__={"polymorphic_on":type,"polymorphic_identity":"base",}classSubClass(BaseClass):subname=Column(String(50),unique=True)__mapper_args__={"polymorphic_identity":"sub"}self.create_tables()sess=self.sessionb1=BaseClass(name="b1")sc=SubClass(name="s1",subname="sc1")sess.add_all([b1,sc])sess.commit()b1.name="b1modified"BaseClassHistory=BaseClass.__history_mapper__.class_SubClassHistory=SubClass.__history_mapper__.class_eq_(sess.query(BaseClassHistory).order_by(BaseClassHistory.id,BaseClassHistory.version).all(),[BaseClassHistory(id=1,name="b1",type="base",version=1)],)sc.name="s1modified"b1.name="b1modified2"eq_(sess.query(BaseClassHistory).order_by(BaseClassHistory.id,BaseClassHistory.version).all(),[BaseClassHistory(id=1,name="b1",type="base",version=1),BaseClassHistory(id=1,name="b1modified",type="base",version=2),SubClassHistory(id=2,name="s1",type="sub",version=1),],)# test the unique constraint on the subclass# columnsc.name="modifyagain"sess.flush()deftest_unique(self):classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)name=Column(String(50),unique=True)data=Column(String(50))self.create_tables()sess=self.sessionsc=SomeClass(name="sc1",data="sc1")sess.add(sc)sess.commit()sc.data="sc1modified"sess.commit()assertsc.version==2sc.data="sc1modified2"sess.commit()assertsc.version==3deftest_relationship(self):classSomeRelated(self.Base,ComparableEntity):__tablename__="somerelated"id=Column(Integer,primary_key=True)classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)name=Column(String(50))related_id=Column(Integer,ForeignKey("somerelated.id"))related=relationship("SomeRelated",backref="classes")SomeClassHistory=SomeClass.__history_mapper__.class_self.create_tables()sess=self.sessionsc=SomeClass(name="sc1")sess.add(sc)sess.commit()assertsc.version==1sr1=SomeRelated()sc.related=sr1sess.commit()assertsc.version==2eq_(sess.query(SomeClassHistory).filter(SomeClassHistory.version==1).all(),[SomeClassHistory(version=1,name="sc1",related_id=None)],)sc.related=Noneeq_(sess.query(SomeClassHistory).order_by(SomeClassHistory.version).all(),[SomeClassHistory(version=1,name="sc1",related_id=None),SomeClassHistory(version=2,name="sc1",related_id=sr1.id),],)assertsc.version==3deftest_backref_relationship(self):classSomeRelated(self.Base,ComparableEntity):__tablename__="somerelated"id=Column(Integer,primary_key=True)name=Column(String(50))related_id=Column(Integer,ForeignKey("sometable.id"))related=relationship("SomeClass",backref="related")classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)self.create_tables()sess=self.sessionsc=SomeClass()sess.add(sc)sess.commit()assertsc.version==1sr=SomeRelated(name="sr",related=sc)sess.add(sr)sess.commit()assertsc.version==1sr.name="sr2"sess.commit()assertsc.version==1sess.delete(sr)sess.commit()assertsc.version==1deftest_create_double_flush(self):classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)name=Column(String(30))other=Column(String(30))self.create_tables()sc=SomeClass()self.session.add(sc)self.session.flush()sc.name="Foo"self.session.flush()assertsc.version==2deftest_mutate_plain_column(self):classDocument(self.Base,Versioned):__tablename__="document"id=Column(Integer,primary_key=True,autoincrement=True)name=Column(String,nullable=True)description_=Column("description",String,nullable=True)self.create_tables()document=Document()self.session.add(document)document.name="Foo"self.session.commit()document.name="Bar"self.session.commit()DocumentHistory=Document.__history_mapper__.class_v2=self.session.query(Document).one()v1=self.session.query(DocumentHistory).one()eq_(v1.id,v2.id)eq_(v2.name,"Bar")eq_(v1.name,"Foo")deftest_mutate_named_column(self):classDocument(self.Base,Versioned):__tablename__="document"id=Column(Integer,primary_key=True,autoincrement=True)name=Column(String,nullable=True)description_=Column("description",String,nullable=True)self.create_tables()document=Document()self.session.add(document)document.description_="Foo"self.session.commit()document.description_="Bar"self.session.commit()DocumentHistory=Document.__history_mapper__.class_v2=self.session.query(Document).one()v1=self.session.query(DocumentHistory).one()eq_(v1.id,v2.id)eq_(v2.description_,"Bar")eq_(v1.description_,"Foo")deftest_unique_identifiers_across_deletes(self):"""Ensure unique integer values are used for the primary table. Checks whether the database assigns the same identifier twice within the span of a table. SQLite will do this if sqlite_autoincrement is not set (e.g. SQLite's AUTOINCREMENT flag). """classSomeClass(Versioned,self.Base,ComparableEntity):__tablename__="sometable"id=Column(Integer,primary_key=True)name=Column(String(50))self.create_tables()sess=self.sessionsc=SomeClass(name="sc1")sess.add(sc)sess.commit()sess.delete(sc)sess.commit()sc2=SomeClass(name="sc2")sess.add(sc2)sess.commit()SomeClassHistory=SomeClass.__history_mapper__.class_# only one entry should exist in the history table; one()# ensures thatscdeleted=sess.query(SomeClassHistory).one()# If sc2 has the same id that deleted sc1 had,# it will fail when modified or deleted# because of the violation of the uniqueness of the primary key on# sometable_historyne_(sc2.id,scdeleted.id)# If previous assertion fails, this will also fail:sc2.name="sc2 modified"sess.commit()deftest_external_id(self):classObjectExternal(Versioned,self.Base,ComparableEntity):__tablename__="externalobjects"id1=Column(String(3),primary_key=True)id2=Column(String(3),primary_key=True)name=Column(String(50))self.create_tables()sess=self.sessionsc=ObjectExternal(id1="aaa",id2="bbb",name="sc1")sess.add(sc)sess.commit()sc.name="sc1modified"sess.commit()assertsc.version==2ObjectExternalHistory=ObjectExternal.__history_mapper__.class_eq_(sess.query(ObjectExternalHistory).all(),[ObjectExternalHistory(version=1,id1="aaa",id2="bbb",name="sc1"),],)sess.delete(sc)sess.commit()assertsess.query(ObjectExternal).count()==0eq_(sess.query(ObjectExternalHistory).all(),[ObjectExternalHistory(version=1,id1="aaa",id2="bbb",name="sc1"),ObjectExternalHistory(version=2,id1="aaa",id2="bbb",name="sc1modified"),],)sc=ObjectExternal(id1="aaa",id2="bbb",name="sc1reappeared")sess.add(sc)sess.commit()assertsc.version==3sc.name="sc1reappearedmodified"sess.commit()assertsc.version==4eq_(sess.query(ObjectExternalHistory).all(),[ObjectExternalHistory(version=1,id1="aaa",id2="bbb",name="sc1"),ObjectExternalHistory(version=2,id1="aaa",id2="bbb",name="sc1modified"),ObjectExternalHistory(version=3,id1="aaa",id2="bbb",name="sc1reappeared"),],)classTestVersioningNewBase(TestVersioning):defmake_base(self):classBase(DeclarativeBase):passself.Base=BaseclassTestVersioningUnittest(TestVersioning,unittest.TestCase):passclassTestVersioningNewBaseUnittest(TestVersioningNewBase,unittest.TestCase):passif__name__=="__main__":unittest.main()