Imports: manage model limitation (don't create items not in the list)

1 files changed, 86 insertions, 32 deletions

diff --git a/archaeological_operations/tests.py b/archaeological_operations/tests.py
index d4134693f..f2126a68e 100644
--- a/archaeological_operations/tests.py
+++ b/archaeological_operations/tests.py
@@ -35,7 +35,7 @@ import models
 from archaeological_operations import views
 
 from ishtar_common.models import OrganizationType, Organization, \
-    ImporterType, IshtarUser, TargetKey, IshtarSiteProfile, Town
+    ImporterType, IshtarUser, TargetKey, ImporterModel, IshtarSiteProfile, Town
 
 from ishtar_common import forms_common
 from ishtar_common.tests import WizardTest, WizardTestFormData as FormData, \
@@ -65,41 +65,22 @@ class ImportOperationTest(TestCase):
         self.username, self.password, self.user = create_superuser()
         self.ishtar_user = IshtarUser.objects.get(pk=self.user.pk)
 
-    def testMCCImportOperation(self, test=True):
-        # MCC opérations
-        if self.test_operations is False:
-            test = False
-        first_ope_nb = models.Operation.objects.count()
-        MCC_OPERATION = ImporterType.objects.get(name=u"MCC - Opérations")
+    def init_ope_import(self):
+        mcc_operation = ImporterType.objects.get(name=u"MCC - Opérations")
         mcc_operation_file = open(
             settings.ROOT_PATH +
             '../archaeological_operations/tests/MCC-operations-example.csv',
             'rb')
         file_dict = {'imported_file': SimpleUploadedFile(
             mcc_operation_file.name, mcc_operation_file.read())}
-        post_dict = {'importer_type': MCC_OPERATION.pk, 'skip_lines': 1,
+        post_dict = {'importer_type': mcc_operation.pk, 'skip_lines': 1,
                      "encoding": 'utf-8'}
-        form = forms_common.NewImportForm(data=post_dict, files=file_dict,
-                                          instance=None)
+        form = forms_common.NewImportForm(data=post_dict, files=file_dict)
         form.is_valid()
-        if test:
-            self.assertTrue(form.is_valid())
-        impt = form.save(self.ishtar_user)
-        target_key_nb = TargetKey.objects.count()
-        impt.initialize()
-        # new key have to be set
-        if test:
-            self.assertTrue(TargetKey.objects.count() > target_key_nb)
+        return mcc_operation, form
 
-        # first try to import
-        impt.importation()
-        current_ope_nb = models.Operation.objects.count()
-        # no new operation imported because of a missing connection for
-        # operation_type value
-        if test:
-            self.assertTrue(current_ope_nb == first_ope_nb)
-
-        # doing manualy connections
+    def init_ope_targetkey(self, imp):
+        # doing manually connections
         tg = TargetKey.objects.filter(target__target='operation_type'
                                       ).order_by('-pk').all()[0]
         tg.value = models.OperationType.objects.get(
@@ -107,18 +88,44 @@ class ImportOperationTest(TestCase):
         tg.is_set = True
         tg.save()
 
-        target = TargetKey.objects.get(key='gallo-romain')
+        target = TargetKey.objects.get(key='gallo-romain',
+                                       associated_import=imp)
         gallo = models.Period.objects.get(txt_idx='gallo-roman')
         target.value = gallo.pk
         target.is_set = True
         target.save()
 
-        target = TargetKey.objects.get(key='age-du-fer')
+        target = TargetKey.objects.get(key='age-du-fer',
+                                       associated_import=imp)
         iron = models.Period.objects.get(txt_idx='iron_age')
         target.value = iron.pk
         target.is_set = True
         target.save()
 
+    def test_mcc_import_operation(self, test=True):
+        # MCC opérations
+        if self.test_operations is False:
+            test = False
+        first_ope_nb = models.Operation.objects.count()
+        importer, form = self.init_ope_import()
+        if test:
+            self.assertTrue(form.is_valid())
+        impt = form.save(self.ishtar_user)
+        target_key_nb = TargetKey.objects.count()
+        impt.initialize()
+        # new key have to be set
+        if test:
+            self.assertTrue(TargetKey.objects.count() > target_key_nb)
+
+        # first try to import
+        impt.importation()
+        current_ope_nb = models.Operation.objects.count()
+        # no new operation imported because of a missing connection for
+        # operation_type value
+        if test:
+            self.assertTrue(current_ope_nb == first_ope_nb)
+        self.init_ope_targetkey(imp=impt)
+
         impt.importation()
         if not test:
             return
@@ -131,8 +138,9 @@ class ImportOperationTest(TestCase):
         self.assertTrue(last_ope.code_patriarche == 4200)
         self.assertTrue(last_ope.operation_type.txt_idx == 'prog_excavation')
         self.assertEqual(last_ope.periods.count(), 2)
-        periods = last_ope.periods.all()
-        self.assertTrue(iron in periods and gallo in periods)
+        periods = [period.txt_idx for period in last_ope.periods.all()]
+        self.assertIn('iron_age', periods)
+        self.assertIn('gallo-roman', periods)
 
         # a second importation will be not possible: no two same patriarche
         # code
@@ -141,10 +149,56 @@ class ImportOperationTest(TestCase):
         self.assertTrue(last_ope ==
                         models.Operation.objects.order_by('-pk').all()[0])
 
+    def test_model_limitation(self):
+        importer, form = self.init_ope_import()
+        importer.created_models.clear()
+        impt = form.save(self.ishtar_user)
+        impt.initialize()
+        self.init_ope_targetkey(imp=impt)
+
+        # no model defined in created_models: normal import
+        init_ope_number = models.Operation.objects.count()
+        impt.importation()
+        current_ope_nb = models.Operation.objects.count()
+        self.assertEqual(current_ope_nb, init_ope_number + 1)
+
+        last_ope = models.Operation.objects.order_by('-pk').all()[0]
+        last_ope.delete()
+
+        importer, form = self.init_ope_import()
+        # add an inadequate model to make created_models non empty
+        importer.created_models.clear()
+        importer.created_models.add(ImporterModel.objects.get(
+            klass='ishtar_common.models.Organization'
+        ))
+        impt = form.save(self.ishtar_user)
+        impt.initialize()
+        self.init_ope_targetkey(imp=impt)
+
+        # no imports
+        impt.importation()
+        current_ope_nb = models.Operation.objects.count()
+        self.assertEqual(current_ope_nb, init_ope_number)
+
+        importer, form = self.init_ope_import()
+        # add operation model to allow creation
+        importer.created_models.clear()
+        importer.created_models.add(ImporterModel.objects.get(
+            klass='archaeological_operations.models.Operation'
+        ))
+        impt = form.save(self.ishtar_user)
+        impt.initialize()
+        self.init_ope_targetkey(imp=impt)
+
+        # import of operations
+        impt.importation()
+        current_ope_nb = models.Operation.objects.count()
+        self.assertEqual(current_ope_nb, init_ope_number + 1)
+
     def testMCCImportParcels(self, test=True):
         if self.test_operations is False:
             test = False
-        self.testMCCImportOperation(test=False)
+        self.test_mcc_import_operation(test=False)
         old_nb = models.Parcel.objects.count()
         MCC_PARCEL = ImporterType.objects.get(name=u"MCC - Parcelles")
         mcc_file = open(