Updated test(s).

Author: vladd-bit

config.py

@@ -9,6 +9,8 @@
 
 ROOT_DIR = os.path.abspath(os.curdir)
 TMP_FILE_DIR = os.path.join(ROOT_DIR, "tmp")
+WORKER_PORT_MAP_FILE_PATH = os.path.join(TMP_FILE_DIR, './worker_process_data.txt')
+
 
 # Should we actually ocr or just extract text from PDFs ? NOTE: OCR IS STILL APPLIED TO IMAGES if detected | possible vals : "OCR", "NO_OCR"
 OPERATION_MODE = os.environ.get("OCR_SERVICE_OPERATION_MODE", "OCR")

ocr_service/tests/process.py

@@ -4,6 +4,7 @@
 import unittest
 
 from ocr_service.app.app import create_app
+from ocr_service.utils.utils import sync_port_mapping
 
 from ..tests.test_utils import *
 
@@ -97,6 +98,8 @@ def _setup_logging(cls):
 
     @staticmethod
     def _setup_app(cls):
+        sync_port_mapping(worker_id=0, worker_pid=os.getpid())
+
         cls.app = create_app()
         cls.client = cls.app.test_client()
 

ocr_service/utils/utils.py

@@ -101,12 +101,12 @@ def get_process_id_by_process_name(process_name: str = "") -> int:
 def sync_port_mapping(worker_id = None, worker_pid = None):
 
 
-      f_path = os.path.join(TMP_FILE_DIR, './worker_process_data.txt')
       open_mode = "r+"
-      if not os.path.exists(f_path):
+
+      if not os.path.exists(WORKER_PORT_MAP_FILE_PATH):
           open_mode = "w+"
 
-      with open(f_path, encoding="utf-8", mode=open_mode) as f:
+      with open(WORKER_PORT_MAP_FILE_PATH, encoding="utf-8", mode=open_mode) as f:
             fcntl.lockf(f, fcntl.LOCK_EX)
 
             port_mapping = {}
@@ -124,7 +124,13 @@ def sync_port_mapping(worker_id = None, worker_pid = None):
 
 def get_assigned_port(current_worker_pid):
     port_mapping = {}
-    with open(os.path.join(TMP_FILE_DIR, './worker_process_data.txt'), encoding="utf-8", mode='r+') as f:
+
+    open_mode = "r+"
+
+    if not os.path.exists(WORKER_PORT_MAP_FILE_PATH):
+          open_mode = "w+"
+    
+    with open(WORKER_PORT_MAP_FILE_PATH, encoding="utf-8", mode=open_mode) as f:
         fcntl.lockf(f, fcntl.LOCK_EX)
         port_mapping = json.loads(f.read())
         fcntl.lockf(f, fcntl.LOCK_UN)