sqlalchemy-migrate/migrate/tests/versioning/test_schemadiff.py

# -*- coding: utf-8 -*-

import os

from sqlalchemy import *

from migrate.versioning import schemadiff

from migrate.tests import fixture

class SchemaDiffBase(fixture.DB):

    level = fixture.DB.CONNECT
    def _make_table(self,*cols,**kw):
        self.table = Table('xtable', self.meta,
            Column('id',Integer(), primary_key=True),
            *cols
        )
        if kw.get('create',True):
            self.table.create()

    def _assert_diff(self,col_A,col_B):
        self._make_table(col_A)
        self.meta.clear()
        self._make_table(col_B,create=False)
        diff = self._run_diff()
        # print diff
        self.assertTrue(diff)
        self.assertEqual(1,len(diff.tables_different))
        td = list(diff.tables_different.values())[0]
        self.assertEqual(1,len(td.columns_different))
        cd = list(td.columns_different.values())[0]
        label_width = max(len(self.name1), len(self.name2))
        self.assertEqual(('Schema diffs:\n'
             '  table with differences: xtable\n'
             '    column with differences: data\n'
             '      %*s: %r\n'
             '      %*s: %r')%(
                label_width,
                self.name1,
                cd.col_A,
                label_width,
                self.name2,
                cd.col_B
                ),str(diff))

class Test_getDiffOfModelAgainstDatabase(SchemaDiffBase):
    name1 = 'model'
    name2 = 'database'

    def _run_diff(self,**kw):
        return schemadiff.getDiffOfModelAgainstDatabase(
            self.meta, self.engine, **kw
            )

    @fixture.usedb()
    def test_table_missing_in_db(self):
        self._make_table(create=False)
        diff = self._run_diff()
        self.assertTrue(diff)
        self.assertEqual('Schema diffs:\n  tables missing from %s: xtable' % self.name2,
            str(diff))

    @fixture.usedb()
    def test_table_missing_in_model(self):
        self._make_table()
        self.meta.clear()
        diff = self._run_diff()
        self.assertTrue(diff)
        self.assertEqual('Schema diffs:\n  tables missing from %s: xtable' % self.name1,
            str(diff))

    @fixture.usedb()
    def test_column_missing_in_db(self):
        # db
        Table('xtable', self.meta,
              Column('id',Integer(), primary_key=True),
              ).create()
        self.meta.clear()
        # model
        self._make_table(
            Column('xcol',Integer()),
            create=False
            )
        # run diff
        diff = self._run_diff()
        self.assertTrue(diff)
        self.assertEqual('Schema diffs:\n'
            '  table with differences: xtable\n'
            '    %s missing these columns: xcol' % self.name2,
            str(diff))

    @fixture.usedb()
    def test_column_missing_in_model(self):
        # db
        self._make_table(
            Column('xcol',Integer()),
            )
        self.meta.clear()
        # model
        self._make_table(
            create=False
            )
        # run diff
        diff = self._run_diff()
        self.assertTrue(diff)
        self.assertEqual('Schema diffs:\n'
            '  table with differences: xtable\n'
            '    %s missing these columns: xcol' % self.name1,
            str(diff))

    @fixture.usedb()
    def test_exclude_tables(self):
        # db
        Table('ytable', self.meta,
              Column('id',Integer(), primary_key=True),
              ).create()
        Table('ztable', self.meta,
              Column('id',Integer(), primary_key=True),
              ).create()
        self.meta.clear()
        # model
        self._make_table(
            create=False
            )
        Table('ztable', self.meta,
              Column('id',Integer(), primary_key=True),
              )
        # run diff
        diff = self._run_diff(excludeTables=('xtable','ytable'))
        # ytable only in database
        # xtable only in model
        # ztable identical on both
        # ...so we expect no diff!
        self.assertFalse(diff)
        self.assertEqual('No schema diffs',str(diff))

    @fixture.usedb()
    def test_identical_just_pk(self):
        self._make_table()
        diff = self._run_diff()
        self.assertFalse(diff)
        self.assertEqual('No schema diffs',str(diff))


    @fixture.usedb()
    def test_different_type(self):
        self._assert_diff(
            Column('data', String(10)),
            Column('data', Integer()),
            )

    @fixture.usedb()
    def test_int_vs_float(self):
        self._assert_diff(
            Column('data', Integer()),
            Column('data', Float()),
            )

    # NOTE(mriedem): The ibm_db_sa driver handles the Float() as a DOUBLE()
    # which extends Numeric() but isn't defined in sqlalchemy.types, so we
    # can't check for it as a special case like is done in schemadiff.ColDiff.
    @fixture.usedb(not_supported='ibm_db_sa')
    def test_float_vs_numeric(self):
        self._assert_diff(
            Column('data', Float()),
            Column('data', Numeric()),
            )

    @fixture.usedb()
    def test_numeric_precision(self):
        self._assert_diff(
            Column('data', Numeric(precision=5)),
            Column('data', Numeric(precision=6)),
            )

    @fixture.usedb()
    def test_numeric_scale(self):
        self._assert_diff(
            Column('data', Numeric(precision=6,scale=0)),
            Column('data', Numeric(precision=6,scale=1)),
            )

    @fixture.usedb()
    def test_string_length(self):
        self._assert_diff(
            Column('data', String(10)),
            Column('data', String(20)),
            )

    @fixture.usedb()
    def test_integer_identical(self):
        self._make_table(
            Column('data', Integer()),
            )
        diff = self._run_diff()
        self.assertEqual('No schema diffs',str(diff))
        self.assertFalse(diff)

    @fixture.usedb()
    def test_string_identical(self):
        self._make_table(
            Column('data', String(10)),
            )
        diff = self._run_diff()
        self.assertEqual('No schema diffs',str(diff))
        self.assertFalse(diff)

    @fixture.usedb()
    def test_text_identical(self):
        self._make_table(
            Column('data', Text),
            )
        diff = self._run_diff()
        self.assertEqual('No schema diffs',str(diff))
        self.assertFalse(diff)

class Test_getDiffOfModelAgainstModel(Test_getDiffOfModelAgainstDatabase):
    name1 = 'metadataA'
    name2 = 'metadataB'

    def _run_diff(self,**kw):
        db_meta= MetaData()
        db_meta.reflect(self.engine)
        return schemadiff.getDiffOfModelAgainstModel(
            self.meta, db_meta, **kw
            )