sanitize notes updates for invalid characters before persisting to database

services/workflow/workflow/server.py

@@ -17,6 +17,7 @@
 # along with Workspaces.  If not, see <https://www.gnu.org/licenses/>.
 # pylint: disable=E0401
 """ This is our workflow server API. """
+import glob
 import http
 import json
 import logging
@@ -183,31 +184,31 @@ class WorkflowWorkingDirRestService:
     def get_qa_notes(self):
         requested_workflow = self.request.info.lookup_workflow_request(self.request.matchdict["request_id"])
         results_path = requested_workflow.results_dir
-        paths = []
-        for root, dirs, _ in os.walk(results_path, followlinks=True):
-            for d in dirs:
-                if d.startswith("pipeline-"):
-                    paths.append(os.path.join(root, d))
 
-        qa_notes_path = paths[0] + "/html/qa_notes.html"
-        qa_notes_text = ""
+        # qa notes need the products directory weblog
+        path = glob.glob(results_path + "/products/pipeline-*/html")[0]
+        qa_notes_path = path + "/qa_notes.html"
 
-        with open(qa_notes_path, "r") as qa_notes:
-            qa_notes_text = qa_notes.read()
+        if Path(qa_notes_path).exists():
+            qa_notes_text = ""
+            with open(qa_notes_path, "r") as qa_notes:
+                qa_notes_text = qa_notes.read()
 
-        return Response(status_int=http.HTTPStatus.OK, json_body={"resp": f"{qa_notes_text}"})
+            return Response(status_int=http.HTTPStatus.OK, json_body={"resp": f"{qa_notes_text}"})
+
+        return Response(
+            status_int=http.HTTPStatus.NOT_FOUND,
+            json_body={"resp": "QA Notes file not found. Please close the window and try again."},
+        )
 
     @view_config(request_method="POST", route_name="get_qa_notes", renderer="json")
     def save_qa_notes(self):
         requested_workflow = self.request.info.lookup_workflow_request(self.request.matchdict["request_id"])
         results_path = requested_workflow.results_dir
-        paths = []
-        for root, dirs, _ in os.walk(results_path, followlinks=True):
-            for d in dirs:
-                if d.startswith("pipeline-"):
-                    paths.append(os.path.join(root, d))
 
-        qa_notes_path = paths[0] + "/html/qa_notes.html"
+        # qa notes need the products directory weblog
+        path = glob.glob(results_path + "/products/pipeline-*/html")[0]
+        qa_notes_path = path + "/qa_notes.html"
 
         with open(qa_notes_path, "w") as qa_notes:
             # sanitize input before writing/persisting