Serialization: serialize types for import/export

2 files changed, 83 insertions, 0 deletions

diff --git a/ishtar_common/serializers.py b/ishtar_common/serializers.py
index d7d998e96..31caa8cb7 100644
--- a/ishtar_common/serializers.py
+++ b/ishtar_common/serializers.py
@@ -1,6 +1,64 @@
+import datetime
+import os
+import tempfile
 from rest_framework import serializers
+from zipfile import ZipFile
+
+from django.apps import apps
+from django.core.serializers import serialize
+
+from . import models
 
 
 class PublicSerializer(serializers.BaseSerializer):
     def to_representation(self, obj):
         return obj.public_representation()
+
+
+def type_model_serialization(archive=False, return_empty_types=False,
+                             archive_name=None):
+    """
+    Serialize all types models to JSON
+    Used for import and export scripts
+
+    :param return_empty_types: if True instead of serialization return empty
+    types (default False)
+    :param archive: if True return a zip file containing all the file serialized
+    (defaukt False)
+    :return: string containing the json serialization of types unless
+    return_empty_types or archive is set to True
+    """
+    if archive and return_empty_types:
+        raise ValueError("archive and return_empty_types are incompatible")
+    result = {}
+    for model in apps.get_models():
+        if not isinstance(model(), models.GeneralType):
+            continue
+        model_name = model.__name__
+        model_name = str(model.__module__).split(".")[0] + "__" + model_name
+        result[model_name] = serialize(
+            "json", model.objects.all(),
+            indent=2,
+            use_natural_foreign_keys=True, use_natural_primary_keys=True
+        )
+    if return_empty_types:
+        return [k for k in result if not result[k]]
+    if not archive:
+        return result
+    if not archive_name:
+        tmpdir = tempfile.mkdtemp(prefix="ishtarexport-") + os.sep
+        archive_name = tmpdir + "ishtar-{}.zip".format(
+            datetime.date.today().strftime("%Y-%m-%d")
+        )
+    if not archive_name.endswith(".zip"):
+        archive_name += ".zip"
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        with ZipFile(archive_name, 'w') as current_zip:
+            for model_name in result:
+                base_filename = model_name + ".json"
+                filename = tmpdirname + os.sep + base_filename
+                with open(filename, "w") as json_file:
+                    json_file.write(result[model_name])
+                current_zip.write(filename, arcname=base_filename)
+    return archive_name
+
diff --git a/ishtar_common/tests.py b/ishtar_common/tests.py
index b056359ec..3289ad8f6 100644
--- a/ishtar_common/tests.py
+++ b/ishtar_common/tests.py
@@ -20,11 +20,13 @@
 from bs4 import BeautifulSoup as Soup
 import csv
 import datetime
+import importlib
 import io
 import json
 import os
 import shutil
 from io import StringIO
+import zipfile
 
 from django.apps import apps
 
@@ -46,6 +48,7 @@ from django.test.runner import DiscoverRunner
 from ishtar_common import models
 from ishtar_common import views
 from ishtar_common.apps import admin_site
+from ishtar_common.serializers import type_model_serialization
 from ishtar_common.utils import post_save_geo, update_data, move_dict_data, \
     rename_and_simplify_media_name, try_fix_file
 
@@ -544,6 +547,28 @@ class CacheTest(TestCase):
         self.assertIn(b'href="/operation_modification/', response.content)
 
 
+class SerializationTest(TestCase):
+    fixtures = [settings.ROOT_PATH +
+                '../fixtures/initial_data-auth-fr.json',
+                settings.ROOT_PATH +
+                '../ishtar_common/fixtures/initial_data-fr.json',]
+
+    def test_type_serialization(self):
+        json_result = type_model_serialization()
+        for k in json_result.keys():
+            module_name, model_name = k.split("__")
+            module = importlib.import_module(module_name + ".models")
+            model = getattr(module, model_name)
+            self.assertEqual(model.objects.count(),
+                             len(json.loads(json_result[k])))
+
+    def test_type_serialization_zip(self):
+        # only check the validity of the zip, the content is tested above
+        zip_filename = type_model_serialization(archive=True)
+        with zipfile.ZipFile(zip_filename, "r") as zip_file:
+            self.assertIsNone(zip_file.testzip())
+
+
 class AccessControlTest(TestCase):
     def test_administrator(self):
         admin, created = models.PersonType.objects.get_or_create(