django/tests/serializers/tests.py

from datetime import datetime
from functools import partialmethod
from io import StringIO
from unittest import mock, skipIf

from django.core import serializers
from django.core.serializers import SerializerDoesNotExist
from django.core.serializers.base import ProgressBar
from django.db import connection, transaction
from django.http import HttpResponse
from django.test import SimpleTestCase, override_settings, skipUnlessDBFeature
from django.test.utils import Approximate

from .models import (
    Actor,
    Article,
    Author,
    AuthorProfile,
    BaseModel,
    Category,
    Child,
    ComplexModel,
    Movie,
    Player,
    ProxyBaseModel,
    ProxyProxyBaseModel,
    Score,
    Team,
)


@override_settings(
    SERIALIZATION_MODULES={
        "json2": "django.core.serializers.json",
    }
)
class SerializerRegistrationTests(SimpleTestCase):
    def setUp(self):
        self.old_serializers = serializers._serializers
        serializers._serializers = {}

    def tearDown(self):
        serializers._serializers = self.old_serializers

    def test_register(self):
        "Registering a new serializer populates the full registry. Refs #14823"
        serializers.register_serializer("json3", "django.core.serializers.json")

        public_formats = serializers.get_public_serializer_formats()
        self.assertIn("json3", public_formats)
        self.assertIn("json2", public_formats)
        self.assertIn("xml", public_formats)

    def test_unregister(self):
        """
        Unregistering a serializer doesn't cause the registry to be
        repopulated.
        """
        serializers.unregister_serializer("xml")
        serializers.register_serializer("json3", "django.core.serializers.json")

        public_formats = serializers.get_public_serializer_formats()

        self.assertNotIn("xml", public_formats)
        self.assertIn("json3", public_formats)

    def test_unregister_unknown_serializer(self):
        with self.assertRaises(SerializerDoesNotExist):
            serializers.unregister_serializer("nonsense")

    def test_builtin_serializers(self):
        "Requesting a list of serializer formats populates the registry"
        all_formats = set(serializers.get_serializer_formats())
        public_formats = set(serializers.get_public_serializer_formats())

        self.assertIn("xml", all_formats)
        self.assertIn("xml", public_formats)

        self.assertIn("json2", all_formats)
        self.assertIn("json2", public_formats)

        self.assertIn("python", all_formats)
        self.assertNotIn("python", public_formats)

    def test_get_unknown_serializer(self):
        """
        #15889: get_serializer('nonsense') raises a SerializerDoesNotExist
        """
        with self.assertRaises(SerializerDoesNotExist):
            serializers.get_serializer("nonsense")

        with self.assertRaises(KeyError):
            serializers.get_serializer("nonsense")

        # SerializerDoesNotExist is instantiated with the nonexistent format
        with self.assertRaisesMessage(SerializerDoesNotExist, "nonsense"):
            serializers.get_serializer("nonsense")

    def test_get_unknown_deserializer(self):
        with self.assertRaises(SerializerDoesNotExist):
            serializers.get_deserializer("nonsense")


class SerializersTestBase:
    serializer_name = None  # Set by subclasses to the serialization format name

    @classmethod
    def setUpTestData(cls):
        sports = Category.objects.create(name="Sports")
        music = Category.objects.create(name="Music")
        op_ed = Category.objects.create(name="Op-Ed")

        cls.joe = Author.objects.create(name="Joe")
        cls.jane = Author.objects.create(name="Jane")

        cls.a1 = Article(
            author=cls.jane,
            headline="Poker has no place on ESPN",
            pub_date=datetime(2006, 6, 16, 11, 00),
        )
        cls.a1.save()
        cls.a1.categories.set([sports, op_ed])

        cls.a2 = Article(
            author=cls.joe,
            headline="Time to reform copyright",
            pub_date=datetime(2006, 6, 16, 13, 00, 11, 345),
        )
        cls.a2.save()
        cls.a2.categories.set([music, op_ed])

    def test_serialize(self):
        """Basic serialization works."""
        serial_str = serializers.serialize(self.serializer_name, Article.objects.all())
        self.assertTrue(self._validate_output(serial_str))

    def test_serializer_roundtrip(self):
        """Serialized content can be deserialized."""
        serial_str = serializers.serialize(self.serializer_name, Article.objects.all())
        models = list(serializers.deserialize(self.serializer_name, serial_str))
        self.assertEqual(len(models), 2)

    def test_serialize_to_stream(self):
        obj = ComplexModel(field1="first", field2="second", field3="third")
        obj.save_base(raw=True)

        # Serialize the test database to a stream
        for stream in (StringIO(), HttpResponse()):
            serializers.serialize(self.serializer_name, [obj], indent=2, stream=stream)

            # Serialize normally for a comparison
            string_data = serializers.serialize(self.serializer_name, [obj], indent=2)

            # The two are the same
            if isinstance(stream, StringIO):
                self.assertEqual(string_data, stream.getvalue())
            else:
                self.assertEqual(string_data, stream.content.decode())

    def test_serialize_specific_fields(self):
        obj = ComplexModel(field1="first", field2="second", field3="third")
        obj.save_base(raw=True)

        # Serialize then deserialize the test database
        serialized_data = serializers.serialize(
            self.serializer_name, [obj], indent=2, fields=("field1", "field3")
        )
        result = next(serializers.deserialize(self.serializer_name, serialized_data))

        # The deserialized object contains data in only the serialized fields.
        self.assertEqual(result.object.field1, "first")
        self.assertEqual(result.object.field2, "")
        self.assertEqual(result.object.field3, "third")

    def test_altering_serialized_output(self):
        """
        The ability to create new objects by modifying serialized content.
        """
        old_headline = "Poker has no place on ESPN"
        new_headline = "Poker has no place on television"
        serial_str = serializers.serialize(self.serializer_name, Article.objects.all())
        serial_str = serial_str.replace(old_headline, new_headline)
        models = list(serializers.deserialize(self.serializer_name, serial_str))

        # Prior to saving, old headline is in place
        self.assertTrue(Article.objects.filter(headline=old_headline))
        self.assertFalse(Article.objects.filter(headline=new_headline))

        for model in models:
            model.save()

        # After saving, new headline is in place
        self.assertTrue(Article.objects.filter(headline=new_headline))
        self.assertFalse(Article.objects.filter(headline=old_headline))

    def test_one_to_one_as_pk(self):
        """
        If you use your own primary key field (such as a OneToOneField), it
        doesn't appear in the serialized field list - it replaces the pk
        identifier.
        """
        AuthorProfile.objects.create(
            author=self.joe, date_of_birth=datetime(1970, 1, 1)
        )
        serial_str = serializers.serialize(
            self.serializer_name, AuthorProfile.objects.all()
        )
        self.assertFalse(self._get_field_values(serial_str, "author"))

        for obj in serializers.deserialize(self.serializer_name, serial_str):
            self.assertEqual(obj.object.pk, self.joe.pk)

    def test_serialize_field_subset(self):
        """Output can be restricted to a subset of fields"""
        valid_fields = ("headline", "pub_date")
        invalid_fields = ("author", "categories")
        serial_str = serializers.serialize(
            self.serializer_name, Article.objects.all(), fields=valid_fields
        )
        for field_name in invalid_fields:
            self.assertFalse(self._get_field_values(serial_str, field_name))

        for field_name in valid_fields:
            self.assertTrue(self._get_field_values(serial_str, field_name))

    def test_serialize_unicode_roundtrip(self):
        """Unicode makes the roundtrip intact"""
        actor_name = "Za\u017c\u00f3\u0142\u0107"
        movie_title = "G\u0119\u015bl\u0105 ja\u017a\u0144"
        ac = Actor(name=actor_name)
        mv = Movie(title=movie_title, actor=ac)
        ac.save()
        mv.save()

        serial_str = serializers.serialize(self.serializer_name, [mv])
        self.assertEqual(self._get_field_values(serial_str, "title")[0], movie_title)
        self.assertEqual(self._get_field_values(serial_str, "actor")[0], actor_name)

        obj_list = list(serializers.deserialize(self.serializer_name, serial_str))
        mv_obj = obj_list[0].object
        self.assertEqual(mv_obj.title, movie_title)

    def test_unicode_serialization(self):
        unicode_name = "יוניקוד"
        data = serializers.serialize(self.serializer_name, [Author(name=unicode_name)])
        self.assertIn(unicode_name, data)
        objs = list(serializers.deserialize(self.serializer_name, data))
        self.assertEqual(objs[0].object.name, unicode_name)

    def test_serialize_progressbar(self):
        fake_stdout = StringIO()
        serializers.serialize(
            self.serializer_name,
            Article.objects.all(),
            progress_output=fake_stdout,
            object_count=Article.objects.count(),
        )
        self.assertTrue(
            fake_stdout.getvalue().endswith(
                "[" + "." * ProgressBar.progress_width + "]\n"
            )
        )

    def test_serialize_superfluous_queries(self):
        """Ensure no superfluous queries are made when serializing ForeignKeys

        #17602
        """
        ac = Actor(name="Actor name")
        ac.save()
        mv = Movie(title="Movie title", actor_id=ac.pk)
        mv.save()

        with self.assertNumQueries(0):
            serializers.serialize(self.serializer_name, [mv])

    def test_serialize_prefetch_related_m2m(self):
        # One query for the Article table and one for each prefetched m2m
        # field.
        with self.assertNumQueries(4):
            serializers.serialize(
                self.serializer_name,
                Article.objects.prefetch_related("categories", "meta_data", "topics"),
            )
        # One query for the Article table, and three m2m queries for each
        # article.
        with self.assertNumQueries(7):
            serializers.serialize(self.serializer_name, Article.objects.all())

    def test_serialize_with_null_pk(self):
        """
        Serialized data with no primary key results
        in a model instance with no id
        """
        category = Category(name="Reference")
        serial_str = serializers.serialize(self.serializer_name, [category])
        pk_value = self._get_pk_values(serial_str)[0]
        self.assertFalse(pk_value)

        cat_obj = list(serializers.deserialize(self.serializer_name, serial_str))[
            0
        ].object
        self.assertIsNone(cat_obj.id)

    def test_float_serialization(self):
        """Float values serialize and deserialize intact"""
        sc = Score(score=3.4)
        sc.save()
        serial_str = serializers.serialize(self.serializer_name, [sc])
        deserial_objs = list(serializers.deserialize(self.serializer_name, serial_str))
        self.assertEqual(deserial_objs[0].object.score, Approximate(3.4, places=1))

    def test_deferred_field_serialization(self):
        author = Author.objects.create(name="Victor Hugo")
        author = Author.objects.defer("name").get(pk=author.pk)
        serial_str = serializers.serialize(self.serializer_name, [author])
        deserial_objs = list(serializers.deserialize(self.serializer_name, serial_str))
        self.assertIsInstance(deserial_objs[0].object, Author)

    def test_custom_field_serialization(self):
        """Custom fields serialize and deserialize intact"""
        team_str = "Spartak Moskva"
        player = Player()
        player.name = "Soslan Djanaev"
        player.rank = 1
        player.team = Team(team_str)
        player.save()
        serial_str = serializers.serialize(self.serializer_name, Player.objects.all())
        team = self._get_field_values(serial_str, "team")
        self.assertTrue(team)
        self.assertEqual(team[0], team_str)

        deserial_objs = list(serializers.deserialize(self.serializer_name, serial_str))
        self.assertEqual(
            deserial_objs[0].object.team.to_string(), player.team.to_string()
        )

    def test_pre_1000ad_date(self):
        """Year values before 1000AD are properly formatted"""
        # Regression for #12524 -- dates before 1000AD get prefixed
        # 0's on the year
        a = Article.objects.create(
            author=self.jane,
            headline="Nobody remembers the early years",
            pub_date=datetime(1, 2, 3, 4, 5, 6),
        )

        serial_str = serializers.serialize(self.serializer_name, [a])
        date_values = self._get_field_values(serial_str, "pub_date")
        self.assertEqual(date_values[0].replace("T", " "), "0001-02-03 04:05:06")

    def test_pkless_serialized_strings(self):
        """
        Serialized strings without PKs can be turned into models
        """
        deserial_objs = list(
            serializers.deserialize(self.serializer_name, self.pkless_str)
        )
        for obj in deserial_objs:
            self.assertFalse(obj.object.id)
            obj.save()
        self.assertEqual(Category.objects.count(), 5)

    def test_deterministic_mapping_ordering(self):
        """Mapping such as fields should be deterministically ordered. (#24558)"""
        output = serializers.serialize(self.serializer_name, [self.a1], indent=2)
        categories = self.a1.categories.values_list("pk", flat=True)
        self.assertEqual(
            output,
            self.mapping_ordering_str
            % {
                "article_pk": self.a1.pk,
                "author_pk": self.a1.author_id,
                "first_category_pk": categories[0],
                "second_category_pk": categories[1],
            },
        )

    def test_deserialize_force_insert(self):
        """Deserialized content can be saved with force_insert as a parameter."""
        serial_str = serializers.serialize(self.serializer_name, [self.a1])
        deserial_obj = list(serializers.deserialize(self.serializer_name, serial_str))[
            0
        ]
        with mock.patch("django.db.models.Model") as mock_model:
            deserial_obj.save(force_insert=False)
            mock_model.save_base.assert_called_with(
                deserial_obj.object, raw=True, using=None, force_insert=False
            )

    @skipUnlessDBFeature("can_defer_constraint_checks")
    def test_serialize_proxy_model(self):
        BaseModel.objects.create(parent_data=1)
        base_objects = BaseModel.objects.all()
        proxy_objects = ProxyBaseModel.objects.all()
        proxy_proxy_objects = ProxyProxyBaseModel.objects.all()
        base_data = serializers.serialize("json", base_objects)
        proxy_data = serializers.serialize("json", proxy_objects)
        proxy_proxy_data = serializers.serialize("json", proxy_proxy_objects)
        self.assertEqual(base_data, proxy_data.replace("proxy", ""))
        self.assertEqual(base_data, proxy_proxy_data.replace("proxy", ""))

    def test_serialize_inherited_fields(self):
        child_1 = Child.objects.create(parent_data="a", child_data="b")
        child_2 = Child.objects.create(parent_data="c", child_data="d")
        child_1.parent_m2m.add(child_2)
        child_data = serializers.serialize(self.serializer_name, [child_1, child_2])
        self.assertEqual(self._get_field_values(child_data, "parent_m2m"), [])
        self.assertEqual(self._get_field_values(child_data, "parent_data"), [])

    def test_serialize_only_pk(self):
        with self.assertNumQueries(7) as ctx:
            serializers.serialize(
                self.serializer_name,
                Article.objects.all(),
                use_natural_foreign_keys=False,
            )

        categories_sql = ctx[1]["sql"]
        self.assertNotIn(connection.ops.quote_name("meta_data_id"), categories_sql)
        meta_data_sql = ctx[2]["sql"]
        self.assertNotIn(connection.ops.quote_name("kind"), meta_data_sql)
        topics_data_sql = ctx[3]["sql"]
        self.assertNotIn(connection.ops.quote_name("category_id"), topics_data_sql)

    def test_serialize_no_only_pk_with_natural_keys(self):
        with self.assertNumQueries(7) as ctx:
            serializers.serialize(
                self.serializer_name,
                Article.objects.all(),
                use_natural_foreign_keys=True,
            )

        categories_sql = ctx[1]["sql"]
        self.assertNotIn(connection.ops.quote_name("meta_data_id"), categories_sql)
        # CategoryMetaData has natural_key().
        meta_data_sql = ctx[2]["sql"]
        self.assertIn(connection.ops.quote_name("kind"), meta_data_sql)
        topics_data_sql = ctx[3]["sql"]
        self.assertNotIn(connection.ops.quote_name("category_id"), topics_data_sql)


class SerializerAPITests(SimpleTestCase):
    def test_stream_class(self):
        class File:
            def __init__(self):
                self.lines = []

            def write(self, line):
                self.lines.append(line)

            def getvalue(self):
                return "".join(self.lines)

        class Serializer(serializers.json.Serializer):
            stream_class = File

        serializer = Serializer()
        data = serializer.serialize([Score(id=1, score=3.4)])
        self.assertIs(serializer.stream_class, File)
        self.assertIsInstance(serializer.stream, File)
        self.assertEqual(
            data, '[{"model": "serializers.score", "pk": 1, "fields": {"score": 3.4}}]'
        )


class SerializersTransactionTestBase:
    available_apps = ["serializers"]

    @skipUnlessDBFeature("supports_forward_references")
    def test_forward_refs(self):
        """
        Objects ids can be referenced before they are
        defined in the serialization data.
        """
        # The deserialization process needs to run in a transaction in order
        # to test forward reference handling.
        with transaction.atomic():
            objs = serializers.deserialize(self.serializer_name, self.fwd_ref_str)
            with connection.constraint_checks_disabled():
                for obj in objs:
                    obj.save()

        for model_cls in (Category, Author, Article):
            self.assertEqual(model_cls.objects.count(), 1)
        art_obj = Article.objects.all()[0]
        self.assertEqual(art_obj.categories.count(), 1)
        self.assertEqual(art_obj.author.name, "Agnes")


def register_tests(test_class, method_name, test_func, exclude=()):
    """
    Dynamically create serializer tests to ensure that all registered
    serializers are automatically tested.
    """
    for format_ in serializers.get_serializer_formats():
        if format_ == "geojson" or format_ in exclude:
            continue
        decorated_func = skipIf(
            isinstance(serializers.get_serializer(format_), serializers.BadSerializer),
            "The Python library for the %s serializer is not installed." % format_,
        )(test_func)
        setattr(
            test_class, method_name % format_, partialmethod(decorated_func, format_)
        )