wendelin_telecom_test: Add constraint tests

bt5/wendelin_telecom_test/TestTemplateItem/portal_components/test.erp5.testWendelinTelecom.py

@@ -83,26 +83,29 @@ class WendelinTelecomTest(SecurityTestCase):
   def beforeTearDown(self):
     self.abort()
 
-    # Clean up all test items
-    for module, portal_type in (
-      (self.portal.project_module, 'Project'),
-      (self.portal.person_module, 'Person'),
-      (self.portal.data_acquisition_unit_module, 'Data Acquisition Unit'),
-      (self.portal.data_supply_module, 'Data Supply'),
-      (self.portal.data_stream_module, 'Data Stream'),
-      (self.portal.data_ingestion_module, 'Data Ingestion'),
-      (self.portal.data_analysis_module, 'Data Analysis'),
-      (self.portal.data_array_module, 'Data Array'),
-    ):
-      object_list = module.objectValues(portal_type=portal_type)
-      if object_list:
-        test_object_id_list = [
-          obj.getId() for obj in object_list \
-          if ('test' in obj.getReference().lower() and 'default' not in obj.getId())
-        ]
-        if test_object_id_list:
-          module.manage_delObjects(ids=test_object_id_list)
-    self.tic()
+    # Manually change this to False to keep testing data
+    cleanup_data = True
+    if cleanup_data:
+      # Clean up all test items
+      for module, portal_type in (
+        (self.portal.project_module, 'Project'),
+        (self.portal.person_module, 'Person'),
+        (self.portal.data_acquisition_unit_module, 'Data Acquisition Unit'),
+        (self.portal.data_supply_module, 'Data Supply'),
+        (self.portal.data_stream_module, 'Data Stream'),
+        (self.portal.data_ingestion_module, 'Data Ingestion'),
+        (self.portal.data_analysis_module, 'Data Analysis'),
+        (self.portal.data_array_module, 'Data Array'),
+      ):
+        object_list = module.objectValues(portal_type=portal_type)
+        if object_list:
+          test_object_id_list = [
+            obj.getId() for obj in object_list \
+            if ('test' in obj.getReference().lower() and 'default' not in obj.getId())
+          ]
+          if test_object_id_list:
+            module.manage_delObjects(ids=test_object_id_list)
+      self.tic()
 
   def _removeDocument(self, document_to_remove):
     path = document_to_remove.getRelativeUrl()
@@ -134,7 +137,13 @@ class WendelinTelecomTest(SecurityTestCase):
 
     return user
 
-  def registerOrs(self, tag_hostname_seed=None, tag_comp_id_seed=None, tag_enb_id_seed=None):
+  def registerOrs(
+    self,
+    tag_hostname_seed=None,
+    tag_comp_id_seed=None,
+    tag_enb_id_seed=None,
+    test_suffix=True
+  ):
     # Create a Data Acquisition Unit and related Data Supply
     # with a tag constructed from the provided seeds.
     # If any seed is NOT defined, it is generated at random.
@@ -151,7 +160,9 @@ class WendelinTelecomTest(SecurityTestCase):
       ors_tag += '_COMP-%s' % tag_comp_id_seed
     if tag_enb_id_seed != '':
       ors_tag += '_e0x%s' % tag_enb_id_seed
-    ors_tag += 'Test'
+    # If test suffix is false, manual cleanup will be needed!
+    if test_suffix:
+      ors_tag += 'Test'
 
     response = self.portal.ERP5Site_registerOrs(ors_tag)
     self.tic()
@@ -1229,7 +1240,235 @@ class WendelinTelecomTest(SecurityTestCase):
       entity_tag
     )
 
-  def test_05_1_updateOrsConfigurationsFastInput(self, username=None):
+  def test_05_1_checkOrsItemConsistency(self):
+    '''
+    Test the Wendelin Telecom constraints defined on ORS Data Acquisition Units and Data Supplies.
+    Check all cases of both consistency and inconsistency.
+    '''
+
+    # Manually create a Data Acquisition Unit without a reference
+    # No need to validate it
+    inconsistent_data_acquisition_unit = self.portal.data_acquisition_unit_module.newContent(
+      portal_type='Data Acquisition Unit'
+    )
+    self.tic()
+    self.addCleanup(self._removeDocument, inconsistent_data_acquisition_unit)
+
+    # Check that there is a consistency error: no reference was found
+    consistency_message_list = self.getConsistencyMessageList(inconsistent_data_acquisition_unit)
+    self.assertTrue(len(consistency_message_list) == 1)
+    self.assertTrue(consistency_message_list[0] == 'Reference has not been set')
+
+    # Now, successively set invalid ORS tags as the reference
+    # and check that it is detected as invalid
+    invalid_tag_list = [
+      'ors123',
+      '_COMP-234_e0xABCDE',
+      'ors123_COMP-_e0xABCDE',
+      'ors123_COMP-456_e0xA'
+    ]
+    for invalid_tag in invalid_tag_list:
+      inconsistent_data_acquisition_unit.setReference(invalid_tag)
+      self.tic()
+      consistency_message_list = self.getConsistencyMessageList(inconsistent_data_acquisition_unit)
+      self.assertTrue(len(consistency_message_list) == 1)
+      self.assertTrue(
+        consistency_message_list[0] == "Reference '%s' is not a valid ORS tag" % invalid_tag
+      )
+
+    # Next, manually create a Data Supply without a reference
+    # No need to validate it either, yet
+    inconsistent_data_supply = self.portal.data_supply_module.newContent(
+      portal_type='Data Supply'
+    )
+    self.tic()
+    self.addCleanup(self._removeDocument, inconsistent_data_supply)
+
+    # Check that there are several consistency errors
+    consistency_message_list = self.getConsistencyMessageList(inconsistent_data_supply)
+    self.assertTrue(len(consistency_message_list) == 2)
+    self.assertTrue(
+      consistency_message_list[0] == 'Data Supply is not related to a Data Acquisition Unit'
+    )
+    self.assertTrue(consistency_message_list[1] == 'Reference has not been set')
+
+    # Now, register an ORS with a valid tag
+    ors_item_dict = self.registerOrs(test_suffix=False)
+    self.addCleanup(self._removeDocument, ors_item_dict['data_acquisition_unit'])
+    self.addCleanup(self._removeDocument, ors_item_dict['data_supply'])
+
+    # Check that the created items are consistent (no error messages)
+    for item_key in ['data_acquisition_unit', 'data_supply']:
+      consistency_message_list = self.getConsistencyMessageList(ors_item_dict[item_key])
+      self.assertFalse(consistency_message_list)
+
+    # Now, change the reference of the Data Supply
+    ors_tag = ors_item_dict['data_acquisition_unit'].getReference()
+    new_radio_id = generateRandomString(length=5, hexadecimal=True)
+    while new_radio_id == ors_tag[20:]:
+      new_radio_id = generateRandomString(length=5, hexadecimal=True)
+    ors_item_dict['data_supply'].setReference(ors_tag[:20] + new_radio_id)
+    self.tic()
+
+    # Check that the Data Supply is now inconsistent
+    consistency_message_list = self.getConsistencyMessageList(ors_item_dict['data_supply'])
+    self.assertTrue(len(consistency_message_list) == 1)
+    self.assertTrue(
+      consistency_message_list[0] == \
+      "Reference does not match the associated Data Acquisition Unit's reference"
+    )
+
+    # Finally, invalidate the Data Supply
+    ors_item_dict['data_supply'].invalidate()
+    self.tic()
+
+    # Check that there is a new consistency error
+    consistency_message_list = self.getConsistencyMessageList(ors_item_dict['data_supply'])
+    self.assertTrue(len(consistency_message_list) == 2)
+    self.assertTrue(
+      consistency_message_list[0] == \
+      "Reference does not match the associated Data Acquisition Unit's reference"
+    )
+    self.assertTrue(
+      consistency_message_list[1] == \
+      "Validation state does not match that of the associated Data Acquisition Unit"
+    )
+
+    # Finally, register a generic eNB device with a non-ORS hostname in its tag
+    # This should be valid
+    enb_tag = 'edgepod1_COMP-1234_e0x1A2D0'
+    enb_data_acquisition_unit = self.portal.data_acquisition_unit_module.newContent(
+      portal_type='Data Acquisition Unit',
+      reference=enb_tag
+    )
+    enb_data_acquisition_unit.validate()
+    enb_data_supply = enb_data_acquisition_unit.DataAcquisitionUnit_createOrsDataSupply(batch=1)
+
+    # Check that the created items are consistent (no error messages)
+    for item in [enb_data_acquisition_unit, enb_data_supply]:
+      consistency_message_list = self.getConsistencyMessageList(item)
+      self.assertFalse(consistency_message_list)
+
+  def test_05_2_checkOrsIngestionItemConsistency(self):
+    '''
+    Test the Wendelin Telecom constraints defined on ORS ingestion items,
+    namely Data Ingestions and Data Analyses.
+    Check all cases of both consistency and inconsistency.
+    '''
+
+    # Register an ORS
+    ors_item_dict = self.registerOrs()
+
+    # Perform a data ingestion for the ORS
+    ingestion_item_dict = self.getOrsLogIngestionItems(
+      self.test_ors_example_log_empty,
+      ors_item_dict['data_acquisition_unit'].getReference()
+    )
+
+    # Check that the Data Ingestion and Data Analysis are consistent
+    for item_key in ['data_ingestion', 'data_analysis']:
+      consistency_message_list = self.getConsistencyMessageList(
+        ingestion_item_dict[item_key]
+      )
+      self.assertFalse(consistency_message_list)
+
+    # Invalidate the Data Supply
+    ors_item_dict['data_supply'].invalidate()
+    self.tic()
+
+    # Check that the Data Analysis (and by extension the Data Ingestion)
+    # now has a consistency error
+    consistency_message_list = self.getConsistencyMessageList(
+      ingestion_item_dict['data_analysis']
+    )
+    self.assertTrue(len(consistency_message_list) == 1)
+    self.assertTrue(
+      consistency_message_list[0] == "Simulation is started but should be stopped"
+    )
+
+    # Now, unlink the Data Supply from the Data Analysis
+    saved_specialise_list = ingestion_item_dict['data_analysis'].getSpecialiseValueList()
+    ingestion_item_dict['data_analysis'].setSpecialiseValueList([])
+    self.tic()
+
+    # Check that the Data Analysis has consistency errors
+    consistency_message_list = self.getConsistencyMessageList(
+      ingestion_item_dict['data_analysis']
+    )
+    self.assertTrue(len(consistency_message_list) == 2)
+    self.assertTrue(
+      consistency_message_list[0] == \
+      "Missing Data Supply: %s" % ors_item_dict['data_supply'].getRelativeUrl()
+    )
+    self.assertTrue(
+      consistency_message_list[1] == "Simulation should be delivered"
+    )
+
+    # Re-validate and relink the Data Supply and stop the Data Ingestion
+    ors_item_dict['data_supply'].validate()
+    ingestion_item_dict['data_analysis'].setSpecialiseValueList(saved_specialise_list)
+    ingestion_item_dict['data_ingestion'].stop()
+    self.tic()
+
+    # Check that the Data Ingestion has a consistency error
+    consistency_message_list = self.getConsistencyMessageList(
+      ingestion_item_dict['data_ingestion']
+    )
+    self.assertTrue(len(consistency_message_list) == 1)
+    self.assertTrue(
+      consistency_message_list[0] == "Simulation is stopped but should be started"
+    )
+
+    # Restart the ingestion
+    ingestion_item_dict['data_ingestion'].start()
+
+    # Register another ORS and link the Data Supply to the ingestion
+    new_ors_item_dict = self.registerOrs()
+    ingestion_item_dict['data_ingestion'].setSpecialiseValueList([
+      ors_item_dict['data_supply'],
+      new_ors_item_dict['data_supply']
+    ])
+    self.tic()
+
+    # Check that the Data Ingestion has another consistency error
+    consistency_message_list = self.getConsistencyMessageList(
+      ingestion_item_dict['data_ingestion']
+    )
+    self.assertTrue(len(consistency_message_list) == 1)
+    self.assertTrue(
+      consistency_message_list[0] == "More than one Data Supply linked to Data Delivery"
+    )
+
+    # Finally, remove the aggregate from the Data Analysis's last line
+    data_analysis_line_list = ingestion_item_dict['data_analysis'].contentValues(
+      portal_type='Data Analysis Line'
+    )
+    data_analysis_last_line = data_analysis_line_list[-1]
+    saved_aggregate_list = data_analysis_last_line.getAggregateValueList()
+    data_analysis_last_line.setAggregateValueList([])
+    self.tic()
+
+    # Check that the other Data Analysis Lines are consistent
+    for line in data_analysis_line_list[:-1]:
+      consistency_message_list = self.getConsistencyMessageList(line)
+      self.assertFalse(consistency_message_list)
+
+    # Check that the Data Analysis line has a consistency error
+    consistency_message_list = self.getConsistencyMessageList(
+      data_analysis_last_line
+    )
+    self.assertTrue(len(consistency_message_list) == 2)
+    self.assertTrue(
+      consistency_message_list[0] == "Item Type Data Stream is missing"
+    )
+    self.assertTrue(
+      consistency_message_list[1] == "Item Type Progress Indicator is missing"
+    )
+
+    data_analysis_last_line.setAggregateValueList(saved_aggregate_list)
+    self.tic()
+
+  def test_06_1_updateOrsConfigurationsFastInput(self, username=None):
     '''
     Test the scripts called by the Update ORS Configurations Fast Input action which updates the title
     of Data Acquisition Units, as well as the client project to which they are indirectly linked.
@@ -1550,7 +1789,7 @@ class WendelinTelecomTest(SecurityTestCase):
       'started',
     )
 
-  def test_05_2_updateOrsConfigurationsFastInputAsAdministrator(self):
+  def test_06_2_updateOrsConfigurationsFastInputAsAdministrator(self):
     '''
     Repeat the previous test, but logged in as an Administrator user to make sure it works
     and that the function has the adequate permissions.
@@ -1563,9 +1802,9 @@ class WendelinTelecomTest(SecurityTestCase):
       'administrator'
     )
 
-    self.test_05_1_updateOrsConfigurationsFastInput(username=admin_username)
+    self.test_06_1_updateOrsConfigurationsFastInput(username=admin_username)
 
-  def test_06_archiveOrsDataIngestionFastInput(self):
+  def test_07_archiveOrsDataIngestionFastInput(self):
     '''
     Test the scripts called by the Archive ORS Data Ingestion Fast Input action which archives
     the selected Data Acquisition Units as well as all related items, in order to restart
@@ -1964,7 +2203,7 @@ class WendelinTelecomTest(SecurityTestCase):
       not in redo_listbox_item_url_list
     )
 
-  def test_07_refreshOrsKpiDataAnalysisFastInput(self):
+  def test_08_refreshOrsKpiDataAnalysisFastInput(self):
     '''
     Test the scripts called by the Refresh ORS KPI Data Analyses Fast Input action which plans
     a full refresh for the selected ongoing Data Analyses.
@@ -2082,7 +2321,7 @@ class WendelinTelecomTest(SecurityTestCase):
       and ingestion_item_dict_list[4]['data_analysis'].getRefreshState() == 'refresh_planned'
     )
 
-  def test_08_wendelinTelecomSecurityModel(self):
+  def test_09_wendelinTelecomSecurityModel(self):
     '''
     Test Wendelin Telecom's custom security model:
     check that different users have the correct permissions according to their function and project.