rendered_metadata_default,
rendered_inspector_default,
):
+
# don't do defaults for SERIAL columns
if (
metadata_column.primary_key
conn_col_default = rendered_inspector_default
+ if conn_col_default and re.match(
+ r"nextval\('(.+?)'::regclass\)", conn_col_default
+ ):
+ conn_col_default = conn_col_default.replace("::regclass", "")
+
defaults_equal = conn_col_default == rendered_metadata_default
if defaults_equal:
return False
metadata_default = literal_column(metadata_default)
# run a real compare against the server
+ # TODO: this seems quite a bad idea for a default that's a SQL
+ # function! SQL functions are not deterministic!
conn = self.connection
assert conn is not None
return not conn.scalar(
def setup_class(cls):
cls.bind = config.db
staging_env()
- cls.migration_context = MigrationContext.configure(
- connection=cls.bind.connect(),
- opts={"compare_type": True, "compare_server_default": True},
- )
def setUp(self):
self.metadata = MetaData()
+ self.migration_context = MigrationContext.configure(
+ connection=self.bind.connect(),
+ opts={"compare_type": True, "compare_server_default": True},
+ )
self.autogen_context = api.AutogenContext(self.migration_context)
@classmethod
clear_staging_env()
def tearDown(self):
+ self.migration_context.connection.close()
+
with config.db.begin() as conn:
self.metadata.drop_all(conn)
def _compare_default_roundtrip(
self, type_, orig_default, alternate=None, diff_expected=None
):
+ # note this only tests compare_server_default including
+ # postgresql.compare_server_default. it does not run PG
+ # autogen_column_reflect() which is involved with omitting SERIAL
+ # columns
diff_expected = (
diff_expected
if diff_expected is not None
)
assert not self._compare_default(t1, t2, t2.c.id, "")
+ def test_non_pk_sequence(self):
+ """test issue #1507"""
+ sequential_id_seq = Sequence(
+ "post_sequential_id_seq", metadata=self.metadata, start=1
+ )
+ sequential_id_seq.create(self.bind)
+ self._compare_default_roundtrip(
+ Integer, sequential_id_seq.next_value()
+ )
+
class PostgresqlDetectSerialTest(TestBase):
__only_on__ = "postgresql"
seq,
)
- @testing.combinations((None,), ("test_schema",))
- def test_numeric(self, schema):
+ @testing.combinations((None,), ("test_schema",), argnames="schema")
+ @testing.variation("use_pk", [True, False])
+ def test_numeric(self, schema, use_pk):
seq = Sequence("x_id_seq", schema=schema)
seq_name = seq.name if schema is None else f"{schema}.{seq.name}"
self._expect_default(
"x",
Numeric(8, 2),
server_default=seq.next_value(),
- primary_key=True,
+ primary_key=bool(use_pk),
),
schema,
seq,