~sirn/fanboi2

8988d5b3a399aeece57dc6a5b8041f242b436480 — Kridsada Thanabulpong 4 years ago 0b82013
Integrate versioned with existing models.
M fanboi2/models/board.py => fanboi2/models/board.py +2 -2
@@ 2,7 2,7 @@ from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import synonym
from sqlalchemy.sql.schema import Column
from sqlalchemy.sql.sqltypes import String, Text, Unicode
from ._base import Base, JsonType
from ._base import Base, JsonType, Versioned


DEFAULT_BOARD_CONFIG = {


@@ 13,7 13,7 @@ DEFAULT_BOARD_CONFIG = {
}


class Board(Base):
class Board(Versioned, Base):
    """Model class for board. This model serve as a category to topic and
    also holds settings regarding how posts are created and displayed. It
    should always be accessed using :attr:`slug`.

M fanboi2/models/post.py => fanboi2/models/post.py +3 -2
@@ 3,10 3,10 @@ from sqlalchemy.orm import backref, relationship
from sqlalchemy.sql import func, select
from sqlalchemy.sql.schema import Column, ForeignKey, UniqueConstraint
from sqlalchemy.sql.sqltypes import Integer, String, Text, Boolean
from ._base import Base
from ._base import Base, Versioned


class Post(Base):
class Post(Versioned, Base):
    """Model class for posts. Each content in a :class:`Topic` and metadata
    regarding its poster are stored here. It has :attr:`number` which is a
    sequential number specifying its position within :class:`Topic`.


@@ 24,6 24,7 @@ class Post(Base):
    topic = relationship('Topic',
                         backref=backref('posts',
                                         lazy='dynamic',
                                         cascade='all,delete',
                                         order_by='Post.number'))



M fanboi2/models/topic.py => fanboi2/models/topic.py +3 -2
@@ 4,11 4,11 @@ from sqlalchemy.orm import backref, column_property, relationship
from sqlalchemy.sql import desc, func, select
from sqlalchemy.sql.schema import Column, ForeignKey
from sqlalchemy.sql.sqltypes import Integer, Enum, Unicode
from ._base import Base, DBSession
from ._base import Base, DBSession, Versioned
from .post import Post


class Topic(Base):
class Topic(Versioned, Base):
    """Model class for topic. This model only holds topic metadata such as
    title or its associated board. The actual content of a topic belongs
    to :class:`Post`.


@@ 22,6 22,7 @@ class Topic(Base):
    board = relationship('Board',
                         backref=backref('topics',
                                         lazy='dynamic',
                                         cascade='all,delete',
                                         order_by="desc(func.coalesce("
                                                  "Topic.bumped_at,"
                                                  "Topic.created_at))"))

M fanboi2/tests/test_models.py => fanboi2/tests/test_models.py +233 -24
@@ 959,14 959,108 @@ class TestVersioned(unittest.TestCase):
        self._dropTable(Base)


class BoardModelTest(ModelMixin, unittest.TestCase):
class TestBoardModel(ModelMixin, unittest.TestCase):

    def test_relations(self):
        board = self._makeBoard(title="Foobar", slug="foo")
        self.assertEqual([], list(board.topics))

    def test_relations_cascade(self):
        from sqlalchemy import inspect
        board1 = self._makeBoard(title="Foobar", slug="foo")
        board2 = self._makeBoard(title="Lorem", slug="lorem")
        topic1 = self._makeTopic(board=board1, title="Heavenly Moon")
        topic2 = self._makeTopic(board=board1, title="Beastie Starter")
        topic3 = self._makeTopic(board=board2, title="Evans")
        post1 = self._makePost(topic=topic1, body='Foobar')
        post2 = self._makePost(topic=topic1, body='Bazz')
        post3 = self._makePost(topic=topic2, body='Hoge')
        post4 = self._makePost(topic=topic3, body='Fuzz')
        self.assertEqual({topic2, topic1}, set(board1.topics))
        self.assertEqual({topic3}, set(board2.topics))
        self.assertEqual({post2, post1}, set(topic1.posts))
        self.assertEqual({post3}, set(topic2.posts))
        self.assertEqual({post4}, set(topic3.posts))
        DBSession.delete(board1)
        DBSession.flush()
        self.assertTrue(inspect(board1).deleted)
        self.assertFalse(inspect(board2).deleted)
        self.assertTrue(inspect(topic1).deleted)
        self.assertTrue(inspect(topic2).deleted)
        self.assertFalse(inspect(topic3).deleted)
        self.assertTrue(inspect(post1).deleted)
        self.assertTrue(inspect(post2).deleted)
        self.assertTrue(inspect(post3).deleted)
        self.assertFalse(inspect(post4).deleted)

    def test_versioned(self):
        from fanboi2.models import Board
        BoardHistory = Board.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        self.assertEqual(board.version, 1)
        self.assertEqual(DBSession.query(BoardHistory).count(), 0)
        board.title = 'Foobar and Baz'
        DBSession.add(board)
        DBSession.flush()
        self.assertEqual(board.version, 2)
        self.assertEqual(DBSession.query(BoardHistory).count(), 1)
        board_v1 = DBSession.query(BoardHistory).filter_by(version=1).one()
        self.assertEqual(board_v1.id, board.id)
        self.assertEqual(board_v1.title, 'Foobar')
        self.assertEqual(board_v1.change_type, 'update')
        self.assertEqual(board_v1.version, 1)
        self.assertIsNotNone(board_v1.changed_at)
        self.assertIsNotNone(board_v1.created_at)
        self.assertIsNone(board_v1.updated_at)

    def test_versioned_deleted(self):
        from sqlalchemy import inspect
        from fanboi2.models import Board
        BoardHistory = Board.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        DBSession.delete(board)
        DBSession.flush()
        self.assertTrue(inspect(board).deleted)
        self.assertEqual(DBSession.query(BoardHistory).count(), 1)
        board_v1 = DBSession.query(BoardHistory).filter_by(version=1).one()
        self.assertEqual(board_v1.id, board.id)
        self.assertEqual(board_v1.title, 'Foobar')
        self.assertEqual(board_v1.change_type, 'delete')
        self.assertEqual(board_v1.version, 1)

    def test_versioned_deleted_cascade(self):
        from sqlalchemy import inspect
        from fanboi2.models import Board, Topic, Post
        BoardHistory = Board.__history_mapper__.class_
        TopicHistory = Topic.__history_mapper__.class_
        PostHistory = Post.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        topic = self._makeTopic(board=board, title='Heavenly Moon')
        post = self._makePost(topic=topic, body='Foobar')
        self.assertEqual(DBSession.query(BoardHistory).count(), 0)
        self.assertEqual(DBSession.query(TopicHistory).count(), 0)
        self.assertEqual(DBSession.query(PostHistory).count(), 0)
        DBSession.delete(board)
        DBSession.flush()
        self.assertEqual(DBSession.query(BoardHistory).count(), 1)
        self.assertEqual(DBSession.query(TopicHistory).count(), 1)
        self.assertEqual(DBSession.query(PostHistory).count(), 1)
        board_v1 = DBSession.query(BoardHistory).filter_by(version=1).one()
        self.assertEqual(board_v1.id, board.id)
        self.assertEqual(board_v1.change_type, 'delete')
        self.assertEqual(board_v1.version, 1)
        topic_v1 = DBSession.query(TopicHistory).filter_by(version=1).one()
        self.assertEqual(topic_v1.id, topic.id)
        self.assertEqual(topic_v1.board_id, board.id)
        self.assertEqual(topic_v1.change_type, 'delete')
        self.assertEqual(topic_v1.version, 1)
        post_v1 = DBSession.query(PostHistory).filter_by(version=1).one()
        self.assertEqual(post_v1.id, post.id)
        self.assertEqual(post_v1.topic_id, topic.id)
        self.assertEqual(post_v1.change_type, 'delete')
        self.assertEqual(post_v1.version, 1)

    def test_settings(self):
        from fanboi2.models import DBSession
        from fanboi2.models.board import DEFAULT_BOARD_CONFIG
        board = self._makeBoard(title="Foobar", slug="Foo")
        self.assertEqual(board.settings, DEFAULT_BOARD_CONFIG)


@@ 981,12 1075,9 @@ class BoardModelTest(ModelMixin, unittest.TestCase):
        from fanboi2.models import Topic
        board1 = self._makeBoard(title="Foobar", slug="foo")
        board2 = self._makeBoard(title="Lorem", slug="lorem")
        topic1 = Topic(board=board1, title="Heavenly Moon")
        topic2 = Topic(board=board1, title="Beastie Starter")
        topic3 = Topic(board=board1, title="Evans")
        DBSession.add(topic1)
        DBSession.add(topic2)
        DBSession.add(topic3)
        topic1 = self._makeTopic(board=board1, title="Heavenly Moon")
        topic2 = self._makeTopic(board=board1, title="Beastie Starter")
        topic3 = self._makeTopic(board=board1, title="Evans")
        DBSession.flush()
        self.assertEqual({topic3, topic2, topic1}, set(board1.topics))
        self.assertEqual([], list(board2.topics))


@@ 995,34 1086,29 @@ class BoardModelTest(ModelMixin, unittest.TestCase):
        from datetime import datetime, timedelta
        from fanboi2.models import Topic, Post
        board = self._makeBoard(title="Foobar", slug="foobar")
        topic1 = Topic(board=board, title="First")
        topic2 = Topic(board=board, title="Second")
        topic3 = Topic(board=board, title="Third")
        topic4 = Topic(board=board, title="Fourth")
        topic5 = Topic(
        topic1 = self._makeTopic(board=board, title="First")
        topic2 = self._makeTopic(board=board, title="Second")
        topic3 = self._makeTopic(board=board, title="Third")
        topic4 = self._makeTopic(board=board, title="Fourth")
        topic5 = self._makeTopic(
            board=board,
            title="Fifth",
            created_at=datetime.now() + timedelta(seconds=10))
        DBSession.add(topic1)
        DBSession.add(topic2)
        DBSession.add(topic3)
        DBSession.flush()
        DBSession.add(Post(topic=topic1, ip_address="1.1.1.1", body="!!1"))
        DBSession.add(Post(topic=topic4, ip_address="1.1.1.1", body="Baz"))
        self._makePost(topic=topic1, ip_address="1.1.1.1", body="!!1")
        self._makePost(topic=topic4, ip_address="1.1.1.1", body="Baz")
        mappings = ((topic3, 3, True), (topic2, 5, True), (topic4, 8, False))
        for obj, offset, bump in mappings:
            DBSession.add(Post(
            self._makePost(
                topic=obj,
                ip_address="1.1.1.1",
                body="Foo",
                created_at=datetime.now() + timedelta(seconds=offset),
                bumped=bump))
        DBSession.add(Post(
                bumped=bump)
        self._makePost(
            topic=topic5,
            ip_address="1.1.1.1",
            body="Hax",
            bumped=False))
        DBSession.flush()
            bumped=False)
        DBSession.refresh(board)
        self.assertEqual([topic5, topic2, topic3, topic1, topic4],
                         list(board.topics))


@@ 1037,6 1123,89 @@ class TestTopicModel(ModelMixin, unittest.TestCase):
        self.assertEqual([], list(topic.posts))
        self.assertEqual([topic], list(board.topics))

    def test_relations_cascade(self):
        from sqlalchemy import inspect
        from fanboi2.models import Post
        board = self._makeBoard(title='Foobar', slug='foo')
        topic1 = self._makeTopic(board=board, title='Shamshir Dance')
        topic2 = self._makeTopic(board=board, title='Nyoah Sword Dance')
        post1 = self._makePost(topic=topic1, body='Lorem ipsum')
        post2 = self._makePost(topic=topic1, body='Dolor sit amet')
        post3 = self._makePost(topic=topic2, body='Quas magnam et')
        self.assertEqual({post2, post1}, set(topic1.posts))
        self.assertEqual({post3}, set(topic2.posts))
        DBSession.delete(topic1)
        DBSession.flush()
        self.assertTrue(inspect(topic1).deleted)
        self.assertFalse(inspect(topic2).deleted)
        self.assertTrue(inspect(post1).deleted)
        self.assertTrue(inspect(post2).deleted)
        self.assertFalse(inspect(post3).deleted)

    def test_versioned(self):
        from fanboi2.models import Topic
        TopicHistory = Topic.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        topic = self._makeTopic(board=board, title='Foobar')
        self.assertEqual(topic.version, 1)
        self.assertEqual(DBSession.query(TopicHistory).count(), 0)
        topic.title = 'Foobar Baz'
        DBSession.add(topic)
        DBSession.flush()
        self.assertEqual(topic.version, 2)
        self.assertEqual(DBSession.query(TopicHistory).count(), 1)
        topic_v1 = DBSession.query(TopicHistory).filter_by(version=1).one()
        self.assertEqual(topic_v1.id, topic.id)
        self.assertEqual(topic_v1.board_id, topic.board_id)
        self.assertEqual(topic_v1.title, 'Foobar')
        self.assertEqual(topic_v1.change_type, 'update')
        self.assertEqual(topic_v1.version, 1)
        self.assertIsNotNone(topic_v1.changed_at)
        self.assertIsNotNone(topic_v1.created_at)
        self.assertIsNone(topic_v1.updated_at)

    def test_versioned_deleted(self):
        from sqlalchemy import inspect
        from fanboi2.models import Topic
        TopicHistory = Topic.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        topic = self._makeTopic(board=board, title='Foobar')
        DBSession.delete(topic)
        DBSession.flush()
        self.assertTrue(inspect(topic).deleted)
        self.assertEqual(DBSession.query(TopicHistory).count(), 1)
        topic_v1 = DBSession.query(TopicHistory).filter_by(version=1).one()
        self.assertEqual(topic_v1.id, topic.id)
        self.assertEqual(topic_v1.board_id, topic.board_id)
        self.assertEqual(topic_v1.title, 'Foobar')
        self.assertEqual(topic_v1.change_type, 'delete')
        self.assertEqual(topic_v1.version, 1)

    def test_versioned_deleted_cascade(self):
        from sqlalchemy import inspect
        from fanboi2.models import Topic, Post
        TopicHistory = Topic.__history_mapper__.class_
        PostHistory = Post.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        topic = self._makeTopic(board=board, title='Cosmic Agenda')
        post = self._makePost(topic=topic, body='Foobar')
        self.assertEqual(DBSession.query(TopicHistory).count(), 0)
        self.assertEqual(DBSession.query(PostHistory).count(), 0)
        DBSession.delete(topic)
        DBSession.flush()
        self.assertEqual(DBSession.query(TopicHistory).count(), 1)
        self.assertEqual(DBSession.query(PostHistory).count(), 1)
        topic_v1 = DBSession.query(TopicHistory).filter_by(version=1).one()
        self.assertEqual(topic_v1.id, topic.id)
        self.assertEqual(topic_v1.board_id, board.id)
        self.assertEqual(topic_v1.change_type, 'delete')
        self.assertEqual(topic_v1.version, 1)
        post_v1 = DBSession.query(PostHistory).filter_by(version=1).one()
        self.assertEqual(post_v1.id, post.id)
        self.assertEqual(post_v1.topic_id, topic.id)
        self.assertEqual(post_v1.change_type, 'delete')
        self.assertEqual(post_v1.version, 1)

    def test_posts(self):
        board = self._makeBoard(title="Foobar", slug="foo")
        topic1 = self._makeTopic(board=board, title="Lorem ipsum dolor")


@@ 1254,6 1423,46 @@ class TestPostModel(ModelMixin, unittest.TestCase):
        self.assertEqual(post.topic, topic)
        self.assertEqual([post], list(topic.posts))

    def test_versioned(self):
        from fanboi2.models import Post
        PostHistory = Post.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        topic = self._makeTopic(board=board, title='Lorem ipsum dolor')
        post = self._makePost(topic=topic, body='Foobar baz')
        self.assertEqual(post.version, 1)
        self.assertEqual(DBSession.query(PostHistory).count(), 0)
        post.body = 'Foobar baz updated'
        DBSession.add(post)
        DBSession.flush()
        self.assertEqual(post.version, 2)
        self.assertEqual(DBSession.query(PostHistory).count(), 1)
        post_v1 = DBSession.query(PostHistory).filter_by(version=1).one()
        self.assertEqual(post_v1.id, post.id)
        self.assertEqual(post_v1.topic_id, topic.id)
        self.assertEqual(post_v1.body, 'Foobar baz')
        self.assertEqual(post_v1.version, 1)
        self.assertIsNotNone(post_v1.changed_at)
        self.assertIsNotNone(post_v1.created_at)
        self.assertIsNone(post_v1.updated_at)

    def test_versioned_deleted(self):
        from sqlalchemy import inspect
        from fanboi2.models import Post
        PostHistory = Post.__history_mapper__.class_
        board = self._makeBoard(title='Foobar', slug='foo')
        topic = self._makeTopic(board=board, title='Lorem ipsum dolor')
        post = self._makePost(topic=topic, body='Foobar baz')
        DBSession.delete(post)
        DBSession.flush()
        self.assertTrue(inspect(post).deleted)
        self.assertEqual(DBSession.query(PostHistory).count(), 1)
        post_v1 = DBSession.query(PostHistory).filter_by(version=1).one()
        self.assertEqual(post_v1.id, post.id)
        self.assertEqual(post_v1.topic_id, topic.id)
        self.assertEqual(post_v1.body, 'Foobar baz')
        self.assertEqual(post_v1.change_type, 'delete')
        self.assertEqual(post_v1.version, 1)

    def test_number(self):
        board = self._makeBoard(title="Foobar", slug="foo")
        topic1 = self._makeTopic(board=board, title="Numbering one")

A migration/versions/0d0f281cc4ec_create_versioning_tables.py => migration/versions/0d0f281cc4ec_create_versioning_tables.py +71 -0
@@ 0,0 1,71 @@
"""create versioning tables

Revision ID: 0d0f281cc4ec
Revises: 84a168aadc17
Create Date: 2016-10-24 12:23:08.537936

"""

# revision identifiers, used by Alembic.
revision = '0d0f281cc4ec'
down_revision = '84a168aadc17'

from alembic import op
from fanboi2.models import JsonType
from sqlalchemy.dialects.postgresql import ENUM
import sqlalchemy as sa


def upgrade():
    op.create_table('board_history',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('slug', sa.String(length=64), nullable=False),
        sa.Column('title', sa.Unicode(length=255), nullable=False),
        sa.Column('agreements', sa.Text, nullable=True),
        sa.Column('description', sa.Text, nullable=True),
        sa.Column('settings', JsonType(), nullable=False),
        sa.Column('version', sa.Integer, nullable=False),
        sa.Column('change_type', sa.String),
        sa.Column('changed_at', sa.DateTime(timezone=True)),
        sa.PrimaryKeyConstraint('id', 'version', name='pk_board'),
    )
    op.create_table('topic_history',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('board_id', sa.Integer(), nullable=False),
        sa.Column('title', sa.Unicode(length=255), nullable=False),
        sa.Column('status',
                  ENUM('open', 'locked', 'archived',
                       name='topic_status',
                       create_type=False),
                  nullable=False),
        sa.Column('version', sa.Integer, nullable=False),
        sa.Column('change_type', sa.String),
        sa.Column('changed_at', sa.DateTime(timezone=True)),
        sa.PrimaryKeyConstraint('id', 'version', name='pk_topic')
    )
    op.create_table('post_history',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
        sa.Column('topic_id', sa.Integer(), nullable=False),
        sa.Column('ip_address', sa.String(), nullable=False),
        sa.Column('ident', sa.String(length=32), nullable=True),
        sa.Column('number', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(), nullable=False),
        sa.Column('body', sa.Text(), nullable=False),
        sa.Column('bumped', sa.Boolean(), nullable=False),
        sa.Column('version', sa.Integer, nullable=False),
        sa.Column('change_type', sa.String),
        sa.Column('changed_at', sa.DateTime(timezone=True)),
        sa.PrimaryKeyConstraint('id', 'version', name='pk_post')
    )


def downgrade():
    op.drop_table('post_history')
    op.drop_table('topic_history')
    op.drop_table('board_history')

M migration/versions/38f5ad30fe6f_create_initial_table.py => migration/versions/38f5ad30fe6f_create_initial_table.py +0 -1
@@ 36,7 36,6 @@ def upgrade():
        sa.Column('title', sa.Unicode(length=255), nullable=False),
        sa.Column('status',
                  sa.Enum('open', 'locked', 'archived', name='topic_status'),
                  default='open',
                  nullable=False),
        sa.ForeignKeyConstraint(['board_id'], ['board.id'], ),
        sa.PrimaryKeyConstraint('id')

A migration/versions/84a168aadc17_add_versioning_columns.py => migration/versions/84a168aadc17_add_versioning_columns.py +34 -0
@@ 0,0 1,34 @@
"""add versioning columns

Revision ID: 84a168aadc17
Revises: c71cae24d1
Create Date: 2016-10-24 11:56:06.244550

"""

# revision identifiers, used by Alembic.
revision = '84a168aadc17'
down_revision = 'c71cae24d1'

from alembic import op
from sqlalchemy import sql
import sqlalchemy as sa


def _add_version_column(table):
    op.add_column(table, sa.Column('version', sa.Integer))
    version_column = sql.table(table, sql.column('version'))
    op.execute(version_column.update().values(version=1))
    op.alter_column(table, 'version', nullable=False)


def upgrade():
    _add_version_column('board')
    _add_version_column('topic')
    _add_version_column('post')


def downgrade():
    op.drop_column('board', 'version')
    op.drop_column('topic', 'version')
    op.drop_column('post', 'version')