summaryrefslogtreecommitdiff
path: root/migrate/tests/versioning/test_script.py
diff options
context:
space:
mode:
authorJim Rollenhagen <jim@jimrollenhagen.com>2019-09-26 09:56:42 -0400
committerJim Rollenhagen <jim@jimrollenhagen.com>2019-09-26 09:56:42 -0400
commit52672a64cc0cab4ea14a4a756fce850eb03315e3 (patch)
treea86024e4e6141aa8983c750f751c58d924f5b11a /migrate/tests/versioning/test_script.py
parent8acab2cd75a5b23ac162e49c8e4fb1e3f958352a (diff)
downloadsqlalchemy-migrate-master.tar.gz
Retire github mirror, repo moved to opendevHEADmaster
Diffstat (limited to 'migrate/tests/versioning/test_script.py')
-rw-r--r--migrate/tests/versioning/test_script.py305
1 files changed, 0 insertions, 305 deletions
diff --git a/migrate/tests/versioning/test_script.py b/migrate/tests/versioning/test_script.py
deleted file mode 100644
index 20e6af0..0000000
--- a/migrate/tests/versioning/test_script.py
+++ /dev/null
@@ -1,305 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-import imp
-import os
-import sys
-import shutil
-
-import six
-from migrate import exceptions
-from migrate.versioning import version, repository
-from migrate.versioning.script import *
-from migrate.versioning.util import *
-
-from migrate.tests import fixture
-from migrate.tests.fixture.models import tmp_sql_table
-
-
-class TestBaseScript(fixture.Pathed):
-
- def test_all(self):
- """Testing all basic BaseScript operations"""
- # verify / source / run
- src = self.tmp()
- open(src, 'w').close()
- bscript = BaseScript(src)
- BaseScript.verify(src)
- self.assertEqual(bscript.source(), '')
- self.assertRaises(NotImplementedError, bscript.run, 'foobar')
-
-
-class TestPyScript(fixture.Pathed, fixture.DB):
- cls = PythonScript
- def test_create(self):
- """We can create a migration script"""
- path = self.tmp_py()
- # Creating a file that doesn't exist should succeed
- self.cls.create(path)
- self.assertTrue(os.path.exists(path))
- # Created file should be a valid script (If not, raises an error)
- self.cls.verify(path)
- # Can't create it again: it already exists
- self.assertRaises(exceptions.PathFoundError,self.cls.create,path)
-
- @fixture.usedb(supported='sqlite')
- def test_run(self):
- script_path = self.tmp_py()
- pyscript = PythonScript.create(script_path)
- pyscript.run(self.engine, 1)
- pyscript.run(self.engine, -1)
-
- self.assertRaises(exceptions.ScriptError, pyscript.run, self.engine, 0)
- self.assertRaises(exceptions.ScriptError, pyscript._func, 'foobar')
-
- # clean pyc file
- if six.PY3:
- os.remove(imp.cache_from_source(script_path))
- else:
- os.remove(script_path + 'c')
-
- # test deprecated upgrade/downgrade with no arguments
- contents = open(script_path, 'r').read()
- f = open(script_path, 'w')
- f.write(contents.replace("upgrade(migrate_engine)", "upgrade()"))
- f.close()
-
- pyscript = PythonScript(script_path)
- pyscript._module = None
- try:
- pyscript.run(self.engine, 1)
- pyscript.run(self.engine, -1)
- except exceptions.ScriptError:
- pass
- else:
- self.fail()
-
- def test_verify_notfound(self):
- """Correctly verify a python migration script: nonexistant file"""
- path = self.tmp_py()
- self.assertFalse(os.path.exists(path))
- # Fails on empty path
- self.assertRaises(exceptions.InvalidScriptError,self.cls.verify,path)
- self.assertRaises(exceptions.InvalidScriptError,self.cls,path)
-
- def test_verify_invalidpy(self):
- """Correctly verify a python migration script: invalid python file"""
- path=self.tmp_py()
- # Create empty file
- f = open(path,'w')
- f.write("def fail")
- f.close()
- self.assertRaises(Exception,self.cls.verify_module,path)
- # script isn't verified on creation, but on module reference
- py = self.cls(path)
- self.assertRaises(Exception,(lambda x: x.module),py)
-
- def test_verify_nofuncs(self):
- """Correctly verify a python migration script: valid python file; no upgrade func"""
- path = self.tmp_py()
- # Create empty file
- f = open(path, 'w')
- f.write("def zergling():\n\tprint('rush')")
- f.close()
- self.assertRaises(exceptions.InvalidScriptError, self.cls.verify_module, path)
- # script isn't verified on creation, but on module reference
- py = self.cls(path)
- self.assertRaises(exceptions.InvalidScriptError,(lambda x: x.module),py)
-
- @fixture.usedb(supported='sqlite')
- def test_preview_sql(self):
- """Preview SQL abstract from ORM layer (sqlite)"""
- path = self.tmp_py()
-
- f = open(path, 'w')
- content = '''
-from migrate import *
-from sqlalchemy import *
-
-metadata = MetaData()
-
-UserGroup = Table('Link', metadata,
- Column('link1ID', Integer),
- Column('link2ID', Integer),
- UniqueConstraint('link1ID', 'link2ID'))
-
-def upgrade(migrate_engine):
- metadata.create_all(migrate_engine)
- '''
- f.write(content)
- f.close()
-
- pyscript = self.cls(path)
- SQL = pyscript.preview_sql(self.url, 1)
- self.assertEqualIgnoreWhitespace("""
- CREATE TABLE "Link"
- ("link1ID" INTEGER,
- "link2ID" INTEGER,
- UNIQUE ("link1ID", "link2ID"))
- """, SQL)
- # TODO: test: No SQL should be executed!
-
- def test_verify_success(self):
- """Correctly verify a python migration script: success"""
- path = self.tmp_py()
- # Succeeds after creating
- self.cls.create(path)
- self.cls.verify(path)
-
- # test for PythonScript.make_update_script_for_model
-
- @fixture.usedb()
- def test_make_update_script_for_model(self):
- """Construct script source from differences of two models"""
-
- self.setup_model_params()
- self.write_file(self.first_model_path, self.base_source)
- self.write_file(self.second_model_path, self.base_source + self.model_source)
-
- source_script = self.pyscript.make_update_script_for_model(
- engine=self.engine,
- oldmodel=load_model('testmodel_first:meta'),
- model=load_model('testmodel_second:meta'),
- repository=self.repo_path,
- )
-
- self.assertTrue("['User'].create()" in source_script)
- self.assertTrue("['User'].drop()" in source_script)
-
- @fixture.usedb()
- def test_make_update_script_for_equal_models(self):
- """Try to make update script from two identical models"""
-
- self.setup_model_params()
- self.write_file(self.first_model_path, self.base_source + self.model_source)
- self.write_file(self.second_model_path, self.base_source + self.model_source)
-
- source_script = self.pyscript.make_update_script_for_model(
- engine=self.engine,
- oldmodel=load_model('testmodel_first:meta'),
- model=load_model('testmodel_second:meta'),
- repository=self.repo_path,
- )
-
- self.assertFalse('User.create()' in source_script)
- self.assertFalse('User.drop()' in source_script)
-
- @fixture.usedb()
- def test_make_update_script_direction(self):
- """Check update scripts go in the right direction"""
-
- self.setup_model_params()
- self.write_file(self.first_model_path, self.base_source)
- self.write_file(self.second_model_path, self.base_source + self.model_source)
-
- source_script = self.pyscript.make_update_script_for_model(
- engine=self.engine,
- oldmodel=load_model('testmodel_first:meta'),
- model=load_model('testmodel_second:meta'),
- repository=self.repo_path,
- )
-
- self.assertTrue(0
- < source_script.find('upgrade')
- < source_script.find("['User'].create()")
- < source_script.find('downgrade')
- < source_script.find("['User'].drop()"))
-
- def setup_model_params(self):
- self.script_path = self.tmp_py()
- self.repo_path = self.tmp()
- self.first_model_path = os.path.join(self.temp_usable_dir, 'testmodel_first.py')
- self.second_model_path = os.path.join(self.temp_usable_dir, 'testmodel_second.py')
-
- self.base_source = """from sqlalchemy import *\nmeta = MetaData()\n"""
- self.model_source = """
-User = Table('User', meta,
- Column('id', Integer, primary_key=True),
- Column('login', Unicode(40)),
- Column('passwd', String(40)),
-)"""
-
- self.repo = repository.Repository.create(self.repo_path, 'repo')
- self.pyscript = PythonScript.create(self.script_path)
- sys.modules.pop('testmodel_first', None)
- sys.modules.pop('testmodel_second', None)
-
- def write_file(self, path, contents):
- f = open(path, 'w')
- f.write(contents)
- f.close()
-
-
-class TestSqlScript(fixture.Pathed, fixture.DB):
-
- @fixture.usedb()
- def test_error(self):
- """Test if exception is raised on wrong script source"""
- src = self.tmp()
-
- f = open(src, 'w')
- f.write("""foobar""")
- f.close()
-
- sqls = SqlScript(src)
- self.assertRaises(Exception, sqls.run, self.engine)
-
- @fixture.usedb()
- def test_success(self):
- """Test sucessful SQL execution"""
- # cleanup and prepare python script
- tmp_sql_table.metadata.drop_all(self.engine, checkfirst=True)
- script_path = self.tmp_py()
- pyscript = PythonScript.create(script_path)
-
- # populate python script
- contents = open(script_path, 'r').read()
- contents = contents.replace("pass", "tmp_sql_table.create(migrate_engine)")
- contents = 'from migrate.tests.fixture.models import tmp_sql_table\n' + contents
- f = open(script_path, 'w')
- f.write(contents)
- f.close()
-
- # write SQL script from python script preview
- pyscript = PythonScript(script_path)
- src = self.tmp()
- f = open(src, 'w')
- f.write(pyscript.preview_sql(self.url, 1))
- f.close()
-
- # run the change
- sqls = SqlScript(src)
- sqls.run(self.engine)
- tmp_sql_table.metadata.drop_all(self.engine, checkfirst=True)
-
- @fixture.usedb()
- def test_transaction_management_statements(self):
- """
- Test that we can successfully execute SQL scripts with transaction
- management statements.
- """
- for script_pattern in (
- "BEGIN TRANSACTION; %s; COMMIT;",
- "BEGIN; %s; END TRANSACTION;",
- "/* comment */BEGIN TRANSACTION; %s; /* comment */COMMIT;",
- "/* comment */ BEGIN TRANSACTION; %s; /* comment */ COMMIT;",
- """
--- comment
-BEGIN TRANSACTION;
-
-%s;
-
--- comment
-COMMIT;""",
- ):
-
- test_statement = ("CREATE TABLE TEST1 (field1 int); "
- "DROP TABLE TEST1")
- script = script_pattern % test_statement
- src = self.tmp()
-
- with open(src, 'wt') as f:
- f.write(script)
-
- sqls = SqlScript(src)
- sqls.run(self.engine)