fix index error on double addition of the same related model (#1346)
* fix index error on double addition of the same related model * connect to db * change to cleanup instead of transaction as a check
This commit is contained in:
@ -325,7 +325,7 @@ class RelationProxy(Generic[T], List[T]):
|
|||||||
:param item: child to add to relation
|
:param item: child to add to relation
|
||||||
:type item: Model
|
:type item: Model
|
||||||
"""
|
"""
|
||||||
new_idx = len(self)
|
new_idx = len(self) if item not in self else self.index(item)
|
||||||
relation_name = self.related_field_name
|
relation_name = self.related_field_name
|
||||||
await self._owner.signals.pre_relation_add.send(
|
await self._owner.signals.pre_relation_add.send(
|
||||||
sender=self._owner.__class__,
|
sender=self._owner.__class__,
|
||||||
|
|||||||
0
tests/test_hashes/__init__.py
Normal file
0
tests/test_hashes/__init__.py
Normal file
100
tests/test_hashes/test_many_to_many.py
Normal file
100
tests/test_hashes/test_many_to_many.py
Normal file
@ -0,0 +1,100 @@
|
|||||||
|
import sqlite3
|
||||||
|
from typing import ForwardRef, List, Optional
|
||||||
|
|
||||||
|
import asyncpg
|
||||||
|
import ormar
|
||||||
|
import pymysql
|
||||||
|
import pytest
|
||||||
|
import pytest_asyncio
|
||||||
|
|
||||||
|
from tests.lifespan import init_tests
|
||||||
|
from tests.settings import create_config
|
||||||
|
|
||||||
|
base_ormar_config = create_config(force_rollback=True)
|
||||||
|
|
||||||
|
|
||||||
|
class Category(ormar.Model):
|
||||||
|
ormar_config = base_ormar_config.copy(tablename="categories")
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(max_length=40)
|
||||||
|
|
||||||
|
|
||||||
|
class Author(ormar.Model):
|
||||||
|
ormar_config = base_ormar_config.copy(tablename="authors")
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
name: str = ormar.String(max_length=40)
|
||||||
|
|
||||||
|
|
||||||
|
class Post(ormar.Model):
|
||||||
|
ormar_config = base_ormar_config.copy(tablename="posts")
|
||||||
|
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
title: str = ormar.String(max_length=200)
|
||||||
|
categories: Optional[List[Category]] = ormar.ManyToMany(Category)
|
||||||
|
authors: Optional[List[Author]] = ormar.ManyToMany(
|
||||||
|
Author, through=ForwardRef("AuthorXPosts")
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class AuthorXPosts(ormar.Model):
|
||||||
|
ormar_config = base_ormar_config.copy(
|
||||||
|
tablename="authors_x_posts", constraints=[ormar.UniqueColumns("author", "post")]
|
||||||
|
)
|
||||||
|
id: int = ormar.Integer(primary_key=True)
|
||||||
|
author: Optional[int] = ormar.Integer(default=None)
|
||||||
|
post: Optional[int] = ormar.Integer(default=None)
|
||||||
|
|
||||||
|
|
||||||
|
Post.update_forward_refs()
|
||||||
|
|
||||||
|
|
||||||
|
create_test_database = init_tests(base_ormar_config)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest_asyncio.fixture(scope="function", autouse=True)
|
||||||
|
async def cleanup():
|
||||||
|
yield
|
||||||
|
async with base_ormar_config.database:
|
||||||
|
await Post.ormar_config.model_fields["categories"].through.objects.delete(
|
||||||
|
each=True
|
||||||
|
)
|
||||||
|
await Post.ormar_config.model_fields["authors"].through.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_adding_same_m2m_model_twice():
|
||||||
|
async with base_ormar_config.database:
|
||||||
|
async with base_ormar_config.database:
|
||||||
|
post = await Post.objects.create(title="Hello, M2M")
|
||||||
|
news = await Category(name="News").save()
|
||||||
|
|
||||||
|
await post.categories.add(news)
|
||||||
|
await post.categories.add(news)
|
||||||
|
|
||||||
|
categories = await post.categories.all()
|
||||||
|
assert categories == [news]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_adding_same_m2m_model_twice_with_unique():
|
||||||
|
async with base_ormar_config.database:
|
||||||
|
async with base_ormar_config.database:
|
||||||
|
post = await Post.objects.create(title="Hello, M2M")
|
||||||
|
redactor = await Author(name="News").save()
|
||||||
|
|
||||||
|
await post.authors.add(redactor)
|
||||||
|
with pytest.raises(
|
||||||
|
(
|
||||||
|
sqlite3.IntegrityError,
|
||||||
|
pymysql.IntegrityError,
|
||||||
|
asyncpg.exceptions.UniqueViolationError,
|
||||||
|
)
|
||||||
|
):
|
||||||
|
await post.authors.add(redactor)
|
||||||
Reference in New Issue
Block a user