# Path: psqlextra/indexes/conditional_unique_index.py
# class ConditionalUniqueIndex(Index):
#     """Creates a partial unique index based on a given condition.
# 
#     Useful, for example, if you need unique combination of foreign keys, but you might want to include
#     NULL as a valid value. In that case, you can just use:
# 
#     >>> class Meta:
#     >>>    indexes = [
#     >>>        ConditionalUniqueIndex(fields=['a', 'b', 'c'], condition='"c" IS NOT NULL'),
#     >>>        ConditionalUniqueIndex(fields=['a', 'b'], condition='"c" IS NULL')
#     >>>    ]
#     """
# 
#     sql_create_index = "CREATE UNIQUE INDEX %(name)s ON %(table)s (%(columns)s)%(extra)s WHERE %(condition)s"
# 
#     def __init__(self, condition: str, fields=[], name=None):
#         """Initializes a new instance of :see:ConditionalUniqueIndex."""
# 
#         super().__init__(fields=fields, name=name)
# 
#         self._condition = condition
# 
#     def create_sql(self, model, schema_editor, using="", **kwargs):
#         """Creates the actual SQL used when applying the migration."""
#         if django.VERSION >= (2, 0):
#             statement = super().create_sql(model, schema_editor, using)
#             statement.template = self.sql_create_index
#             statement.parts["condition"] = self._condition
#             return statement
#         else:
#             sql_create_index = self.sql_create_index
#             sql_parameters = {
#                 **Index.get_sql_create_template_values(
#                     self, model, schema_editor, using
#                 ),
#                 "condition": self._condition,
#             }
#             return sql_create_index % sql_parameters
# 
#     def deconstruct(self):
#         """Serializes the :see:ConditionalUniqueIndex for the migrations
#         file."""
#         path = "%s.%s" % (self.__class__.__module__, self.__class__.__name__)
#         path = path.replace("django.db.models.indexes", "django.db.models")
#         return (
#             path,
#             (),
#             {
#                 "fields": self.fields,
#                 "name": self.name,
#                 "condition": self._condition,
#             },
#         )
# 
# Path: tests/fake_model.py
# def get_fake_model(fields=None, model_base=PostgresModel, meta_options={}):
#     """Defines a fake model and creates it in the database."""
# 
#     model = define_fake_model(fields, model_base, meta_options)
# 
#     with connection.schema_editor() as schema_editor:
#         schema_editor.create_model(model)
# 
#     return model
# 
# Path: tests/migrations.py
# def apply_migration(operations, state=None, backwards: bool = False):
#     """Executes the specified migration operations using the specified schema
#     editor.
# 
#     Arguments:
#         operations:
#             The migration operations to execute.
# 
#         state:
#             The state state to use during the
#             migrations.
# 
#         backwards:
#             Whether to apply the operations
#             in reverse (backwards).
#     """
# 
#     state = state or migrations.state.ProjectState.from_apps(apps)
# 
#     class Migration(migrations.Migration):
#         pass
# 
#     Migration.operations = operations
# 
#     migration = Migration("migration", "tests")
#     executor = MigrationExecutor(connection)
# 
#     if not backwards:
#         executor.apply_migration(state, migration)
#     else:
#         executor.unapply_migration(state, migration)
# 
#     return migration
# 
# @contextmanager
# def filtered_schema_editor(*filters: List[str]):
#     """Gets a schema editor, but filters executed SQL statements based on the
#     specified text filters.
# 
#     Arguments:
#         filters:
#             List of strings to filter SQL
#             statements on.
#     """
# 
#     with connection.schema_editor() as schema_editor:
#         wrapper_for = schema_editor.execute
#         with mock.patch.object(
#             PostgresSchemaEditor, "execute", wraps=wrapper_for
#         ) as execute:
#             filter_results = {}
#             yield filter_results
# 
#     for filter_text in filters:
#         filter_results[filter_text] = [
#             call for call in execute.mock_calls if filter_text in str(call)
#         ]


import pytest
from django.db import IntegrityError, models, transaction
from django.db.migrations import AddIndex, CreateModel
from psqlextra.indexes import ConditionalUniqueIndex
from .fake_model import get_fake_model
from .migrations import apply_migration, filtered_schema_editor
    model.objects.upsert(
        conflict_target=["a", "b"],
        index_predicate='"b" IS NOT NULL',
        fields=dict(a=1, b=1, c=1),
    )
    assert model.objects.all().count() == 2
    assert model.objects.filter(a=1, c=2).count() == 1
    assert model.objects.filter(a=1, b=1, c=1).count() == 1

    model.objects.upsert(
        conflict_target=["a", "b"],
        index_predicate='"b" IS NOT NULL',
        fields=dict(a=1, b=1, c=2),
    )
    assert model.objects.all().count() == 2
    assert model.objects.filter(a=1, c=1).count() == 0
    assert model.objects.filter(a=1, b=1, c=2).count() == 1


def test_cui_inserting():
    """Tests inserting respects the :see:ConditionalUniqueIndex rules."""

    model = get_fake_model(
        fields={
            "a": models.IntegerField(),
            "b": models.IntegerField(null=True),
            "c": models.IntegerField(),
        },
        meta_options={
            "indexes": [

# Path: psqlextra/indexes/conditional_unique_index.py
# class ConditionalUniqueIndex(Index):
#     """Creates a partial unique index based on a given condition.
# 
#     Useful, for example, if you need unique combination of foreign keys, but you might want to include
#     NULL as a valid value. In that case, you can just use:
# 
#     >>> class Meta:
#     >>>    indexes = [
#     >>>        ConditionalUniqueIndex(fields=['a', 'b', 'c'], condition='"c" IS NOT NULL'),
#     >>>        ConditionalUniqueIndex(fields=['a', 'b'], condition='"c" IS NULL')
#     >>>    ]
#     """
# 
#     sql_create_index = "CREATE UNIQUE INDEX %(name)s ON %(table)s (%(columns)s)%(extra)s WHERE %(condition)s"
# 
#     def __init__(self, condition: str, fields=[], name=None):
#         """Initializes a new instance of :see:ConditionalUniqueIndex."""
# 
#         super().__init__(fields=fields, name=name)
# 
#         self._condition = condition
# 
#     def create_sql(self, model, schema_editor, using="", **kwargs):
#         """Creates the actual SQL used when applying the migration."""
#         if django.VERSION >= (2, 0):
#             statement = super().create_sql(model, schema_editor, using)
#             statement.template = self.sql_create_index
#             statement.parts["condition"] = self._condition
#             return statement
#         else:
#             sql_create_index = self.sql_create_index
#             sql_parameters = {
#                 **Index.get_sql_create_template_values(
#                     self, model, schema_editor, using
#                 ),
#                 "condition": self._condition,
#             }
#             return sql_create_index % sql_parameters
# 
#     def deconstruct(self):
#         """Serializes the :see:ConditionalUniqueIndex for the migrations
#         file."""
#         path = "%s.%s" % (self.__class__.__module__, self.__class__.__name__)
#         path = path.replace("django.db.models.indexes", "django.db.models")
#         return (
#             path,
#             (),
#             {
#                 "fields": self.fields,
#                 "name": self.name,
#                 "condition": self._condition,
#             },
#         )
# 
# Path: tests/fake_model.py
# def get_fake_model(fields=None, model_base=PostgresModel, meta_options={}):
#     """Defines a fake model and creates it in the database."""
# 
#     model = define_fake_model(fields, model_base, meta_options)
# 
#     with connection.schema_editor() as schema_editor:
#         schema_editor.create_model(model)
# 
#     return model
# 
# Path: tests/migrations.py
# def apply_migration(operations, state=None, backwards: bool = False):
#     """Executes the specified migration operations using the specified schema
#     editor.
# 
#     Arguments:
#         operations:
#             The migration operations to execute.
# 
#         state:
#             The state state to use during the
#             migrations.
# 
#         backwards:
#             Whether to apply the operations
#             in reverse (backwards).
#     """
# 
#     state = state or migrations.state.ProjectState.from_apps(apps)
# 
#     class Migration(migrations.Migration):
#         pass
# 
#     Migration.operations = operations
# 
#     migration = Migration("migration", "tests")
#     executor = MigrationExecutor(connection)
# 
#     if not backwards:
#         executor.apply_migration(state, migration)
#     else:
#         executor.unapply_migration(state, migration)
# 
#     return migration
# 
# @contextmanager
# def filtered_schema_editor(*filters: List[str]):
#     """Gets a schema editor, but filters executed SQL statements based on the
#     specified text filters.
# 
#     Arguments:
#         filters:
#             List of strings to filter SQL
#             statements on.
#     """
# 
#     with connection.schema_editor() as schema_editor:
#         wrapper_for = schema_editor.execute
#         with mock.patch.object(
#             PostgresSchemaEditor, "execute", wraps=wrapper_for
#         ) as execute:
#             filter_results = {}
#             yield filter_results
# 
#     for filter_text in filters:
#         filter_results[filter_text] = [
#             call for call in execute.mock_calls if filter_text in str(call)
#         ]


# Path: tests/test_conditional_unique_index.py
import pytest
from django.db import IntegrityError, models, transaction
from django.db.migrations import AddIndex, CreateModel
from psqlextra.indexes import ConditionalUniqueIndex
from .fake_model import get_fake_model
from .migrations import apply_migration, filtered_schema_editor

    model.objects.upsert(
        conflict_target=["a", "b"],
        index_predicate='"b" IS NOT NULL',
        fields=dict(a=1, b=1, c=1),
    )
    assert model.objects.all().count() == 2
    assert model.objects.filter(a=1, c=2).count() == 1
    assert model.objects.filter(a=1, b=1, c=1).count() == 1

    model.objects.upsert(
        conflict_target=["a", "b"],
        index_predicate='"b" IS NOT NULL',
        fields=dict(a=1, b=1, c=2),
    )
    assert model.objects.all().count() == 2
    assert model.objects.filter(a=1, c=1).count() == 0
    assert model.objects.filter(a=1, b=1, c=2).count() == 1


def test_cui_inserting():
    """Tests inserting respects the :see:ConditionalUniqueIndex rules."""

    model = get_fake_model(
        fields={
            "a": models.IntegerField(),
            "b": models.IntegerField(null=True),
            "c": models.IntegerField(),
        },
        meta_options={
            "indexes": [

                ConditionalUniqueIndex(