- reorganize tests that deal with running scripts

and reading scripts; try to put tests against common themes together, improve names

diff --git a/tests/test_command.py b/tests/test_command.py
index 6d70f5ad225df80fe66b4e33191531689091f783..58fed8934196e69c8b9337487f313caed5f1b36b 100644 (file)
--- a/tests/test_command.py
+++ b/tests/test_command.py
@@ -1,13 +1,13 @@
 from alembic import command
 from io import TextIOWrapper, BytesIO
 from alembic.script import ScriptDirectory
-from alembic.testing.fixtures import TestBase
+from alembic.testing.fixtures import TestBase, capture_context_buffer
 from alembic.testing.env import staging_env, _sqlite_testing_config, \
-    three_rev_fixture, clear_staging_env
+    three_rev_fixture, clear_staging_env, _no_sql_testing_config
 from alembic.testing import eq_
 
 
-class StdoutCommandTest(TestBase):
+class HistoryTest(TestBase):
 
     @classmethod
     def setup_class(cls):
@@ -81,3 +81,62 @@ class StdoutCommandTest(TestBase):
         self.cfg.stdout = buf = self._buf_fixture()
         command.history(self.cfg, "current:")
         self._eq_cmd_output(buf, [self.c, self.b, self.a])
+
+
+class UpgradeDowngradeStampTest(TestBase):
+
+    def setUp(self):
+        self.env = staging_env()
+        self.cfg = cfg = _no_sql_testing_config()
+        cfg.set_main_option('dialect_name', 'sqlite')
+        cfg.remove_main_option('url')
+
+        self.a, self.b, self.c = three_rev_fixture(cfg)
+
+    def tearDown(self):
+        clear_staging_env()
+
+    def test_version_from_none_insert(self):
+        with capture_context_buffer() as buf:
+            command.upgrade(self.cfg, self.a, sql=True)
+        assert "CREATE TABLE alembic_version" in buf.getvalue()
+        assert "INSERT INTO alembic_version" in buf.getvalue()
+        assert "CREATE STEP 1" in buf.getvalue()
+        assert "CREATE STEP 2" not in buf.getvalue()
+        assert "CREATE STEP 3" not in buf.getvalue()
+
+    def test_version_from_middle_update(self):
+        with capture_context_buffer() as buf:
+            command.upgrade(self.cfg, "%s:%s" % (self.b, self.c), sql=True)
+        assert "CREATE TABLE alembic_version" not in buf.getvalue()
+        assert "UPDATE alembic_version" in buf.getvalue()
+        assert "CREATE STEP 1" not in buf.getvalue()
+        assert "CREATE STEP 2" not in buf.getvalue()
+        assert "CREATE STEP 3" in buf.getvalue()
+
+    def test_version_to_none(self):
+        with capture_context_buffer() as buf:
+            command.downgrade(self.cfg, "%s:base" % self.c, sql=True)
+        assert "CREATE TABLE alembic_version" not in buf.getvalue()
+        assert "INSERT INTO alembic_version" not in buf.getvalue()
+        assert "DROP TABLE alembic_version" in buf.getvalue()
+        assert "DROP STEP 3" in buf.getvalue()
+        assert "DROP STEP 2" in buf.getvalue()
+        assert "DROP STEP 1" in buf.getvalue()
+
+    def test_version_to_middle(self):
+        with capture_context_buffer() as buf:
+            command.downgrade(self.cfg, "%s:%s" % (self.c, self.a), sql=True)
+        assert "CREATE TABLE alembic_version" not in buf.getvalue()
+        assert "INSERT INTO alembic_version" not in buf.getvalue()
+        assert "DROP TABLE alembic_version" not in buf.getvalue()
+        assert "DROP STEP 3" in buf.getvalue()
+        assert "DROP STEP 2" in buf.getvalue()
+        assert "DROP STEP 1" not in buf.getvalue()
+
+    def test_stamp(self):
+        with capture_context_buffer() as buf:
+            command.stamp(self.cfg, "head", sql=True)
+        assert "UPDATE alembic_version "\
+            "SET version_num='%s';" % self.c in buf.getvalue()
+

diff --git a/tests/test_versioning.py b/tests/test_script_consumption.py
similarity index 64%
rename from tests/test_versioning.py
rename to tests/test_script_consumption.py
index d933f9353334e558bb8c11bfb968d93cd8bd892a..5bf77bbb0f58ffe1fe0da3a7a41899b9c7395399 100644 (file)
--- a/tests/test_versioning.py
+++ b/tests/test_script_consumption.py
@@ -1,16 +1,18 @@
+# coding: utf-8
+
 import os
+import re
 
-from alembic import command, util
-from alembic.script import ScriptDirectory
+from alembic import command, util, compat
+from alembic.script import ScriptDirectory, Script
 from alembic.testing.env import clear_staging_env, staging_env, \
-    _sqlite_testing_config, write_script, _sqlite_file_db
+    _sqlite_testing_config, write_script, _sqlite_file_db, \
+    three_rev_fixture, _no_sql_testing_config
 from alembic.testing import eq_, assert_raises_message
-from alembic.testing.fixtures import TestBase
-
-a = b = c = None
+from alembic.testing.fixtures import TestBase, capture_context_buffer
 
 
-class VersioningTest(TestBase):
+class ApplyVersionsFunctionalTest(TestBase):
     __only_on__ = 'sqlite'
 
     sourceless = False
@@ -32,10 +34,9 @@ class VersioningTest(TestBase):
         self._test_006_upgrade_again()
 
     def _test_001_revisions(self):
-        global a, b, c
-        a = util.rev_id()
-        b = util.rev_id()
-        c = util.rev_id()
+        self.a = a = util.rev_id()
+        self.b = b = util.rev_id()
+        self.c = c = util.rev_id()
 
         script = ScriptDirectory.from_config(self.cfg)
         script.generate_revision(a, None, refresh=True)
@@ -84,14 +85,14 @@ class VersioningTest(TestBase):
     """ % (c, b), sourceless=self.sourceless)
 
     def _test_002_upgrade(self):
-        command.upgrade(self.cfg, c)
+        command.upgrade(self.cfg, self.c)
         db = self.bind
         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(self):
-        command.downgrade(self.cfg, a)
+        command.downgrade(self.cfg, self.a)
         db = self.bind
         assert db.dialect.has_table(db.connect(), 'foo')
         assert not db.dialect.has_table(db.connect(), 'bar')
@@ -105,14 +106,98 @@ class VersioningTest(TestBase):
         assert not db.dialect.has_table(db.connect(), 'bat')
 
     def _test_005_upgrade(self):
-        command.upgrade(self.cfg, b)
+        command.upgrade(self.cfg, self.b)
         db = self.bind
         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')
 
     def _test_006_upgrade_again(self):
-        command.upgrade(self.cfg, b)
+        command.upgrade(self.cfg, self.b)
+        db = self.bind
+        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')
+
+
+class SourcelessApplyVersionsTest(ApplyVersionsFunctionalTest):
+    sourceless = True
+
+
+class TransactionalDDLTest(TestBase):
+    def setUp(self):
+        self.env = staging_env()
+        self.cfg = cfg = _no_sql_testing_config()
+        cfg.set_main_option('dialect_name', 'sqlite')
+        cfg.remove_main_option('url')
+
+        self.a, self.b, self.c = three_rev_fixture(cfg)
+
+    def tearDown(self):
+        clear_staging_env()
+
+    def test_begin_commit_transactional_ddl(self):
+        with capture_context_buffer(transactional_ddl=True) as buf:
+            command.upgrade(self.cfg, self.c, sql=True)
+        assert re.match(
+            (r"^BEGIN;\s+CREATE TABLE.*?%s.*" % self.a) +
+            (r".*%s" % self.b) +
+            (r".*%s.*?COMMIT;.*$" % self.c),
+
+            buf.getvalue(), re.S)
+
+    def test_begin_commit_nontransactional_ddl(self):
+        with capture_context_buffer(transactional_ddl=False) as buf:
+            command.upgrade(self.cfg, self.a, sql=True)
+        assert re.match(r"^CREATE TABLE.*?\n+$", buf.getvalue(), re.S)
+        assert "COMMIT;" not in buf.getvalue()
+
+    def test_begin_commit_per_rev_ddl(self):
+        with capture_context_buffer(transaction_per_migration=True) as buf:
+            command.upgrade(self.cfg, self.c, sql=True)
+        assert re.match(
+            (r"^BEGIN;\s+CREATE TABLE.*%s.*?COMMIT;.*" % self.a) +
+            (r"BEGIN;.*?%s.*?COMMIT;.*" % self.b) +
+            (r"BEGIN;.*?%s.*?COMMIT;.*$" % self.c),
+
+            buf.getvalue(), re.S)
+
+
+class EncodingTest(TestBase):
+
+    def setUp(self):
+        self.env = staging_env()
+        self.cfg = cfg = _no_sql_testing_config()
+        cfg.set_main_option('dialect_name', 'sqlite')
+        cfg.remove_main_option('url')
+        self.a = util.rev_id()
+        script = ScriptDirectory.from_config(cfg)
+        script.generate_revision(self.a, "revision a", refresh=True)
+        write_script(script, self.a, (compat.u("""# coding: utf-8
+from __future__ import unicode_literals
+revision = '%s'
+down_revision = None
+
+from alembic import op
+
+def upgrade():
+    op.execute("« S’il vous plaît…")
+
+def downgrade():
+    op.execute("drôle de petite voix m’a réveillé")
+
+""") % self.a), encoding='utf-8')
+
+    def tearDown(self):
+        clear_staging_env()
+
+    def test_encode(self):
+        with capture_context_buffer(
+            bytes_io=True,
+            output_encoding='utf-8'
+        ) as buf:
+            command.upgrade(self.cfg, self.a, sql=True)
+        assert compat.u("« S’il vous plaît…").encode("utf-8") in buf.getvalue()
 
 
 class VersionNameTemplateTest(TestBase):
@@ -176,23 +261,30 @@ class VersionNameTemplateTest(TestBase):
         script = ScriptDirectory.from_config(self.cfg)
         a = util.rev_id()
         script.generate_revision(a, "foobar", refresh=True)
+
+        path = script._revision_map[a].path
+        with open(path, 'w') as fp:
+            fp.write("""
+down_revision = None
+
+from alembic import op
+
+def upgrade():
+    op.execute("CREATE TABLE foo(id integer)")
+
+def downgrade():
+    op.execute("DROP TABLE foo")
+""")
+        pyc_path = util.pyc_file_from_path(path)
+        if os.access(pyc_path, os.F_OK):
+            os.unlink(pyc_path)
+
         assert_raises_message(
             util.CommandError,
             "Could not determine revision id from filename foobar_%s.py. "
             "Be sure the 'revision' variable is declared "
             "inside the script." % a,
-            write_script, script, a, """
-        down_revision = None
-
-        from alembic import op
-
-        def upgrade():
-            op.execute("CREATE TABLE foo(id integer)")
-
-        def downgrade():
-            op.execute("DROP TABLE foo")
-
-        """)
+            Script._from_path, script, path)
 
 
 class IgnoreInitTest(TestBase):
@@ -237,10 +329,6 @@ class SourcelessIgnoreInitTest(IgnoreInitTest):
     sourceless = True
 
 
-class SourcelessVersioningTest(VersioningTest):
-    sourceless = True
-
-
 class SourcelessNeedsFlagTest(TestBase):
 
     def setUp(self):

diff --git a/tests/test_revision_create.py b/tests/test_script_production.py
similarity index 100%
rename from tests/test_revision_create.py
rename to tests/test_script_production.py

diff --git a/tests/test_sql_script.py b/tests/test_sql_script.py
deleted file mode 100644 (file)
index 4633c6a..0000000
--- a/tests/test_sql_script.py
+++ /dev/null
@@ -1,139 +0,0 @@
-# coding: utf-8
-
-from __future__ import unicode_literals
-
-from alembic.testing.fixtures import TestBase
-from alembic import util
-
-from alembic.testing.env import staging_env, _no_sql_testing_config, \
-    three_rev_fixture, clear_staging_env, write_script
-from alembic.testing.fixtures import capture_context_buffer
-from alembic import command
-from alembic.script import ScriptDirectory
-import re
-
-cfg = None
-a, b, c = None, None, None
-
-
-class ThreeRevTest(TestBase):
-
-    def setUp(self):
-        global cfg, env
-        env = staging_env()
-        cfg = _no_sql_testing_config()
-        cfg.set_main_option('dialect_name', 'sqlite')
-        cfg.remove_main_option('url')
-        global a, b, c
-        a, b, c = three_rev_fixture(cfg)
-
-    def tearDown(self):
-        clear_staging_env()
-
-    def test_begin_commit_transactional_ddl(self):
-        with capture_context_buffer(transactional_ddl=True) as buf:
-            command.upgrade(cfg, c, sql=True)
-        assert re.match(
-            (r"^BEGIN;\s+CREATE TABLE.*?%s.*" % a) +
-            (r".*%s" % b) +
-            (r".*%s.*?COMMIT;.*$" % c),
-
-            buf.getvalue(), re.S)
-
-    def test_begin_commit_nontransactional_ddl(self):
-        with capture_context_buffer(transactional_ddl=False) as buf:
-            command.upgrade(cfg, a, sql=True)
-        assert re.match(r"^CREATE TABLE.*?\n+$", buf.getvalue(), re.S)
-        assert "COMMIT;" not in buf.getvalue()
-
-    def test_begin_commit_per_rev_ddl(self):
-        with capture_context_buffer(transaction_per_migration=True) as buf:
-            command.upgrade(cfg, c, sql=True)
-        assert re.match(
-            (r"^BEGIN;\s+CREATE TABLE.*%s.*?COMMIT;.*" % a) +
-            (r"BEGIN;.*?%s.*?COMMIT;.*" % b) +
-            (r"BEGIN;.*?%s.*?COMMIT;.*$" % c),
-
-            buf.getvalue(), re.S)
-
-    def test_version_from_none_insert(self):
-        with capture_context_buffer() as buf:
-            command.upgrade(cfg, a, sql=True)
-        assert "CREATE TABLE alembic_version" in buf.getvalue()
-        assert "INSERT INTO alembic_version" in buf.getvalue()
-        assert "CREATE STEP 1" in buf.getvalue()
-        assert "CREATE STEP 2" not in buf.getvalue()
-        assert "CREATE STEP 3" not in buf.getvalue()
-
-    def test_version_from_middle_update(self):
-        with capture_context_buffer() as buf:
-            command.upgrade(cfg, "%s:%s" % (b, c), sql=True)
-        assert "CREATE TABLE alembic_version" not in buf.getvalue()
-        assert "UPDATE alembic_version" in buf.getvalue()
-        assert "CREATE STEP 1" not in buf.getvalue()
-        assert "CREATE STEP 2" not in buf.getvalue()
-        assert "CREATE STEP 3" in buf.getvalue()
-
-    def test_version_to_none(self):
-        with capture_context_buffer() as buf:
-            command.downgrade(cfg, "%s:base" % c, sql=True)
-        assert "CREATE TABLE alembic_version" not in buf.getvalue()
-        assert "INSERT INTO alembic_version" not in buf.getvalue()
-        assert "DROP TABLE alembic_version" in buf.getvalue()
-        assert "DROP STEP 3" in buf.getvalue()
-        assert "DROP STEP 2" in buf.getvalue()
-        assert "DROP STEP 1" in buf.getvalue()
-
-    def test_version_to_middle(self):
-        with capture_context_buffer() as buf:
-            command.downgrade(cfg, "%s:%s" % (c, a), sql=True)
-        assert "CREATE TABLE alembic_version" not in buf.getvalue()
-        assert "INSERT INTO alembic_version" not in buf.getvalue()
-        assert "DROP TABLE alembic_version" not in buf.getvalue()
-        assert "DROP STEP 3" in buf.getvalue()
-        assert "DROP STEP 2" in buf.getvalue()
-        assert "DROP STEP 1" not in buf.getvalue()
-
-    def test_stamp(self):
-        with capture_context_buffer() as buf:
-            command.stamp(cfg, "head", sql=True)
-        assert "UPDATE alembic_version "\
-            "SET version_num='%s';" % c in buf.getvalue()
-
-
-class EncodingTest(TestBase):
-
-    def setUp(self):
-        global cfg, env, a
-        env = staging_env()
-        cfg = _no_sql_testing_config()
-        cfg.set_main_option('dialect_name', 'sqlite')
-        cfg.remove_main_option('url')
-        a = util.rev_id()
-        script = ScriptDirectory.from_config(cfg)
-        script.generate_revision(a, "revision a", refresh=True)
-        write_script(script, a, ("""# coding: utf-8
-from __future__ import unicode_literals
-revision = '%s'
-down_revision = None
-
-from alembic import op
-
-def upgrade():
-    op.execute("« S’il vous plaît…")
-
-def downgrade():
-    op.execute("drôle de petite voix m’a réveillé")
-
-""" % a), encoding='utf-8')
-
-    def tearDown(self):
-        clear_staging_env()
-
-    def test_encode(self):
-        with capture_context_buffer(
-            bytes_io=True,
-            output_encoding='utf-8'
-        ) as buf:
-            command.upgrade(cfg, a, sql=True)
-        assert "« S’il vous plaît…".encode("utf-8") in buf.getvalue()

diff --git a/tests/test_revision_paths.py b/tests/test_version_traversal.py
similarity index 100%
rename from tests/test_revision_paths.py
rename to tests/test_version_traversal.py