Skip to content

Commit

Permalink
test(sqla_factory): added association_proxy tests for SQLAlchemy v1.4
Browse files Browse the repository at this point in the history
  • Loading branch information
nisemenov committed Jan 14, 2025
1 parent a71d048 commit 35f379c
Show file tree
Hide file tree
Showing 5 changed files with 1,590 additions and 554 deletions.
38 changes: 0 additions & 38 deletions tests/sqlalchemy_factory/models.py

This file was deleted.

77 changes: 77 additions & 0 deletions tests/sqlalchemy_factory/test_association_proxy_v1.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
from typing import Optional

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import relationship
from sqlalchemy.orm.decl_api import DeclarativeMeta, registry

from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory

_registry = registry()


class Base(metaclass=DeclarativeMeta):
__abstract__ = True
__allow_unmapped__ = True

registry = _registry
metadata = _registry.metadata


class User(Base):
__tablename__ = "users"

id = Column(Integer, primary_key=True)
name = Column(String)

user_keyword_associations = relationship(
"UserKeywordAssociation",
back_populates="user",
)
keywords = association_proxy(
"user_keyword_associations", "keyword", creator=lambda keyword_obj: UserKeywordAssociation(keyword=keyword_obj)
)


class UserKeywordAssociation(Base):
__tablename__ = "user_keyword"

user_id = Column(Integer, ForeignKey("users.id"), primary_key=True)
keyword_id = Column(Integer, ForeignKey("keywords.id"), primary_key=True)

user = relationship(User, back_populates="user_keyword_associations")
keyword = relationship("Keyword")

# for prevent mypy error: Unexpected keyword argument "keyword" for "UserKeywordAssociation" [call-arg]
def __init__(self, keyword: Optional["Keyword"] = None):
self.keyword = keyword


class Keyword(Base):
__tablename__ = "keywords"

id = Column(Integer, primary_key=True)
keyword = Column(String)


def test_association_proxy() -> None:
class UserFactory(SQLAlchemyFactory[User]):
__set_association_proxy__ = True

user = UserFactory.build()
assert isinstance(user.keywords[0], Keyword)
assert isinstance(user.user_keyword_associations[0], UserKeywordAssociation)


async def test_complex_association_proxy() -> None:
class KeywordFactory(SQLAlchemyFactory[Keyword]): ...

class ComplexUserFactory(SQLAlchemyFactory[User]):
__set_association_proxy__ = True

keywords = KeywordFactory.batch_async(1)

user = await ComplexUserFactory.build_async()
assert isinstance(user, User)
assert isinstance(user.keywords[0], Keyword)
assert isinstance(user.user_keyword_associations[0], UserKeywordAssociation)
67 changes: 67 additions & 0 deletions tests/sqlalchemy_factory/test_association_proxy_v2.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
import pytest
from sqlalchemy import ForeignKey, __version__, orm
from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy
from sqlalchemy.orm import Mapped, relationship

from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory

if __version__.startswith("1"):
pytest.importorskip("SQLAlchemy", "2")


class Base(orm.DeclarativeBase):
pass


class User(Base):
__tablename__ = "users"

id: Mapped[int] = orm.mapped_column(primary_key=True)
name: Mapped[str]

user_keyword_associations: Mapped[list["UserKeywordAssociation"]] = relationship(
back_populates="user",
)
keywords: AssociationProxy[list["Keyword"]] = association_proxy(
"user_keyword_associations",
"keyword",
creator=lambda keyword_obj: UserKeywordAssociation(keyword=keyword_obj),
)


class UserKeywordAssociation(Base):
__tablename__ = "user_keyword"
user_id: Mapped[int] = orm.mapped_column(ForeignKey("users.id"), primary_key=True)
keyword_id: Mapped[int] = orm.mapped_column(ForeignKey("keywords.id"), primary_key=True)

user: Mapped[User] = relationship(back_populates="user_keyword_associations")
keyword: Mapped["Keyword"] = relationship()


class Keyword(Base):
__tablename__ = "keywords"
id: Mapped[int] = orm.mapped_column(primary_key=True)
keyword: Mapped[str]


def test_association_proxy() -> None:
class UserFactory(SQLAlchemyFactory[User]):
__set_association_proxy__ = True

user = UserFactory.build()
assert isinstance(user.keywords[0], Keyword)
assert isinstance(user.user_keyword_associations[0], UserKeywordAssociation)


async def test_complex_association_proxy() -> None:
class KeywordFactory(SQLAlchemyFactory[Keyword]): ...

class ComplexUserFactory(SQLAlchemyFactory[User]):
__set_association_proxy__ = True

keywords = KeywordFactory.batch_async(1)

user = await ComplexUserFactory.build_async()
assert isinstance(user, User)
assert isinstance(user.keywords[0], Keyword)
assert isinstance(user.user_keyword_associations[0], UserKeywordAssociation)
25 changes: 0 additions & 25 deletions tests/sqlalchemy_factory/test_sqlalchemy_factory_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,8 +33,6 @@
from polyfactory.factories.sqlalchemy_factory import SQLAlchemyFactory
from polyfactory.fields import Ignore

from .models import Keyword, User, UserKeywordAssociation


@pytest.fixture()
def engine() -> Engine:
Expand Down Expand Up @@ -252,29 +250,6 @@ class AuthorFactory(SQLAlchemyFactory[Author]):
assert isinstance(result.books[0], Book)


def test_association_proxy() -> None:
class UserFactory(SQLAlchemyFactory[User]):
__set_association_proxy__ = True

user = UserFactory.build()
assert isinstance(user.keywords[0], Keyword)
assert isinstance(user.user_keyword_associations[0], UserKeywordAssociation)


async def test_complex_association_proxy() -> None:
class KeywordFactory(SQLAlchemyFactory[Keyword]): ...

class ComplexUserFactory(SQLAlchemyFactory[User]):
__set_association_proxy__ = True

keywords = KeywordFactory.batch_async(1)

user = await ComplexUserFactory.build_async()
assert isinstance(user, User)
assert isinstance(user.keywords[0], Keyword)
assert isinstance(user.user_keyword_associations[0], UserKeywordAssociation)


def test_sqla_factory_create(engine: Engine) -> None:
Base.metadata.create_all(engine)

Expand Down
Loading

0 comments on commit 35f379c

Please sign in to comment.