fix: Ensure base model compatibility with DeclarativeBase subclasses. (…

…#197)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pytest-mock-resources"
-version = "2.9.1"
+version = "2.9.2"
 description = "A pytest plugin for easily instantiating reproducible mock resources."
 authors = [
  "Omar Khan <oakhan3@gmail.com>",

src/pytest_mock_resources/compat/sqlalchemy.py

@@ -1,10 +1,14 @@
+from typing import Optional
+
 import sqlalchemy
 import sqlalchemy.engine.url
+from sqlalchemy.schema import MetaData
 
 from pytest_mock_resources.compat.import_ import ImportAdaptor
 
 version = getattr(sqlalchemy, "__version__", "")
 
+
 if version.startswith("1.4") or version.startswith("2."):
  from sqlalchemy.ext import asyncio
  from sqlalchemy.orm import declarative_base, DeclarativeMeta
@@ -29,6 +33,14 @@ def _select(*args, **kwargs):
  select = _select
 
 
+def extract_model_base_metadata(base) -> Optional[sqlalchemy.MetaData]:
+ metadata = getattr(base, "metadata", None)
+ if isinstance(metadata, MetaData):
+ return metadata
+
+ return None
+
+
 __all__ = [
  "asyncio",
  "declarative_base",

src/pytest_mock_resources/sqlalchemy.py

@@ -247,9 +247,9 @@ def normalize_actions(
  unique_metadata: Set[MetaData] = set()
  normalized_actions: List[T] = []
  for action in ordered_actions:
- if isinstance(action, compat.sqlalchemy.DeclarativeMeta):
-  action = action.metadata
- normalized_actions.append(action)
+ metadata = compat.sqlalchemy.extract_model_base_metadata(action)
+ if metadata:
+ normalized_actions.append(metadata)
 
  elif isinstance(action, Rows):
  new_metadata = {row.metadata for row in action.rows} - unique_metadata

tests/__init__.py

@@ -4,6 +4,7 @@
 
 from pytest_mock_resources.compat import sqlalchemy
 
+is_at_least_sqlalchemy2 = sqlalchemy.version.startswith("2.")
 is_sqlalchemy2 = sqlalchemy.version.startswith("1.4") or sqlalchemy.version.startswith("2.")
 skip_if_sqlalchemy2 = pytest.mark.skipif(
  is_sqlalchemy2,

tests/fixture/postgres/test_sqlalchemy2_base_class.py

@@ -0,0 +1,21 @@
+from tests import is_at_least_sqlalchemy2
+
+if is_at_least_sqlalchemy2:
+ from sqlalchemy import Column, Integer
+ from sqlalchemy.orm import DeclarativeBase # type: ignore
+
+ from pytest_mock_resources import create_postgres_fixture
+
+ class Base(DeclarativeBase):
+ ...
+
+ class Thing(Base):
+ __tablename__ = "thing"
+
+ id = Column(Integer, autoincrement=True, primary_key=True)
+
+ pg = create_postgres_fixture(Base, session=True)
+
+ def test_creates_ddl(pg):
+ rows = pg.query(Thing).all()
+ assert len(rows) == 0