add tests for plain API, fixed some small bugs

9 files changed, 246 insertions, 28 deletions

diff --git a/test/fixture/models.py b/test/fixture/models.py
index 4454bf5..40e8e48 100644
--- a/test/fixture/models.py
+++ b/test/fixture/models.py
@@ -3,9 +3,12 @@ from sqlalchemy import *
 # test rundiffs in shell
 meta_old_rundiffs = MetaData()
 meta_rundiffs = MetaData()
+meta = MetaData()
 
 tmp_account_rundiffs = Table('tmp_account_rundiffs', meta_rundiffs,
     Column('id', Integer, primary_key=True),
     Column('login', String(40)),
     Column('passwd', String(40)),
 )
+
+tmp_sql_table = Table('tmp_sql_table', meta, Column('id', Integer))
diff --git a/test/versioning/test_api.py b/test/versioning/test_api.py
index f13ea1a..0356a85 100644
--- a/test/versioning/test_api.py
+++ b/test/versioning/test_api.py
@@ -5,6 +5,8 @@ from migrate.versioning import api
 from migrate.versioning.exceptions import *
 
 from test.fixture.pathed import *
+from test.fixture import models
+from test import fixture
 
 
 class TestAPI(Pathed):
@@ -15,14 +17,104 @@ class TestAPI(Pathed):
         self.assertRaises(UsageError, api.help, 'foobar')
         self.assert_(isinstance(api.help('create'), str))
 
-    def test_help_commands(self):
-        pass
+        # test that all commands return some text
+        for cmd in api.__all__:
+            content = api.help(cmd)
+            self.assertTrue(content)
 
     def test_create(self):
-        pass
+        tmprepo = self.tmp_repos()
+        api.create(tmprepo, 'temp')
+
+        # repository already exists
+        self.assertRaises(KnownError, api.create, tmprepo, 'temp')
 
     def test_script(self):
-        pass
+        repo = self.tmp_repos()
+        api.create(repo, 'temp')
+        api.script('first version', repo)
 
     def test_script_sql(self):
-        pass
+        repo = self.tmp_repos()
+        api.create(repo, 'temp')
+        api.script_sql('postgres', repo)
+
+    def test_version(self):
+        repo = self.tmp_repos()
+        api.create(repo, 'temp')
+        api.version(repo)
+
+    def test_source(self):
+        repo = self.tmp_repos()
+        api.create(repo, 'temp')
+        api.script('first version', repo)
+        api.script_sql('default', repo)
+
+        # no repository
+        self.assertRaises(UsageError, api.source, 1)
+
+        # stdout
+        out = api.source(1, dest=None, repository=repo)
+        self.assertTrue(out)
+
+        # file
+        out = api.source(1, dest=self.tmp_repos(), repository=repo)
+        self.assertFalse(out)
+
+    def test_manage(self):
+        output = api.manage(os.path.join(self.temp_usable_dir, 'manage.py'))
+
+
+class TestSchemaAPI(fixture.DB, Pathed):
+
+    def _setup(self, url):
+        super(TestSchemaAPI, self)._setup(url)
+        self.repo = self.tmp_repos()
+        api.create(self.repo, 'temp')
+        self.schema = api.version_control(url, self.repo)
+
+    def _teardown(self):
+        self.schema = api.drop_version_control(self.url, self.repo)
+        super(TestSchemaAPI, self)._teardown()
+
+    @fixture.usedb()
+    def test_workflow(self):
+        self.assertEqual(api.db_version(self.url, self.repo), 0)
+        api.script('First Version', self.repo)
+        self.assertEqual(api.db_version(self.url, self.repo), 0)
+        api.upgrade(self.url, self.repo, 1)
+        self.assertEqual(api.db_version(self.url, self.repo), 1)
+        api.downgrade(self.url, self.repo, 0)
+        self.assertEqual(api.db_version(self.url, self.repo), 0)
+        api.test(self.url, self.repo)
+        self.assertEqual(api.db_version(self.url, self.repo), 0)
+
+        # preview
+        # TODO: test output
+        out = api.upgrade(self.url, self.repo, preview_py=True)
+        out = api.upgrade(self.url, self.repo, preview_sql=True)
+
+        api.upgrade(self.url, self.repo, 1)
+        api.script_sql('default', self.repo)
+        self.assertRaises(UsageError, api.upgrade, self.url, self.repo, 2, preview_py=True)
+        out = api.upgrade(self.url, self.repo, 2, preview_sql=True)
+
+        # cant upgrade to version 1, already at version 1
+        self.assertEqual(api.db_version(self.url, self.repo), 1)
+        self.assertRaises(KnownError, api.upgrade, self.url, self.repo, 0)
+
+    @fixture.usedb()
+    def test_compare_model_to_db(self):
+        diff = api.compare_model_to_db(self.url, self.repo, models.meta)
+
+    @fixture.usedb()
+    def test_create_model(self):
+        model = api.create_model(self.url, self.repo)
+
+    @fixture.usedb()
+    def test_make_update_script_for_model(self):
+        model = api.make_update_script_for_model(self.url, self.repo, models.meta_old_rundiffs, models.meta_rundiffs)
+
+    @fixture.usedb()
+    def test_update_db_from_model(self):
+        model = api.update_db_from_model(self.url, self.repo, models.meta_rundiffs)
diff --git a/test/versioning/test_keyedinstance.py b/test/versioning/test_keyedinstance.py
index eaedf01..9bc9202 100644
--- a/test/versioning/test_keyedinstance.py
+++ b/test/versioning/test_keyedinstance.py
@@ -1,3 +1,6 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
 from test import fixture
 from migrate.versioning.util.keyedinstance import *
 
@@ -38,3 +41,5 @@ class TestKeydInstance(fixture.Base):
         Uniq1.clear()
         a12 = Uniq1('a')
         self.assert_(a10 is not a12)
+
+        self.assertRaises(NotImplementedError, KeyedInstance._key)
diff --git a/test/versioning/test_repository.py b/test/versioning/test_repository.py
index 4d95e31..6c10454 100644
--- a/test/versioning/test_repository.py
+++ b/test/versioning/test_repository.py
@@ -101,13 +101,15 @@ class TestVersionedRepository(fixture.Pathed):
         # Load repository and commit script
         repo = Repository(self.path_repos)
         repo.create_script('')
-
-        # Get script object
-        source = repo.version(1).script().source()
+        repo.create_script_sql('postgres')
 
         # Source is valid: script must have an upgrade function
         # (not a very thorough test, but should be plenty)
-        self.assert_(source.find('def upgrade') >= 0)
+        source = repo.version(1).script().source()
+        self.assertTrue(source.find('def upgrade') >= 0)
+
+        source = repo.version(2).script('postgres', 'upgrade').source()
+        self.assertEqual(source.strip(), '')
 
     def test_latestversion(self):
         """Repository.version() (no params) returns the latest version"""
diff --git a/test/versioning/test_schema.py b/test/versioning/test_schema.py
index 9f84217..90d4411 100644
--- a/test/versioning/test_schema.py
+++ b/test/versioning/test_schema.py
@@ -16,11 +16,10 @@ class TestControlledSchema(fixture.Pathed, fixture.DB):
     # Transactions break postgres in this test; we'll clean up after ourselves
     level = fixture.DB.CONNECT
 
-    
     def setUp(self):
         super(TestControlledSchema, self).setUp()
-        path_repos = self.temp_usable_dir + '/repo/'
-        self.repos = Repository.create(path_repos, 'repo_name')
+        self.path_repos = self.temp_usable_dir + '/repo/'
+        self.repos = Repository.create(self.path_repos, 'repo_name')
 
     def _setup(self, url):
         self.setUp()
@@ -45,6 +44,19 @@ class TestControlledSchema(fixture.Pathed, fixture.DB):
         super(TestControlledSchema, self).tearDown()
 
     @fixture.usedb()
+    def test_schema_table_fail(self):
+        """Test scenarios when loading schema should fail"""
+        dbcontrol = ControlledSchema.create(self.engine, self.path_repos)
+        dbcontrol.table.drop()
+
+        try:
+            dbcontrol.load()
+        except exceptions.DatabaseNotControlledError:
+            pass
+        else:
+            self.fail()
+
+    @fixture.usedb()
     def test_version_control(self):
         """Establish version control on a particular database"""
         # Establish version control on this database
@@ -116,7 +128,7 @@ class TestControlledSchema(fixture.Pathed, fixture.DB):
             #self.assertRaises(ControlledSchema.InvalidVersionError,
             # Can't have custom errors with assertRaises...
             try:
-                ControlledSchema.create(self.engine,self.repos,version)
+                ControlledSchema.create(self.engine, self.repos,version)
                 self.assert_(False, repr(version))
             except exceptions.InvalidVersionError:
                 pass
diff --git a/test/versioning/test_script.py b/test/versioning/test_script.py
index 16fa166..9272f96 100644
--- a/test/versioning/test_script.py
+++ b/test/versioning/test_script.py
@@ -10,6 +10,7 @@ from migrate.versioning.script import *
 from migrate.versioning.util import *
 
 from test import fixture
+from test.fixture.models import tmp_sql_table
 
 
 class TestBaseScript(fixture.Pathed):
@@ -48,6 +49,25 @@ class TestPyScript(fixture.Pathed, fixture.DB):
         self.assertRaises(exceptions.ScriptError, pyscript.run, self.engine, 0)
         self.assertRaises(exceptions.ScriptError, pyscript._func, 'foobar')
 
+        # clean pyc file
+        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 TypeError:
+            pass
+        else:
+            self.fail()
+
     def test_verify_notfound(self):
         """Correctly verify a python migration script: nonexistant file"""
         path = self.tmp_py()
@@ -86,7 +106,7 @@ class TestPyScript(fixture.Pathed, fixture.DB):
         path = self.tmp_py()
 
         f = open(path, 'w')
-        content = """
+        content = '''
 from migrate import *
 from sqlalchemy import *
 
@@ -99,7 +119,7 @@ UserGroup = Table('Link', metadata,
 
 def upgrade(migrate_engine):
     metadata.create_all(migrate_engine)
-        """
+        '''
         f.write(content)
         f.close()
 
@@ -130,7 +150,6 @@ def upgrade(migrate_engine):
         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'),
@@ -195,3 +214,31 @@ class TestSqlScript(fixture.Pathed, fixture.DB):
 
         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 test.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, executemany=False)
+        tmp_sql_table.metadata.drop_all(self.engine, checkfirst=True)
diff --git a/test/versioning/test_shell.py b/test/versioning/test_shell.py
index a4d2b64..d9e3364 100644
--- a/test/versioning/test_shell.py
+++ b/test/versioning/test_shell.py
@@ -3,13 +3,14 @@
 
 import os
 import tempfile
+from runpy import run_module
 
 from sqlalchemy import MetaData, Table
 
-from migrate.versioning import genmodel, shell, api
 from migrate.versioning.repository import Repository
 from migrate.versioning.exceptions import *
 from test.fixture import *
+from migrate.versioning import genmodel, shell, api
 
 
 class TestShellCommands(Shell):
@@ -30,6 +31,36 @@ class TestShellCommands(Shell):
             self.assertTrue(result.stdout)
             self.assertFalse(result.stderr)
 
+    def test_main(self):
+        """Test main() function"""
+        # TODO: test output?
+        try:
+            run_module('migrate.versioning.shell', run_name='__main__')
+        except:
+            pass
+        repos = self.tmp_repos()
+        shell.main(['help'])
+        shell.main(['help', 'create'])
+        shell.main(['create', 'repo_name', '--preview_sql'], repository=repos)
+        shell.main(['version', '--', '--repository=%s' % repos])
+        shell.main(['version', '-d', '--repository=%s' % repos, '--version=2'])
+        try:
+            shell.main(['foobar'])
+        except SystemExit, e:
+            pass
+        try:
+            shell.main(['create', 'f', 'o', 'o'])
+        except SystemExit, e:
+            pass
+        try:
+            shell.main(['create'])
+        except SystemExit, e:
+            pass
+        try:
+            shell.main(['create', 'repo_name'], repository=repos)
+        except SystemExit, e:
+            pass
+
     def test_create(self):
         """Repositories are created successfully"""
         repos = self.tmp_repos()
@@ -333,7 +364,7 @@ class TestShellDatabase(Shell, DB):
 
         # Empty script should succeed
         result = self.env.run('migrate script Desc %s' % repos_path)
-        result = self.env.run('migrate test %s %s' % (repos_path, self.url))
+        result = self.env.run('migrate test %s %s' % (self.url, repos_path))
         self.assertEquals(self.run_version(repos_path), 1)
         self.assertEquals(self.run_db_version(self.url, repos_path), 0)
 
@@ -355,7 +386,7 @@ class TestShellDatabase(Shell, DB):
         file.write(script_text)
         file.close()
 
-        result = self.env.run('migrate test %s %s bla' % (repos_path, self.url), expect_error=True)
+        result = self.env.run('migrate test %s %s bla' % (self.url, repos_path), expect_error=True)
         self.assertEqual(result.returncode, 2)
         self.assertEquals(self.run_version(repos_path), 1)
         self.assertEquals(self.run_db_version(self.url, repos_path), 0)
@@ -384,7 +415,7 @@ class TestShellDatabase(Shell, DB):
         file = open(script_path, 'w')
         file.write(script_text)
         file.close()
-        result = self.env.run('migrate test %s %s' % (repos_path, self.url))
+        result = self.env.run('migrate test %s %s' % (self.url, repos_path))
         self.assertEquals(self.run_version(repos_path), 1)
         self.assertEquals(self.run_db_version(self.url, repos_path), 0)
         
@@ -426,12 +457,12 @@ class TestShellDatabase(Shell, DB):
         
         # Update db to latest model.
         result = self.env.run('migrate update_db_from_model %s %s %s'\
-            % (self.url, model_module, repos_path))
+            % (self.url, repos_path, model_module))
         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
 
         result = self.env.run('migrate compare_model_to_db %s %s %s'\
-            % (self.url, model_module, repos_path))
+            % (self.url, repos_path, model_module))
         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)
@@ -448,9 +479,9 @@ class TestShellDatabase(Shell, DB):
         self.assertTrue('Not enough arguments' in result.stderr)
 
         result_script = self.env.run('migrate make_update_script_for_model %s %s %s %s'\
-            % (self.url, old_model_module, model_module, repos_path))
+            % (self.url, repos_path, old_model_module, model_module))
         self.assertEqualsIgnoreWhitespace(result_script.stdout,
-        """from sqlalchemy import *
+        '''from sqlalchemy import *
         from migrate import *
 
         meta = MetaData()
@@ -469,7 +500,7 @@ class TestShellDatabase(Shell, DB):
         def downgrade(migrate_engine):
             # Operations to reverse the above upgrade go here.
             meta.bind = migrate_engine
-            tmp_account_rundiffs.drop()""")
+            tmp_account_rundiffs.drop()''')
     
         # Save the upgrade script.
         result = self.env.run('migrate script Desc %s' % repos_path)
@@ -477,7 +508,7 @@ class TestShellDatabase(Shell, DB):
         open(upgrade_script_path, 'w').write(result_script.stdout)
 
         result = self.env.run('migrate compare_model_to_db %s %s %s'\
-            % (self.url, model_module, repos_path))
+            % (self.url, repos_path, model_module))
         self.assert_("No schema diffs" in result.stdout)
 
         self.meta.drop_all()  # in case junk tables are lying around in the test database
diff --git a/test/versioning/test_util.py b/test/versioning/test_util.py
index a3d4834..7471651 100644
--- a/test/versioning/test_util.py
+++ b/test/versioning/test_util.py
@@ -36,10 +36,13 @@ class TestUtil(fixture.Pathed):
             engine_arg_assert_unicode=True)
         self.assertTrue(engine.dialect.assert_unicode)
 
-        # deprecated echo= parameter
+        # deprecated echo=True parameter
         engine = construct_engine(url, echo='True')
         self.assertTrue(engine.echo)
 
+        # unsupported argument
+        self.assertRaises(ValueError, construct_engine, 1)
+
     def test_asbool(self):
         """test asbool parsing"""
         result = asbool(True)
diff --git a/test/versioning/test_version.py b/test/versioning/test_version.py
index a4855fe..7f79f66 100644
--- a/test/versioning/test_version.py
+++ b/test/versioning/test_version.py
@@ -3,6 +3,7 @@
 
 from test import fixture
 from migrate.versioning.version import *
+from migrate.versioning.exceptions import *
 
 
 class TestVerNum(fixture.Base):
@@ -12,6 +13,11 @@ class TestVerNum(fixture.Base):
         for version in versions:
             self.assertRaises(ValueError, VerNum, version)
 
+    def test_str(self):
+        """Test str and repr version numbers"""
+        self.assertEqual(str(VerNum(2)), '2')
+        self.assertEqual(repr(VerNum(2)), '<VerNum(2)>')
+
     def test_is(self):
         """Two version with the same number should be equal"""
         a = VerNum(1)
@@ -62,6 +68,7 @@ class TestVerNum(fixture.Base):
         self.assert_(VerNum(2) >= 1)
         self.assertFalse(VerNum(1) >= 2)
         
+
 class TestVersion(fixture.Pathed):
 
     def setUp(self):
@@ -69,6 +76,7 @@ class TestVersion(fixture.Pathed):
 
     def test_str_to_filename(self):
         self.assertEquals(str_to_filename(''), '')
+        self.assertEquals(str_to_filename(''), '')
         self.assertEquals(str_to_filename('__'), '_')
         self.assertEquals(str_to_filename('a'), 'a')
         self.assertEquals(str_to_filename('Abc Def'), 'Abc_Def')
@@ -91,12 +99,18 @@ class TestVersion(fixture.Pathed):
         coll2 = Collection(self.temp_usable_dir)
         self.assertEqual(coll.versions, coll2.versions)
 
+        Collection.clear()
+
+    def test_old_repository(self):
+        open(os.path.join(self.temp_usable_dir, '1'), 'w')
+        self.assertRaises(Exception, Collection, self.temp_usable_dir)
+
     #def test_collection_unicode(self):
     #    pass
 
     def test_create_new_python_version(self):
         coll = Collection(self.temp_usable_dir)
-        coll.create_new_python_version("foo bar")
+        coll.create_new_python_version("'")
 
         ver = coll.version()
         self.assert_(ver.script().source())
@@ -140,3 +154,12 @@ class TestVersion(fixture.Pathed):
 
         ver = Version(1, path, [sqlite_upgrade_file, python_file])
         self.assertEquals(os.path.basename(ver.script('postgres', 'upgrade').path), python_file)
+
+    def test_bad_version(self):
+        ver = Version(1, self.temp_usable_dir, [])
+        self.assertRaises(ScriptError, ver.add_script, '123.sql')
+
+        pyscript = os.path.join(self.temp_usable_dir, 'bla.py')
+        open(pyscript, 'w')
+        ver.add_script(pyscript)
+        self.assertRaises(ScriptError, ver.add_script, 'bla.py')