From 1739141e600d82766a079d57c36038518c96f2fa Mon Sep 17 00:00:00 2001
From: chausman <chausman@nrao.edu>
Date: Fri, 9 Jul 2021 16:33:53 -0600
Subject: [PATCH] refactoring casa-envoy
---
.../pexable/casa_envoy/casa_envoy/auditor.py | 81 ++++---
.../casa_envoy/casa_envoy/interfaces.py | 19 +-
.../casa_envoy/casa_envoy/launchers.py | 224 +++++++++++-------
.../pexable/casa_envoy/casa_envoy/palaver.py | 101 ++------
.../casa_envoy/test/image-metadata.json | 1 +
.../pexable/casa_envoy/test/test.json | 3 +-
.../pexable/casa_envoy/test/test_auditor.py | 99 +++++---
.../casa_envoy/test/test_casa_envoy.py | 57 +++--
.../pexable/casa_envoy/test/test_launchers.py | 99 ++++----
9 files changed, 382 insertions(+), 302 deletions(-)
diff --git a/apps/cli/executables/pexable/casa_envoy/casa_envoy/auditor.py b/apps/cli/executables/pexable/casa_envoy/casa_envoy/auditor.py
index 078367eb8..f6c85c0d2 100644
--- a/apps/cli/executables/pexable/casa_envoy/casa_envoy/auditor.py
+++ b/apps/cli/executables/pexable/casa_envoy/casa_envoy/auditor.py
@@ -14,49 +14,57 @@ from casa_envoy.interfaces import AuditorIF
from casa_envoy.schema import AbstractTextFile
-def get_fields_for(filename: str) -> list:
- metadata_list = ["fileSetIds", "workflowName", "systemId", "productLocator", "destinationDirectory"]
+def get_fields_for(product_type: str, filename: str) -> list:
+ cal_metadata_list = [
+ "fileSetIds",
+ "workflowName",
+ "systemId",
+ "creationTime",
+ "productLocator",
+ "destinationDirectory",
+ ]
+ img_metadata_list = [
+ "fileSetIds",
+ "workflowName",
+ "systemId",
+ "creationTime",
+ "productLocator",
+ "destinationDirectory",
+ "cms_path",
+ "sdmId",
+ ]
+
ppr_list = ["RootDirectory", "RelativePath", "SdmIdentifier"]
if ".xml" in filename:
return ppr_list
- if ".json" in filename:
- return metadata_list
+ if ".json" in filename and "cal" in product_type:
+ return cal_metadata_list
+ elif ".json" in filename and "img" in product_type:
+ return img_metadata_list
-def get_xml_content(filename: str):
- with open(filename) as file:
- return BeautifulSoup(file.read(), "xml")
+def get_xml_content(file: AbstractTextFile):
+ return BeautifulSoup(file.content, "xml")
-def get_value_for(filename: str, key: str) -> str:
- if ".xml" in filename:
- ppr_content = get_xml_content(filename)
+def get_value_for(file: AbstractTextFile, key: str) -> str:
+ if ".xml" in file.filename:
+ ppr_content = get_xml_content(file)
return ppr_content.find(key).string
- if ".json" in filename:
- with open(filename) as file:
- metadata = json.loads(file.read())
- return metadata[key]
+ if ".json" in file.filename:
+ content = json.loads(file.content)
+ return content[key]
class AuditFiles(AuditorIF):
- def __init__(self, files: List[str], settings: Dict[str, str]):
+ def __init__(self, files: List[AbstractTextFile], settings: Dict[str, str]):
+ self.logger = logging.getLogger("casa_envoy")
self.files = files
self.settings = settings
- self.logger = logging.getLogger("casa_envoy")
-
- def read_file(self, filename: str) -> AbstractTextFile:
- if os.path.isfile(filename):
- with open(filename) as file:
- if ".json" in filename:
- metadata = json.loads(file.read())
- return AbstractTextFile(filename, json.dumps(metadata))
- else:
- if ".xml" in filename:
- ppr = file.read()
- return AbstractTextFile(filename, ppr)
+ self.product_type = settings.get("product_type")
def check_required_fields(self, file: AbstractTextFile, fields: list) -> bool:
missing = []
@@ -83,7 +91,7 @@ class AuditFiles(AuditorIF):
shutil.copy(ppr.filename, "./working")
os.chdir("./working")
- parsed_xml = get_xml_content(ppr.filename)
+ parsed_xml = get_xml_content(ppr)
parsed_xml.find("RootDirectory").string = self.settings["rootDirectory"]
parsed_xml.find("RelativePath").string = self.settings["processingDirectory"]
@@ -97,15 +105,16 @@ class AuditFiles(AuditorIF):
invalid_files = []
for file in self.files:
self.logger.info(f"Auditing file {file}...")
- f = self.read_file(file)
- if f.filename == "PPR.xml":
+ if file.filename == "PPR.xml":
self.logger.info("Correcting PPR.xml for condor processing...")
- f = self.correct_for_condor(f)
+ file = self.correct_for_condor(file)
- valid = self.check_required_fields(file=f, fields=get_fields_for(f.filename))
+ valid = self.check_required_fields(
+ file=file, fields=get_fields_for(self.product_type, file.filename)
+ )
if not valid:
- invalid_files.append(f.filename)
+ invalid_files.append(file.filename)
if len(invalid_files) != 0:
self.logger.info(f"INVALID FILE FOUND: {invalid_files}")
@@ -115,7 +124,7 @@ class AuditFiles(AuditorIF):
class AuditDirectories(AuditorIF):
- def __init__(self, ppr: str, settings: Dict[str, str]):
+ def __init__(self, ppr: AbstractTextFile, settings: Dict[str, str]):
self.logger = logging.getLogger("casa_envoy")
self.rootDirectory = settings["rootDirectory"]
self.relative_path = settings["processingDirectory"]
@@ -142,4 +151,6 @@ class AuditDirectories(AuditorIF):
self.logger.info("FAILURE: data not found in rawdata/")
return False
else:
- self.logger.info("DIRECTORY ERROR: A directory is missing from the processing root directory.")
+ self.logger.info(
+ "DIRECTORY ERROR: A directory is missing from the processing root directory."
+ )
diff --git a/apps/cli/executables/pexable/casa_envoy/casa_envoy/interfaces.py b/apps/cli/executables/pexable/casa_envoy/casa_envoy/interfaces.py
index 597a2f128..6721bdb07 100644
--- a/apps/cli/executables/pexable/casa_envoy/casa_envoy/interfaces.py
+++ b/apps/cli/executables/pexable/casa_envoy/casa_envoy/interfaces.py
@@ -1,27 +1,30 @@
"""
Interfaces for casa_envoy
"""
+import abc
from abc import ABC
+from typing import Dict
-class CasaLauncherIF(ABC):
+class LauncherIF(ABC):
"""
- Generic CASA Launcher methods.
+ Generic Launcher methods.
Should be implemented for any type of CASA processing.
"""
- def run(self):
- raise NotImplementedError
- def setup_environment(self, parameters: dict):
- raise NotImplementedError
+ @abc.abstractmethod
+ def launch_casa(self):
+ pass
- def check_logs(self, parent_path: str):
- raise NotImplementedError
+ @abc.abstractmethod
+ def run_audit(self, parameters: Dict[str, str]):
+ pass
class AuditorIF(ABC):
"""
Generic functionality implementation for auditor classes
"""
+
def audit(self):
raise NotImplementedError
diff --git a/apps/cli/executables/pexable/casa_envoy/casa_envoy/launchers.py b/apps/cli/executables/pexable/casa_envoy/casa_envoy/launchers.py
index 911e4b473..ccc21a1b3 100644
--- a/apps/cli/executables/pexable/casa_envoy/casa_envoy/launchers.py
+++ b/apps/cli/executables/pexable/casa_envoy/casa_envoy/launchers.py
@@ -4,106 +4,84 @@ import glob
import sys
import os
import subprocess
+from typing import Dict
-from casa_envoy.interfaces import CasaLauncherIF
-from casa_envoy.schema import AbstractTextFile
+import json
+
+from casa_envoy.auditor import AuditFiles, AuditDirectories
+from casa_envoy.interfaces import LauncherIF
+from casa_envoy.schema import AbstractTextFile
-def get_base_environment(parameters: dict):
- os.environ["SCIPIPE_ROOTDIR"] = parameters["rootDirectory"]
- os.environ["CASA_HOME"] = parameters["homeForReprocessing"]
- os.environ["LANG"] = "en_US.UTF-8"
+def get_abs_file(filename: str) -> AbstractTextFile:
+ with open(filename) as file:
+ if ".json" in filename:
+ content = json.loads(file.read())
+ elif ".xml" in filename:
+ content = file.read()
-def check_processing_env(logger: logging.Logger):
- logger.info("Checking processing environment:")
- env1 = os.environ.get("SCIPIPE_ROOTDIR")
- logger.info(f"SCIPIPE_ROOTDIR: {env1}")
- env2 = os.environ.get("CASA_HOME")
- logger.info(f"CASA_HOME: {env2}")
- env3 = os.environ.get("PPR_FILENAME")
- logger.info(f"PPR_FILENAME: {env3}")
- env4 = os.environ.get("LANG")
- logger.info(f"LANG: {env4}")
- if "None" in [env1, env2, env3, env4]:
- logger.info("Environment setup Failed!")
- sys.exit(1)
- else:
- logger.info("Environment ready for processing")
+ return AbstractTextFile(filename, content)
-class CalibrationLauncher(CasaLauncherIF):
- def __init__(self, ppr: AbstractTextFile, metadata: AbstractTextFile):
+class CasaLauncher:
+ def __init__(self, parameters: dict):
self.logger = logging.getLogger("casa_envoy")
- self.ppr = ppr
- self.metadata = metadata
+ self.parameters = parameters
- def run(self):
- self.logger.info("RUNNING CASA CALIBRATION!")
- os.chdir("./working")
- result = subprocess.Popen(
- "PYTHONPATH='' xvfb-run -e ${PWD}/xvfb-run.err.txt -d -s \"-screen 0 800x600x16\" "
- "${CASA_HOME}/bin/casa --pipeline --nogui --nologger -c "
- "${CASA_HOME}/pipeline/pipeline/runvlapipeline.py ${PPR_FILENAME} || true",
- shell=True,
- executable="/bin/bash",
- stdout=sys.stdout,
- stderr=sys.stderr,
- )
- return result.communicate()
-
- def setup_environment(self, parameters: dict):
- get_base_environment(parameters)
- os.environ["PPR_FILENAME"] = str(self.ppr)
-
- check_processing_env(self.logger)
-
- def check_logs(self, parent_path: str):
- self.logger.info("CHECKING CASA CALIBRATION LOGS!")
- # make sure we are in the correct directory to find log file
- if not os.getcwd().endswith("/working"):
- os.chdir(parent_path + "/working")
+ def setup_environment(self):
+ os.environ["SCIPIPE_ROOTDIR"] = self.parameters["rootDirectory"]
+ os.environ["CASA_HOME"] = self.parameters["homeForReprocessing"]
+ os.environ["LANG"] = "en_US.UTF-8"
+ os.environ["PPR_FILENAME"] = self.parameters["ppr"]
- casa_logs = glob.glob("casa-*.log")
+ self.check_processing_env()
- for file in casa_logs:
- if re.match("^.*SEVERE\sflagmanager.*$", open(file).read()):
- self.logger.error("CASA ERROR!")
- else:
- self.logger.info("CASA Success!")
+ def check_processing_env(self):
+ self.logger.info("Checking processing environment:")
+ env_list = ["SCIPIPE_ROOTDIR", "CASA_HOME", "PPR_FILENAME", "LANG"]
+ result_list = []
-class ImagingLauncher(CasaLauncherIF):
- def __init__(self, ppr: AbstractTextFile, metadata: AbstractTextFile):
- self.logger = logging.getLogger("casa_envoy")
- self.ppr = ppr
- self.metadata = metadata
+ for var in env_list:
+ env = os.environ.get(var)
+ result_list.append(env)
+ self.logger.info(f"{var}: {env}")
+
+ if "None" in result_list:
+ self.logger.info("Environment setup Failed!")
+ sys.exit(1)
+ else:
+ self.logger.info("Environment ready for processing")
def run(self):
- self.logger.info("RUNNING CASA IMAGING!")
- os.chdir("./working")
- result = subprocess.Popen(
- "PYTHONPATH='' xvfb-run -e ${PWD}/xvfb-run.err.txt -d -s \"-screen 0 800x600x16\" "
- "${CASA_HOME}/bin/casa --pipeline --nogui --nologger -c "
- "${CASA_HOME}/pipeline/pipeline/runvlapipeline.py ${PPR_FILENAME} || true",
- shell=True,
- executable="/bin/bash",
- stdout=sys.stdout,
- stderr=sys.stderr,
- )
- return result.communicate()
-
- def setup_environment(self, parameters: dict):
- get_base_environment(parameters)
- os.environ["PPR_FILENAME"] = str(self.ppr)
-
- check_processing_env(self.logger)
-
- def check_logs(self, parent_path: str):
- self.logger.info("CHECKING CASA IMAGING LOGS!")
+ self.setup_environment()
+
+ if self.parameters.get("useCasa"):
+ self.logger.info("RUNNING CASA!")
+ os.chdir("./working")
+ result = subprocess.Popen(
+ "PYTHONPATH='' xvfb-run -e ${PWD}/xvfb-run.err.txt -d -s \"-screen 0 800x600x16\" "
+ "${CASA_HOME}/bin/casa --pipeline --nogui --nologger -c "
+ "${CASA_HOME}/pipeline/pipeline/runvlapipeline.py ${PPR_FILENAME} || true",
+ shell=True,
+ executable="/bin/bash",
+ stdout=sys.stdout,
+ stderr=sys.stderr,
+ )
+ return result.communicate()
+ else:
+ self.logger.info("RUNNING VELA!")
+ run_type = self.parameters.get("product_type")
+ metadata = self.parameters.get("metadata")
+ ppr = self.parameters.get("ppr")
+ subprocess.run(["./vela", run_type, metadata, ppr])
+
+ def check_logs(self):
+ self.logger.info("CHECKING CASA LOGS!")
# make sure we are in the correct directory to find log file
if not os.getcwd().endswith("/working"):
- os.chdir(parent_path + "/working")
+ os.chdir(self.parameters.get("parent_path") + "/working")
casa_logs = glob.glob("casa-*.log")
@@ -112,3 +90,83 @@ class ImagingLauncher(CasaLauncherIF):
self.logger.error("CASA ERROR!")
else:
self.logger.info("CASA Success!")
+
+
+class CalibrationLauncher(LauncherIF):
+ def __init__(self, parameters: dict):
+ self.logger = logging.getLogger("casa_envoy")
+ self.parameters = parameters
+ self.ppr = get_abs_file(parameters.get("ppr"))
+ self.metadata = get_abs_file(parameters.get("metadata"))
+
+ def launch_casa(self):
+ if self.check_calibratable():
+ self.run_audit(self.parameters)
+ CasaLauncher(self.parameters).run()
+ else:
+ self.logger.error("ERROR: Provided SPL is not type execution block!")
+ sys.exit(1)
+
+ def check_calibratable(self) -> bool:
+ spl = self.metadata.content["productLocator"]
+ if "execblock" in spl:
+ return True
+ else:
+ self.logger.info("SPL ERROR: This product locator is not calibratable!")
+ return False
+
+ def run_audit(self, parameters: Dict[str, str]):
+ dir_audit = AuditDirectories(self.ppr, parameters).audit()
+ if dir_audit:
+ self.logger.info("Directory audit successful!")
+ else:
+ self.logger.error("FAILURE: directory structure audit was unsuccessful!")
+ sys.exit(1)
+
+ audit = AuditFiles([self.ppr, self.metadata], parameters).audit()
+ if audit:
+ self.logger.info("File audit successful!")
+ else:
+ self.logger.error("FAILURE: file audit was unsuccessful!")
+ sys.exit(1)
+
+
+class ImagingLauncher(LauncherIF):
+ def __init__(self, parameters: dict):
+ self.logger = logging.getLogger("casa_envoy")
+ self.parameters = parameters
+ self.ppr = get_abs_file(parameters.get("ppr"))
+ self.metadata = get_abs_file(parameters.get("metadata"))
+
+ def launch_casa(self):
+ if self.check_imageable():
+ self.run_audit(self.parameters)
+ CasaLauncher(self.parameters).run()
+ else:
+ self.logger.error("ERROR: CMS information missing or incorrect!")
+ sys.exit(1)
+
+ def check_imageable(self) -> bool:
+ content = self.metadata.content
+ cms_name = content["cmsName"]
+ cms_path = content["calibrationSourceDirectory"]
+ if cms_name is not None and cms_path is not None and cms_name[-3:] == ".ms":
+ return True
+ else:
+ self.logger.info("CMS ERROR: Imaging requires a valid CMS name and location!")
+ return False
+
+ def run_audit(self, parameters: Dict[str, str]):
+ dir_audit = AuditDirectories(self.ppr, parameters).audit()
+ if dir_audit:
+ self.logger.info("Directory audit successful!")
+ else:
+ self.logger.info("FAILURE: directory structure audit was unsuccessful!")
+ sys.exit(1)
+
+ audit = AuditFiles([self.ppr, self.metadata], parameters).audit()
+ if audit:
+ self.logger.info("File audit successful!")
+ else:
+ self.logger.info("FAILURE: file audit was unsuccessful!")
+ sys.exit(1)
diff --git a/apps/cli/executables/pexable/casa_envoy/casa_envoy/palaver.py b/apps/cli/executables/pexable/casa_envoy/casa_envoy/palaver.py
index 0aa55ca27..8be5536d8 100644
--- a/apps/cli/executables/pexable/casa_envoy/casa_envoy/palaver.py
+++ b/apps/cli/executables/pexable/casa_envoy/casa_envoy/palaver.py
@@ -5,14 +5,10 @@ palaver definition: an unnecessarily elaborate or complex procedure
import argparse
import logging
import pathlib
-import subprocess
import sys
import os
-import json
-from typing import Dict
from pycapo import CapoConfig
-from casa_envoy.auditor import AuditDirectories, AuditFiles
from casa_envoy.launchers import CalibrationLauncher, ImagingLauncher
"""
@@ -24,20 +20,27 @@ logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler(sys.stdout))
-def _get_settings(cwd: pathlib.Path):
+def _get_settings(cwd: pathlib.Path, args: list):
use_casa = CapoConfig().getboolean("edu.nrao.archive.workspaces.ProcessingSettings.useCasa")
casa_home = (
CapoConfig().settings("edu.nrao.archive.workflow.config.CasaVersions").homeForReprocessing
)
+ parent_path = cwd
root_dir = str(cwd.parent)
processing_dir = str(cwd.stem)
+ metadata = args[0]
+ ppr = args[1]
+
return {
"useCasa": use_casa,
"homeForReprocessing": casa_home,
"rootDirectory": root_dir,
"processingDirectory": processing_dir,
+ "parent_path": parent_path,
+ "metadata": metadata,
+ "ppr": ppr,
}
@@ -63,87 +66,23 @@ def arg_parser() -> argparse.ArgumentParser:
return parser
-def check_calibratable(metadata_filename: str) -> bool:
- with open(metadata_filename) as json_meta:
- metadata = json.loads(json_meta.read())
- spl = metadata["productLocator"]
- if "execblock" in spl:
- return True
- else:
- logger.info("SPL ERROR: This product locator is not calibratable!")
- return False
-
-
-def check_imagable(metadata_filename: str) -> bool:
- with open(metadata_filename) as json_meta:
- metadata = json.loads(json_meta.read())
- cms_name = metadata["cmsName"]
- cms_path = metadata["calibrationSourceDirectory"]
- if cms_name is not None and cms_path is not None and cms_name[-3:] == ".ms":
- return True
- else:
- logger.info("CMS ERROR: Imaging requires a valid CMS name and location!")
- return False
-
-
-def run_audit(ppr: str, metadata: str, settings: Dict[str, str]):
- dir_audit = AuditDirectories(ppr, settings).audit()
- if dir_audit:
- logger.info("Directory audit successful!")
- else:
- logger.info("FAILURE: directory structure audit was unsuccessful!")
- sys.exit(1)
-
- audit = AuditFiles([ppr, metadata], settings).audit()
- if audit:
- logger.info("File audit successful!")
- else:
- logger.info("FAILURE: file audit was unsuccessful!")
- sys.exit(1)
-
-
def main():
args = arg_parser().parse_args()
path = os.getcwd()
- settings = _get_settings(pathlib.Path(path))
if args.standard_cal is not None:
- metadata = args.standard_cal[0]
- ppr = args.standard_cal[1]
-
- run_audit(ppr, metadata, settings)
- if check_calibratable(metadata):
- if settings.get("useCasa"):
- launcher = CalibrationLauncher(ppr, metadata)
- launcher.setup_environment(settings)
- launcher.run()
- launcher.check_logs(parent_path=path)
- # make sure we return to the parent directory after processing
- os.chdir(path)
- else:
- logger.info("RUNNING VELA!")
- subprocess.run(["./vela", "--standard-cal", metadata, ppr])
- else:
- logger.error("ERROR: Provided SPL is not type execution block!")
- sys.exit(1)
+ parameters = _get_settings(pathlib.Path(path), args.standard_cal)
+ parameters["product_type"] = "standard-cal"
+
+ CalibrationLauncher(parameters).launch_casa()
+ # make sure we return to the parent directory after processing
+ os.chdir(path)
elif args.standard_img is not None:
- metadata = args.standard_img[0]
- ppr = args.standard_img[1]
-
- run_audit(ppr, metadata, settings)
- if check_imagable(metadata):
- if settings.get("useCasa"):
- launcher = ImagingLauncher(ppr, metadata)
- launcher.setup_environment(settings)
- launcher.run()
- launcher.check_logs(parent_path=path)
- # return to parent directory after processing
- os.chdir(path)
- else:
- logger.info("RUNNING VELA!")
- subprocess.run(["./vela", "--standard-img", metadata, ppr])
- else:
- logger.error("ERROR: CMS information missing or incorrect!")
- sys.exit(1)
+ parameters = _get_settings(pathlib.Path(path), args.standard_img)
+ parameters["product_type"] = "standard-img"
+
+ ImagingLauncher(parameters).launch_casa()
+ # return to parent directory after processing
+ os.chdir(path)
diff --git a/apps/cli/executables/pexable/casa_envoy/test/image-metadata.json b/apps/cli/executables/pexable/casa_envoy/test/image-metadata.json
index e2b19ffab..340c903ac 100644
--- a/apps/cli/executables/pexable/casa_envoy/test/image-metadata.json
+++ b/apps/cli/executables/pexable/casa_envoy/test/image-metadata.json
@@ -2,6 +2,7 @@
"fileSetIds": "brain_000.58099.67095825232",
"workflowName": "std_cms_imaging",
"systemId": "4",
+ "creationTime": "2021-07-01T21:14:00",
"productLocator": "uid://evla/execblock/ec082e65-452d-4fec-ad88-f5b4af1f9e36",
"projectMetadata": {
"projectCode": "Operations",
diff --git a/apps/cli/executables/pexable/casa_envoy/test/test.json b/apps/cli/executables/pexable/casa_envoy/test/test.json
index dad157e97..e84dbb8e0 100644
--- a/apps/cli/executables/pexable/casa_envoy/test/test.json
+++ b/apps/cli/executables/pexable/casa_envoy/test/test.json
@@ -2,6 +2,7 @@
"fileSetIds": "brain_001.58099.678886747686",
"workflowName": "std_calibration",
"systemId": "5",
+ "creationTime": "2021-05-06T17:34:36",
"productLocator": "uid://evla/execblock/9cb0964b-ad6b-40ed-bd87-d08c502503e2",
"projectMetadata": {
"projectCode": "Operations",
@@ -10,4 +11,4 @@
"observer": "VLA Operations"
},
"destinationDirectory": "/tmp/workspaces_tmp/testing"
-}
\ No newline at end of file
+}
diff --git a/apps/cli/executables/pexable/casa_envoy/test/test_auditor.py b/apps/cli/executables/pexable/casa_envoy/test/test_auditor.py
index af33bfdfc..87b69ed65 100644
--- a/apps/cli/executables/pexable/casa_envoy/test/test_auditor.py
+++ b/apps/cli/executables/pexable/casa_envoy/test/test_auditor.py
@@ -13,104 +13,139 @@ from casa_envoy.auditor import (
)
from casa_envoy.schema import AbstractTextFile
-settings = {
+cal_settings = {
"useCasa": False,
"homeForReprocessing": "/home/casa/packages/pipeline/current",
"rootDirectory": "/tmp/workspaces_tmp/",
- "processingDirectory": "tmpiox5trbp"
+ "processingDirectory": "tmpiox5trbp",
+ "metadata": "test/test.json",
+ "ppr": "test/PPR.xml",
+ "product_type": "standard-cal",
}
test_ppr = AbstractTextFile(filename="test/PPR.xml", content=Path("test/PPR.xml").read_text())
+test_cal_metadata = AbstractTextFile(
+ filename="test/test.json", content=Path("test/test.json").read_text()
+)
+
+img_settings = {
+ "useCasa": False,
+ "homeForReprocessing": "/home/casa/packages/pipeline/current",
+ "rootDirectory": "/tmp/workspaces_tmp/",
+ "processingDirectory": "tmpiox5trbp",
+ "metadata": "test/image-metadata.json",
+ "ppr": "test/cmsimage-PPR.xml",
+ "product_type": "standard-img",
+}
+test_img_ppr = AbstractTextFile(
+ filename="test/cmsimage-PPR.xml", content=Path("test/cmsimage-PPR.xml").read_text()
+)
+test_img_metadata = AbstractTextFile(
+ filename="test/image-metadata.json", content=Path("test/image-metadata.json").read_text()
+)
def test_get_fields_for():
- fields = ["fileSetIds", "workflowName", "systemId", "productLocator", "destinationDirectory"]
- result = get_fields_for("test/test.json")
+ product_type = "standard-cal"
+ fields = [
+ "fileSetIds",
+ "workflowName",
+ "systemId",
+ "creationTime",
+ "productLocator",
+ "destinationDirectory",
+ ]
+ result = get_fields_for(product_type=product_type, filename=test_cal_metadata.filename)
assert result == fields
fields2 = ["RootDirectory", "RelativePath", "SdmIdentifier"]
- result2 = get_fields_for("test/PPR.xml")
+ result2 = get_fields_for(product_type=product_type, filename=test_ppr.filename)
assert result2 == fields2
+ product_type2 = "standard-img"
+ img_fields = [
+ "fileSetIds",
+ "workflowName",
+ "systemId",
+ "creationTime",
+ "productLocator",
+ "destinationDirectory",
+ "cms_path",
+ "sdmId",
+ ]
+ result = get_fields_for(product_type=product_type2, filename=test_img_metadata.filename)
+ assert result == img_fields
+
def test_get_xml_content():
- content = get_xml_content("test/PPR.xml")
+ content = get_xml_content(test_ppr)
assert content.find("RelativePath").string == "tmpiox5trbp"
assert content.find("SdmIdentifier").string == "brain_001.58099.678886747686"
+ assert content.find("CreationTime").string == "2021-05-06T17:34:36"
def test_get_value_for():
- value = get_value_for("test/PPR.xml", "RelativePath")
+ value = get_value_for(test_ppr, "RelativePath")
assert value == "tmpiox5trbp"
- value = get_value_for("test/PPR.xml", "SdmIdentifier")
+ value = get_value_for(test_ppr, "SdmIdentifier")
assert value == "brain_001.58099.678886747686"
- value = get_value_for("test/test.json", "workflowName")
+ value = get_value_for(test_cal_metadata, "workflowName")
assert value == "std_calibration"
- value = get_value_for("test/test.json", "productLocator")
+ value = get_value_for(test_cal_metadata, "productLocator")
assert value == "uid://evla/execblock/9cb0964b-ad6b-40ed-bd87-d08c502503e2"
class TestAuditFiles:
- def test_read_file(self):
- file = AuditFiles(["test/test.json", "test/PPR.xml"], settings).read_file("test/test.json")
- assert file.filename == "test/test.json"
-
- file2 = AuditFiles(["test/test.json", "test/PPR.xml"], settings).read_file("test/PPR.xml")
- assert file2.filename == "test/PPR.xml"
-
def test_check_required_fields(self):
- file = AuditFiles(["test/test.json", "test/PPR.xml"], settings).read_file("test/test.json")
+ audit_files = AuditFiles([test_cal_metadata, test_ppr], cal_settings)
fields = [
"fileSetIds",
"workflowName",
"systemId",
+ "creationTime",
"productLocator",
"destinationDirectory",
]
- result = AuditFiles(["test/test.json", "test/PPR.xml"], settings).check_required_fields(file, fields)
+ result = audit_files.check_required_fields(test_cal_metadata, fields)
assert result is True
fieldsF = [
"fileSetIds",
"workflowName",
"systemId",
+ "creationTime",
"productLocator",
"destinationDirectory",
"FakeField",
]
- resultF = AuditFiles(["test/test.json", "test/PPR.xml"], settings).check_required_fields(
- file, fieldsF
- )
+ resultF = audit_files.check_required_fields(test_cal_metadata, fieldsF)
assert resultF is False
- file2 = AuditFiles(["test/test.json", "test/PPR.xml"], settings).read_file("test/PPR.xml")
fields2 = ["RootDirectory", "RelativePath", "SdmIdentifier"]
- result2 = AuditFiles(["test/test.json", "test/PPR.xml"], settings).check_required_fields(
- file2, fields2
- )
+ result2 = audit_files.check_required_fields(test_ppr, fields2)
assert result2 is True
fields2F = ["RootDirectory", "RelativePath", "SdmIdentifier", "FakeField"]
- result2F = AuditFiles(["test/test.json", "test/PPR.xml"], settings).check_required_fields(
- file2, fields2F
- )
+ result2F = audit_files.check_required_fields(test_ppr, fields2F)
assert result2F is False
@patch("os.chdir")
@patch("shutil.copy")
def test_correct_for_condor(self, mock_copy, mock_os):
- ppr = AuditFiles(["test/test.json", "test/PPR.xml"], settings).correct_for_condor(ppr=test_ppr)
+ ppr = AuditFiles([test_cal_metadata, test_ppr], cal_settings).correct_for_condor(
+ ppr=test_ppr
+ )
assert ppr.filename == "test/PPR.xml"
def test_audit(self):
- result = AuditFiles(["test/test.json", "test/PPR.xml"], settings=settings).audit()
+ result = AuditFiles([test_cal_metadata, test_ppr], cal_settings).audit()
assert result is True
class TestAuditDirectories:
def test_audit(self):
- result = AuditDirectories("test/PPR.xml", settings).audit()
+ result = AuditDirectories(test_ppr, cal_settings).audit()
assert result is False
diff --git a/apps/cli/executables/pexable/casa_envoy/test/test_casa_envoy.py b/apps/cli/executables/pexable/casa_envoy/test/test_casa_envoy.py
index 4089a39f0..c0ff840ec 100644
--- a/apps/cli/executables/pexable/casa_envoy/test/test_casa_envoy.py
+++ b/apps/cli/executables/pexable/casa_envoy/test/test_casa_envoy.py
@@ -4,32 +4,59 @@ Tests for casa_envoy.palaver
import argparse
from unittest.mock import patch, MagicMock
-from casa_envoy.palaver import check_calibratable, check_imagable, _get_settings
+import casa_envoy.palaver as palaver
expected_settings = {
"useCasa": False,
"homeForReprocessing": "/home/casa/packages/pipeline/current",
"rootDirectory": "/lustre/aoc/cluster/pipeline/docker/workspaces/spool",
"processingDirectory": "tmpo1ca1pp_",
+ "parent_path": "/lustre/aoc/cluster/pipeline/docker/workspaces/spool/tmpo1ca1pp_",
+ "metadata": "test/test.json",
+ "ppr": "test/PPR.xml",
+ "product_type": "standard-cal",
}
args = argparse.Namespace()
-def test_check_calibratable():
- check = check_calibratable("test/test.json")
- assert check is True
+class TestPalaver:
+ def test_get_settings(self):
+ args.standard_cal = ["test/test.json", "test/PPR.xml"]
+ with patch(
+ "pathlib.Path.cwd",
+ MagicMock(
+ return_value="/lustre/aoc/cluster/pipeline/docker/workspaces/spool/tmpo1ca1pp_"
+ ),
+ ) as cwd:
+ settings = palaver._get_settings(cwd, args.standard_cal)
+ assert settings["useCasa"] == expected_settings["useCasa"]
+ assert settings["homeForReprocessing"] == expected_settings["homeForReprocessing"]
+ assert settings["metadata"] == expected_settings["metadata"]
+ assert settings["ppr"] == expected_settings["ppr"]
-def test_check_imagable():
- check = check_imagable("test/image-metadata.json")
- assert check is True
+ args.standard_cal = None
+ @patch("os.chdir")
+ @patch("os.getcwd")
+ def test_main_cal(self, mock_cwd, mock_chdir):
+ args.standard_cal = ["test/test.json", "test/PPR.xml"]
-def test_get_settings():
- with patch(
- "pathlib.Path.cwd",
- MagicMock(return_value="/lustre/aoc/cluster/pipeline/docker/workspaces/spool/tmpo1ca1pp_"),
- ) as cwd:
- settings = _get_settings(cwd)
- assert settings["useCasa"] == expected_settings["useCasa"]
- assert settings["homeForReprocessing"] == expected_settings["homeForReprocessing"]
+ with patch("argparse.ArgumentParser.parse_args", MagicMock(return_value=args)) as mock_args:
+ with patch("casa_envoy.launchers.CalibrationLauncher.launch_casa") as cal_launcher:
+ palaver.main()
+ assert cal_launcher.call_count == 1
+
+ args.standard_cal = None
+
+ @patch("os.chdir")
+ @patch("os.getcwd")
+ def test_main_img(self, mock_cwd, mock_chdir):
+ args.standard_img = ["test/image-metadata.json", "test/cmsimage-PPR.xml"]
+
+ with patch("argparse.ArgumentParser.parse_args", MagicMock(return_value=args)) as mock_args:
+ with patch("casa_envoy.launchers.ImagingLauncher.launch_casa") as img_launcher:
+ palaver.main()
+ assert img_launcher.call_count == 1
+
+ args.standard_img = None
diff --git a/apps/cli/executables/pexable/casa_envoy/test/test_launchers.py b/apps/cli/executables/pexable/casa_envoy/test/test_launchers.py
index dfe285a1c..b3f714042 100644
--- a/apps/cli/executables/pexable/casa_envoy/test/test_launchers.py
+++ b/apps/cli/executables/pexable/casa_envoy/test/test_launchers.py
@@ -1,68 +1,73 @@
import os
from unittest.mock import mock_open, patch
-from casa_envoy.launchers import CalibrationLauncher, ImagingLauncher
+from casa_envoy.launchers import CalibrationLauncher, ImagingLauncher, CasaLauncher
+cal_parameters = {
+ "useCasa": False,
+ "homeForReprocessing": "/home/casa/packages/pipeline/current",
+ "rootDirectory": "/tmp/workspaces_tmp/",
+ "processingDirectory": "tmpiox5trbp",
+ "metadata": "test/test.json",
+ "ppr": "test/PPR.xml",
+ "product_type": "standard-cal",
+}
+img_parameters = {
+ "useCasa": False,
+ "homeForReprocessing": "/home/casa/packages/pipeline/current",
+ "rootDirectory": "/tmp/workspaces_tmp/",
+ "processingDirectory": "tmpiox5trbp",
+ "metadata": "test/image-metadata.json",
+ "ppr": "test/cmsimage-PPR.xml",
+ "product_type": "standard-img",
+}
-class TestCalibrationLauncher:
- @patch("subprocess.Popen")
+
+class TestCasaLauncher:
+ def test_setup_environment(self):
+ CasaLauncher(parameters=cal_parameters).setup_environment()
+ assert os.environ.get("SCIPIPE_ROOTDIR") == cal_parameters["rootDirectory"]
+ assert os.environ.get("CASA_HOME") == cal_parameters["homeForReprocessing"]
+ assert os.environ.get("PPR_FILENAME") == "test/PPR.xml"
+
+ @patch("subprocess.run")
@patch("os.chdir")
def test_run(self, mock_os, mock_subprocess):
- CalibrationLauncher("test/PPR.xml", "test/test.json").run()
+ CasaLauncher(parameters=cal_parameters).run()
assert mock_subprocess.call_count == 1
- def test_setup_environment(self):
- parameters = {
- "useCasa": False,
- "homeForReprocessing": "/home/casa/packages/pipeline/current",
- "rootDirectory": "/tmp/workspaces_tmp/",
- "processingDirectory": "tmpiox5trbp",
- }
- CalibrationLauncher("PPR.xml", "test.json").setup_environment(parameters=parameters)
- assert os.environ.get("SCIPIPE_ROOTDIR") == parameters["rootDirectory"]
- assert os.environ.get("CASA_HOME") == parameters["homeForReprocessing"]
- assert os.environ.get("PPR_FILENAME") == "PPR.xml"
-
@patch("builtins.open")
@patch("glob.glob")
@patch("os.chdir")
@patch("os.getcwd")
def test_check_logs(self, mock_os_cwd, mock_os_dir, mock_glob, mock_open):
- CalibrationLauncher("PPR.xml", "test.json").check_logs(parent_path=".")
+ CasaLauncher(parameters=cal_parameters).check_logs()
assert mock_os_cwd.call_count == 1
assert mock_os_dir.call_count == 0
assert mock_glob.call_count == 1
-class TestImagingLauncher:
- @patch("os.chdir")
- def test_run(self, mock_os):
- with patch("subprocess.Popen") as sp:
- ImagingLauncher("test/cmsimage-PPR.xml", "test/image-metadata.json").run()
- assert sp.call_count == 1
+class TestCalibrationLauncher:
+ @patch("casa_envoy.launchers.CalibrationLauncher.run_audit")
+ @patch("casa_envoy.launchers.CasaLauncher.run")
+ def test_launch_casa(self, mock_run, mock_audit):
+ CalibrationLauncher(parameters=cal_parameters).launch_casa()
+ assert mock_run.call_count == 1
+ assert mock_audit.call_count == 1
- def test_setup_environment(self):
- parameters = {
- "useCasa": False,
- "homeForReprocessing": "/home/casa/packages/pipeline/current",
- "rootDirectory": "/tmp/workspaces_tmp/",
- "processingDirectory": "tmpiox5trbp",
- }
- ppr = "cmsimage-PPR.xml"
- metadata = "image-metadata.json"
- ImagingLauncher(ppr, metadata).setup_environment(parameters=parameters)
- assert os.environ.get("SCIPIPE_ROOTDIR") == parameters["rootDirectory"]
- assert os.environ.get("CASA_HOME") == parameters["homeForReprocessing"]
- assert os.environ.get("PPR_FILENAME") == ppr
+ def test_check_calibratable(self):
+ check = CalibrationLauncher(parameters=cal_parameters).check_calibratable()
+ assert check is True
- @patch("builtins.open")
- @patch("glob.glob")
- @patch("os.chdir")
- @patch("os.getcwd")
- def test_check_logs(self, mock_os_cwd, mock_os_dir, mock_glob, mock_open):
- ppr = "cmsimage-PPR.xml"
- metadata = "image-metadata.json"
- ImagingLauncher(ppr, metadata).check_logs(parent_path=".")
- assert mock_os_cwd.call_count == 1
- assert mock_os_dir.call_count == 0
- assert mock_glob.call_count == 1
+
+class TestImagingLauncher:
+ @patch("casa_envoy.launchers.CalibrationLauncher.run_audit")
+ @patch("casa_envoy.launchers.CasaLauncher.run")
+ def test_launch_casa(self, mock_run, mock_audit):
+ CalibrationLauncher(parameters=img_parameters).launch_casa()
+ assert mock_run.call_count == 1
+ assert mock_audit.call_count == 1
+
+ def test_check_imageable(self):
+ check = ImagingLauncher(parameters=img_parameters).check_imageable()
+ assert check is True
--
GitLab