django/tests/postgres_tests/test_hstore.py

import json

from django.core import checks, exceptions, serializers
from django.db import connection
from django.db.models import OuterRef, Subquery
from django.db.models.expressions import RawSQL
from django.forms import Form
from django.test.utils import CaptureQueriesContext, isolate_apps

from . import PostgreSQLSimpleTestCase, PostgreSQLTestCase
from .models import HStoreModel, PostgreSQLModel

try:
    from django.contrib.postgres import forms
    from django.contrib.postgres.fields import HStoreField
    from django.contrib.postgres.fields.hstore import KeyTransform
    from django.contrib.postgres.validators import KeysValidator
except ImportError:
    pass


class SimpleTests(PostgreSQLTestCase):
    def test_save_load_success(self):
        value = {'a': 'b'}
        instance = HStoreModel(field=value)
        instance.save()
        reloaded = HStoreModel.objects.get()
        self.assertEqual(reloaded.field, value)

    def test_null(self):
        instance = HStoreModel(field=None)
        instance.save()
        reloaded = HStoreModel.objects.get()
        self.assertIsNone(reloaded.field)

    def test_value_null(self):
        value = {'a': None}
        instance = HStoreModel(field=value)
        instance.save()
        reloaded = HStoreModel.objects.get()
        self.assertEqual(reloaded.field, value)

    def test_key_val_cast_to_string(self):
        value = {'a': 1, 'b': 'B', 2: 'c', 'ï': 'ê'}
        expected_value = {'a': '1', 'b': 'B', '2': 'c', 'ï': 'ê'}

        instance = HStoreModel.objects.create(field=value)
        instance = HStoreModel.objects.get()
        self.assertEqual(instance.field, expected_value)

        instance = HStoreModel.objects.get(field__a=1)
        self.assertEqual(instance.field, expected_value)

        instance = HStoreModel.objects.get(field__has_keys=[2, 'a', 'ï'])
        self.assertEqual(instance.field, expected_value)

    def test_array_field(self):
        value = [
            {'a': 1, 'b': 'B', 2: 'c', 'ï': 'ê'},
            {'a': 1, 'b': 'B', 2: 'c', 'ï': 'ê'},
        ]
        expected_value = [
            {'a': '1', 'b': 'B', '2': 'c', 'ï': 'ê'},
            {'a': '1', 'b': 'B', '2': 'c', 'ï': 'ê'},
        ]
        instance = HStoreModel.objects.create(array_field=value)
        instance.refresh_from_db()
        self.assertEqual(instance.array_field, expected_value)


class TestQuerying(PostgreSQLTestCase):

    @classmethod
    def setUpTestData(cls):
        cls.objs = HStoreModel.objects.bulk_create([
            HStoreModel(field={'a': 'b'}),
            HStoreModel(field={'a': 'b', 'c': 'd'}),
            HStoreModel(field={'c': 'd'}),
            HStoreModel(field={}),
            HStoreModel(field=None),
        ])

    def test_exact(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__exact={'a': 'b'}),
            self.objs[:1]
        )

    def test_contained_by(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__contained_by={'a': 'b', 'c': 'd'}),
            self.objs[:4]
        )

    def test_contains(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__contains={'a': 'b'}),
            self.objs[:2]
        )

    def test_in_generator(self):
        def search():
            yield {'a': 'b'}
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__in=search()),
            self.objs[:1]
        )

    def test_has_key(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__has_key='c'),
            self.objs[1:3]
        )

    def test_has_keys(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__has_keys=['a', 'c']),
            self.objs[1:2]
        )

    def test_has_any_keys(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__has_any_keys=['a', 'c']),
            self.objs[:3]
        )

    def test_key_transform(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__a='b'),
            self.objs[:2]
        )

    def test_key_transform_raw_expression(self):
        expr = RawSQL('%s::hstore', ['x => b, y => c'])
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__a=KeyTransform('x', expr)),
            self.objs[:2]
        )

    def test_keys(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__keys=['a']),
            self.objs[:1]
        )

    def test_values(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__values=['b']),
            self.objs[:1]
        )

    def test_field_chaining(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__a__contains='b'),
            self.objs[:2]
        )

    def test_order_by_field(self):
        more_objs = (
            HStoreModel.objects.create(field={'g': '637'}),
            HStoreModel.objects.create(field={'g': '002'}),
            HStoreModel.objects.create(field={'g': '042'}),
            HStoreModel.objects.create(field={'g': '981'}),
        )
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__has_key='g').order_by('field__g'),
            [more_objs[1], more_objs[2], more_objs[0], more_objs[3]]
        )

    def test_keys_contains(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__keys__contains=['a']),
            self.objs[:2]
        )

    def test_values_overlap(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__values__overlap=['b', 'd']),
            self.objs[:3]
        )

    def test_key_isnull(self):
        obj = HStoreModel.objects.create(field={'a': None})
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__a__isnull=True),
            self.objs[2:5] + [obj]
        )
        self.assertSequenceEqual(
            HStoreModel.objects.filter(field__a__isnull=False),
            self.objs[:2]
        )

    def test_usage_in_subquery(self):
        self.assertSequenceEqual(
            HStoreModel.objects.filter(id__in=HStoreModel.objects.filter(field__a='b')),
            self.objs[:2]
        )

    def test_key_sql_injection(self):
        with CaptureQueriesContext(connection) as queries:
            self.assertFalse(
                HStoreModel.objects.filter(**{
                    "field__test' = 'a') OR 1 = 1 OR ('d": 'x',
                }).exists()
            )
        self.assertIn(
            """."field" -> 'test'' = ''a'') OR 1 = 1 OR (''d') = 'x' """,
            queries[0]['sql'],
        )

    def test_obj_subquery_lookup(self):
        qs = HStoreModel.objects.annotate(
            value=Subquery(HStoreModel.objects.filter(pk=OuterRef('pk')).values('field')),
        ).filter(value__a='b')
        self.assertSequenceEqual(qs, self.objs[:2])


@isolate_apps('postgres_tests')
class TestChecks(PostgreSQLSimpleTestCase):

    def test_invalid_default(self):
        class MyModel(PostgreSQLModel):
            field = HStoreField(default={})

        model = MyModel()
        self.assertEqual(model.check(), [
            checks.Warning(
                msg=(
                    "HStoreField default should be a callable instead of an "
                    "instance so that it's not shared between all field "
                    "instances."
                ),
                hint='Use a callable instead, e.g., use `dict` instead of `{}`.',
                obj=MyModel._meta.get_field('field'),
                id='fields.E010',
            )
        ])

    def test_valid_default(self):
        class MyModel(PostgreSQLModel):
            field = HStoreField(default=dict)

        self.assertEqual(MyModel().check(), [])


class TestSerialization(PostgreSQLSimpleTestCase):
    test_data = json.dumps([{
        'model': 'postgres_tests.hstoremodel',
        'pk': None,
        'fields': {
            'field': json.dumps({'a': 'b'}),
            'array_field': json.dumps([
                json.dumps({'a': 'b'}),
                json.dumps({'b': 'a'}),
            ]),
        },
    }])

    def test_dumping(self):
        instance = HStoreModel(field={'a': 'b'}, array_field=[{'a': 'b'}, {'b': 'a'}])
        data = serializers.serialize('json', [instance])
        self.assertEqual(json.loads(data), json.loads(self.test_data))

    def test_loading(self):
        instance = list(serializers.deserialize('json', self.test_data))[0].object
        self.assertEqual(instance.field, {'a': 'b'})
        self.assertEqual(instance.array_field, [{'a': 'b'}, {'b': 'a'}])

    def test_roundtrip_with_null(self):
        instance = HStoreModel(field={'a': 'b', 'c': None})
        data = serializers.serialize('json', [instance])
        new_instance = list(serializers.deserialize('json', data))[0].object
        self.assertEqual(instance.field, new_instance.field)


class TestValidation(PostgreSQLSimpleTestCase):

    def test_not_a_string(self):
        field = HStoreField()
        with self.assertRaises(exceptions.ValidationError) as cm:
            field.clean({'a': 1}, None)
        self.assertEqual(cm.exception.code, 'not_a_string')
        self.assertEqual(cm.exception.message % cm.exception.params, 'The value of “a” is not a string or null.')

    def test_none_allowed_as_value(self):
        field = HStoreField()
        self.assertEqual(field.clean({'a': None}, None), {'a': None})


class TestFormField(PostgreSQLSimpleTestCase):

    def test_valid(self):
        field = forms.HStoreField()
        value = field.clean('{"a": "b"}')
        self.assertEqual(value, {'a': 'b'})

    def test_invalid_json(self):
        field = forms.HStoreField()
        with self.assertRaises(exceptions.ValidationError) as cm:
            field.clean('{"a": "b"')
        self.assertEqual(cm.exception.messages[0], 'Could not load JSON data.')
        self.assertEqual(cm.exception.code, 'invalid_json')

    def test_non_dict_json(self):
        field = forms.HStoreField()
        msg = 'Input must be a JSON dictionary.'
        with self.assertRaisesMessage(exceptions.ValidationError, msg) as cm:
            field.clean('["a", "b", 1]')
        self.assertEqual(cm.exception.code, 'invalid_format')

    def test_not_string_values(self):
        field = forms.HStoreField()
        value = field.clean('{"a": 1}')
        self.assertEqual(value, {'a': '1'})

    def test_none_value(self):
        field = forms.HStoreField()
        value = field.clean('{"a": null}')
        self.assertEqual(value, {'a': None})

    def test_empty(self):
        field = forms.HStoreField(required=False)
        value = field.clean('')
        self.assertEqual(value, {})

    def test_model_field_formfield(self):
        model_field = HStoreField()
        form_field = model_field.formfield()
        self.assertIsInstance(form_field, forms.HStoreField)

    def test_field_has_changed(self):
        class HStoreFormTest(Form):
            f1 = forms.HStoreField()
        form_w_hstore = HStoreFormTest()
        self.assertFalse(form_w_hstore.has_changed())

        form_w_hstore = HStoreFormTest({'f1': '{"a": 1}'})
        self.assertTrue(form_w_hstore.has_changed())

        form_w_hstore = HStoreFormTest({'f1': '{"a": 1}'}, initial={'f1': '{"a": 1}'})
        self.assertFalse(form_w_hstore.has_changed())

        form_w_hstore = HStoreFormTest({'f1': '{"a": 2}'}, initial={'f1': '{"a": 1}'})
        self.assertTrue(form_w_hstore.has_changed())

        form_w_hstore = HStoreFormTest({'f1': '{"a": 1}'}, initial={'f1': {"a": 1}})
        self.assertFalse(form_w_hstore.has_changed())

        form_w_hstore = HStoreFormTest({'f1': '{"a": 2}'}, initial={'f1': {"a": 1}})
        self.assertTrue(form_w_hstore.has_changed())


class TestValidator(PostgreSQLSimpleTestCase):

    def test_simple_valid(self):
        validator = KeysValidator(keys=['a', 'b'])
        validator({'a': 'foo', 'b': 'bar', 'c': 'baz'})

    def test_missing_keys(self):
        validator = KeysValidator(keys=['a', 'b'])
        with self.assertRaises(exceptions.ValidationError) as cm:
            validator({'a': 'foo', 'c': 'baz'})
        self.assertEqual(cm.exception.messages[0], 'Some keys were missing: b')
        self.assertEqual(cm.exception.code, 'missing_keys')

    def test_strict_valid(self):
        validator = KeysValidator(keys=['a', 'b'], strict=True)
        validator({'a': 'foo', 'b': 'bar'})

    def test_extra_keys(self):
        validator = KeysValidator(keys=['a', 'b'], strict=True)
        with self.assertRaises(exceptions.ValidationError) as cm:
            validator({'a': 'foo', 'b': 'bar', 'c': 'baz'})
        self.assertEqual(cm.exception.messages[0], 'Some unknown keys were provided: c')
        self.assertEqual(cm.exception.code, 'extra_keys')

    def test_custom_messages(self):
        messages = {
            'missing_keys': 'Foobar',
        }
        validator = KeysValidator(keys=['a', 'b'], strict=True, messages=messages)
        with self.assertRaises(exceptions.ValidationError) as cm:
            validator({'a': 'foo', 'c': 'baz'})
        self.assertEqual(cm.exception.messages[0], 'Foobar')
        self.assertEqual(cm.exception.code, 'missing_keys')
        with self.assertRaises(exceptions.ValidationError) as cm:
            validator({'a': 'foo', 'b': 'bar', 'c': 'baz'})
        self.assertEqual(cm.exception.messages[0], 'Some unknown keys were provided: c')
        self.assertEqual(cm.exception.code, 'extra_keys')

    def test_deconstruct(self):
        messages = {
            'missing_keys': 'Foobar',
        }
        validator = KeysValidator(keys=['a', 'b'], strict=True, messages=messages)
        path, args, kwargs = validator.deconstruct()
        self.assertEqual(path, 'django.contrib.postgres.validators.KeysValidator')
        self.assertEqual(args, ())
        self.assertEqual(kwargs, {'keys': ['a', 'b'], 'strict': True, 'messages': messages})