Merge pull request #1 from EVOLVED-5G/verdict

Verdict

CHANGELOG.md

@@ -1,3 +1,8 @@
+**03/02/2022** [Version 3.2.0]
+
+ - Implement Verdict handling
+ - Add Evaluate, UpgradeVerdict tasks
+
 **11/11/2021** [Version 3.1.0]
 
  - Add NEF Emulator handling

Executor/Tasks/Evolved5g/jenkins_api.py

@@ -56,6 +56,7 @@ def Run(self):
             self.Publish(self.params["PublishKey"], jobId)
         except Exception as e:
             self.Log(Level.ERROR, f"Unable to trigger job: {e}")
+            self.SetVerdictOnError()
 
 
 class JenkinsStatus(JenkinsBase):
@@ -79,3 +80,4 @@ def Run(self):
             self.Publish(self.params["PublishKey"], status)
         except Exception as e:
             self.Log(Level.ERROR, f"Unable to check job '{jobId}' status: {e}")
+            self.SetVerdictOnError()

Executor/Tasks/Evolved5g/nef_emulator.py

@@ -50,3 +50,4 @@ def Run(self):
             self.Log(Level.INFO, f"Response: {msg}")
         except Exception as e:
             self.Log(Level.ERROR, str(e))
+            self.SetVerdictOnError()

Executor/Tasks/PostRun/farewell.py

@@ -7,7 +7,7 @@
 
 class Farewell(Task):
     def __init__(self, logMethod, parent):
-        super().__init__("Farewell", parent, logMethod, None)
+        super().__init__("Farewell", parent, None, logMethod, None)
 
     def Run(self):
         remote = self.parent.Descriptor.Remote

Executor/Tasks/PreRun/coordinate.py

@@ -7,7 +7,7 @@
 
 class Coordinate(Task):
     def __init__(self, logMethod, parent):
-        super().__init__("Coordinate", parent, logMethod, None)
+        super().__init__("Coordinate", parent, None, logMethod, None)
 
     def Run(self):
         remote = self.parent.Descriptor.Remote
@@ -18,7 +18,6 @@ def Run(self):
                 if host is not None:
                     remoteApi = RemoteApi(host, port)
                     self.parent.RemoteApi = remoteApi
-                    # TODO: Why are these messages not visible in the logs?
                     self.Log(Level.INFO, 'Remote connection configured. Waiting for remote Execution ID...')
 
                     timeout = eastWest.Timeout or 120

Executor/Tasks/Remote/getValue.py

@@ -13,6 +13,7 @@ def Run(self):
 
         if valueName is None:
             self.Log(Level.ERROR, "'Value' not defined, please review the Task configuration.")
+            self.SetVerdictOnError()
             return
 
         value = None
@@ -22,6 +23,7 @@ def Run(self):
             value = self.remoteApi.GetValue(self.remoteId, valueName)
             if value is None:
                 if self.timeout <= 0:
+                    self.SetVerdictOnError()
                     raise RuntimeError(f"Timeout reached while waiting for remote remote value '{valueName}'.")
                 sleep(5)
                 self.timeout -= 5

Executor/Tasks/Remote/waitForMilestone.py

@@ -13,6 +13,7 @@ def Run(self):
 
         if milestone is None:
             self.Log(Level.ERROR, "'Milestone' not defined, please review the Task configuration.")
+            self.SetVerdictOnError()
             return
 
         milestones = []
@@ -23,10 +24,12 @@ def Run(self):
             self.Log(Level.DEBUG, f"Status: '{status}'; Milestones: {milestones}")
 
             if status in [ExecutorStatus.Cancelled, ExecutorStatus.Errored]:
+                self.SetVerdictOnError()
                 raise RuntimeError(f"Execution on remote side has been terminated with status: {status.name}")
 
             if milestone not in milestones:
                 if self.timeout <= 0:
+                    self.SetVerdictOnError()
                     raise RuntimeError(f"Timeout reached while waiting for milestone '{milestone}'.")
                 sleep(5)
                 self.timeout -= 5

Executor/Tasks/Run/__init__.py

@@ -11,3 +11,5 @@
 from .add_milestone import AddMilestone
 from .publish_from_source import PublishFromPreviousTaskLog, PublishFromFile
 from .rest_api import RestApi
+from .upgrade_verdict import UpgradeVerdict
+from .evaluate import Evaluate

Executor/Tasks/Run/compress_files.py

@@ -31,4 +31,5 @@ def Run(self):
             self.Log(Level.INFO, "File created")
             self.parent.GeneratedFiles.append(output)
         except Exception as e:
+            self.SetVerdictOnError()
             self.Log(Level.ERROR, f"Exception while creating zip file: {e}")

Executor/Tasks/Run/csvToInflux.py

@@ -34,10 +34,12 @@ def Run(self):
             self.Log(Level.DEBUG, f"Payload: {payload}")
         except Exception as e:
             self.Log(Level.ERROR, f"Exception while converting CSV: {e}")
+            self.SetVerdictOnError()
             return
 
         try:
             self.Log(Level.INFO, "Sending payload to InfluxDb")
             InfluxDb.Send(payload)
         except Exception as e:
             self.Log(Level.ERROR, f"Exception while sending CSV values to Influx: {e}")
+            self.SetVerdictOnError()

Executor/Tasks/Run/delay.py

@@ -16,6 +16,7 @@ def Run(self):
                 raise ValueError
         except ValueError:
             self.Log(Level.ERROR, f"{value} is not a valid number of seconds")
+            self.SetVerdictOnError()
             return
 
         self.Log(Level.INFO, f'Waiting for {time} seconds')

Executor/Tasks/Run/evaluate.py

@@ -0,0 +1,25 @@
+from Task import Task
+from Helper import Level
+import re
+
+
+class Evaluate(Task):
+    def __init__(self, logMethod, parent, params):
+        super().__init__("Evaluate", parent, params, logMethod, None)
+        self.paramRules = {
+            'Key': (None, True),
+            'Expression': (None, True)
+        }
+
+    def Run(self):
+        expression = self.params["Expression"]
+        key = self.params["Key"]
+        self.Log(Level.INFO, f"Evaluating '{key} = {expression}'")
+
+        try:
+            result = eval(expression)
+        except Exception as e:
+            self.Log(Level.ERROR, f"Exception while evaluating expression '{expression}'")
+            raise e
+
+        self.Publish(key, str(result))

Executor/Tasks/Run/publish.py

@@ -1,6 +1,4 @@
 from Task import Task
-from Helper import Level
-from time import sleep
 
 
 class Publish(Task):
@@ -9,4 +7,6 @@ def __init__(self, logMethod, parent, params):
 
     def Run(self):
         for key, value in self.params.items():
+            if key in ["VerdictOnError"]:
+                continue  # Keys common to all tasks are ignored
             self.Publish(key, value)

Executor/Tasks/Run/publish_from_source.py

@@ -26,6 +26,7 @@ def Run(self):
             for index, key in keys:
                 self.Log(Level.DEBUG, f"  {index}: {key}")
         except Exception as e:
+            self.SetVerdictOnError()
             raise RuntimeError(f"Invalid 'Keys' definition: {e}")
 
         regex = re.compile(pattern)
@@ -41,6 +42,7 @@ def generator(self, params: Dict):
         raise NotImplementedError()
 
     def raiseConfigError(self, variable: str):
+        self.SetVerdictOnError()
         raise RuntimeError(f"'{variable}' not defined, please review the Task configuration.")
 
 
@@ -49,7 +51,7 @@ def __init__(self, logMethod, parent, params):
         super().__init__("Publish From Previous Task Log", parent, params, logMethod)
 
     def generator(self, params: Dict):
-        logMessages = self.parent.Params["PreviousTaskLog"]
+        logMessages = self.parent.PreviousTaskLog
         for message in logMessages:
             yield message
 

Executor/Tasks/Run/rest_api.py

@@ -75,4 +75,5 @@ def Run(self):
             if status not in statusCodes:
                 message = f"Unexpected status code received: {status}"
                 self.Log(Level.ERROR, message)
+                self.SetVerdictOnError()
                 raise RuntimeError(message)

Executor/Tasks/Run/slice_creation_time.py

@@ -34,6 +34,7 @@ def Run(self):
                 nestData = json.load(input)
         except Exception as e:
             self.Log(Level.ERROR, f"Exception while reading NEST file: {e}")
+            self.SetVerdictOnError()
             return
 
         from Helper import InfluxDb, InfluxPayload, InfluxPoint  # Delayed to avoid cyclic imports
@@ -51,6 +52,7 @@ def Run(self):
                 sliceId = response
             except Exception as e:
                 self.Log(Level.ERROR, f"Exception on instantiation, skipping iteration: {e}")
+                self.SetVerdictOnError()
                 sleep(pollTime)
                 continue
 
@@ -86,6 +88,7 @@ def Run(self):
                                  f"Deployment time for slice {sliceId} (Iteration {iteration}): {ns_depl_time}")
                     except Exception as e:
                         self.Log(Level.ERROR, f"Exception while calculating deployment time, skipping iteration: {e}")
+                        self.SetVerdictOnError()
                         break
 
                     point = InfluxPoint(datetime.now(timezone.utc))
@@ -113,13 +116,15 @@ def Run(self):
                         self.Log(Level.DEBUG, f"Waiting for slice deletion.")
             except Exception as e:
                 self.Log(Level.ERROR, f"Exception while deleting slice: {e}")
+                self.SetVerdictOnError()
 
         self.Log(Level.DEBUG, f"Payload: {payload}")
         self.Log(Level.INFO, f"Sending results to InfluxDb")
         try:
             InfluxDb.Send(payload)
         except Exception as e:
             self.Log(Level.ERROR, f"Exception while sending payload: {e}")
+            self.SetVerdictOnError()
             if csvFile is None:
                 self.Log(Level.INFO, "Forcing creation of CSV file")
                 csvFile = join(self.parent.TempFolder, f"SliceCreationTime.csv")

Executor/Tasks/Run/tap_execute.py

@@ -20,6 +20,7 @@ def Run(self):
 
         if not config.Enabled:
             self.Log(Level.CRITICAL, "Trying to run TapExecute Task while TAP is not enabled")
+            self.SetVerdictOnError()
         else:
             tapPlan = self.params['TestPlan']
             externals = self.params['Externals']
@@ -42,6 +43,7 @@ def Run(self):
                             self.parent.GeneratedFiles.append(output)
                             self.Log(Level.INFO, f"Saved {len(files)} files to {output}")
                     except Exception as e:
+                        self.SetVerdictOnError()
                         self.Log(Level.ERROR, f"Exception while compressing results: {e}")
                 else:
                     self.Log(Level.WARNING, f"Results path ({path}) does not exist.")

Executor/Tasks/Run/upgrade_verdict.py

@@ -0,0 +1,52 @@
+from Task import Task
+from Helper import Level
+import re
+
+
+class UpgradeVerdict(Task):
+    def __init__(self, logMethod, parent, params):
+        super().__init__("Upgrade Verdict", parent, params, logMethod, None)
+        self.paramRules = {
+            'Key': (None, True),
+            'Pattern': (None, True),
+            'VerdictOnMissing': ("NotSet", False),
+            'VerdictOnMatch': ("NotSet", False),
+            'VerdictOnNoMatch': ("NotSet", False),
+        }
+
+    def Run(self):
+        from Executor import Verdict
+
+        def _assignVerdict(name: str) -> Verdict | None:
+            try:
+                return Verdict[name]
+            except KeyError:
+                self.SetVerdictOnError()
+                self.Log(Level.ERROR, f"Unrecognized Verdict '{name}'")
+                return None
+
+        onMiss = _assignVerdict(self.params["VerdictOnMissing"])
+        onMatch = _assignVerdict(self.params["VerdictOnMatch"])
+        onNoMatch = _assignVerdict(self.params["VerdictOnNoMatch"])
+        if None in [onMiss, onMatch, onNoMatch]: return
+
+        key = self.params["Key"]
+        regex = re.compile(self.params["Pattern"])
+        collection = self.parent.Params
+
+        if key not in collection.keys():
+            self.Log(Level.WARNING, f"Key '{key}' not found. Setting Verdict to '{onMiss.name}'")
+            self.Log(Level.DEBUG, f"Available keys: {list(collection.keys())}")
+            self.Verdict = onMiss
+        else:
+            value = str(collection[key])
+            if regex.match(value):
+                condition = "matches"
+                verdict = onMatch
+            else:
+                condition = "does not match"
+                verdict = onNoMatch
+
+            self.Log(Level.INFO, f"'{key}'='{value}' {condition} pattern. Setting Verdict to '{verdict.name}'")
+            self.Verdict = verdict
+

Executor/__init__.py

@@ -2,4 +2,4 @@
 from .pre_runner import PreRunner
 from .executor import Executor
 from .post_runner import PostRunner
-from .status import Status as ExecutorStatus
+from .enums import Status as ExecutorStatus, Verdict

Executor/status.py → Executor/enums.py

@@ -10,3 +10,12 @@ def label(self):
         if self.name == 'Errored': return 'label-danger'
         if self.name == 'Finished': return 'label-success'
         return 'label-info'
+
+
+@unique
+class Verdict(Enum):
+    NotSet, Pass, Inconclusive, Fail, Cancel, Error = range(6)
+
+    @staticmethod
+    def Max(a, b):
+        return a if a.value > b.value else b

Executor/executor.py

@@ -1,9 +1,8 @@
 from Helper import Level
 from typing import Dict
-from time import sleep
 from .executor_base import ExecutorBase
 from Task import Task
-from .status import Status
+from .enums import Status, Verdict
 from tempfile import TemporaryDirectory
 from math import floor
 
@@ -19,20 +18,29 @@ def Run(self):
 
         tasks = self.Configuration.RunTasks
         self.params['PreviousTaskLog'] = []
+        self.params['Verdict'] = Verdict.NotSet
         for i, task in enumerate(tasks, start=1):
             if self.stopRequested:
                 self.LogAndMessage(Level.INFO, "Received stop request, exiting")
                 self.Status = Status.Cancelled
+                self.params['Verdict'] = Verdict.Cancel
                 break
             taskInstance: Task = task.Task(self.Log, self, Expander.ExpandDict(task.Params, self))
             self.AddMessage(f'Starting task {taskInstance.name}')
-            taskInstance.Start()
+
+            try:
+                taskInstance.Start()
+            except Exception as e:
+                self.params['Verdict'] = Verdict.Error
+                raise e
 
             # Add the values generated by the task to the global dictionary
             self.params.update(taskInstance.Vault)
             self.params['PreviousTaskLog'] = taskInstance.LogMessages
+            self.params['Verdict'] = Verdict.Max(self.Verdict, taskInstance.Verdict)
 
-            self.AddMessage(f'Task {taskInstance.name} finished', int(floor(10 + ((i / len(tasks)) * 80))))
+            self.AddMessage(f"Task '{taskInstance.name}' finished with verdict '{taskInstance.Verdict.name}'",
+                            int(floor(10 + ((i / len(tasks)) * 90))))
         else:
             self.Status = Status.Finished