split tests into packages
This commit is contained in:
collerek
0
tests/test_relations/__init__.py
Normal file
0
tests/test_relations/__init__.py
Normal file
156
tests/test_relations/test_cascades.py
Normal file
156
tests/test_relations/test_cascades.py
Normal file
@ -0,0 +1,156 @@
|
||||
from typing import Optional
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
|
||||
import ormar
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class Band(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "bands"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
|
||||
|
||||
class ArtistsBands(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "artists_x_bands"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
|
||||
|
||||
class Artist(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "artists"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
bands = ormar.ManyToMany(Band, through=ArtistsBands)
|
||||
|
||||
|
||||
class Album(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "albums"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
artist: Optional[Artist] = ormar.ForeignKey(Artist, ondelete="CASCADE")
|
||||
|
||||
|
||||
class Track(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "tracks"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
album: Optional[Album] = ormar.ForeignKey(Album, ondelete="CASCADE")
|
||||
title: str = ormar.String(max_length=100)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = sqlalchemy.create_engine(DATABASE_URL)
|
||||
metadata.drop_all(engine)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
async def cleanup():
|
||||
yield
|
||||
async with database:
|
||||
await Band.objects.delete(each=True)
|
||||
await Artist.objects.delete(each=True)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_simple_cascade(cleanup):
|
||||
async with database:
|
||||
artist = await Artist(name="Dr Alban").save()
|
||||
await Album(name="Jamaica", artist=artist).save()
|
||||
await Artist.objects.delete(id=artist.id)
|
||||
artists = await Artist.objects.all()
|
||||
assert len(artists) == 0
|
||||
|
||||
albums = await Album.objects.all()
|
||||
assert len(albums) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_nested_cascade(cleanup):
|
||||
async with database:
|
||||
artist = await Artist(name="Dr Alban").save()
|
||||
album = await Album(name="Jamaica", artist=artist).save()
|
||||
await Track(title="Yuhu", album=album).save()
|
||||
|
||||
await Artist.objects.delete(id=artist.id)
|
||||
|
||||
artists = await Artist.objects.all()
|
||||
assert len(artists) == 0
|
||||
|
||||
albums = await Album.objects.all()
|
||||
assert len(albums) == 0
|
||||
|
||||
tracks = await Track.objects.all()
|
||||
assert len(tracks) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_many_to_many_cascade(cleanup):
|
||||
async with database:
|
||||
artist = await Artist(name="Dr Alban").save()
|
||||
band = await Band(name="Scorpions").save()
|
||||
await artist.bands.add(band)
|
||||
|
||||
check = await Artist.objects.select_related("bands").get()
|
||||
assert check.bands[0].name == "Scorpions"
|
||||
|
||||
await Artist.objects.delete(id=artist.id)
|
||||
|
||||
artists = await Artist.objects.all()
|
||||
assert len(artists) == 0
|
||||
|
||||
bands = await Band.objects.all()
|
||||
assert len(bands) == 1
|
||||
|
||||
connections = await ArtistsBands.objects.all()
|
||||
assert len(connections) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reverse_many_to_many_cascade(cleanup):
|
||||
async with database:
|
||||
artist = await Artist(name="Dr Alban").save()
|
||||
band = await Band(name="Scorpions").save()
|
||||
await artist.bands.add(band)
|
||||
|
||||
check = await Artist.objects.select_related("bands").get()
|
||||
assert check.bands[0].name == "Scorpions"
|
||||
|
||||
await Band.objects.delete(id=band.id)
|
||||
|
||||
artists = await Artist.objects.all()
|
||||
assert len(artists) == 1
|
||||
|
||||
connections = await ArtistsBands.objects.all()
|
||||
assert len(connections) == 0
|
||||
|
||||
bands = await Band.objects.all()
|
||||
assert len(bands) == 0
|
||||
55
tests/test_relations/test_database_fk_creation.py
Normal file
55
tests/test_relations/test_database_fk_creation.py
Normal file
@ -0,0 +1,55 @@
|
||||
from typing import Optional
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
|
||||
import ormar
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
engine = sqlalchemy.create_engine(DATABASE_URL)
|
||||
|
||||
|
||||
class Artist(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "artists"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
|
||||
|
||||
class Album(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "albums"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
artist: Optional[Artist] = ormar.ForeignKey(Artist, ondelete="CASCADE")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
metadata.drop_all(engine)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
def test_simple_cascade():
|
||||
inspector = sqlalchemy.inspect(engine)
|
||||
columns = inspector.get_columns("albums")
|
||||
assert len(columns) == 3
|
||||
col_names = [col.get("name") for col in columns]
|
||||
assert sorted(["id", "name", "artist"]) == sorted(col_names)
|
||||
fks = inspector.get_foreign_keys("albums")
|
||||
assert len(fks) == 1
|
||||
assert fks[0]["name"] == "fk_albums_artists_id_artist"
|
||||
assert fks[0]["constrained_columns"][0] == "artist"
|
||||
assert fks[0]["referred_columns"][0] == "id"
|
||||
assert fks[0]["options"].get("ondelete") == "CASCADE"
|
||||
429
tests/test_relations/test_foreign_keys.py
Normal file
429
tests/test_relations/test_foreign_keys.py
Normal file
@ -0,0 +1,429 @@
|
||||
from typing import Optional
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
|
||||
import ormar
|
||||
from ormar.exceptions import NoMatch, MultipleMatches, RelationshipInstanceError
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL, force_rollback=True)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class Album(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "albums"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
is_best_seller: bool = ormar.Boolean(default=False)
|
||||
|
||||
|
||||
class Track(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "tracks"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
album: Optional[Album] = ormar.ForeignKey(Album)
|
||||
title: str = ormar.String(max_length=100)
|
||||
position: int = ormar.Integer()
|
||||
play_count: int = ormar.Integer(nullable=True, default=0)
|
||||
|
||||
|
||||
class Cover(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "covers"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
album: Optional[Album] = ormar.ForeignKey(Album, related_name="cover_pictures")
|
||||
title: str = ormar.String(max_length=100)
|
||||
|
||||
|
||||
class Organisation(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "org"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
ident: str = ormar.String(max_length=100, choices=["ACME Ltd", "Other ltd"])
|
||||
|
||||
|
||||
class Team(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "teams"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
org: Optional[Organisation] = ormar.ForeignKey(Organisation)
|
||||
name: str = ormar.String(max_length=100)
|
||||
|
||||
|
||||
class Member(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "members"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
team: Optional[Team] = ormar.ForeignKey(Team)
|
||||
email: str = ormar.String(max_length=100)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = sqlalchemy.create_engine(DATABASE_URL)
|
||||
metadata.drop_all(engine)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wrong_query_foreign_key_type():
|
||||
async with database:
|
||||
with pytest.raises(RelationshipInstanceError):
|
||||
Track(title="The Error", album="wrong_pk_type")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setting_explicitly_empty_relation():
|
||||
async with database:
|
||||
track = Track(album=None, title="The Bird", position=1)
|
||||
assert track.album is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_related_name():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
album = await Album.objects.create(name="Vanilla")
|
||||
await Cover.objects.create(album=album, title="The cover file")
|
||||
assert len(album.cover_pictures) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_model_crud():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
album = Album(name="Jamaica")
|
||||
await album.save()
|
||||
track1 = Track(album=album, title="The Bird", position=1)
|
||||
track2 = Track(album=album, title="Heart don't stand a chance", position=2)
|
||||
track3 = Track(album=album, title="The Waters", position=3)
|
||||
await track1.save()
|
||||
await track2.save()
|
||||
await track3.save()
|
||||
|
||||
track = await Track.objects.get(title="The Bird")
|
||||
assert track.album.pk == album.pk
|
||||
assert isinstance(track.album, ormar.Model)
|
||||
assert track.album.name is None
|
||||
await track.album.load()
|
||||
assert track.album.name == "Jamaica"
|
||||
|
||||
assert len(album.tracks) == 3
|
||||
assert album.tracks[1].title == "Heart don't stand a chance"
|
||||
|
||||
album1 = await Album.objects.get(name="Jamaica")
|
||||
assert album1.pk == album.pk
|
||||
assert album1.tracks == []
|
||||
|
||||
await Track.objects.create(
|
||||
album={"id": track.album.pk}, title="The Bird2", position=4
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_select_related():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
album = Album(name="Malibu")
|
||||
await album.save()
|
||||
track1 = Track(album=album, title="The Bird", position=1)
|
||||
track2 = Track(album=album, title="Heart don't stand a chance", position=2)
|
||||
track3 = Track(album=album, title="The Waters", position=3)
|
||||
await track1.save()
|
||||
await track2.save()
|
||||
await track3.save()
|
||||
|
||||
fantasies = Album(name="Fantasies")
|
||||
await fantasies.save()
|
||||
track4 = Track(album=fantasies, title="Help I'm Alive", position=1)
|
||||
track5 = Track(album=fantasies, title="Sick Muse", position=2)
|
||||
track6 = Track(album=fantasies, title="Satellite Mind", position=3)
|
||||
await track4.save()
|
||||
await track5.save()
|
||||
await track6.save()
|
||||
|
||||
track = await Track.objects.select_related("album").get(title="The Bird")
|
||||
assert track.album.name == "Malibu"
|
||||
|
||||
tracks = await Track.objects.select_related("album").all()
|
||||
assert len(tracks) == 6
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_model_removal_from_relations():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
album = Album(name="Chichi")
|
||||
await album.save()
|
||||
track1 = Track(album=album, title="The Birdman", position=1)
|
||||
track2 = Track(album=album, title="Superman", position=2)
|
||||
track3 = Track(album=album, title="Wonder Woman", position=3)
|
||||
await track1.save()
|
||||
await track2.save()
|
||||
await track3.save()
|
||||
|
||||
assert len(album.tracks) == 3
|
||||
await album.tracks.remove(track1)
|
||||
assert len(album.tracks) == 2
|
||||
assert track1.album is None
|
||||
|
||||
await track1.update()
|
||||
track1 = await Track.objects.get(title="The Birdman")
|
||||
assert track1.album is None
|
||||
|
||||
await album.tracks.add(track1)
|
||||
assert len(album.tracks) == 3
|
||||
assert track1.album == album
|
||||
|
||||
await track1.update()
|
||||
track1 = await Track.objects.select_related("album__tracks").get(
|
||||
title="The Birdman"
|
||||
)
|
||||
album = await Album.objects.select_related("tracks").get(name="Chichi")
|
||||
assert track1.album == album
|
||||
|
||||
track1.remove(album, name="album")
|
||||
assert track1.album is None
|
||||
assert len(album.tracks) == 2
|
||||
|
||||
track2.remove(album, name="album")
|
||||
assert track2.album is None
|
||||
assert len(album.tracks) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fk_filter():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
malibu = Album(name="Malibu%")
|
||||
await malibu.save()
|
||||
await Track.objects.create(album=malibu, title="The Bird", position=1)
|
||||
await Track.objects.create(
|
||||
album=malibu, title="Heart don't stand a chance", position=2
|
||||
)
|
||||
await Track.objects.create(album=malibu, title="The Waters", position=3)
|
||||
|
||||
fantasies = await Album.objects.create(name="Fantasies")
|
||||
await Track.objects.create(
|
||||
album=fantasies, title="Help I'm Alive", position=1
|
||||
)
|
||||
await Track.objects.create(album=fantasies, title="Sick Muse", position=2)
|
||||
await Track.objects.create(
|
||||
album=fantasies, title="Satellite Mind", position=3
|
||||
)
|
||||
|
||||
tracks = (
|
||||
await Track.objects.select_related("album")
|
||||
.filter(album__name="Fantasies")
|
||||
.all()
|
||||
)
|
||||
assert len(tracks) == 3
|
||||
for track in tracks:
|
||||
assert track.album.name == "Fantasies"
|
||||
|
||||
tracks = (
|
||||
await Track.objects.select_related("album")
|
||||
.filter(album__name__icontains="fan")
|
||||
.all()
|
||||
)
|
||||
assert len(tracks) == 3
|
||||
for track in tracks:
|
||||
assert track.album.name == "Fantasies"
|
||||
|
||||
tracks = await Track.objects.filter(album__name__contains="Fan").all()
|
||||
assert len(tracks) == 3
|
||||
for track in tracks:
|
||||
assert track.album.name == "Fantasies"
|
||||
|
||||
tracks = await Track.objects.filter(album__name__contains="Malibu%").all()
|
||||
assert len(tracks) == 3
|
||||
|
||||
tracks = (
|
||||
await Track.objects.filter(album=malibu).select_related("album").all()
|
||||
)
|
||||
assert len(tracks) == 3
|
||||
for track in tracks:
|
||||
assert track.album.name == "Malibu%"
|
||||
|
||||
tracks = await Track.objects.select_related("album").all(album=malibu)
|
||||
assert len(tracks) == 3
|
||||
for track in tracks:
|
||||
assert track.album.name == "Malibu%"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_fk():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
acme = await Organisation.objects.create(ident="ACME Ltd")
|
||||
red_team = await Team.objects.create(org=acme, name="Red Team")
|
||||
blue_team = await Team.objects.create(org=acme, name="Blue Team")
|
||||
await Member.objects.create(team=red_team, email="a@example.org")
|
||||
await Member.objects.create(team=red_team, email="b@example.org")
|
||||
await Member.objects.create(team=blue_team, email="c@example.org")
|
||||
await Member.objects.create(team=blue_team, email="d@example.org")
|
||||
|
||||
other = await Organisation.objects.create(ident="Other ltd")
|
||||
team = await Team.objects.create(org=other, name="Green Team")
|
||||
await Member.objects.create(team=team, email="e@example.org")
|
||||
|
||||
members = (
|
||||
await Member.objects.select_related("team__org")
|
||||
.filter(team__org__ident="ACME Ltd")
|
||||
.all()
|
||||
)
|
||||
assert len(members) == 4
|
||||
for member in members:
|
||||
assert member.team.org.ident == "ACME Ltd"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wrong_choices():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
with pytest.raises(ValueError):
|
||||
await Organisation.objects.create(ident="Test 1")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pk_filter():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
fantasies = await Album.objects.create(name="Test")
|
||||
track = await Track.objects.create(
|
||||
album=fantasies, title="Test1", position=1
|
||||
)
|
||||
await Track.objects.create(album=fantasies, title="Test2", position=2)
|
||||
await Track.objects.create(album=fantasies, title="Test3", position=3)
|
||||
tracks = (
|
||||
await Track.objects.select_related("album").filter(pk=track.pk).all()
|
||||
)
|
||||
assert len(tracks) == 1
|
||||
|
||||
tracks = (
|
||||
await Track.objects.select_related("album")
|
||||
.filter(position=2, album__name="Test")
|
||||
.all()
|
||||
)
|
||||
assert len(tracks) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_limit_and_offset():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
fantasies = await Album.objects.create(name="Limitless")
|
||||
await Track.objects.create(
|
||||
id=None, album=fantasies, title="Sample", position=1
|
||||
)
|
||||
await Track.objects.create(album=fantasies, title="Sample2", position=2)
|
||||
await Track.objects.create(album=fantasies, title="Sample3", position=3)
|
||||
|
||||
tracks = await Track.objects.limit(1).all()
|
||||
assert len(tracks) == 1
|
||||
assert tracks[0].title == "Sample"
|
||||
|
||||
tracks = await Track.objects.limit(1).offset(1).all()
|
||||
assert len(tracks) == 1
|
||||
assert tracks[0].title == "Sample2"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_exceptions():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
fantasies = await Album.objects.create(name="Test")
|
||||
|
||||
with pytest.raises(NoMatch):
|
||||
await Album.objects.get(name="Test2")
|
||||
|
||||
await Track.objects.create(album=fantasies, title="Test1", position=1)
|
||||
await Track.objects.create(album=fantasies, title="Test2", position=2)
|
||||
await Track.objects.create(album=fantasies, title="Test3", position=3)
|
||||
with pytest.raises(MultipleMatches):
|
||||
await Track.objects.select_related("album").get(album=fantasies)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_wrong_model_passed_as_fk():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
with pytest.raises(RelationshipInstanceError):
|
||||
org = await Organisation.objects.create(ident="ACME Ltd")
|
||||
await Track.objects.create(album=org, title="Test1", position=1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_model_with_no_children():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
album = await Album.objects.create(name="Test")
|
||||
album.name = "Test2"
|
||||
await Album.objects.bulk_update([album], columns=["name"])
|
||||
|
||||
updated_album = await Album.objects.get(id=album.id)
|
||||
assert updated_album.name == "Test2"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_model_with_children():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
best_seller = await Album.objects.create(name="to_be_best_seller")
|
||||
best_seller2 = await Album.objects.create(name="to_be_best_seller2")
|
||||
not_best_seller = await Album.objects.create(name="unpopular")
|
||||
await Track.objects.create(
|
||||
album=best_seller, title="t1", position=1, play_count=100
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=best_seller2, title="t2", position=1, play_count=100
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=not_best_seller, title="t3", position=1, play_count=3
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=best_seller, title="t4", position=1, play_count=500
|
||||
)
|
||||
|
||||
tracks = (
|
||||
await Track.objects.select_related("album")
|
||||
.filter(play_count__gt=10)
|
||||
.all()
|
||||
)
|
||||
best_seller_albums = {}
|
||||
for track in tracks:
|
||||
album = track.album
|
||||
if album.id in best_seller_albums:
|
||||
continue
|
||||
album.is_best_seller = True
|
||||
best_seller_albums[album.id] = album
|
||||
await Album.objects.bulk_update(
|
||||
best_seller_albums.values(), columns=["is_best_seller"]
|
||||
)
|
||||
best_seller_albums_db = await Album.objects.filter(
|
||||
is_best_seller=True
|
||||
).all()
|
||||
assert len(best_seller_albums_db) == 2
|
||||
377
tests/test_relations/test_m2m_through_fields.py
Normal file
377
tests/test_relations/test_m2m_through_fields.py
Normal file
@ -0,0 +1,377 @@
|
||||
from typing import Any, Sequence, cast
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
from pydantic.typing import ForwardRef
|
||||
|
||||
import ormar
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL, force_rollback=True)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class BaseMeta(ormar.ModelMeta):
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
|
||||
class Category(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
tablename = "categories"
|
||||
|
||||
id = ormar.Integer(primary_key=True)
|
||||
name = ormar.String(max_length=40)
|
||||
|
||||
|
||||
class PostCategory(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
tablename = "posts_x_categories"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
sort_order: int = ormar.Integer(nullable=True)
|
||||
param_name: str = ormar.String(default="Name", max_length=200)
|
||||
|
||||
|
||||
class Blog(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
pass
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
|
||||
|
||||
class Post(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
pass
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
categories = ormar.ManyToMany(Category, through=PostCategory)
|
||||
blog = ormar.ForeignKey(Blog)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = sqlalchemy.create_engine(DATABASE_URL)
|
||||
metadata.drop_all(engine)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
class PostCategory2(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
tablename = "posts_x_categories2"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
sort_order: int = ormar.Integer(nullable=True)
|
||||
|
||||
|
||||
class Post2(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
pass
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
categories = ormar.ManyToMany(Category, through=ForwardRef("PostCategory2"))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forward_ref_is_updated():
|
||||
async with database:
|
||||
assert Post2.Meta.requires_ref_update
|
||||
Post2.update_forward_refs()
|
||||
|
||||
assert Post2.Meta.model_fields["postcategory2"].to == PostCategory2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setting_fields_on_through_model():
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
category = await Category(name="Test category").save()
|
||||
await post.categories.add(category)
|
||||
|
||||
assert hasattr(post.categories[0], "postcategory")
|
||||
assert post.categories[0].postcategory is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setting_additional_fields_on_through_model_in_add():
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
category = await Category(name="Test category").save()
|
||||
await post.categories.add(category, sort_order=1)
|
||||
postcat = await PostCategory.objects.get()
|
||||
assert postcat.sort_order == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setting_additional_fields_on_through_model_in_create():
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 2}
|
||||
)
|
||||
postcat = await PostCategory.objects.get()
|
||||
assert postcat.sort_order == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_getting_additional_fields_from_queryset() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1", postcategory={"sort_order": 1}
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 2}
|
||||
)
|
||||
|
||||
await post.categories.all()
|
||||
assert post.postcategory is None
|
||||
assert post.categories[0].postcategory.sort_order == 1
|
||||
assert post.categories[1].postcategory.sort_order == 2
|
||||
|
||||
post2 = await Post.objects.select_related("categories").get(
|
||||
categories__name="Test category2"
|
||||
)
|
||||
assert post2.categories[0].postcategory.sort_order == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_only_one_side_has_through() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1", postcategory={"sort_order": 1}
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 2}
|
||||
)
|
||||
|
||||
post2 = await Post.objects.select_related("categories").get()
|
||||
assert post2.postcategory is None
|
||||
assert post2.categories[0].postcategory is not None
|
||||
|
||||
await post2.categories.all()
|
||||
assert post2.postcategory is None
|
||||
assert post2.categories[0].postcategory is not None
|
||||
|
||||
categories = await Category.objects.select_related("posts").all()
|
||||
assert isinstance(categories[0], Category)
|
||||
assert categories[0].postcategory is None
|
||||
assert categories[0].posts[0].postcategory is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_filtering_by_through_model() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1",
|
||||
postcategory={"sort_order": 1, "param_name": "volume"},
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 2, "param_name": "area"}
|
||||
)
|
||||
|
||||
post2 = (
|
||||
await Post.objects.select_related("categories")
|
||||
.filter(postcategory__sort_order__gt=1)
|
||||
.get()
|
||||
)
|
||||
assert len(post2.categories) == 1
|
||||
assert post2.categories[0].postcategory.sort_order == 2
|
||||
|
||||
post3 = await Post.objects.filter(
|
||||
categories__postcategory__param_name="volume"
|
||||
).get()
|
||||
assert len(post3.categories) == 1
|
||||
assert post3.categories[0].postcategory.param_name == "volume"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_deep_filtering_by_through_model() -> Any:
|
||||
async with database:
|
||||
blog = await Blog(title="My Blog").save()
|
||||
post = await Post(title="Test post", blog=blog).save()
|
||||
|
||||
await post.categories.create(
|
||||
name="Test category1",
|
||||
postcategory={"sort_order": 1, "param_name": "volume"},
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 2, "param_name": "area"}
|
||||
)
|
||||
|
||||
blog2 = (
|
||||
await Blog.objects.select_related("posts__categories")
|
||||
.filter(posts__postcategory__sort_order__gt=1)
|
||||
.get()
|
||||
)
|
||||
assert len(blog2.posts) == 1
|
||||
assert len(blog2.posts[0].categories) == 1
|
||||
assert blog2.posts[0].categories[0].postcategory.sort_order == 2
|
||||
|
||||
blog3 = await Blog.objects.filter(
|
||||
posts__categories__postcategory__param_name="volume"
|
||||
).get()
|
||||
assert len(blog3.posts) == 1
|
||||
assert len(blog3.posts[0].categories) == 1
|
||||
assert blog3.posts[0].categories[0].postcategory.param_name == "volume"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ordering_by_through_model() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1",
|
||||
postcategory={"sort_order": 2, "param_name": "volume"},
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 1, "param_name": "area"}
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category3",
|
||||
postcategory={"sort_order": 3, "param_name": "velocity"},
|
||||
)
|
||||
|
||||
post2 = (
|
||||
await Post.objects.select_related("categories")
|
||||
.order_by("-postcategory__sort_order")
|
||||
.get()
|
||||
)
|
||||
assert len(post2.categories) == 3
|
||||
assert post2.categories[0].name == "Test category3"
|
||||
assert post2.categories[2].name == "Test category2"
|
||||
|
||||
post3 = (
|
||||
await Post.objects.select_related("categories")
|
||||
.order_by("categories__postcategory__param_name")
|
||||
.get()
|
||||
)
|
||||
assert len(post3.categories) == 3
|
||||
assert post3.categories[0].postcategory.param_name == "area"
|
||||
assert post3.categories[2].postcategory.param_name == "volume"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_through_models_from_queryset_on_through() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1",
|
||||
postcategory={"sort_order": 2, "param_name": "volume"},
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 1, "param_name": "area"}
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category3",
|
||||
postcategory={"sort_order": 3, "param_name": "velocity"},
|
||||
)
|
||||
|
||||
await PostCategory.objects.filter(param_name="volume", post=post.id).update(
|
||||
sort_order=4
|
||||
)
|
||||
post2 = (
|
||||
await Post.objects.select_related("categories")
|
||||
.order_by("-postcategory__sort_order")
|
||||
.get()
|
||||
)
|
||||
assert len(post2.categories) == 3
|
||||
assert post2.categories[0].postcategory.param_name == "volume"
|
||||
assert post2.categories[2].postcategory.param_name == "area"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_through_model_after_load() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1",
|
||||
postcategory={"sort_order": 2, "param_name": "volume"},
|
||||
)
|
||||
post2 = await Post.objects.select_related("categories").get()
|
||||
assert len(post2.categories) == 1
|
||||
|
||||
await post2.categories[0].postcategory.load()
|
||||
await post2.categories[0].postcategory.update(sort_order=3)
|
||||
|
||||
post3 = await Post.objects.select_related("categories").get()
|
||||
assert len(post3.categories) == 1
|
||||
assert post3.categories[0].postcategory.sort_order == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_through_from_related() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1",
|
||||
postcategory={"sort_order": 2, "param_name": "volume"},
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 1, "param_name": "area"}
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category3",
|
||||
postcategory={"sort_order": 3, "param_name": "velocity"},
|
||||
)
|
||||
|
||||
await post.categories.filter(name="Test category3").update(
|
||||
postcategory={"sort_order": 4}
|
||||
)
|
||||
|
||||
post2 = (
|
||||
await Post.objects.select_related("categories")
|
||||
.order_by("postcategory__sort_order")
|
||||
.get()
|
||||
)
|
||||
assert len(post2.categories) == 3
|
||||
assert post2.categories[2].postcategory.sort_order == 4
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_excluding_fields_on_through_model() -> Any:
|
||||
async with database:
|
||||
post = await Post(title="Test post").save()
|
||||
await post.categories.create(
|
||||
name="Test category1",
|
||||
postcategory={"sort_order": 2, "param_name": "volume"},
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category2", postcategory={"sort_order": 1, "param_name": "area"}
|
||||
)
|
||||
await post.categories.create(
|
||||
name="Test category3",
|
||||
postcategory={"sort_order": 3, "param_name": "velocity"},
|
||||
)
|
||||
|
||||
post2 = (
|
||||
await Post.objects.select_related("categories")
|
||||
.exclude_fields("postcategory__param_name")
|
||||
.order_by("postcategory__sort_order")
|
||||
.get()
|
||||
)
|
||||
assert len(post2.categories) == 3
|
||||
assert post2.categories[0].postcategory.param_name is None
|
||||
assert post2.categories[0].postcategory.sort_order == 1
|
||||
|
||||
assert post2.categories[2].postcategory.param_name is None
|
||||
assert post2.categories[2].postcategory.sort_order == 3
|
||||
|
||||
post3 = (
|
||||
await Post.objects.select_related("categories")
|
||||
.fields({"postcategory": ..., "title": ...})
|
||||
.exclude_fields({"postcategory": {"param_name", "sort_order"}})
|
||||
.get()
|
||||
)
|
||||
assert len(post3.categories) == 3
|
||||
for category in post3.categories:
|
||||
assert category.postcategory.param_name is None
|
||||
assert category.postcategory.sort_order is None
|
||||
240
tests/test_relations/test_many_to_many.py
Normal file
240
tests/test_relations/test_many_to_many.py
Normal file
@ -0,0 +1,240 @@
|
||||
import asyncio
|
||||
from typing import List, Optional
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
|
||||
import ormar
|
||||
from ormar.exceptions import ModelPersistenceError, NoMatch, RelationshipInstanceError
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL, force_rollback=True)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class Author(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "authors"
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
first_name: str = ormar.String(max_length=80)
|
||||
last_name: str = ormar.String(max_length=80)
|
||||
|
||||
|
||||
class Category(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "categories"
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=40)
|
||||
|
||||
|
||||
class Post(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "posts"
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
categories: Optional[List[Category]] = ormar.ManyToMany(Category)
|
||||
author: Optional[Author] = ormar.ForeignKey(Author)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def event_loop():
|
||||
loop = asyncio.get_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
async def create_test_database():
|
||||
engine = sqlalchemy.create_engine(DATABASE_URL)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
async def cleanup():
|
||||
yield
|
||||
async with database:
|
||||
PostCategory = Post.Meta.model_fields["categories"].through
|
||||
await PostCategory.objects.delete(each=True)
|
||||
await Post.objects.delete(each=True)
|
||||
await Category.objects.delete(each=True)
|
||||
await Author.objects.delete(each=True)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_not_saved_raises_error(cleanup):
|
||||
async with database:
|
||||
guido = await Author(first_name="Guido", last_name="Van Rossum").save()
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
news = Category(name="News")
|
||||
|
||||
with pytest.raises(ModelPersistenceError):
|
||||
await post.categories.add(news)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_not_existing_raises_error(cleanup):
|
||||
async with database:
|
||||
guido = await Author(first_name="Guido", last_name="Van Rossum").save()
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
|
||||
with pytest.raises(NoMatch):
|
||||
await post.categories.get()
|
||||
|
||||
assert await post.categories.get_or_none() is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_assigning_related_objects(cleanup):
|
||||
async with database:
|
||||
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
news = await Category.objects.create(name="News")
|
||||
|
||||
# Add a category to a post.
|
||||
await post.categories.add(news)
|
||||
# or from the other end:
|
||||
await news.posts.add(post)
|
||||
|
||||
assert await post.categories.get_or_none(name="no exist") is None
|
||||
assert await post.categories.get_or_none(name="News") == news
|
||||
|
||||
# Creating columns object from instance:
|
||||
await post.categories.create(name="Tips")
|
||||
assert len(post.categories) == 2
|
||||
|
||||
post_categories = await post.categories.all()
|
||||
assert len(post_categories) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_quering_of_the_m2m_models(cleanup):
|
||||
async with database:
|
||||
# orm can do this already.
|
||||
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
news = await Category.objects.create(name="News")
|
||||
# tl;dr: `post.categories` exposes the QuerySet API.
|
||||
|
||||
await post.categories.add(news)
|
||||
|
||||
post_categories = await post.categories.all()
|
||||
assert len(post_categories) == 1
|
||||
|
||||
assert news == await post.categories.get(name="News")
|
||||
|
||||
num_posts = await news.posts.count()
|
||||
assert num_posts == 1
|
||||
|
||||
posts_about_m2m = await news.posts.filter(title__contains="M2M").all()
|
||||
assert len(posts_about_m2m) == 1
|
||||
assert posts_about_m2m[0] == post
|
||||
posts_about_python = await Post.objects.filter(categories__name="python").all()
|
||||
assert len(posts_about_python) == 0
|
||||
|
||||
# Traversal of relationships: which categories has Guido contributed to?
|
||||
category = await Category.objects.filter(posts__author=guido).get()
|
||||
assert category == news
|
||||
# or:
|
||||
category2 = await Category.objects.filter(
|
||||
posts__author__first_name="Guido"
|
||||
).get()
|
||||
assert category2 == news
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_removal_of_the_relations(cleanup):
|
||||
async with database:
|
||||
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
news = await Category.objects.create(name="News")
|
||||
await post.categories.add(news)
|
||||
assert len(await post.categories.all()) == 1
|
||||
await post.categories.remove(news)
|
||||
assert len(await post.categories.all()) == 0
|
||||
# or:
|
||||
await news.posts.add(post)
|
||||
assert len(await news.posts.all()) == 1
|
||||
await news.posts.remove(post)
|
||||
assert len(await news.posts.all()) == 0
|
||||
|
||||
# Remove all columns objects:
|
||||
await post.categories.add(news)
|
||||
await post.categories.clear()
|
||||
assert len(await post.categories.all()) == 0
|
||||
|
||||
# post would also lose 'news' category when running:
|
||||
await post.categories.add(news)
|
||||
await news.delete()
|
||||
assert len(await post.categories.all()) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_selecting_related(cleanup):
|
||||
async with database:
|
||||
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
news = await Category.objects.create(name="News")
|
||||
recent = await Category.objects.create(name="Recent")
|
||||
await post.categories.add(news)
|
||||
await post.categories.add(recent)
|
||||
assert len(await post.categories.all()) == 2
|
||||
# Loads categories and posts (2 queries) and perform the join in Python.
|
||||
categories = await Category.objects.select_related("posts").all()
|
||||
# No extra queries needed => no more `await`s required.
|
||||
for category in categories:
|
||||
assert category.posts[0] == post
|
||||
|
||||
news_posts = await news.posts.select_related("author").all()
|
||||
assert news_posts[0].author == guido
|
||||
|
||||
assert (await post.categories.limit(1).all())[0] == news
|
||||
assert (await post.categories.offset(1).limit(1).all())[0] == recent
|
||||
|
||||
assert await post.categories.first() == news
|
||||
|
||||
assert await post.categories.exists()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_selecting_related_fail_without_saving(cleanup):
|
||||
async with database:
|
||||
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
|
||||
post = Post(title="Hello, M2M", author=guido)
|
||||
with pytest.raises(RelationshipInstanceError):
|
||||
await post.categories.all()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_adding_unsaved_related(cleanup):
|
||||
async with database:
|
||||
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
news = Category(name="News")
|
||||
with pytest.raises(ModelPersistenceError):
|
||||
await post.categories.add(news)
|
||||
|
||||
await news.save()
|
||||
await post.categories.add(news)
|
||||
assert len(await post.categories.all()) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_removing_unsaved_related(cleanup):
|
||||
async with database:
|
||||
guido = await Author.objects.create(first_name="Guido", last_name="Van Rossum")
|
||||
post = await Post.objects.create(title="Hello, M2M", author=guido)
|
||||
news = Category(name="News")
|
||||
with pytest.raises(NoMatch):
|
||||
await post.categories.remove(news)
|
||||
389
tests/test_relations/test_prefetch_related.py
Normal file
389
tests/test_relations/test_prefetch_related.py
Normal file
@ -0,0 +1,389 @@
|
||||
from typing import List, Optional
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
|
||||
import ormar
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL, force_rollback=True)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class RandomSet(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "randoms"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(name="random_id", primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
|
||||
|
||||
class Tonation(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "tonations"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(name="tonation_name", max_length=100)
|
||||
rand_set: Optional[RandomSet] = ormar.ForeignKey(RandomSet)
|
||||
|
||||
|
||||
class Division(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "divisions"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(name="division_id", primary_key=True)
|
||||
name: str = ormar.String(max_length=100, nullable=True)
|
||||
|
||||
|
||||
class Shop(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "shops"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100, nullable=True)
|
||||
division: Optional[Division] = ormar.ForeignKey(Division)
|
||||
|
||||
|
||||
class AlbumShops(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "albums_x_shops"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class Album(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "albums"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100, nullable=True)
|
||||
shops: List[Shop] = ormar.ManyToMany(to=Shop, through=AlbumShops)
|
||||
|
||||
|
||||
class Track(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "tracks"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(name="track_id", primary_key=True)
|
||||
album: Optional[Album] = ormar.ForeignKey(Album)
|
||||
title: str = ormar.String(max_length=100)
|
||||
position: int = ormar.Integer()
|
||||
tonation: Optional[Tonation] = ormar.ForeignKey(Tonation, name="tonation_id")
|
||||
|
||||
|
||||
class Cover(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "covers"
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
album: Optional[Album] = ormar.ForeignKey(
|
||||
Album, related_name="cover_pictures", name="album_id"
|
||||
)
|
||||
title: str = ormar.String(max_length=100)
|
||||
artist: str = ormar.String(max_length=200, nullable=True)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = sqlalchemy.create_engine(DATABASE_URL)
|
||||
metadata.drop_all(engine)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prefetch_related():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
album = Album(name="Malibu")
|
||||
await album.save()
|
||||
ton1 = await Tonation.objects.create(name="B-mol")
|
||||
await Track.objects.create(
|
||||
album=album, title="The Bird", position=1, tonation=ton1
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=album,
|
||||
title="Heart don't stand a chance",
|
||||
position=2,
|
||||
tonation=ton1,
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=album, title="The Waters", position=3, tonation=ton1
|
||||
)
|
||||
await Cover.objects.create(title="Cover1", album=album, artist="Artist 1")
|
||||
await Cover.objects.create(title="Cover2", album=album, artist="Artist 2")
|
||||
|
||||
fantasies = Album(name="Fantasies")
|
||||
await fantasies.save()
|
||||
await Track.objects.create(
|
||||
album=fantasies, title="Help I'm Alive", position=1
|
||||
)
|
||||
await Track.objects.create(album=fantasies, title="Sick Muse", position=2)
|
||||
await Track.objects.create(
|
||||
album=fantasies, title="Satellite Mind", position=3
|
||||
)
|
||||
await Cover.objects.create(
|
||||
title="Cover3", album=fantasies, artist="Artist 3"
|
||||
)
|
||||
await Cover.objects.create(
|
||||
title="Cover4", album=fantasies, artist="Artist 4"
|
||||
)
|
||||
|
||||
album = (
|
||||
await Album.objects.filter(name="Malibu")
|
||||
.prefetch_related(["tracks__tonation", "cover_pictures"])
|
||||
.get()
|
||||
)
|
||||
assert len(album.tracks) == 3
|
||||
assert album.tracks[0].title == "The Bird"
|
||||
assert len(album.cover_pictures) == 2
|
||||
assert album.cover_pictures[0].title == "Cover1"
|
||||
assert (
|
||||
album.tracks[0].tonation.name
|
||||
== album.tracks[2].tonation.name
|
||||
== "B-mol"
|
||||
)
|
||||
|
||||
albums = await Album.objects.prefetch_related("tracks").all()
|
||||
assert len(albums[0].tracks) == 3
|
||||
assert len(albums[1].tracks) == 3
|
||||
assert albums[0].tracks[0].title == "The Bird"
|
||||
assert albums[1].tracks[0].title == "Help I'm Alive"
|
||||
|
||||
track = await Track.objects.prefetch_related(["album__cover_pictures"]).get(
|
||||
title="The Bird"
|
||||
)
|
||||
assert track.album.name == "Malibu"
|
||||
assert len(track.album.cover_pictures) == 2
|
||||
assert track.album.cover_pictures[0].artist == "Artist 1"
|
||||
|
||||
track = (
|
||||
await Track.objects.prefetch_related(["album__cover_pictures"])
|
||||
.exclude_fields("album__cover_pictures__artist")
|
||||
.get(title="The Bird")
|
||||
)
|
||||
assert track.album.name == "Malibu"
|
||||
assert len(track.album.cover_pictures) == 2
|
||||
assert track.album.cover_pictures[0].artist is None
|
||||
|
||||
tracks = await Track.objects.prefetch_related("album").all()
|
||||
assert len(tracks) == 6
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prefetch_related_with_many_to_many():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
div = await Division.objects.create(name="Div 1")
|
||||
shop1 = await Shop.objects.create(name="Shop 1", division=div)
|
||||
shop2 = await Shop.objects.create(name="Shop 2", division=div)
|
||||
album = Album(name="Malibu")
|
||||
await album.save()
|
||||
await album.shops.add(shop1)
|
||||
await album.shops.add(shop2)
|
||||
|
||||
await Track.objects.create(album=album, title="The Bird", position=1)
|
||||
await Track.objects.create(
|
||||
album=album, title="Heart don't stand a chance", position=2
|
||||
)
|
||||
await Track.objects.create(album=album, title="The Waters", position=3)
|
||||
await Cover.objects.create(title="Cover1", album=album, artist="Artist 1")
|
||||
await Cover.objects.create(title="Cover2", album=album, artist="Artist 2")
|
||||
|
||||
track = await Track.objects.prefetch_related(
|
||||
["album__cover_pictures", "album__shops__division"]
|
||||
).get(title="The Bird")
|
||||
assert track.album.name == "Malibu"
|
||||
assert len(track.album.cover_pictures) == 2
|
||||
assert track.album.cover_pictures[0].artist == "Artist 1"
|
||||
|
||||
assert len(track.album.shops) == 2
|
||||
assert track.album.shops[0].name == "Shop 1"
|
||||
assert track.album.shops[0].division.name == "Div 1"
|
||||
|
||||
album2 = Album(name="Malibu 2")
|
||||
await album2.save()
|
||||
await album2.shops.add(shop1)
|
||||
await album2.shops.add(shop2)
|
||||
await Track.objects.create(album=album2, title="The Bird 2", position=1)
|
||||
|
||||
tracks = await Track.objects.prefetch_related(["album__shops"]).all()
|
||||
assert tracks[0].album.name == "Malibu"
|
||||
assert tracks[0].album.shops[0].name == "Shop 1"
|
||||
assert tracks[3].album.name == "Malibu 2"
|
||||
assert tracks[3].album.shops[0].name == "Shop 1"
|
||||
|
||||
assert tracks[0].album.shops[0] == tracks[3].album.shops[0]
|
||||
assert id(tracks[0].album.shops[0]) == id(tracks[3].album.shops[0])
|
||||
tracks[0].album.shops[0].name = "Dummy"
|
||||
assert tracks[0].album.shops[0].name == tracks[3].album.shops[0].name
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prefetch_related_empty():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
await Track.objects.create(title="The Bird", position=1)
|
||||
track = await Track.objects.prefetch_related(["album__cover_pictures"]).get(
|
||||
title="The Bird"
|
||||
)
|
||||
assert track.title == "The Bird"
|
||||
assert track.album is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prefetch_related_with_select_related():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
div = await Division.objects.create(name="Div 1")
|
||||
shop1 = await Shop.objects.create(name="Shop 1", division=div)
|
||||
shop2 = await Shop.objects.create(name="Shop 2", division=div)
|
||||
album = Album(name="Malibu")
|
||||
await album.save()
|
||||
await album.shops.add(shop1)
|
||||
await album.shops.add(shop2)
|
||||
|
||||
await Cover.objects.create(title="Cover1", album=album, artist="Artist 1")
|
||||
await Cover.objects.create(title="Cover2", album=album, artist="Artist 2")
|
||||
|
||||
album = (
|
||||
await Album.objects.select_related(["tracks", "shops"])
|
||||
.filter(name="Malibu")
|
||||
.prefetch_related(["cover_pictures", "shops__division"])
|
||||
.first()
|
||||
)
|
||||
|
||||
assert len(album.tracks) == 0
|
||||
assert len(album.cover_pictures) == 2
|
||||
assert album.shops[0].division.name == "Div 1"
|
||||
|
||||
rand_set = await RandomSet.objects.create(name="Rand 1")
|
||||
ton1 = await Tonation.objects.create(name="B-mol", rand_set=rand_set)
|
||||
await Track.objects.create(
|
||||
album=album, title="The Bird", position=1, tonation=ton1
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=album,
|
||||
title="Heart don't stand a chance",
|
||||
position=2,
|
||||
tonation=ton1,
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=album, title="The Waters", position=3, tonation=ton1
|
||||
)
|
||||
|
||||
album = (
|
||||
await Album.objects.select_related("tracks__tonation__rand_set")
|
||||
.filter(name="Malibu")
|
||||
.prefetch_related(["cover_pictures", "shops__division"])
|
||||
.order_by(
|
||||
["-shops__name", "-cover_pictures__artist", "shops__division__name"]
|
||||
)
|
||||
.get()
|
||||
)
|
||||
assert len(album.tracks) == 3
|
||||
assert album.tracks[0].tonation == album.tracks[2].tonation == ton1
|
||||
assert len(album.cover_pictures) == 2
|
||||
assert album.cover_pictures[0].artist == "Artist 2"
|
||||
|
||||
assert len(album.shops) == 2
|
||||
assert album.shops[0].name == "Shop 2"
|
||||
assert album.shops[0].division.name == "Div 1"
|
||||
|
||||
track = (
|
||||
await Track.objects.select_related("album")
|
||||
.prefetch_related(["album__cover_pictures", "album__shops__division"])
|
||||
.get(title="The Bird")
|
||||
)
|
||||
assert track.album.name == "Malibu"
|
||||
assert len(track.album.cover_pictures) == 2
|
||||
assert track.album.cover_pictures[0].artist == "Artist 1"
|
||||
|
||||
assert len(track.album.shops) == 2
|
||||
assert track.album.shops[0].name == "Shop 1"
|
||||
assert track.album.shops[0].division.name == "Div 1"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prefetch_related_with_select_related_and_fields():
|
||||
async with database:
|
||||
async with database.transaction(force_rollback=True):
|
||||
div = await Division.objects.create(name="Div 1")
|
||||
shop1 = await Shop.objects.create(name="Shop 1", division=div)
|
||||
shop2 = await Shop.objects.create(name="Shop 2", division=div)
|
||||
album = Album(name="Malibu")
|
||||
await album.save()
|
||||
await album.shops.add(shop1)
|
||||
await album.shops.add(shop2)
|
||||
await Cover.objects.create(title="Cover1", album=album, artist="Artist 1")
|
||||
await Cover.objects.create(title="Cover2", album=album, artist="Artist 2")
|
||||
rand_set = await RandomSet.objects.create(name="Rand 1")
|
||||
ton1 = await Tonation.objects.create(name="B-mol", rand_set=rand_set)
|
||||
await Track.objects.create(
|
||||
album=album, title="The Bird", position=1, tonation=ton1
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=album,
|
||||
title="Heart don't stand a chance",
|
||||
position=2,
|
||||
tonation=ton1,
|
||||
)
|
||||
await Track.objects.create(
|
||||
album=album, title="The Waters", position=3, tonation=ton1
|
||||
)
|
||||
|
||||
album = (
|
||||
await Album.objects.select_related("tracks__tonation__rand_set")
|
||||
.filter(name="Malibu")
|
||||
.prefetch_related(["cover_pictures", "shops__division"])
|
||||
.exclude_fields({"shops": {"division": {"name"}}})
|
||||
.get()
|
||||
)
|
||||
assert len(album.tracks) == 3
|
||||
assert album.tracks[0].tonation == album.tracks[2].tonation == ton1
|
||||
assert len(album.cover_pictures) == 2
|
||||
assert album.cover_pictures[0].artist == "Artist 1"
|
||||
|
||||
assert len(album.shops) == 2
|
||||
assert album.shops[0].name == "Shop 1"
|
||||
assert album.shops[0].division.name is None
|
||||
|
||||
album = (
|
||||
await Album.objects.select_related("tracks")
|
||||
.filter(name="Malibu")
|
||||
.prefetch_related(["cover_pictures", "shops__division"])
|
||||
.fields(
|
||||
{
|
||||
"name": ...,
|
||||
"shops": {"division"},
|
||||
"cover_pictures": {"id": ..., "title": ...},
|
||||
}
|
||||
)
|
||||
.exclude_fields({"shops": {"division": {"name"}}})
|
||||
.get()
|
||||
)
|
||||
assert len(album.tracks) == 3
|
||||
assert len(album.cover_pictures) == 2
|
||||
assert album.cover_pictures[0].artist is None
|
||||
assert album.cover_pictures[0].title is not None
|
||||
|
||||
assert len(album.shops) == 2
|
||||
assert album.shops[0].name is None
|
||||
assert album.shops[0].division is not None
|
||||
assert album.shops[0].division.name is None
|
||||
@ -0,0 +1,100 @@
|
||||
from typing import List, Optional
|
||||
|
||||
import databases
|
||||
import sqlalchemy
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
import ormar
|
||||
import pytest
|
||||
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
db = databases.Database(DATABASE_URL)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class User(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = db
|
||||
tablename = "test_users"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=50)
|
||||
|
||||
|
||||
class Signup(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = db
|
||||
tablename = "test_signup"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
|
||||
|
||||
class Session(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = db
|
||||
tablename = "test_sessions"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=255, index=True)
|
||||
some_text: str = ormar.Text()
|
||||
some_other_text: Optional[str] = ormar.Text(nullable=True)
|
||||
teacher: Optional[User] = ormar.ForeignKey(
|
||||
User, nullable=True, related_name="teaching"
|
||||
)
|
||||
students: Optional[List[User]] = ormar.ManyToMany(
|
||||
User, through=Signup, related_name="attending"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = create_engine(DATABASE_URL)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_students():
|
||||
async with db:
|
||||
for user_id in [1, 2, 3, 4, 5]:
|
||||
await User.objects.create(name=f"User {user_id}")
|
||||
|
||||
for name, some_text, some_other_text in [
|
||||
("Session 1", "Some text 1", "Some other text 1"),
|
||||
("Session 2", "Some text 2", "Some other text 2"),
|
||||
("Session 3", "Some text 3", "Some other text 3"),
|
||||
("Session 4", "Some text 4", "Some other text 4"),
|
||||
("Session 5", "Some text 5", "Some other text 5"),
|
||||
]:
|
||||
await Session(
|
||||
name=name, some_text=some_text, some_other_text=some_other_text
|
||||
).save()
|
||||
|
||||
s1 = await Session.objects.get(pk=1)
|
||||
s2 = await Session.objects.get(pk=2)
|
||||
|
||||
users = {}
|
||||
for i in range(1, 6):
|
||||
user = await User.objects.get(pk=i)
|
||||
users[f"user_{i}"] = user
|
||||
if i % 2 == 0:
|
||||
await s1.students.add(user)
|
||||
else:
|
||||
await s2.students.add(user)
|
||||
|
||||
assert len(s1.students) > 0
|
||||
assert len(s2.students) > 0
|
||||
|
||||
user = await User.objects.select_related("attending").get(pk=1)
|
||||
|
||||
assert user.attending is not None
|
||||
assert len(user.attending) > 0
|
||||
|
||||
query = Session.objects.prefetch_related(["students", "teacher",])
|
||||
sessions = await query.all()
|
||||
assert len(sessions) == 5
|
||||
65
tests/test_relations/test_relations_default_exception.py
Normal file
65
tests/test_relations/test_relations_default_exception.py
Normal file
@ -0,0 +1,65 @@
|
||||
# type: ignore
|
||||
from typing import List, Optional
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
|
||||
import ormar
|
||||
from ormar.exceptions import ModelDefinitionError
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL, force_rollback=True)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class Author(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "authors"
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
first_name: str = ormar.String(max_length=80)
|
||||
last_name: str = ormar.String(max_length=80)
|
||||
|
||||
|
||||
class Category(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "categories"
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=40)
|
||||
|
||||
|
||||
def test_fk_error():
|
||||
with pytest.raises(ModelDefinitionError):
|
||||
|
||||
class Post(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "posts"
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
categories: Optional[List[Category]] = ormar.ManyToMany(Category)
|
||||
author: Optional[Author] = ormar.ForeignKey(Author, default="aa")
|
||||
|
||||
|
||||
def test_m2m_error():
|
||||
with pytest.raises(ModelDefinitionError):
|
||||
|
||||
class Post(ormar.Model):
|
||||
class Meta:
|
||||
tablename = "posts"
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
categories: Optional[List[Category]] = ormar.ManyToMany(
|
||||
Category, default="aa"
|
||||
)
|
||||
158
tests/test_relations/test_select_related_with_limit.py
Normal file
158
tests/test_relations/test_select_related_with_limit.py
Normal file
@ -0,0 +1,158 @@
|
||||
from typing import List, Optional
|
||||
|
||||
import databases
|
||||
import sqlalchemy
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
import ormar
|
||||
import pytest
|
||||
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
db = databases.Database(DATABASE_URL, force_rollback=True)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class Keyword(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = db
|
||||
tablename = "keywords"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=50)
|
||||
|
||||
|
||||
class KeywordPrimaryModel(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = db
|
||||
tablename = "primary_models_keywords"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
|
||||
|
||||
class PrimaryModel(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = db
|
||||
tablename = "primary_models"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=255, index=True)
|
||||
some_text: str = ormar.Text()
|
||||
some_other_text: Optional[str] = ormar.Text(nullable=True)
|
||||
keywords: Optional[List[Keyword]] = ormar.ManyToMany(
|
||||
Keyword, through=KeywordPrimaryModel
|
||||
)
|
||||
|
||||
|
||||
class SecondaryModel(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = db
|
||||
tablename = "secondary_models"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=100)
|
||||
primary_model: PrimaryModel = ormar.ForeignKey(
|
||||
PrimaryModel, related_name="secondary_models",
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_primary_models():
|
||||
async with db:
|
||||
for name, some_text, some_other_text in [
|
||||
("Primary 1", "Some text 1", "Some other text 1"),
|
||||
("Primary 2", "Some text 2", "Some other text 2"),
|
||||
("Primary 3", "Some text 3", "Some other text 3"),
|
||||
("Primary 4", "Some text 4", "Some other text 4"),
|
||||
("Primary 5", "Some text 5", "Some other text 5"),
|
||||
("Primary 6", "Some text 6", "Some other text 6"),
|
||||
("Primary 7", "Some text 7", "Some other text 7"),
|
||||
("Primary 8", "Some text 8", "Some other text 8"),
|
||||
("Primary 9", "Some text 9", "Some other text 9"),
|
||||
("Primary 10", "Some text 10", "Some other text 10"),
|
||||
]:
|
||||
await PrimaryModel(
|
||||
name=name, some_text=some_text, some_other_text=some_other_text
|
||||
).save()
|
||||
|
||||
for tag_id in [1, 2, 3, 4, 5]:
|
||||
await Keyword.objects.create(name=f"Tag {tag_id}")
|
||||
|
||||
p1 = await PrimaryModel.objects.get(pk=1)
|
||||
p2 = await PrimaryModel.objects.get(pk=2)
|
||||
for i in range(1, 6):
|
||||
keyword = await Keyword.objects.get(pk=i)
|
||||
if i % 2 == 0:
|
||||
await p1.keywords.add(keyword)
|
||||
else:
|
||||
await p2.keywords.add(keyword)
|
||||
models = await PrimaryModel.objects.select_related("keywords").limit(5).all()
|
||||
|
||||
assert len(models) == 5
|
||||
assert len(models[0].keywords) == 2
|
||||
assert len(models[1].keywords) == 3
|
||||
assert len(models[2].keywords) == 0
|
||||
|
||||
models2 = (
|
||||
await PrimaryModel.objects.select_related("keywords")
|
||||
.limit(5)
|
||||
.offset(3)
|
||||
.all()
|
||||
)
|
||||
assert len(models2) == 5
|
||||
assert [x.name for x in models2] != [x.name for x in models]
|
||||
assert [x.name for x in models2] == [
|
||||
"Primary 4",
|
||||
"Primary 5",
|
||||
"Primary 6",
|
||||
"Primary 7",
|
||||
"Primary 8",
|
||||
]
|
||||
|
||||
models3 = (
|
||||
await PrimaryModel.objects.select_related("keywords")
|
||||
.limit(5, limit_raw_sql=True)
|
||||
.all()
|
||||
)
|
||||
|
||||
assert len(models3) == 2
|
||||
assert len(models3[0].keywords) == 2
|
||||
assert len(models3[1].keywords) == 3
|
||||
|
||||
models4 = (
|
||||
await PrimaryModel.objects.offset(1)
|
||||
.select_related("keywords")
|
||||
.limit(5, limit_raw_sql=True)
|
||||
.all()
|
||||
)
|
||||
|
||||
assert len(models4) == 3
|
||||
assert [x.name for x in models4] == ["Primary 1", "Primary 2", "Primary 3"]
|
||||
assert len(models4[0].keywords) == 1
|
||||
assert len(models4[1].keywords) == 3
|
||||
assert len(models4[2].keywords) == 0
|
||||
|
||||
models5 = (
|
||||
await PrimaryModel.objects.select_related("keywords")
|
||||
.offset(2, limit_raw_sql=True)
|
||||
.limit(5)
|
||||
.all()
|
||||
)
|
||||
|
||||
assert len(models5) == 3
|
||||
assert [x.name for x in models5] == ["Primary 2", "Primary 3", "Primary 4"]
|
||||
assert len(models5[0].keywords) == 3
|
||||
assert len(models5[1].keywords) == 0
|
||||
assert len(models5[2].keywords) == 0
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = create_engine(DATABASE_URL)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
@ -0,0 +1,130 @@
|
||||
# type: ignore
|
||||
from datetime import date
|
||||
from typing import List, Optional, Union
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
import ormar
|
||||
from ormar import ModelDefinitionError
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class MainMeta(ormar.ModelMeta):
|
||||
metadata = metadata
|
||||
database = database
|
||||
|
||||
|
||||
class Role(ormar.Model):
|
||||
class Meta(MainMeta):
|
||||
pass
|
||||
|
||||
name: str = ormar.String(primary_key=True, max_length=1000)
|
||||
order: int = ormar.Integer(default=0, name="sort_order")
|
||||
description: str = ormar.Text()
|
||||
|
||||
|
||||
class Company(ormar.Model):
|
||||
class Meta(MainMeta):
|
||||
pass
|
||||
|
||||
name: str = ormar.String(primary_key=True, max_length=1000)
|
||||
|
||||
|
||||
class UserRoleCompany(ormar.Model):
|
||||
class Meta(MainMeta):
|
||||
pass
|
||||
|
||||
|
||||
class User(ormar.Model):
|
||||
class Meta(MainMeta):
|
||||
pass
|
||||
|
||||
registrationnumber: str = ormar.String(primary_key=True, max_length=1000)
|
||||
company: Company = ormar.ForeignKey(Company)
|
||||
company2: Company = ormar.ForeignKey(Company, related_name="secondary_users")
|
||||
name: str = ormar.Text()
|
||||
role: Optional[Role] = ormar.ForeignKey(Role)
|
||||
roleforcompanies: Optional[Union[Company, List[Company]]] = ormar.ManyToMany(
|
||||
Company, through=UserRoleCompany, related_name="role_users"
|
||||
)
|
||||
lastupdate: date = ormar.DateTime(server_default=sqlalchemy.func.now())
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = create_engine(DATABASE_URL)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
def test_wrong_model():
|
||||
with pytest.raises(ModelDefinitionError):
|
||||
|
||||
class User(ormar.Model):
|
||||
class Meta(MainMeta):
|
||||
pass
|
||||
|
||||
registrationnumber: str = ormar.Text(primary_key=True)
|
||||
company: Company = ormar.ForeignKey(Company)
|
||||
company2: Company = ormar.ForeignKey(Company)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_primary_models():
|
||||
async with database:
|
||||
await Role.objects.create(
|
||||
name="user", order=0, description="no administration right"
|
||||
)
|
||||
role_1 = await Role.objects.create(
|
||||
name="admin", order=1, description="standard administration right"
|
||||
)
|
||||
await Role.objects.create(
|
||||
name="super_admin", order=2, description="super administration right"
|
||||
)
|
||||
assert await Role.objects.count() == 3
|
||||
|
||||
company_0 = await Company.objects.create(name="Company")
|
||||
company_1 = await Company.objects.create(name="Subsidiary Company 1")
|
||||
company_2 = await Company.objects.create(name="Subsidiary Company 2")
|
||||
company_3 = await Company.objects.create(name="Subsidiary Company 3")
|
||||
assert await Company.objects.count() == 4
|
||||
|
||||
user = await User.objects.create(
|
||||
registrationnumber="00-00000", company=company_0, name="admin", role=role_1
|
||||
)
|
||||
assert await User.objects.count() == 1
|
||||
|
||||
await user.delete()
|
||||
assert await User.objects.count() == 0
|
||||
|
||||
user = await User.objects.create(
|
||||
registrationnumber="00-00000",
|
||||
company=company_0,
|
||||
company2=company_3,
|
||||
name="admin",
|
||||
role=role_1,
|
||||
)
|
||||
await user.roleforcompanies.add(company_1)
|
||||
await user.roleforcompanies.add(company_2)
|
||||
|
||||
users = await User.objects.select_related(
|
||||
["company", "company2", "roleforcompanies"]
|
||||
).all()
|
||||
assert len(users) == 1
|
||||
assert len(users[0].roleforcompanies) == 2
|
||||
assert len(users[0].roleforcompanies[0].role_users) == 1
|
||||
assert users[0].company.name == "Company"
|
||||
assert len(users[0].company.users) == 1
|
||||
assert users[0].company2.name == "Subsidiary Company 3"
|
||||
assert len(users[0].company2.secondary_users) == 1
|
||||
|
||||
users = await User.objects.select_related("roleforcompanies").all()
|
||||
assert len(users) == 1
|
||||
assert len(users[0].roleforcompanies) == 2
|
||||
93
tests/test_relations/test_selecting_proper_table_prefix.py
Normal file
93
tests/test_relations/test_selecting_proper_table_prefix.py
Normal file
@ -0,0 +1,93 @@
|
||||
from typing import List, Optional
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
import ormar
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
class User(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = database
|
||||
tablename = "test_users"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=50)
|
||||
|
||||
|
||||
class Signup(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = database
|
||||
tablename = "test_signup"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
|
||||
|
||||
class Session(ormar.Model):
|
||||
class Meta:
|
||||
metadata = metadata
|
||||
database = database
|
||||
tablename = "test_sessions"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
name: str = ormar.String(max_length=255, index=True)
|
||||
some_text: str = ormar.Text()
|
||||
some_other_text: Optional[str] = ormar.Text(nullable=True)
|
||||
students: Optional[List[User]] = ormar.ManyToMany(User, through=Signup)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="module")
|
||||
def create_test_database():
|
||||
engine = create_engine(DATABASE_URL)
|
||||
metadata.create_all(engine)
|
||||
yield
|
||||
metadata.drop_all(engine)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_sessions_for_user():
|
||||
async with database:
|
||||
for user_id in [1, 2, 3, 4, 5]:
|
||||
await User.objects.create(name=f"User {user_id}")
|
||||
|
||||
for name, some_text, some_other_text in [
|
||||
("Session 1", "Some text 1", "Some other text 1"),
|
||||
("Session 2", "Some text 2", "Some other text 2"),
|
||||
("Session 3", "Some text 3", "Some other text 3"),
|
||||
("Session 4", "Some text 4", "Some other text 4"),
|
||||
("Session 5", "Some text 5", "Some other text 5"),
|
||||
]:
|
||||
await Session(
|
||||
name=name, some_text=some_text, some_other_text=some_other_text
|
||||
).save()
|
||||
|
||||
s1 = await Session.objects.get(pk=1)
|
||||
s2 = await Session.objects.get(pk=2)
|
||||
|
||||
users = {}
|
||||
for i in range(1, 6):
|
||||
user = await User.objects.get(pk=i)
|
||||
users[f"user_{i}"] = user
|
||||
if i % 2 == 0:
|
||||
await s1.students.add(user)
|
||||
else:
|
||||
await s2.students.add(user)
|
||||
|
||||
assert len(s1.students) == 2
|
||||
assert len(s2.students) == 3
|
||||
|
||||
assert [x.pk for x in s1.students] == [2, 4]
|
||||
assert [x.pk for x in s2.students] == [1, 3, 5]
|
||||
|
||||
user = await User.objects.select_related("sessions").get(pk=1)
|
||||
|
||||
assert user.sessions is not None
|
||||
assert len(user.sessions) > 0
|
||||
51
tests/test_relations/test_through_relations_fail.py
Normal file
51
tests/test_relations/test_through_relations_fail.py
Normal file
@ -0,0 +1,51 @@
|
||||
# type: ignore
|
||||
|
||||
import databases
|
||||
import pytest
|
||||
import sqlalchemy
|
||||
|
||||
import ormar
|
||||
from ormar import ModelDefinitionError
|
||||
from tests.settings import DATABASE_URL
|
||||
|
||||
database = databases.Database(DATABASE_URL, force_rollback=True)
|
||||
metadata = sqlalchemy.MetaData()
|
||||
|
||||
|
||||
def test_through_with_relation_fails():
|
||||
class BaseMeta(ormar.ModelMeta):
|
||||
database = database
|
||||
metadata = metadata
|
||||
|
||||
class Category(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
tablename = "categories"
|
||||
|
||||
id = ormar.Integer(primary_key=True)
|
||||
name = ormar.String(max_length=40)
|
||||
|
||||
class Blog(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
pass
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
|
||||
class PostCategory(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
tablename = "posts_x_categories"
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
sort_order: int = ormar.Integer(nullable=True)
|
||||
param_name: str = ormar.String(default="Name", max_length=200)
|
||||
blog = ormar.ForeignKey(Blog)
|
||||
|
||||
with pytest.raises(ModelDefinitionError):
|
||||
|
||||
class Post(ormar.Model):
|
||||
class Meta(BaseMeta):
|
||||
pass
|
||||
|
||||
id: int = ormar.Integer(primary_key=True)
|
||||
title: str = ormar.String(max_length=200)
|
||||
categories = ormar.ManyToMany(Category, through=PostCategory)
|
||||
Reference in New Issue
Block a user