fix test to improve compat with windows

Change-Id: I765fd9fb406705261a23085585b2a7667d97afaf

alembic/testing/env.py

@@ -1,5 +1,6 @@
 import importlib.machinery
 import os
+from pathlib import Path
 import shutil
 import textwrap
 
@@ -16,15 +17,15 @@
 
 def _get_staging_directory():
     if provision.FOLLOWER_IDENT:
-        return "scratch_%s" % provision.FOLLOWER_IDENT
+        return f"scratch_{provision.FOLLOWER_IDENT}"
     else:
         return "scratch"
 
 
 def staging_env(create=True, template="generic", sourceless=False):
     cfg = _testing_config()
     if create:
-        path = os.path.join(_get_staging_directory(), "scripts")
+        path = _join_path(_get_staging_directory(), "scripts")
         assert not os.path.exists(path), (
             "staging directory %s already exists; poor cleanup?" % path
         )
@@ -47,7 +48,7 @@ def staging_env(create=True, template="generic", sourceless=False):
                 "pep3147_everything",
             ), sourceless
             make_sourceless(
-                os.path.join(path, "env.py"),
+                _join_path(path, "env.py"),
                 "pep3147" if "pep3147" in sourceless else "simple",
             )
 
@@ -63,14 +64,14 @@ def clear_staging_env():
 
 
 def script_file_fixture(txt):
-    dir_ = os.path.join(_get_staging_directory(), "scripts")
-    path = os.path.join(dir_, "script.py.mako")
+    dir_ = _join_path(_get_staging_directory(), "scripts")
+    path = _join_path(dir_, "script.py.mako")
     with open(path, "w") as f:
         f.write(txt)
 
 
 def env_file_fixture(txt):
-    dir_ = os.path.join(_get_staging_directory(), "scripts")
+    dir_ = _join_path(_get_staging_directory(), "scripts")
     txt = (
         """
 from alembic import context
@@ -80,7 +81,7 @@ def env_file_fixture(txt):
         + txt
     )
 
-    path = os.path.join(dir_, "env.py")
+    path = _join_path(dir_, "env.py")
     pyc_path = util.pyc_file_from_path(path)
     if pyc_path:
         os.unlink(pyc_path)
@@ -90,26 +91,26 @@ def env_file_fixture(txt):
 
 
 def _sqlite_file_db(tempname="foo.db", future=False, scope=None, **options):
-    dir_ = os.path.join(_get_staging_directory(), "scripts")
+    dir_ = _join_path(_get_staging_directory(), "scripts")
     url = "sqlite:///%s/%s" % (dir_, tempname)
     if scope and util.sqla_14:
         options["scope"] = scope
     return testing_util.testing_engine(url=url, future=future, options=options)
 
 
 def _sqlite_testing_config(sourceless=False, future=False):
-    dir_ = os.path.join(_get_staging_directory(), "scripts")
-    url = "sqlite:///%s/foo.db" % dir_
+    dir_ = _join_path(_get_staging_directory(), "scripts")
+    url = f"sqlite:///{dir_}/foo.db"
 
     sqlalchemy_future = future or ("future" in config.db.__class__.__module__)
 
     return _write_config_file(
-        """
+        f"""
 [alembic]
-script_location = %s
-sqlalchemy.url = %s
-sourceless = %s
-%s
+script_location = {dir_}
+sqlalchemy.url = {url}
+sourceless = {"true" if sourceless else "false"}
+{"sqlalchemy.future = true" if sqlalchemy_future else ""}
 
 [loggers]
 keys = root,sqlalchemy
@@ -140,29 +141,24 @@ class = StreamHandler
 format = %%(levelname)-5.5s [%%(name)s] %%(message)s
 datefmt = %%H:%%M:%%S
     """
-        % (
-            dir_,
-            url,
-            "true" if sourceless else "false",
-            "sqlalchemy.future = true" if sqlalchemy_future else "",
-        )
     )
 
 
 def _multi_dir_testing_config(sourceless=False, extra_version_location=""):
-    dir_ = os.path.join(_get_staging_directory(), "scripts")
+    dir_ = _join_path(_get_staging_directory(), "scripts")
     sqlalchemy_future = "future" in config.db.__class__.__module__
 
     url = "sqlite:///%s/foo.db" % dir_
 
     return _write_config_file(
-        """
+        f"""
 [alembic]
-script_location = %s
-sqlalchemy.url = %s
-sqlalchemy.future = %s
-sourceless = %s
-version_locations = %%(here)s/model1/ %%(here)s/model2/ %%(here)s/model3/ %s
+script_location = {dir_}
+sqlalchemy.url = {url}
+sqlalchemy.future = {"true" if sqlalchemy_future else "false"}
+sourceless = {"true" if sourceless else "false"}
+version_locations = %(here)s/model1/ %(here)s/model2/ %(here)s/model3/ \
+{extra_version_location}
 
 [loggers]
 keys = root
@@ -188,26 +184,19 @@ class = StreamHandler
 format = %%(levelname)-5.5s [%%(name)s] %%(message)s
 datefmt = %%H:%%M:%%S
     """
-        % (
-            dir_,
-            url,
-            "true" if sqlalchemy_future else "false",
-            "true" if sourceless else "false",
-            extra_version_location,
-        )
     )
 
 
 def _no_sql_testing_config(dialect="postgresql", directives=""):
     """use a postgresql url with no host so that
     connections guaranteed to fail"""
-    dir_ = os.path.join(_get_staging_directory(), "scripts")
+    dir_ = _join_path(_get_staging_directory(), "scripts")
     return _write_config_file(
-        """
+        f"""
 [alembic]
-script_location = %s
-sqlalchemy.url = %s://
-%s
+script_location ={dir_}
+sqlalchemy.url = {dialect}://
+{directives}
 
 [loggers]
 keys = root
@@ -234,7 +223,6 @@ class = StreamHandler
 datefmt = %%H:%%M:%%S
 
 """
-        % (dir_, dialect, directives)
     )
 
 
@@ -250,7 +238,7 @@ def _testing_config():
 
     if not os.access(_get_staging_directory(), os.F_OK):
         os.mkdir(_get_staging_directory())
-    return Config(os.path.join(_get_staging_directory(), "test_alembic.ini"))
+    return Config(_join_path(_get_staging_directory(), "test_alembic.ini"))
 
 
 def write_script(
@@ -271,7 +259,7 @@ def write_script(
     old = scriptdir.revision_map.get_revision(script.revision)
     if old.down_revision != script.down_revision:
         raise Exception(
-            "Can't change down_revision " "on a refresh operation."
+            "Can't change down_revision on a refresh operation."
         )
     scriptdir.revision_map.add_revision(script, _replace=True)
 
@@ -312,9 +300,9 @@ def three_rev_fixture(cfg):
     write_script(
         script,
         a,
-        """\
+        f"""\
 "Rev A"
-revision = '%s'
+revision = '{a}'
 down_revision = None
 
 from alembic import op
@@ -327,8 +315,7 @@ def upgrade():
 def downgrade():
     op.execute("DROP STEP 1")
 
-"""
-        % a,
+""",
     )
 
     script.generate_revision(b, "revision b", refresh=True, head=a)
@@ -358,10 +345,10 @@ def downgrade():
     write_script(
         script,
         c,
-        """\
+        f"""\
 "Rev C"
-revision = '%s'
-down_revision = '%s'
+revision = '{c}'
+down_revision = '{b}'
 
 from alembic import op
 
@@ -373,8 +360,7 @@ def upgrade():
 def downgrade():
     op.execute("DROP STEP 3")
 
-"""
-        % (c, b),
+""",
     )
     return a, b, c
 
@@ -396,10 +382,10 @@ def multi_heads_fixture(cfg, a, b, c):
     write_script(
         script,
         d,
-        """\
+        f"""\
 "Rev D"
-revision = '%s'
-down_revision = '%s'
+revision = '{d}'
+down_revision = '{b}'
 
 from alembic import op
 
@@ -412,7 +398,6 @@ def downgrade():
     op.execute("DROP STEP 4")
 
 """
-        % (d, b),
     )
 
     script.generate_revision(
@@ -421,10 +406,10 @@ def downgrade():
     write_script(
         script,
         e,
-        """\
+        f"""\
 "Rev E"
-revision = '%s'
-down_revision = '%s'
+revision = '{e}'
+down_revision = '{d}'
 
 from alembic import op
 
@@ -436,8 +421,7 @@ def upgrade():
 def downgrade():
     op.execute("DROP STEP 5")
 
-"""
-        % (e, d),
+""",
     )
 
     script.generate_revision(
@@ -446,10 +430,10 @@ def downgrade():
     write_script(
         script,
         f,
-        """\
+        f"""\
 "Rev F"
-revision = '%s'
-down_revision = '%s'
+revision = '{f}'
+down_revision = '{b}'
 
 from alembic import op
 
@@ -461,8 +445,7 @@ def upgrade():
 def downgrade():
     op.execute("DROP STEP 6")
 
-"""
-        % (f, b),
+""",
     )
 
     return d, e, f
@@ -471,25 +454,25 @@ def downgrade():
 def _multidb_testing_config(engines):
     """alembic.ini fixture to work exactly with the 'multidb' template"""
 
-    dir_ = os.path.join(_get_staging_directory(), "scripts")
+    dir_ = _join_path(_get_staging_directory(), "scripts")
 
     sqlalchemy_future = "future" in config.db.__class__.__module__
 
     databases = ", ".join(engines.keys())
     engines = "\n\n".join(
-        "[%s]\n" "sqlalchemy.url = %s" % (key, value.url)
+        f"[{key}]\nsqlalchemy.url = {value.url}"
         for key, value in engines.items()
     )
 
     return _write_config_file(
-        """
+        f"""
 [alembic]
-script_location = %s
+script_location = {dir_}
 sourceless = false
-sqlalchemy.future = %s
-databases = %s
+sqlalchemy.future = {"true" if sqlalchemy_future else "false"}
+databases = {databases}
 
-%s
+{engines}
 [loggers]
 keys = root
 
@@ -514,5 +497,8 @@ class = StreamHandler
 format = %%(levelname)-5.5s [%%(name)s] %%(message)s
 datefmt = %%H:%%M:%%S
     """
-        % (dir_, "true" if sqlalchemy_future else "false", databases, engines)
     )
+
+
+def _join_path(base: str, *more: str):
+    return str(Path(base).joinpath(*more).as_posix())

tests/test_command.py

@@ -612,20 +612,22 @@ def tearDown(self):
 
     def _env_fixture(self, version_table_pk=True):
         env_file_fixture(
-            """
+            f"""
 
 from sqlalchemy import MetaData, engine_from_config
 target_metadata = MetaData()
 
 engine = engine_from_config(
     config.get_section(config.config_ini_section),
-    prefix='sqlalchemy.')
+    prefix='sqlalchemy.'
+)
 
 connection = engine.connect()
 
 context.configure(
-    connection=connection, target_metadata=target_metadata,
-    version_table_pk=%r
+    connection=connection,
+    target_metadata=target_metadata,
+    version_table_pk={version_table_pk}
 )
 
 try:
@@ -636,7 +638,6 @@ def _env_fixture(self, version_table_pk=True):
     engine.dispose()
 
 """
-            % (version_table_pk,)
         )
 
     def test_check_no_changes(self):