some sqlite round trip tests

alembic/script.py

@@ -100,12 +100,16 @@ def _revision_map(self):
         map_[None] = None
         return map_
 
-    def rev_path(self, rev_id):
+    def _rev_path(self, rev_id):
         filename = "%s.py" % rev_id
         return os.path.join(self.versions, filename)
 
-    def refresh(self, rev_id):
-        script = Script.from_path(self.rev_path(rev_id))
+    def write(self, rev_id, content):
+        path = self._rev_path(rev_id)
+        file(path, 'w').write(content)
+        if os.access(path + "c", os.F_OK):
+            os.unlink(path + "c")
+        script = Script.from_path(path)
         old = self._revision_map[script.revision]
         if old.down_revision != script.down_revision:
             raise Exception("Can't change down_revision on a refresh operation.")
@@ -151,7 +155,7 @@ def copy_file(self, src, dest):
 
     def generate_rev(self, revid, message):
         current_head = self._current_head()
-        path = self.rev_path(revid)
+        path = self._rev_path(revid)
         self.generate_template(
             os.path.join(self.dir, "script.py.mako"),
             path,

tests/__init__.py

@@ -4,6 +4,7 @@
 import shutil
 import os
 import itertools
+from sqlalchemy import create_engine
 
 staging_directory = os.path.join(os.path.dirname(__file__), 'scratch')
 
@@ -19,11 +20,52 @@ def assert_compiled(element, assert_string, dialect=None):
     dialect = _get_dialect(dialect)
     eq_(unicode(element.compile(dialect=dialect)), assert_string)
 
-def _testing_config(**kw):
+def _testing_config():
     from alembic.config import Config
     if not os.access(staging_directory, os.F_OK):
         os.mkdir(staging_directory)
     return Config(os.path.join(staging_directory, 'test_alembic.ini'))
+
+def _sqlite_testing_config():
+    cfg = _testing_config()
+    dir_ = os.path.join(staging_directory, 'scripts')
+    file(cfg.config_file_name, 'w').write("""
+[alembic]
+script_location = %s
+sqlalchemy.url = sqlite:///%s/foo.db
+
+[loggers]
+keys = root
+
+[handlers]
+keys = console
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatters]
+keys = generic
+
+[formatter_generic]
+format = %%(levelname)-5.5s [%%(name)s] %%(message)s
+datefmt = %%H:%%M:%%S
+    """ % (dir_, dir_))
+    return cfg
+
+def sqlite_db():
+    # sqlite caches table pragma info 
+    # per connection, so create a new
+    # engine for each assertion
+    dir_ = os.path.join(staging_directory, 'scripts')
+    return create_engine('sqlite:///%s/foo.db' % dir_)
 
 def staging_env(create=True):
     from alembic import command, script

tests/test_versioning.py

@@ -0,0 +1,94 @@
+from tests import clear_staging_env, staging_env, _sqlite_testing_config, sqlite_db, eq_, ne_
+from alembic import command, util
+from alembic.script import ScriptDirectory
+import time
+
+def test_001_revisions():
+    global a, b, c
+    a = util.rev_id()
+    b = util.rev_id()
+    c = util.rev_id()
+
+    script = ScriptDirectory.from_config(cfg)
+    script.generate_rev(a, None)
+    script.write(a, """
+down_revision = None
+
+from alembic.op import *
+
+def upgrade():
+    execute("CREATE TABLE foo(id integer)")
+
+def downgrade():
+    execute("DROP TABLE foo")
+
+""")
+
+    script.generate_rev(b, None)
+    script.write(b, """
+down_revision = '%s'
+
+from alembic.op import *
+
+def upgrade():
+    execute("CREATE TABLE bar(id integer)")
+
+def downgrade():
+    execute("DROP TABLE bar")
+
+""" % a)
+
+    script.generate_rev(c, None)
+    script.write(c, """
+down_revision = '%s'
+
+from alembic.op import *
+
+def upgrade():
+    execute("CREATE TABLE bat(id integer)")
+
+def downgrade():
+    execute("DROP TABLE bat")
+
+""" % b)
+
+
+def test_002_upgrade():
+    command.upgrade(cfg, c)
+    db = sqlite_db()
+    assert db.dialect.has_table(db.connect(), 'foo')
+    assert db.dialect.has_table(db.connect(), 'bar')
+    assert db.dialect.has_table(db.connect(), 'bat')
+
+def test_003_downgrade():
+    command.downgrade(cfg, a)
+    db = sqlite_db()
+    assert db.dialect.has_table(db.connect(), 'foo')
+    assert not db.dialect.has_table(db.connect(), 'bar')
+    assert not db.dialect.has_table(db.connect(), 'bat')
+
+def test_004_downgrade():
+    command.downgrade(cfg, 'base')
+    db = sqlite_db()
+    assert not db.dialect.has_table(db.connect(), 'foo')
+    assert not db.dialect.has_table(db.connect(), 'bar')
+    assert not db.dialect.has_table(db.connect(), 'bat')
+
+def test_005_upgrade():
+    command.upgrade(cfg, b)
+    db = sqlite_db()
+    assert db.dialect.has_table(db.connect(), 'foo')
+    assert db.dialect.has_table(db.connect(), 'bar')
+    assert not db.dialect.has_table(db.connect(), 'bat')
+
+# TODO: test some invalid movements
+
+
+def setup():
+    global cfg, env
+    env = staging_env()
+    cfg = _sqlite_testing_config()
+
+
+def teardown():
+    clear_staging_env()