x/sqlalchemy-migrate
Code Issues Proposed changes

run shell diff tests with correct environment

Browse Source
This commit is contained in:
iElectric 2010-05-03 12:03:46 +02:00
parent aeb2b72b20
commit 6ddf8e0212
3 changed files with 43 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/fixture/models.py
View File

@ -7,8 +7,8 @@ meta = MetaData()
tmp_account_rundiffs = Table('tmp_account_rundiffs', meta_rundiffs, tmp_account_rundiffs = Table('tmp_account_rundiffs', meta_rundiffs,
Column('id', Integer, primary_key=True), Column('id', Integer, primary_key=True),
Column('login', String(40)), Column('login', Text()),
Column('passwd', String(40)), Column('passwd', Text()),
) )
tmp_sql_table = Table('tmp_sql_table', meta, Column('id', Integer)) tmp_sql_table = Table('tmp_sql_table', meta, Column('id', Integer))

35
tests/versioning/test_schemadiff.py
View File

@ -1,3 +1,5 @@
# -*- coding: utf-8 -*-
import os import os
import sqlalchemy import sqlalchemy
@ -11,35 +13,44 @@ from tests import fixture
class TestSchemaDiff(fixture.DB): class TestSchemaDiff(fixture.DB):
level=fixture.DB.CONNECT
table_name = 'tmp_schemadiff' table_name = 'tmp_schemadiff'
level = fixture.DB.CONNECT
def _setup(self, url): def _setup(self, url):
super(TestSchemaDiff, self)._setup(url) super(TestSchemaDiff, self)._setup(url)
self.meta = MetaData(self.engine, reflect=True) self.meta = MetaData(self.engine, reflect=True)
self.meta.drop_all() # in case junk tables are lying around in the test database self.meta.drop_all() # in case junk tables are lying around in the test database
self.meta = MetaData(self.engine, reflect=True) # needed if we just deleted some tables self.meta = MetaData(self.engine, reflect=True) # needed if we just deleted some tables
self.table = Table(self.table_name,self.meta, self.table = Table(self.table_name, self.meta,
Column('id',Integer(),primary_key=True), Column('id',Integer(), primary_key=True),
Column('name',UnicodeText()), Column('name', UnicodeText()),
Column('data',UnicodeText()), Column('data', UnicodeText()),
) )
WANT_ENGINE_ECHO = os.environ.get('WANT_ENGINE_ECHO', 'F') # to get debugging: set this to T and run py.test with --pdb
if WANT_ENGINE_ECHO == 'T':
self.engine.echo = True
def _teardown(self): def _teardown(self):
if self.table.exists(): if self.table.exists():
#self.table.drop() # bummer, this doesn't work because the list of tables is out of date, but calling reflect didn't work
self.meta = MetaData(self.engine, reflect=True) self.meta = MetaData(self.engine, reflect=True)
self.meta.drop_all() self.meta.drop_all()
super(TestSchemaDiff, self)._teardown() super(TestSchemaDiff, self)._teardown()
def _applyLatestModel(self): def _applyLatestModel(self):
diff = schemadiff.getDiffOfModelAgainstDatabase(self.meta, self.engine, excludeTables=['migrate_version']) diff = schemadiff.getDiffOfModelAgainstDatabase(self.meta, self.engine, excludeTables=['migrate_version'])
genmodel.ModelGenerator(diff).applyModel() genmodel.ModelGenerator(diff).applyModel()
# TODO: support for diff/generator to extract differences between columns
#@fixture.usedb()
#def test_type_diff(self):
#"""Basic test for correct type diff"""
#self.table.create()
#self.meta = MetaData(self.engine)
#self.table = Table(self.table_name, self.meta,
#Column('id', Integer(), primary_key=True),
#Column('name', Unicode(45)),
#Column('data', Integer),
#)
#diff = schemadiff.getDiffOfModelAgainstDatabase\
#(self.meta, self.engine, excludeTables=['migrate_version'])
@fixture.usedb() @fixture.usedb()
def test_rundiffs(self): def test_rundiffs(self):

31
tests/versioning/test_shell.py
View File

@ -19,6 +19,8 @@ from tests.fixture import Shell, DB, usedb
from tests.fixture import models from tests.fixture import models
HERE = os.path.dirname(os.path.abspath(__file__))
class TestShellCommands(Shell): class TestShellCommands(Shell):
"""Tests migrate.py commands""" """Tests migrate.py commands"""
@ -424,8 +426,8 @@ class TestShellDatabase(Shell, DB):
meta = MetaData(migrate_engine) meta = MetaData(migrate_engine)
account = Table('account', meta, account = Table('account', meta,
Column('id', Integer, primary_key=True), Column('id', Integer, primary_key=True),
Column('login', String(40)), Column('login', Text),
Column('passwd', String(40)), Column('passwd', Text),
) )
def upgrade(): def upgrade():
# Upgrade operations go here. Don't create your own engine; use the engine # Upgrade operations go here. Don't create your own engine; use the engine
@ -468,25 +470,27 @@ class TestShellDatabase(Shell, DB):
result = self.env.run('migrate manage %s --repository=%s --url=%s --model=%s'\ result = self.env.run('migrate manage %s --repository=%s --url=%s --model=%s'\
% (script_path, repos_path, self.url, model_module)) % (script_path, repos_path, self.url, model_module))
self.assert_(os.path.exists(script_path)) self.assert_(os.path.exists(script_path))
# Model is defined but database is empty. # Model is defined but database is empty.
result = self.env.run('migrate compare_model_to_db %s %s --model=%s' % (self.url, repos_path, model_module)) result = self.env.run('migrate compare_model_to_db %s %s --model=%s' \
% (self.url, repos_path, model_module), cwd=HERE)
self.assert_("tables missing in database: tmp_account_rundiffs" in result.stdout) self.assert_("tables missing in database: tmp_account_rundiffs" in result.stdout)
# Test Deprecation # Test Deprecation
result = self.env.run('migrate compare_model_to_db %s %s --model=%s' % (self.url, repos_path, model_module.replace(":", ".")), expect_error=True) result = self.env.run('migrate compare_model_to_db %s %s --model=%s' \
% (self.url, repos_path, model_module.replace(":", ".")), cwd=HERE, expect_error=True)
self.assertEqual(result.returncode, 0) self.assertEqual(result.returncode, 0)
self.assertTrue("DeprecationWarning" in result.stderr) self.assertTrue("DeprecationWarning" in result.stderr)
self.assert_("tables missing in database: tmp_account_rundiffs" in result.stdout) self.assert_("tables missing in database: tmp_account_rundiffs" in result.stdout)
# Update db to latest model. # Update db to latest model.
result = self.env.run('migrate update_db_from_model %s %s %s'\ result = self.env.run('migrate update_db_from_model %s %s %s'\
% (self.url, repos_path, model_module)) % (self.url, repos_path, model_module), cwd=HERE)
self.assertEquals(self.run_version(repos_path), 0) self.assertEquals(self.run_version(repos_path), 0)
self.assertEquals(self.run_db_version(self.url, repos_path), 0) # version did not get bumped yet because new version not yet created self.assertEquals(self.run_db_version(self.url, repos_path), 0) # version did not get bumped yet because new version not yet created
result = self.env.run('migrate compare_model_to_db %s %s %s'\ result = self.env.run('migrate compare_model_to_db %s %s %s'\
% (self.url, repos_path, model_module)) % (self.url, repos_path, model_module), cwd=HERE)
self.assert_("No schema diffs" in result.stdout) self.assert_("No schema diffs" in result.stdout)
result = self.env.run('migrate drop_version_control %s %s' % (self.url, repos_path), expect_error=True) result = self.env.run('migrate drop_version_control %s %s' % (self.url, repos_path), expect_error=True)
@ -496,6 +500,7 @@ class TestShellDatabase(Shell, DB):
temp_dict = dict() temp_dict = dict()
exec result.stdout in temp_dict exec result.stdout in temp_dict
# TODO: compare whole table
self.compare_columns_equal(models.tmp_account_rundiffs.c, temp_dict['tmp_account_rundiffs'].c) self.compare_columns_equal(models.tmp_account_rundiffs.c, temp_dict['tmp_account_rundiffs'].c)
#self.assertTrue("""tmp_account_rundiffs = Table('tmp_account_rundiffs', meta, #self.assertTrue("""tmp_account_rundiffs = Table('tmp_account_rundiffs', meta,
#Column('id', Integer(), primary_key=True, nullable=False), #Column('id', Integer(), primary_key=True, nullable=False),
@ -503,11 +508,11 @@ class TestShellDatabase(Shell, DB):
#Column('passwd', String(length=None, convert_unicode=False, assert_unicode=None))""" in result.stdout) #Column('passwd', String(length=None, convert_unicode=False, assert_unicode=None))""" in result.stdout)
# We're happy with db changes, make first db upgrade script to go from version 0 -> 1. # We're happy with db changes, make first db upgrade script to go from version 0 -> 1.
result = self.env.run('migrate make_update_script_for_model', expect_error=True) result = self.env.run('migrate make_update_script_for_model', expect_error=True, cwd=HERE)
self.assertTrue('Not enough arguments' in result.stderr) self.assertTrue('Not enough arguments' in result.stderr)
result_script = self.env.run('migrate make_update_script_for_model %s %s %s %s'\ result_script = self.env.run('migrate make_update_script_for_model %s %s %s %s'\
% (self.url, repos_path, old_model_module, model_module)) % (self.url, repos_path, old_model_module, model_module), cwd=HERE)
self.assertEqualsIgnoreWhitespace(result_script.stdout, self.assertEqualsIgnoreWhitespace(result_script.stdout,
'''from sqlalchemy import * '''from sqlalchemy import *
from migrate import * from migrate import *
@ -517,8 +522,8 @@ class TestShellDatabase(Shell, DB):
meta = MetaData() meta = MetaData()
tmp_account_rundiffs = Table('tmp_account_rundiffs', meta, tmp_account_rundiffs = Table('tmp_account_rundiffs', meta,
Column('id', Integer(), primary_key=True, nullable=False), Column('id', Integer(), primary_key=True, nullable=False),
Column('login', String(length=40, convert_unicode=False, assert_unicode=None)), Column('login', String(length=None, convert_unicode=False, assert_unicode=None)),
Column('passwd', String(length=40, convert_unicode=False, assert_unicode=None)), Column('passwd', String(length=None, convert_unicode=False, assert_unicode=None)),
) )
def upgrade(migrate_engine): def upgrade(migrate_engine):
@ -538,7 +543,7 @@ class TestShellDatabase(Shell, DB):
open(upgrade_script_path, 'w').write(result_script.stdout) open(upgrade_script_path, 'w').write(result_script.stdout)
result = self.env.run('migrate compare_model_to_db %s %s %s'\ result = self.env.run('migrate compare_model_to_db %s %s %s'\
% (self.url, repos_path, model_module)) % (self.url, repos_path, model_module), cwd=HERE)
self.assert_("No schema diffs" in result.stdout) self.assert_("No schema diffs" in result.stdout)
self.meta.drop_all() # in case junk tables are lying around in the test database self.meta.drop_all() # in case junk tables are lying around in the test database