mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-06-03 22:35:46 -04:00
some work on the #1401 branch has discovered that this very old test is pointless;
it slightly exercises post_update but not much compared to modern tests, and also illustrates a mapping that doesn't actually work currently.
This commit is contained in:
Mike Bayer
@@ -277,154 +277,6 @@ class CompositeSelfRefFKTest(fixtures.MappedTest):
|
||||
assert sess.query(Employee).\
|
||||
get([c2.company_id, 3]).reports_to.name == 'emp5'
|
||||
|
||||
class ComplexPostUpdateTest(fixtures.MappedTest):
|
||||
@classmethod
|
||||
def define_tables(cls, metadata):
|
||||
Table("jobs", metadata,
|
||||
Column("jobno", sa.Unicode(15), primary_key=True),
|
||||
Column("created", sa.DateTime, nullable=False,
|
||||
default=datetime.datetime.now),
|
||||
Column("deleted", sa.Boolean, nullable=False, default=False))
|
||||
|
||||
Table("pageversions", metadata,
|
||||
Column("jobno", sa.Unicode(15), primary_key=True),
|
||||
Column("pagename", sa.Unicode(30), primary_key=True),
|
||||
Column("version", Integer, primary_key=True, default=1),
|
||||
Column("created", sa.DateTime, nullable=False,
|
||||
default=datetime.datetime.now),
|
||||
Column("md5sum", String(32)),
|
||||
Column("width", Integer, nullable=False, default=0),
|
||||
Column("height", Integer, nullable=False, default=0),
|
||||
sa.ForeignKeyConstraint(
|
||||
["jobno", "pagename"],
|
||||
["pages.jobno", "pages.pagename"]))
|
||||
|
||||
Table("pages", metadata,
|
||||
Column("jobno", sa.Unicode(15), ForeignKey("jobs.jobno"),
|
||||
primary_key=True),
|
||||
Column("pagename", sa.Unicode(30), primary_key=True),
|
||||
Column("created", sa.DateTime, nullable=False,
|
||||
default=datetime.datetime.now),
|
||||
Column("deleted", sa.Boolean, nullable=False, default=False),
|
||||
Column("current_version", Integer))
|
||||
|
||||
Table("pagecomments", metadata,
|
||||
Column("jobno", sa.Unicode(15), primary_key=True),
|
||||
Column("pagename", sa.Unicode(30), primary_key=True),
|
||||
Column("comment_id", Integer, primary_key=True,
|
||||
autoincrement=False),
|
||||
Column("content", sa.UnicodeText),
|
||||
sa.ForeignKeyConstraint(
|
||||
["jobno", "pagename"],
|
||||
["pages.jobno", "pages.pagename"]))
|
||||
|
||||
@classmethod
|
||||
def setup_mappers(cls):
|
||||
pageversions, pagecomments, jobs, pages = (cls.tables.pageversions,
|
||||
cls.tables.pagecomments,
|
||||
cls.tables.jobs,
|
||||
cls.tables.pages)
|
||||
|
||||
class Job(cls.Basic):
|
||||
def create_page(self, pagename):
|
||||
return Page(job=self, pagename=pagename)
|
||||
class PageVersion(cls.Basic):
|
||||
def __init__(self, page=None, version=None):
|
||||
self.page = page
|
||||
self.version = version
|
||||
class Page(cls.Basic):
|
||||
def __init__(self, job=None, pagename=None):
|
||||
self.job = job
|
||||
self.pagename = pagename
|
||||
self.currentversion = PageVersion(self, 1)
|
||||
def add_version(self):
|
||||
self.currentversion = PageVersion(
|
||||
page=self, version=self.currentversion.version+1)
|
||||
comment = self.add_comment()
|
||||
comment.closeable = False
|
||||
comment.content = u'some content'
|
||||
return self.currentversion
|
||||
def add_comment(self):
|
||||
nextnum = max([-1] +
|
||||
[c.comment_id for c in self.comments]) + 1
|
||||
newcomment = PageComment()
|
||||
newcomment.comment_id = nextnum
|
||||
self.comments.append(newcomment)
|
||||
newcomment.created_version = self.currentversion.version
|
||||
return newcomment
|
||||
class PageComment(cls.Basic):
|
||||
pass
|
||||
|
||||
mapper(Job, jobs)
|
||||
mapper(PageVersion, pageversions)
|
||||
mapper(Page, pages, properties={
|
||||
'job': relationship(
|
||||
Job,
|
||||
backref=backref('pages',
|
||||
cascade="all, delete-orphan",
|
||||
order_by=pages.c.pagename)),
|
||||
'currentversion': relationship(
|
||||
PageVersion,
|
||||
uselist=False,
|
||||
primaryjoin=sa.and_(
|
||||
pages.c.jobno==pageversions.c.jobno,
|
||||
pages.c.pagename==pageversions.c.pagename,
|
||||
pages.c.current_version==pageversions.c.version),
|
||||
post_update=True),
|
||||
'versions': relationship(
|
||||
PageVersion,
|
||||
cascade="all, delete-orphan",
|
||||
primaryjoin=sa.and_(pages.c.jobno==pageversions.c.jobno,
|
||||
pages.c.pagename==pageversions.c.pagename),
|
||||
order_by=pageversions.c.version,
|
||||
backref=backref('page',lazy='joined')
|
||||
)})
|
||||
mapper(PageComment, pagecomments, properties={
|
||||
'page': relationship(
|
||||
Page,
|
||||
primaryjoin=sa.and_(pages.c.jobno==pagecomments.c.jobno,
|
||||
pages.c.pagename==pagecomments.c.pagename),
|
||||
backref=backref("comments",
|
||||
cascade="all, delete-orphan",
|
||||
order_by=pagecomments.c.comment_id))})
|
||||
|
||||
def test_basic(self):
|
||||
"""A combination of complicated join conditions with post_update."""
|
||||
|
||||
Job = self.classes.Job
|
||||
|
||||
|
||||
j1 = Job(jobno=u'somejob')
|
||||
j1.create_page(u'page1')
|
||||
j1.create_page(u'page2')
|
||||
j1.create_page(u'page3')
|
||||
|
||||
j2 = Job(jobno=u'somejob2')
|
||||
j2.create_page(u'page1')
|
||||
j2.create_page(u'page2')
|
||||
j2.create_page(u'page3')
|
||||
|
||||
j2.pages[0].add_version()
|
||||
j2.pages[0].add_version()
|
||||
j2.pages[1].add_version()
|
||||
|
||||
s = create_session()
|
||||
s.add_all((j1, j2))
|
||||
|
||||
s.flush()
|
||||
|
||||
s.expunge_all()
|
||||
j = s.query(Job).filter_by(jobno=u'somejob').one()
|
||||
oldp = list(j.pages)
|
||||
j.pages = []
|
||||
|
||||
s.flush()
|
||||
|
||||
s.expunge_all()
|
||||
j = s.query(Job).filter_by(jobno=u'somejob2').one()
|
||||
j.pages[1].current_version = 12
|
||||
s.delete(j)
|
||||
s.flush()
|
||||
|
||||
class FKsAsPksTest(fixtures.MappedTest):
|
||||
"""Syncrules on foreign keys that are also primary"""
|
||||
|
||||
Reference in New Issue
Block a user