import operator
import uuid
from unittest import mock, skipIf, skipUnless

from django import forms
from django.core import serializers
from django.core.exceptions import ValidationError
from django.core.serializers.json import DjangoJSONEncoder
from django.db import (
    DataError, IntegrityError, NotSupportedError, OperationalError, connection,
    models,
)
from django.db.models import Count, F, OuterRef, Q, Subquery, Transform, Value
from django.db.models.expressions import RawSQL
from django.db.models.fields.json import (
    KeyTextTransform, KeyTransform, KeyTransformFactory,
    KeyTransformTextLookupMixin,
)
from django.db.models.functions import Cast
from django.test import SimpleTestCase, TestCase, skipUnlessDBFeature
from django.test.utils import CaptureQueriesContext

from .models import CustomJSONDecoder, JSONModel, NullableJSONModel


@skipUnlessDBFeature('supports_json_field')
class JSONFieldTests(TestCase):
    def test_invalid_value(self):
        msg = 'is not JSON serializable'
        with self.assertRaisesMessage(TypeError, msg):
            NullableJSONModel.objects.create(value={
                'uuid': uuid.UUID('d85e2076-b67c-4ee7-8c3a-2bf5a2cc2475'),
            })

    def test_custom_encoder_decoder(self):
        value = {'uuid': uuid.UUID('{d85e2076-b67c-4ee7-8c3a-2bf5a2cc2475}')}
        obj = NullableJSONModel(value_custom=value)
        obj.clean_fields()
        obj.save()
        obj.refresh_from_db()
        self.assertEqual(obj.value_custom, value)

    def test_db_check_constraints(self):
        value = '{@!invalid json value 123 $!@#'
        with mock.patch.object(DjangoJSONEncoder, 'encode', return_value=value):
            with self.assertRaises((IntegrityError, DataError, OperationalError)):
                NullableJSONModel.objects.create(value_custom=value)


class TestMethods(SimpleTestCase):
    def test_deconstruct(self):
        field = models.JSONField()
        name, path, args, kwargs = field.deconstruct()
        self.assertEqual(path, 'django.db.models.JSONField')
        self.assertEqual(args, [])
        self.assertEqual(kwargs, {})

    def test_deconstruct_custom_encoder_decoder(self):
        field = models.JSONField(encoder=DjangoJSONEncoder, decoder=CustomJSONDecoder)
        name, path, args, kwargs = field.deconstruct()
        self.assertEqual(kwargs['encoder'], DjangoJSONEncoder)
        self.assertEqual(kwargs['decoder'], CustomJSONDecoder)

    def test_get_transforms(self):
        @models.JSONField.register_lookup
        class MyTransform(Transform):
            lookup_name = 'my_transform'
        field = models.JSONField()
        transform = field.get_transform('my_transform')
        self.assertIs(transform, MyTransform)
        models.JSONField._unregister_lookup(MyTransform)
        models.JSONField._clear_cached_lookups()
        transform = field.get_transform('my_transform')
        self.assertIsInstance(transform, KeyTransformFactory)

    def test_key_transform_text_lookup_mixin_non_key_transform(self):
        transform = Transform('test')
        msg = (
            'Transform should be an instance of KeyTransform in order to use '
            'this lookup.'
        )
        with self.assertRaisesMessage(TypeError, msg):
            KeyTransformTextLookupMixin(transform)


class TestValidation(SimpleTestCase):
    def test_invalid_encoder(self):
        msg = 'The encoder parameter must be a callable object.'
        with self.assertRaisesMessage(ValueError, msg):
            models.JSONField(encoder=DjangoJSONEncoder())

    def test_invalid_decoder(self):
        msg = 'The decoder parameter must be a callable object.'
        with self.assertRaisesMessage(ValueError, msg):
            models.JSONField(decoder=CustomJSONDecoder())

    def test_validation_error(self):
        field = models.JSONField()
        msg = 'Value must be valid JSON.'
        value = uuid.UUID('{d85e2076-b67c-4ee7-8c3a-2bf5a2cc2475}')
        with self.assertRaisesMessage(ValidationError, msg):
            field.clean({'uuid': value}, None)

    def test_custom_encoder(self):
        field = models.JSONField(encoder=DjangoJSONEncoder)
        value = uuid.UUID('{d85e2076-b67c-4ee7-8c3a-2bf5a2cc2475}')
        field.clean({'uuid': value}, None)


class TestFormField(SimpleTestCase):
    def test_formfield(self):
        model_field = models.JSONField()
        form_field = model_field.formfield()
        self.assertIsInstance(form_field, forms.JSONField)

    def test_formfield_custom_encoder_decoder(self):
        model_field = models.JSONField(encoder=DjangoJSONEncoder, decoder=CustomJSONDecoder)
        form_field = model_field.formfield()
        self.assertIs(form_field.encoder, DjangoJSONEncoder)
        self.assertIs(form_field.decoder, CustomJSONDecoder)


class TestSerialization(SimpleTestCase):
    test_data = (
        '[{"fields": {"value": %s}, '
        '"model": "model_fields.jsonmodel", "pk": null}]'
    )
    test_values = (
        # (Python value, serialized value),
        ({'a': 'b', 'c': None}, '{"a": "b", "c": null}'),
        ('abc', '"abc"'),
        ('{"a": "a"}', '"{\\"a\\": \\"a\\"}"'),
    )

    def test_dumping(self):
        for value, serialized in self.test_values:
            with self.subTest(value=value):
                instance = JSONModel(value=value)
                data = serializers.serialize('json', [instance])
                self.assertJSONEqual(data, self.test_data % serialized)

    def test_loading(self):
        for value, serialized in self.test_values:
            with self.subTest(value=value):
                instance = list(
                    serializers.deserialize('json', self.test_data % serialized)
                )[0].object
                self.assertEqual(instance.value, value)


@skipUnlessDBFeature('supports_json_field')
class TestSaveLoad(TestCase):
    def test_null(self):
        obj = NullableJSONModel(value=None)
        obj.save()
        obj.refresh_from_db()
        self.assertIsNone(obj.value)

    @skipUnlessDBFeature('supports_primitives_in_json_field')
    def test_json_null_different_from_sql_null(self):
        json_null = NullableJSONModel.objects.create(value=Value('null'))
        json_null.refresh_from_db()
        sql_null = NullableJSONModel.objects.create(value=None)
        sql_null.refresh_from_db()
        # 'null' is not equal to NULL in the database.
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value=Value('null')),
            [json_null],
        )
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value=None),
            [json_null],
        )
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__isnull=True),
            [sql_null],
        )
        # 'null' is equal to NULL in Python (None).
        self.assertEqual(json_null.value, sql_null.value)

    @skipUnlessDBFeature('supports_primitives_in_json_field')
    def test_primitives(self):
        values = [
            True,
            1,
            1.45,
            'String',
            '',
        ]
        for value in values:
            with self.subTest(value=value):
                obj = JSONModel(value=value)
                obj.save()
                obj.refresh_from_db()
                self.assertEqual(obj.value, value)

    def test_dict(self):
        values = [
            {},
            {'name': 'John', 'age': 20, 'height': 180.3},
            {'a': True, 'b': {'b1': False, 'b2': None}},
        ]
        for value in values:
            with self.subTest(value=value):
                obj = JSONModel.objects.create(value=value)
                obj.refresh_from_db()
                self.assertEqual(obj.value, value)

    def test_list(self):
        values = [
            [],
            ['John', 20, 180.3],
            [True, [False, None]],
        ]
        for value in values:
            with self.subTest(value=value):
                obj = JSONModel.objects.create(value=value)
                obj.refresh_from_db()
                self.assertEqual(obj.value, value)

    def test_realistic_object(self):
        value = {
            'name': 'John',
            'age': 20,
            'pets': [
                {'name': 'Kit', 'type': 'cat', 'age': 2},
                {'name': 'Max', 'type': 'dog', 'age': 1},
            ],
            'courses': [
                ['A1', 'A2', 'A3'],
                ['B1', 'B2'],
                ['C1'],
            ],
        }
        obj = JSONModel.objects.create(value=value)
        obj.refresh_from_db()
        self.assertEqual(obj.value, value)


@skipUnlessDBFeature('supports_json_field')
class TestQuerying(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.primitives = [True, False, 'yes', 7, 9.6]
        values = [
            None,
            [],
            {},
            {'a': 'b', 'c': 14},
            {
                'a': 'b',
                'c': 14,
                'd': ['e', {'f': 'g'}],
                'h': True,
                'i': False,
                'j': None,
                'k': {'l': 'm'},
                'n': [None],
            },
            [1, [2]],
            {'k': True, 'l': False},
            {
                'foo': 'bar',
                'baz': {'a': 'b', 'c': 'd'},
                'bar': ['foo', 'bar'],
                'bax': {'foo': 'bar'},
            },
        ]
        cls.objs = [
            NullableJSONModel.objects.create(value=value)
            for value in values
        ]
        if connection.features.supports_primitives_in_json_field:
            cls.objs.extend([
                NullableJSONModel.objects.create(value=value)
                for value in cls.primitives
            ])
        cls.raw_sql = '%s::jsonb' if connection.vendor == 'postgresql' else '%s'

    def test_exact(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__exact={}),
            [self.objs[2]],
        )

    def test_exact_complex(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__exact={'a': 'b', 'c': 14}),
            [self.objs[3]],
        )

    def test_isnull(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__isnull=True),
            [self.objs[0]],
        )

    def test_ordering_by_transform(self):
        objs = [
            NullableJSONModel.objects.create(value={'ord': 93, 'name': 'bar'}),
            NullableJSONModel.objects.create(value={'ord': 22.1, 'name': 'foo'}),
            NullableJSONModel.objects.create(value={'ord': -1, 'name': 'baz'}),
            NullableJSONModel.objects.create(value={'ord': 21.931902, 'name': 'spam'}),
            NullableJSONModel.objects.create(value={'ord': -100291029, 'name': 'eggs'}),
        ]
        query = NullableJSONModel.objects.filter(value__name__isnull=False).order_by('value__ord')
        expected = [objs[4], objs[2], objs[3], objs[1], objs[0]]
        mariadb = connection.vendor == 'mysql' and connection.mysql_is_mariadb
        if mariadb or connection.vendor == 'oracle':
            # MariaDB and Oracle return JSON values as strings.
            expected = [objs[2], objs[4], objs[3], objs[1], objs[0]]
        self.assertSequenceEqual(query, expected)

    def test_ordering_grouping_by_key_transform(self):
        base_qs = NullableJSONModel.objects.filter(value__d__0__isnull=False)
        for qs in (
            base_qs.order_by('value__d__0'),
            base_qs.annotate(key=KeyTransform('0', KeyTransform('d', 'value'))).order_by('key'),
        ):
            self.assertSequenceEqual(qs, [self.objs[4]])
        qs = NullableJSONModel.objects.filter(value__isnull=False)
        self.assertQuerysetEqual(
            qs.filter(value__isnull=False).annotate(
                key=KeyTextTransform('f', KeyTransform('1', KeyTransform('d', 'value'))),
            ).values('key').annotate(count=Count('key')).order_by('count'),
            [(None, 0), ('g', 1)],
            operator.itemgetter('key', 'count'),
        )

    @skipIf(connection.vendor == 'oracle', "Oracle doesn't support grouping by LOBs, see #24096.")
    def test_ordering_grouping_by_count(self):
        qs = NullableJSONModel.objects.filter(
            value__isnull=False,
        ).values('value__d__0').annotate(count=Count('value__d__0')).order_by('count')
        self.assertQuerysetEqual(qs, [1, 11], operator.itemgetter('count'))

    def test_key_transform_raw_expression(self):
        expr = RawSQL(self.raw_sql, ['{"x": "bar"}'])
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__foo=KeyTransform('x', expr)),
            [self.objs[7]],
        )

    def test_nested_key_transform_raw_expression(self):
        expr = RawSQL(self.raw_sql, ['{"x": {"y": "bar"}}'])
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__foo=KeyTransform('y', KeyTransform('x', expr))),
            [self.objs[7]],
        )

    def test_key_transform_expression(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__d__0__isnull=False).annotate(
                key=KeyTransform('d', 'value'),
                chain=KeyTransform('0', 'key'),
                expr=KeyTransform('0', Cast('key', models.JSONField())),
            ).filter(chain=F('expr')),
            [self.objs[4]],
        )

    def test_nested_key_transform_expression(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__d__0__isnull=False).annotate(
                key=KeyTransform('d', 'value'),
                chain=KeyTransform('f', KeyTransform('1', 'key')),
                expr=KeyTransform('f', KeyTransform('1', Cast('key', models.JSONField()))),
            ).filter(chain=F('expr')),
            [self.objs[4]],
        )

    def test_has_key(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__has_key='a'),
            [self.objs[3], self.objs[4]],
        )

    def test_has_key_null_value(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__has_key='j'),
            [self.objs[4]],
        )

    def test_has_key_deep(self):
        tests = [
            (Q(value__baz__has_key='a'), self.objs[7]),
            (Q(value__has_key=KeyTransform('a', KeyTransform('baz', 'value'))), self.objs[7]),
            (Q(value__has_key=KeyTransform('c', KeyTransform('baz', 'value'))), self.objs[7]),
            (Q(value__d__1__has_key='f'), self.objs[4]),
            (
                Q(value__has_key=KeyTransform('f', KeyTransform('1', KeyTransform('d', 'value')))),
                self.objs[4],
            )
        ]
        for condition, expected in tests:
            with self.subTest(condition=condition):
                self.assertSequenceEqual(
                    NullableJSONModel.objects.filter(condition),
                    [expected],
                )

    def test_has_key_list(self):
        obj = NullableJSONModel.objects.create(value=[{'a': 1}, {'b': 'x'}])
        tests = [
            Q(value__1__has_key='b'),
            Q(value__has_key=KeyTransform('b', KeyTransform(1, 'value'))),
            Q(value__has_key=KeyTransform('b', KeyTransform('1', 'value'))),
        ]
        for condition in tests:
            with self.subTest(condition=condition):
                self.assertSequenceEqual(
                    NullableJSONModel.objects.filter(condition),
                    [obj],
                )

    def test_has_keys(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__has_keys=['a', 'c', 'h']),
            [self.objs[4]],
        )

    def test_has_any_keys(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__has_any_keys=['c', 'l']),
            [self.objs[3], self.objs[4], self.objs[6]],
        )

    def test_contains(self):
        tests = [
            ({}, self.objs[2:5] + self.objs[6:8]),
            ({'baz': {'a': 'b', 'c': 'd'}}, [self.objs[7]]),
            ({'k': True, 'l': False}, [self.objs[6]]),
            ({'d': ['e', {'f': 'g'}]}, [self.objs[4]]),
            ([1, [2]], [self.objs[5]]),
            ({'n': [None]}, [self.objs[4]]),
            ({'j': None}, [self.objs[4]]),
        ]
        for value, expected in tests:
            with self.subTest(value=value):
                qs = NullableJSONModel.objects.filter(value__contains=value)
                self.assertSequenceEqual(qs, expected)

    @skipUnlessDBFeature('supports_primitives_in_json_field')
    def test_contains_primitives(self):
        for value in self.primitives:
            with self.subTest(value=value):
                qs = NullableJSONModel.objects.filter(value__contains=value)
                self.assertIs(qs.exists(), True)

    @skipIf(
        connection.vendor == 'oracle',
        "Oracle doesn't support contained_by lookup.",
    )
    def test_contained_by(self):
        qs = NullableJSONModel.objects.filter(value__contained_by={'a': 'b', 'c': 14, 'h': True})
        self.assertSequenceEqual(qs, self.objs[2:4])

    @skipUnless(
        connection.vendor == 'oracle',
        "Oracle doesn't support contained_by lookup.",
    )
    def test_contained_by_unsupported(self):
        msg = 'contained_by lookup is not supported on Oracle.'
        with self.assertRaisesMessage(NotSupportedError, msg):
            NullableJSONModel.objects.filter(value__contained_by={'a': 'b'}).get()

    def test_deep_values(self):
        qs = NullableJSONModel.objects.values_list('value__k__l')
        expected_objs = [(None,)] * len(self.objs)
        expected_objs[4] = ('m',)
        self.assertSequenceEqual(qs, expected_objs)

    @skipUnlessDBFeature('can_distinct_on_fields')
    def test_deep_distinct(self):
        query = NullableJSONModel.objects.distinct('value__k__l').values_list('value__k__l')
        self.assertSequenceEqual(query, [('m',), (None,)])

    def test_isnull_key(self):
        # key__isnull=False works the same as has_key='key'.
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__a__isnull=True),
            self.objs[:3] + self.objs[5:],
        )
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__a__isnull=False),
            [self.objs[3], self.objs[4]],
        )
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__j__isnull=False),
            [self.objs[4]],
        )

    def test_isnull_key_or_none(self):
        obj = NullableJSONModel.objects.create(value={'a': None})
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(Q(value__a__isnull=True) | Q(value__a=None)),
            self.objs[:3] + self.objs[5:] + [obj],
        )

    def test_none_key(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__j=None),
            [self.objs[4]],
        )

    def test_none_key_exclude(self):
        obj = NullableJSONModel.objects.create(value={'j': 1})
        if connection.vendor == 'oracle':
            # Oracle supports filtering JSON objects with NULL keys, but the
            # current implementation doesn't support it.
            self.assertSequenceEqual(
                NullableJSONModel.objects.exclude(value__j=None),
                self.objs[1:4] + self.objs[5:] + [obj],
            )
        else:
            self.assertSequenceEqual(NullableJSONModel.objects.exclude(value__j=None), [obj])

    def test_shallow_list_lookup(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__0=1),
            [self.objs[5]],
        )

    def test_shallow_obj_lookup(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__a='b'),
            [self.objs[3], self.objs[4]],
        )

    def test_obj_subquery_lookup(self):
        qs = NullableJSONModel.objects.annotate(
            field=Subquery(NullableJSONModel.objects.filter(pk=OuterRef('pk')).values('value')),
        ).filter(field__a='b')
        self.assertSequenceEqual(qs, [self.objs[3], self.objs[4]])

    def test_deep_lookup_objs(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__k__l='m'),
            [self.objs[4]],
        )

    def test_shallow_lookup_obj_target(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__k={'l': 'm'}),
            [self.objs[4]],
        )

    def test_deep_lookup_array(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__1__0=2),
            [self.objs[5]],
        )

    def test_deep_lookup_mixed(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__d__1__f='g'),
            [self.objs[4]],
        )

    def test_deep_lookup_transform(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__c__gt=2),
            [self.objs[3], self.objs[4]],
        )
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__c__gt=2.33),
            [self.objs[3], self.objs[4]],
        )
        self.assertIs(NullableJSONModel.objects.filter(value__c__lt=5).exists(), False)

    @skipIf(
        connection.vendor == 'oracle',
        'Raises ORA-00600: internal error code on Oracle 18.',
    )
    def test_usage_in_subquery(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(
                id__in=NullableJSONModel.objects.filter(value__c=14),
            ),
            self.objs[3:5],
        )

    def test_key_iexact(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__iexact='BaR').exists(), True)
        self.assertIs(NullableJSONModel.objects.filter(value__foo__iexact='"BaR"').exists(), False)

    def test_key_contains(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__contains='ar').exists(), True)

    def test_key_icontains(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__icontains='Ar').exists(), True)

    def test_key_startswith(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__startswith='b').exists(), True)

    def test_key_istartswith(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__istartswith='B').exists(), True)

    def test_key_endswith(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__endswith='r').exists(), True)

    def test_key_iendswith(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__iendswith='R').exists(), True)

    def test_key_regex(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__regex=r'^bar$').exists(), True)

    def test_key_iregex(self):
        self.assertIs(NullableJSONModel.objects.filter(value__foo__iregex=r'^bAr$').exists(), True)

    @skipUnless(connection.vendor == 'postgresql', 'kwargs are crafted for PostgreSQL.')
    def test_key_sql_injection(self):
        with CaptureQueriesContext(connection) as queries:
            self.assertIs(
                NullableJSONModel.objects.filter(**{
                    """value__test' = '"a"') OR 1 = 1 OR ('d""": 'x',
                }).exists(),
                False,
            )
        self.assertIn(
            """."value" -> 'test'' = ''"a"'') OR 1 = 1 OR (''d') = '"x"' """,
            queries[0]['sql'],
        )

    @skipIf(connection.vendor == 'postgresql', 'PostgreSQL uses operators not functions.')
    def test_key_sql_injection_escape(self):
        query = str(JSONModel.objects.filter(**{
            """value__test") = '"a"' OR 1 = 1 OR ("d""": 'x',
        }).query)
        self.assertIn('"test\\"', query)
        self.assertIn('\\"d', query)

    def test_key_escape(self):
        obj = NullableJSONModel.objects.create(value={'%total': 10})
        self.assertEqual(NullableJSONModel.objects.filter(**{'value__%total': 10}).get(), obj)

    def test_none_key_and_exact_lookup(self):
        self.assertSequenceEqual(
            NullableJSONModel.objects.filter(value__a='b', value__j=None),
            [self.objs[4]],
        )

    def test_lookups_with_key_transform(self):
        tests = (
            ('value__d__contains', 'e'),
            ('value__baz__has_key', 'c'),
            ('value__baz__has_keys', ['a', 'c']),
            ('value__baz__has_any_keys', ['a', 'x']),
            ('value__contains', KeyTransform('bax', 'value')),
            ('value__has_key', KeyTextTransform('foo', 'value')),
        )
        # contained_by lookup is not supported on Oracle.
        if connection.vendor != 'oracle':
            tests += (
                ('value__baz__contained_by', {'a': 'b', 'c': 'd', 'e': 'f'}),
                (
                    'value__contained_by',
                    KeyTransform('x', RawSQL(
                        self.raw_sql,
                        ['{"x": {"a": "b", "c": 1, "d": "e"}}'],
                    )),
                ),
            )
        for lookup, value in tests:
            with self.subTest(lookup=lookup):
                self.assertIs(NullableJSONModel.objects.filter(
                    **{lookup: value},
                ).exists(), True)
