From afaba66ea550d9aefc823b7025d0e2b56d4a37fa Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Mon, 10 Nov 2014 13:20:40 -0500 Subject: [PATCH] - reorganize tests that deal with running scripts and reading scripts; try to put tests against common themes together, improve names --- tests/test_command.py | 65 +++++++- ...rsioning.py => test_script_consumption.py} | 150 ++++++++++++++---- ...on_create.py => test_script_production.py} | 0 tests/test_sql_script.py | 139 ---------------- ...ion_paths.py => test_version_traversal.py} | 0 5 files changed, 181 insertions(+), 173 deletions(-) rename tests/{test_versioning.py => test_script_consumption.py} (64%) rename tests/{test_revision_create.py => test_script_production.py} (100%) delete mode 100644 tests/test_sql_script.py rename tests/{test_revision_paths.py => test_version_traversal.py} (100%) diff --git a/tests/test_command.py b/tests/test_command.py index 6d70f5ad..58fed893 100644 --- 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 d933f935..5bf77bbb 100644 --- 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 index 4633c6ad..00000000 --- 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 -- 2.47.2