NikolajDanger/meow_base
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

reformted jobs being passed to conductors so they only get a job directory and have to read the definitions from the appropriate files

Browse Source
This commit is contained in:
PatchOfScotland
2023-02-09 15:22:26 +01:00
parent 1eb022f79e
commit a2df62c693
19 changed files with 528 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -51,7 +51,7 @@ coverage.xml
.hypothesis/
.pytest_cache/
tests/test_monitor_base
tests/test_handler_base
tests/test_job_queue_dir_dir_dir_dir_dir_dir_dir_dir_dir_dir
tests/test_job_output
# Translations

46
conductors/local_python_conductor.py
View File

@ -12,20 +12,26 @@ from datetime import datetime
from typing import Any, Tuple, Dict
from core.correctness.vars import JOB_TYPE_PYTHON, PYTHON_FUNC, JOB_STATUS, \
STATUS_RUNNING, JOB_START_TIME, PYTHON_EXECUTION_BASE, JOB_ID, META_FILE, \
STATUS_DONE, JOB_END_TIME, STATUS_FAILED, JOB_ERROR, PYTHON_OUTPUT_DIR, \
JOB_TYPE, JOB_TYPE_PAPERMILL
from core.correctness.validation import valid_job
from core.functionality import read_yaml, write_yaml
STATUS_RUNNING, JOB_START_TIME, JOB_ID, META_FILE, \
STATUS_DONE, JOB_END_TIME, STATUS_FAILED, JOB_ERROR, \
JOB_TYPE, JOB_TYPE_PAPERMILL, DEFAULT_JOB_QUEUE_DIR, DEFAULT_JOB_OUTPUT_DIR
from core.correctness.validation import valid_job, valid_dir_path
from core.functionality import read_yaml, write_yaml, make_dir
from core.meow import BaseConductor
class LocalPythonConductor(BaseConductor):
def __init__(self)->None:
def __init__(self, job_queue_dir:str=DEFAULT_JOB_QUEUE_DIR,
job_output_dir:str=DEFAULT_JOB_OUTPUT_DIR)->None:
"""LocalPythonConductor Constructor. This should be used to execute
Python jobs, and will then pass any internal job runner files to the
output directory."""
output directory. Note that if this handler is given to a MeowRunner
object, the job_queue_dir and job_output_dir will be overwridden."""
super().__init__()
self._is_valid_job_queue_dir(job_queue_dir)
self.job_queue_dir = job_queue_dir
self._is_valid_job_output_dir(job_output_dir)
self.job_output_dir = job_output_dir
def valid_execute_criteria(self, job:Dict[str,Any])->Tuple[bool,str]:
"""Function to determine given an job defintion, if this conductor can
@ -38,16 +44,17 @@ class LocalPythonConductor(BaseConductor):
pass
return False, str(e)
def execute(self, job:Dict[str,Any])->None:
def execute(self, job_dir:str)->None:
"""Function to actually execute a Python job. This will read job
defintions from its meta file, update the meta file and attempt to
execute. Some unspecific feedback will be given on execution failure,
but depending on what it is it may be up to the job itself to provide
more detailed feedback."""
valid_job(job)
valid_dir_path(job_dir, must_exist=True)
job_dir = os.path.join(job[PYTHON_EXECUTION_BASE], job[JOB_ID])
meta_file = os.path.join(job_dir, META_FILE)
job = read_yaml(meta_file)
valid_job(job)
# update the status file with running status
job[JOB_STATUS] = STATUS_RUNNING
@ -57,7 +64,7 @@ class LocalPythonConductor(BaseConductor):
# execute the job
try:
job_function = job[PYTHON_FUNC]
job_function(job)
job_function(job_dir)
# get up to date job data
job = read_yaml(meta_file)
@ -83,5 +90,20 @@ class LocalPythonConductor(BaseConductor):
# Move the contents of the execution directory to the final output
# directory.
job_output_dir = os.path.join(job[PYTHON_OUTPUT_DIR], job[JOB_ID])
job_output_dir = \
os.path.join(self.job_output_dir, os.path.basename(job_dir))
shutil.move(job_dir, job_output_dir)
def _is_valid_job_queue_dir(self, job_queue_dir)->None:
"""Validation check for 'job_queue_dir' variable from main
constructor."""
valid_dir_path(job_queue_dir, must_exist=False)
if not os.path.exists(job_queue_dir):
make_dir(job_queue_dir)
def _is_valid_job_output_dir(self, job_output_dir)->None:
"""Validation check for 'job_output_dir' variable from main
constructor."""
valid_dir_path(job_output_dir, must_exist=False)
if not os.path.exists(job_output_dir):
make_dir(job_output_dir)

31
core/correctness/validation.py
View File

@ -44,7 +44,7 @@ JOB_KEYS = {
}
def check_type(variable:Any, expected_type:Type, alt_types:List[Type]=[],
or_none:bool=False)->None:
or_none:bool=False, hint:str="")->None:
"""Checks if a given variable is of the expected type. Raises TypeError or
ValueError as appropriate if any issues are encountered."""
@ -57,9 +57,11 @@ def check_type(variable:Any, expected_type:Type, alt_types:List[Type]=[],
# Only accept None if explicitly allowed
if variable is None:
if or_none == False:
raise TypeError(
f'Not allowed None for variable. Expected {expected_type}.'
)
if hint:
msg = f"Not allowed None for {hint}. Expected {expected_type}."
else:
msg = f"Not allowed None. Expected {expected_type}."
raise TypeError(msg)
else:
return
@ -69,10 +71,12 @@ def check_type(variable:Any, expected_type:Type, alt_types:List[Type]=[],
# Check that variable type is within the accepted type list
if not isinstance(variable, tuple(type_list)):
raise TypeError(
'Expected type(s) are %s, got %s'
% (get_args(expected_type), type(variable))
)
if hint:
msg = f"Expected type(s) for {hint} are '{type_list}', " \
f"got {type(variable)}"
else:
msg = f"Expected type(s) are '{type_list}', got {type(variable)}"
raise TypeError(msg)
def check_callable(call:Any)->None:
"""Checks if a given variable is a callable function. Raises TypeError if
@ -216,16 +220,19 @@ def valid_existing_file_path(variable:str, allow_base:bool=False,
raise ValueError(
f"Requested file '{variable}' is not a file.")
def valid_existing_dir_path(variable:str, allow_base:bool=False):
"""Check the given string is a path to an existing directory."""
def valid_dir_path(variable:str, must_exist:bool=False, allow_base:bool=False
)->None:
"""Check the given string is a valid directory path, either to an existing
one or a location that could contain one."""
# Check that the string is a path
valid_path(variable, allow_base=allow_base, extension="")
# Check the path exists
if not exists(variable):
does_exist = exists(variable)
if must_exist and not does_exist:
raise FileNotFoundError(
f"Requested dir path '{variable}' does not exist.")
# Check it is a directory
if not isdir(variable):
if does_exist and not isdir(variable):
raise ValueError(
f"Requested dir '{variable}' is not a directory.")

6
core/correctness/vars.py
View File

@ -77,13 +77,15 @@ DIR_EVENTS = [
DIR_RETROACTIVE_EVENT
]
# runner defaults
DEFAULT_JOB_QUEUE_DIR = "job_queue"
DEFAULT_JOB_OUTPUT_DIR = "job_output"
# meow jobs
JOB_TYPE = "job_type"
JOB_TYPE_PYTHON = "python"
JOB_TYPE_PAPERMILL = "papermill"
PYTHON_FUNC = "func"
PYTHON_EXECUTION_BASE = "exection_base"
PYTHON_OUTPUT_DIR = "output_dir"
JOB_TYPES = {
JOB_TYPE_PAPERMILL: [

10
core/functionality.py
View File

@ -88,8 +88,8 @@ def _get_file_sha256(file_path):
return sha256_hash.hexdigest()
def get_file_hash(file_path:str, hash:str):
check_type(hash, str)
def get_file_hash(file_path:str, hash:str, hint:str=""):
check_type(hash, str, hint=hint)
valid_existing_file_path(file_path)
@ -204,7 +204,8 @@ def write_notebook(source:Dict[str,Any], filename:str):
def parameterize_jupyter_notebook(jupyter_notebook:Dict[str,Any],
parameters:Dict[str,Any], expand_env_values:bool=False)->Dict[str,Any]:
nbformat.validate(jupyter_notebook)
check_type(parameters, Dict)
check_type(parameters, Dict,
hint="parameterize_jupyter_notebook.parameters")
if jupyter_notebook["nbformat"] != 4:
raise Warning(
@ -277,7 +278,8 @@ def parameterize_jupyter_notebook(jupyter_notebook:Dict[str,Any],
def parameterize_python_script(script:List[str], parameters:Dict[str,Any],
expand_env_values:bool=False)->Dict[str,Any]:
check_script(script)
check_type(parameters, Dict)
check_type(parameters, Dict
,hint="parameterize_python_script.parameters")
output_script = copy.deepcopy(script)

56
core/meow.py
View File

@ -142,7 +142,7 @@ class BasePattern:
"""Validation check for 'sweep' variable from main constructor. This
function is implemented to check for the types given in the signature,
and must be overridden if these differ."""
check_type(sweep, Dict)
check_type(sweep, Dict, hint="BasePattern.sweep")
if not sweep:
return
for _, v in sweep.items():
@ -152,11 +152,23 @@ class BasePattern:
], strict=True)
check_type(
v[SWEEP_START], expected_type=int, alt_types=[float, complex])
v[SWEEP_START],
expected_type=int,
alt_types=[float, complex],
hint=f"BasePattern.sweep[{SWEEP_START}]"
)
check_type(
v[SWEEP_STOP], expected_type=int, alt_types=[float, complex])
v[SWEEP_STOP],
expected_type=int,
alt_types=[float, complex],
hint=f"BasePattern.sweep[{SWEEP_STOP}]"
)
check_type(
v[SWEEP_JUMP], expected_type=int, alt_types=[float, complex])
v[SWEEP_JUMP],
expected_type=int,
alt_types=[float, complex],
hint=f"BasePattern.sweep[{SWEEP_JUMP}]"
)
# Try to check that this loop is not infinite
if v[SWEEP_JUMP] == 0:
raise ValueError(
@ -215,8 +227,8 @@ class BaseRule:
self.pattern = pattern
self._is_valid_recipe(recipe)
self.recipe = recipe
check_type(pattern, BasePattern)
check_type(recipe, BaseRecipe)
check_type(pattern, BasePattern, hint="BaseRule.pattern")
check_type(recipe, BaseRecipe, hint="BaseRule.recipe")
if pattern.recipe != recipe.name:
raise ValueError(f"Cannot create Rule {name}. Pattern "
f"{pattern.name} does not identify Recipe {recipe.name}. It "
@ -369,10 +381,14 @@ class BaseMonitor:
class BaseHandler:
# A channel for sending messages to the runner. Note that this is not
# initialised within the constructor, but within the runner when passed the
# handler is passed to it.
# A channel for sending messages to the runner. Note that this will be
# overridden by a MeowRunner, if a handler instance is passed to it, and so
# does not need to be initialised within the handler itself.
to_runner: VALID_CHANNELS
# Directory where queued jobs are initially written to. Note that this
# will be overridden by a MeowRunner, if a handler instance is passed to
# it, and so does not need to be initialised within the handler itself.
job_queue_dir:str
def __init__(self)->None:
"""BaseHandler Constructor. This will check that any class inheriting
from it implements its validation functions."""
@ -399,6 +415,14 @@ class BaseHandler:
class BaseConductor:
# Directory where queued jobs are initially written to. Note that this
# will be overridden by a MeowRunner, if a handler instance is passed to
# it, and so does not need to be initialised within the handler itself.
job_queue_dir:str
# Directory where completed jobs are finally written to. Note that this
# will be overridden by a MeowRunner, if a handler instance is passed to
# it, and so does not need to be initialised within the handler itself.
job_output_dir:str
def __init__(self)->None:
"""BaseConductor Constructor. This will check that any class inheriting
from it implements its validation functions."""
@ -418,9 +442,9 @@ class BaseConductor:
process it or not. Must be implemented by any child process."""
pass
def execute(self, job:Dict[str,Any])->None:
"""Function to execute a given job. Must be implemented by any child
process."""
def execute(self, job_dir:str)->None:
"""Function to execute a given job directory. Must be implemented by
any child process."""
pass
@ -433,8 +457,8 @@ def create_rules(patterns:Union[Dict[str,BasePattern],List[BasePattern]],
provided pattern and recipe dictionaries must be keyed with the
corresponding pattern and recipe names."""
# Validation of inputs
check_type(patterns, Dict, alt_types=[List])
check_type(recipes, Dict, alt_types=[List])
check_type(patterns, Dict, alt_types=[List], hint="create_rules.patterns")
check_type(recipes, Dict, alt_types=[List], hint="create_rules.recipes")
valid_list(new_rules, BaseRule, min_length=0)
# Convert a pattern list to a dictionary
@ -481,8 +505,8 @@ def create_rule(pattern:BasePattern, recipe:BaseRecipe,
"""Function to create a valid rule from a given pattern and recipe. All
inbuilt rule types are considered, with additional definitions provided
through the 'new_rules' variable."""
check_type(pattern, BasePattern)
check_type(recipe, BaseRecipe)
check_type(pattern, BasePattern, hint="create_rule.pattern")
check_type(recipe, BaseRecipe, hint="create_rule.recipe")
valid_list(new_rules, BaseRule, min_length=0)
# Imported here to avoid circular imports at top of file

91
core/runner.py
View File

@ -15,10 +15,11 @@ from random import randrange
from typing import Any, Union, Dict, List
from core.correctness.vars import DEBUG_WARNING, DEBUG_INFO, EVENT_TYPE, \
VALID_CHANNELS, JOB_ID, META_FILE
VALID_CHANNELS, JOB_ID, META_FILE, DEFAULT_JOB_OUTPUT_DIR, \
DEFAULT_JOB_QUEUE_DIR
from core.correctness.validation import setup_debugging, check_type, \
valid_list
from core.functionality import print_debug, wait, read_yaml
valid_list, valid_dir_path
from core.functionality import print_debug, wait, read_yaml, make_dir
from core.meow import BaseHandler, BaseMonitor, BaseConductor
@ -33,18 +34,32 @@ class MeowRunner:
from_monitors: List[VALID_CHANNELS]
# A collection of all channels from each handler
from_handlers: List[VALID_CHANNELS]
# Directory where queued jobs are initially written to
job_queue_dir:str
# Directory where completed jobs are finally written to
job_output_dir:str
def __init__(self, monitors:Union[BaseMonitor,List[BaseMonitor]],
handlers:Union[BaseHandler,List[BaseHandler]],
conductors:Union[BaseConductor,List[BaseConductor]],
job_queue_dir:str=DEFAULT_JOB_QUEUE_DIR,
job_output_dir:str=DEFAULT_JOB_OUTPUT_DIR,
print:Any=sys.stdout, logging:int=0)->None:
"""MeowRunner constructor. This connects all provided monitors,
handlers and conductors according to what events and jobs they produce
or consume."""
self._is_valid_job_queue_dir(job_queue_dir)
self._is_valid_job_output_dir(job_output_dir)
self._is_valid_conductors(conductors)
# If conductors isn't a list, make it one
if not type(conductors) == list:
conductors = [conductors]
for conductor in conductors:
conductor.job_output_dir = job_output_dir
conductor.job_queue_dir = job_queue_dir
self.conductors = conductors
self._is_valid_handlers(handlers)
@ -56,6 +71,7 @@ class MeowRunner:
# Create a channel from the handler back to this runner
handler_to_runner_reader, handler_to_runner_writer = Pipe()
handler.to_runner = handler_to_runner_writer
handler.job_queue_dir = job_queue_dir
self.from_handlers.append(handler_to_runner_reader)
self.handlers = handlers
@ -170,13 +186,13 @@ class MeowRunner:
# If we've only one conductor, use that
if len(valid_conductors) == 1:
conductor = valid_conductors[0]
self.execute_job(conductor, job)
self.execute_job(conductor, job_dir)
# If multiple handlers then randomly pick one
else:
conductor = valid_conductors[
randrange(len(valid_conductors))
]
self.execute_job(conductor, job)
self.execute_job(conductor, job_dir)
def handle_event(self, handler:BaseHandler, event:Dict[str,Any])->None:
"""Function for a given handler to handle a given event, without
@ -193,19 +209,31 @@ class MeowRunner:
"Something went wrong during handling for event "
f"'{event[EVENT_TYPE]}'. {e}", DEBUG_INFO)
def execute_job(self, conductor:BaseConductor, job:Dict[str,Any])->None:
def execute_job(self, conductor:BaseConductor, job_dir:str)->None:
"""Function for a given conductor to execute a given job, without
crashing the runner in the event of a problem."""
print_debug(self._print_target, self.debug_level,
f"Starting execution for job: '{job[JOB_ID]}'", DEBUG_INFO)
job_id = os.path.basename(job_dir)
print_debug(
self._print_target,
self.debug_level,
f"Starting execution for job: '{job_id}'",
DEBUG_INFO
)
try:
conductor.execute(job)
print_debug(self._print_target, self.debug_level,
f"Completed execution for job: '{job[JOB_ID]}'", DEBUG_INFO)
conductor.execute(job_dir)
print_debug(
self._print_target,
self.debug_level,
f"Completed execution for job: '{job_id}'",
DEBUG_INFO
)
except Exception as e:
print_debug(self._print_target, self.debug_level,
"Something went wrong during execution for job "
f"'{job[JOB_ID]}'. {e}", DEBUG_INFO)
print_debug(
self._print_target,
self.debug_level,
f"Something went wrong in execution of job '{job_id}'. {e}",
DEBUG_INFO
)
def start(self)->None:
"""Function to start the runner by starting all of the constituent
@ -308,20 +336,49 @@ class MeowRunner:
def _is_valid_monitors(self,
monitors:Union[BaseMonitor,List[BaseMonitor]])->None:
"""Validation check for 'monitors' variable from main constructor."""
check_type(monitors, BaseMonitor, alt_types=[List])
check_type(
monitors,
BaseMonitor,
alt_types=[List],
hint="MeowRunner.monitors"
)
if type(monitors) == list:
valid_list(monitors, BaseMonitor, min_length=1)
def _is_valid_handlers(self,
handlers:Union[BaseHandler,List[BaseHandler]])->None:
"""Validation check for 'handlers' variable from main constructor."""
check_type(handlers, BaseHandler, alt_types=[List])
check_type(
handlers,
BaseHandler,
alt_types=[List],
hint="MeowRunner.handlers"
)
if type(handlers) == list:
valid_list(handlers, BaseHandler, min_length=1)
def _is_valid_conductors(self,
conductors:Union[BaseConductor,List[BaseConductor]])->None:
"""Validation check for 'conductors' variable from main constructor."""
check_type(conductors, BaseConductor, alt_types=[List])
check_type(
conductors,
BaseConductor,
alt_types=[List],
hint="MeowRunner.conductors"
)
if type(conductors) == list:
valid_list(conductors, BaseConductor, min_length=1)
def _is_valid_job_queue_dir(self, job_queue_dir)->None:
"""Validation check for 'job_queue_dir' variable from main
constructor."""
valid_dir_path(job_queue_dir, must_exist=False)
if not os.path.exists(job_queue_dir):
make_dir(job_queue_dir)
def _is_valid_job_output_dir(self, job_output_dir)->None:
"""Validation check for 'job_output_dir' variable from main
constructor."""
valid_dir_path(job_output_dir, must_exist=False)
if not os.path.exists(job_output_dir):
make_dir(job_output_dir)

29
patterns/file_event_pattern.py
View File

@ -19,8 +19,7 @@ from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler
from core.correctness.validation import check_type, valid_string, \
valid_dict, valid_list, valid_path, valid_existing_dir_path, \
setup_debugging
valid_dict, valid_list, valid_path, valid_dir_path, setup_debugging
from core.correctness.vars import VALID_RECIPE_NAME_CHARS, \
VALID_VARIABLE_NAME_CHARS, FILE_EVENTS, FILE_CREATE_EVENT, \
FILE_MODIFY_EVENT, FILE_MOVED_EVENT, DEBUG_INFO, \
@ -134,7 +133,7 @@ class WatchdogMonitor(BaseMonitor):
super().__init__(patterns, recipes)
self._is_valid_base_dir(base_dir)
self.base_dir = base_dir
check_type(settletime, int)
check_type(settletime, int, hint="WatchdogMonitor.settletime")
self._print_target, self.debug_level = setup_debugging(print, logging)
self._patterns_lock = threading.Lock()
self._recipes_lock = threading.Lock()
@ -218,7 +217,7 @@ class WatchdogMonitor(BaseMonitor):
"""Function to add a pattern to the current definitions. Any rules
that can be possibly created from that pattern will be automatically
created."""
check_type(pattern, FileEventPattern)
check_type(pattern, FileEventPattern, hint="add_pattern.pattern")
self._patterns_lock.acquire()
try:
if pattern.name in self._patterns:
@ -235,14 +234,19 @@ class WatchdogMonitor(BaseMonitor):
def update_pattern(self, pattern:FileEventPattern)->None:
"""Function to update a pattern in the current definitions. Any rules
created from that pattern will be automatically updated."""
check_type(pattern, FileEventPattern)
check_type(pattern, FileEventPattern, hint="update_pattern.pattern")
self.remove_pattern(pattern.name)
self.add_pattern(pattern)
def remove_pattern(self, pattern: Union[str,FileEventPattern])->None:
"""Function to remove a pattern from the current definitions. Any rules
that will be no longer valid will be automatically removed."""
check_type(pattern, str, alt_types=[FileEventPattern])
check_type(
pattern,
str,
alt_types=[FileEventPattern],
hint="remove_pattern.pattern"
)
lookup_key = pattern
if isinstance(lookup_key, FileEventPattern):
lookup_key = pattern.name
@ -280,7 +284,7 @@ class WatchdogMonitor(BaseMonitor):
"""Function to add a recipe to the current definitions. Any rules
that can be possibly created from that recipe will be automatically
created."""
check_type(recipe, BaseRecipe)
check_type(recipe, BaseRecipe, hint="add_recipe.recipe")
self._recipes_lock.acquire()
try:
if recipe.name in self._recipes:
@ -297,14 +301,19 @@ class WatchdogMonitor(BaseMonitor):
def update_recipe(self, recipe: BaseRecipe)->None:
"""Function to update a recipe in the current definitions. Any rules
created from that recipe will be automatically updated."""
check_type(recipe, BaseRecipe)
check_type(recipe, BaseRecipe, hint="update_recipe.recipe")
self.remove_recipe(recipe.name)
self.add_recipe(recipe)
def remove_recipe(self, recipe:Union[str,BaseRecipe])->None:
"""Function to remove a recipe from the current definitions. Any rules
that will be no longer valid will be automatically removed."""
check_type(recipe, str, alt_types=[BaseRecipe])
check_type(
recipe,
str,
alt_types=[BaseRecipe],
hint="remove_recipe.recipe"
)
lookup_key = recipe
if isinstance(lookup_key, BaseRecipe):
lookup_key = recipe.name
@ -449,7 +458,7 @@ class WatchdogMonitor(BaseMonitor):
def _is_valid_base_dir(self, base_dir:str)->None:
"""Validation check for 'base_dir' variable from main constructor. Is
automatically called during initialisation."""
valid_existing_dir_path(base_dir)
valid_dir_path(base_dir, must_exist=True)
def _is_valid_patterns(self, patterns:Dict[str,FileEventPattern])->None:
"""Validation check for 'patterns' variable from main constructor. Is

54
recipes/jupyter_notebook_recipe.py
View File

@ -12,14 +12,14 @@ import sys
from typing import Any, Tuple, Dict
from core.correctness.validation import check_type, valid_string, \
valid_dict, valid_path, valid_existing_dir_path, setup_debugging, \
valid_dict, valid_path, valid_dir_path, setup_debugging, \
valid_event
from core.correctness.vars import VALID_VARIABLE_NAME_CHARS, PYTHON_FUNC, \
DEBUG_INFO, EVENT_TYPE_WATCHDOG, JOB_HASH, PYTHON_EXECUTION_BASE, \
DEBUG_INFO, EVENT_TYPE_WATCHDOG, JOB_HASH, DEFAULT_JOB_QUEUE_DIR, \
EVENT_PATH, JOB_TYPE_PAPERMILL, WATCHDOG_HASH, JOB_PARAMETERS, \
PYTHON_OUTPUT_DIR, JOB_ID, WATCHDOG_BASE, META_FILE, \
JOB_ID, WATCHDOG_BASE, META_FILE, \
PARAMS_FILE, JOB_STATUS, STATUS_QUEUED, EVENT_RULE, EVENT_TYPE, \
EVENT_RULE, get_base_file, get_job_file, get_result_file
EVENT_RULE, get_base_file
from core.functionality import print_debug, create_job, replace_keywords, \
make_dir, write_yaml, write_notebook, read_notebook
from core.meow import BaseRecipe, BaseHandler
@ -44,7 +44,7 @@ class JupyterNotebookRecipe(BaseRecipe):
def _is_valid_recipe(self, recipe:Dict[str,Any])->None:
"""Validation check for 'recipe' variable from main constructor.
Called within parent BaseRecipe constructor."""
check_type(recipe, Dict)
check_type(recipe, Dict, hint="JupyterNotebookRecipe.recipe")
nbformat.validate(recipe)
def _is_valid_parameters(self, parameters:Dict[str,Any])->None:
@ -62,26 +62,20 @@ class JupyterNotebookRecipe(BaseRecipe):
valid_string(k, VALID_VARIABLE_NAME_CHARS)
class PapermillHandler(BaseHandler):
# handler directory to setup jobs in
handler_base:str
# TODO move me to conductor?
# Final location for job output to be placed
output_dir:str
# Config option, above which debug messages are ignored
debug_level:int
# Where print messages are sent
_print_target:Any
def __init__(self, handler_base:str, output_dir:str, print:Any=sys.stdout,
logging:int=0)->None:
def __init__(self, job_queue_dir:str=DEFAULT_JOB_QUEUE_DIR,
print:Any=sys.stdout, logging:int=0)->None:
"""PapermillHandler Constructor. This creats jobs to be executed using
the papermill module. This does not run as a continuous thread to
handle execution, but is invoked according to a factory pattern using
the handle function."""
the handle function. Note that if this handler is given to a MeowRunner
object, the job_queue_dir will be overwridden."""
super().__init__()
self._is_valid_handler_base(handler_base)
self.handler_base = handler_base
self._is_valid_output_dir(output_dir)
self.output_dir = output_dir
self._is_valid_job_queue_dir(job_queue_dir)
self.job_queue_dir = job_queue_dir
self._print_target, self.debug_level = setup_debugging(print, logging)
print_debug(self._print_target, self.debug_level,
"Created new PapermillHandler instance", DEBUG_INFO)
@ -125,15 +119,12 @@ class PapermillHandler(BaseHandler):
pass
return False, str(e)
def _is_valid_handler_base(self, handler_base)->None:
"""Validation check for 'handler_base' variable from main
def _is_valid_job_queue_dir(self, job_queue_dir)->None:
"""Validation check for 'job_queue_dir' variable from main
constructor."""
valid_existing_dir_path(handler_base)
def _is_valid_output_dir(self, output_dir)->None:
"""Validation check for 'output_dir' variable from main
constructor."""
valid_existing_dir_path(output_dir, allow_base=True)
valid_dir_path(job_queue_dir, must_exist=False)
if not os.path.exists(job_queue_dir):
make_dir(job_queue_dir)
def setup_job(self, event:Dict[str,Any], yaml_dict:Dict[str,Any])->None:
"""Function to set up new job dict and send it to the runner to be
@ -145,8 +136,6 @@ class PapermillHandler(BaseHandler):
JOB_PARAMETERS:yaml_dict,
JOB_HASH: event[WATCHDOG_HASH],
PYTHON_FUNC:papermill_job_func,
PYTHON_OUTPUT_DIR:self.output_dir,
PYTHON_EXECUTION_BASE:self.handler_base
}
)
print_debug(self._print_target, self.debug_level,
@ -162,8 +151,7 @@ class PapermillHandler(BaseHandler):
)
# Create a base job directory
job_dir = os.path.join(
meow_job[PYTHON_EXECUTION_BASE], meow_job[JOB_ID])
job_dir = os.path.join(self.job_queue_dir, meow_job[JOB_ID])
make_dir(job_dir)
# write a status file to the job directory
@ -187,7 +175,7 @@ class PapermillHandler(BaseHandler):
self.to_runner.send(job_dir)
# Papermill job execution code, to be run within the conductor
def papermill_job_func(job):
def papermill_job_func(job_dir):
# Requires own imports as will be run in its own execution environment
import os
import papermill
@ -197,11 +185,11 @@ def papermill_job_func(job):
from core.correctness.vars import JOB_EVENT, JOB_ID, \
EVENT_PATH, META_FILE, PARAMS_FILE, \
JOB_STATUS, JOB_HASH, SHA256, STATUS_SKIPPED, JOB_END_TIME, \
JOB_ERROR, STATUS_FAILED, PYTHON_EXECUTION_BASE, get_job_file, \
JOB_ERROR, STATUS_FAILED, get_job_file, \
get_result_file
# Identify job files
job_dir = os.path.join(job[PYTHON_EXECUTION_BASE], job[JOB_ID])
meta_file = os.path.join(job_dir, META_FILE)
# TODO fix these paths so they are dynamic
base_file = os.path.join(job_dir, get_base_file(JOB_TYPE_PAPERMILL))
@ -209,6 +197,8 @@ def papermill_job_func(job):
result_file = os.path.join(job_dir, get_result_file(JOB_TYPE_PAPERMILL))
param_file = os.path.join(job_dir, PARAMS_FILE)
# Get job defintions
job = read_yaml(meta_file)
yaml_dict = read_yaml(param_file)
# Check the hash of the triggering file, if present. This addresses

52
recipes/python_recipe.py
View File

@ -11,11 +11,11 @@ import sys
from typing import Any, Tuple, Dict, List
from core.correctness.validation import check_script, valid_string, \
valid_dict, valid_event, valid_existing_dir_path, setup_debugging
valid_dict, valid_event, valid_dir_path, setup_debugging
from core.correctness.vars import VALID_VARIABLE_NAME_CHARS, PYTHON_FUNC, \
DEBUG_INFO, EVENT_TYPE_WATCHDOG, JOB_HASH, PYTHON_EXECUTION_BASE, \
DEBUG_INFO, EVENT_TYPE_WATCHDOG, JOB_HASH, DEFAULT_JOB_QUEUE_DIR, \
EVENT_RULE, EVENT_PATH, JOB_TYPE_PYTHON, WATCHDOG_HASH, JOB_PARAMETERS, \
PYTHON_OUTPUT_DIR, JOB_ID, WATCHDOG_BASE, META_FILE, \
JOB_ID, WATCHDOG_BASE, META_FILE, \
PARAMS_FILE, JOB_STATUS, STATUS_QUEUED, EVENT_TYPE, EVENT_RULE, \
get_base_file
from core.functionality import print_debug, create_job, replace_keywords, \
@ -51,27 +51,20 @@ class PythonRecipe(BaseRecipe):
class PythonHandler(BaseHandler):
# TODO move me to base handler
# handler directory to setup jobs in
handler_base:str
# TODO move me to conductor?
# Final location for job output to be placed
output_dir:str
# Config option, above which debug messages are ignored
debug_level:int
# Where print messages are sent
_print_target:Any
def __init__(self, handler_base:str, output_dir:str, print:Any=sys.stdout,
logging:int=0)->None:
def __init__(self, job_queue_dir:str=DEFAULT_JOB_QUEUE_DIR,
print:Any=sys.stdout, logging:int=0)->None:
"""PythonHandler Constructor. This creates jobs to be executed as
python functions. This does not run as a continuous thread to
handle execution, but is invoked according to a factory pattern using
the handle function."""
the handle function. Note that if this handler is given to a MeowRunner
object, the job_queue_dir will be overwridden but its"""
super().__init__()
self._is_valid_handler_base(handler_base)
self.handler_base = handler_base
self._is_valid_output_dir(output_dir)
self.output_dir = output_dir
self._is_valid_job_queue_dir(job_queue_dir)
self.job_queue_dir = job_queue_dir
self._print_target, self.debug_level = setup_debugging(print, logging)
print_debug(self._print_target, self.debug_level,
"Created new PythonHandler instance", DEBUG_INFO)
@ -115,15 +108,12 @@ class PythonHandler(BaseHandler):
pass
return False, str(e)
def _is_valid_handler_base(self, handler_base)->None:
"""Validation check for 'handler_base' variable from main
def _is_valid_job_queue_dir(self, job_queue_dir)->None:
"""Validation check for 'job_queue_dir' variable from main
constructor."""
valid_existing_dir_path(handler_base)
def _is_valid_output_dir(self, output_dir)->None:
"""Validation check for 'output_dir' variable from main
constructor."""
valid_existing_dir_path(output_dir, allow_base=True)
valid_dir_path(job_queue_dir, must_exist=False)
if not os.path.exists(job_queue_dir):
make_dir(job_queue_dir)
def setup_job(self, event:Dict[str,Any], yaml_dict:Dict[str,Any])->None:
"""Function to set up new job dict and send it to the runner to be
@ -134,9 +124,7 @@ class PythonHandler(BaseHandler):
extras={
JOB_PARAMETERS:yaml_dict,
JOB_HASH: event[WATCHDOG_HASH],
PYTHON_FUNC:python_job_func,
PYTHON_OUTPUT_DIR:self.output_dir,
PYTHON_EXECUTION_BASE:self.handler_base
PYTHON_FUNC:python_job_func
}
)
print_debug(self._print_target, self.debug_level,
@ -152,8 +140,7 @@ class PythonHandler(BaseHandler):
)
# Create a base job directory
job_dir = os.path.join(
meow_job[PYTHON_EXECUTION_BASE], meow_job[JOB_ID])
job_dir = os.path.join(self.job_queue_dir, meow_job[JOB_ID])
make_dir(job_dir)
# write a status file to the job directory
@ -178,7 +165,7 @@ class PythonHandler(BaseHandler):
# Papermill job execution code, to be run within the conductor
def python_job_func(job):
def python_job_func(job_dir):
# Requires own imports as will be run in its own execution environment
import sys
import os
@ -189,17 +176,18 @@ def python_job_func(job):
from core.correctness.vars import JOB_EVENT, JOB_ID, \
EVENT_PATH, META_FILE, PARAMS_FILE, \
JOB_STATUS, JOB_HASH, SHA256, STATUS_SKIPPED, JOB_END_TIME, \
JOB_ERROR, STATUS_FAILED, PYTHON_EXECUTION_BASE, get_base_file, \
JOB_ERROR, STATUS_FAILED, get_base_file, \
get_job_file, get_result_file
# Identify job files
job_dir = os.path.join(job[PYTHON_EXECUTION_BASE], job[JOB_ID])
meta_file = os.path.join(job_dir, META_FILE)
base_file = os.path.join(job_dir, get_base_file(JOB_TYPE_PYTHON))
job_file = os.path.join(job_dir, get_job_file(JOB_TYPE_PYTHON))
result_file = os.path.join(job_dir, get_result_file(JOB_TYPE_PYTHON))
param_file = os.path.join(job_dir, PARAMS_FILE)
# Get job defintions
job = read_yaml(meta_file)
yaml_dict = read_yaml(param_file)
# Check the hash of the triggering file, if present. This addresses

12
rules/file_event_jupyter_notebook_rule.py
View File

@ -26,9 +26,17 @@ class FileEventJupyterNotebookRule(BaseRule):
def _is_valid_pattern(self, pattern:FileEventPattern)->None:
"""Validation check for 'pattern' variable from main constructor. Is
automatically called during initialisation."""
check_type(pattern, FileEventPattern)
check_type(
pattern,
FileEventPattern,
hint="FileEventJupyterNotebookRule.pattern"
)
def _is_valid_recipe(self, recipe:JupyterNotebookRecipe)->None:
"""Validation check for 'recipe' variable from main constructor. Is
automatically called during initialisation."""
check_type(recipe, JupyterNotebookRecipe)
check_type(
recipe,
JupyterNotebookRecipe,
hint="FileEventJupyterNotebookRule.recipe"
)

12
rules/file_event_python_rule.py
View File

@ -22,9 +22,17 @@ class FileEventPythonRule(BaseRule):
def _is_valid_pattern(self, pattern:FileEventPattern)->None:
"""Validation check for 'pattern' variable from main constructor. Is
automatically called during initialisation."""
check_type(pattern, FileEventPattern)
check_type(
pattern,
FileEventPattern,
hint="FileEventPythonRule.pattern"
)
def _is_valid_recipe(self, recipe:PythonRecipe)->None:
"""Validation check for 'recipe' variable from main constructor. Is
automatically called during initialisation."""
check_type(recipe, PythonRecipe)
check_type(
recipe,
PythonRecipe,
hint="FileEventPythonRule.recipe"
)

14
tests/shared.py
View File

@ -5,23 +5,31 @@ Author(s): David Marchant
"""
import os
from core.correctness.vars import DEFAULT_JOB_OUTPUT_DIR, DEFAULT_JOB_QUEUE_DIR
from core.functionality import make_dir, rmtree
# testing
TEST_DIR = "test_files"
TEST_MONITOR_BASE = "test_monitor_base"
TEST_HANDLER_BASE = "test_handler_base"
TEST_JOB_QUEUE = "test_job_queue_dir"
TEST_JOB_OUTPUT = "test_job_output"
def setup():
make_dir(TEST_DIR, ensure_clean=True)
make_dir(TEST_MONITOR_BASE, ensure_clean=True)
make_dir(TEST_HANDLER_BASE, ensure_clean=True)
make_dir(TEST_JOB_QUEUE, ensure_clean=True)
make_dir(TEST_JOB_OUTPUT, ensure_clean=True)
make_dir(DEFAULT_JOB_OUTPUT_DIR, ensure_clean=True)
make_dir(DEFAULT_JOB_QUEUE_DIR, ensure_clean=True)
def teardown():
rmtree(TEST_DIR)
rmtree(TEST_MONITOR_BASE)
rmtree(TEST_HANDLER_BASE)
rmtree(TEST_JOB_QUEUE)
rmtree(TEST_JOB_OUTPUT)
rmtree(DEFAULT_JOB_OUTPUT_DIR)
rmtree(DEFAULT_JOB_QUEUE_DIR)
rmtree("first")
# Recipe funcs

230
tests/test_conductors.py
View File

@ -5,7 +5,7 @@ import unittest
from typing import Dict
from core.correctness.vars import JOB_TYPE_PYTHON, SHA256, JOB_PARAMETERS, \
JOB_HASH, PYTHON_FUNC, PYTHON_OUTPUT_DIR, PYTHON_EXECUTION_BASE, JOB_ID, \
JOB_HASH, PYTHON_FUNC, JOB_ID, \
META_FILE, PARAMS_FILE, JOB_STATUS, JOB_ERROR, \
STATUS_DONE, JOB_TYPE_PAPERMILL, get_base_file, get_result_file, \
get_job_file
@ -19,7 +19,7 @@ from recipes.jupyter_notebook_recipe import JupyterNotebookRecipe, \
papermill_job_func
from recipes.python_recipe import PythonRecipe, python_job_func
from shared import setup, teardown, TEST_MONITOR_BASE, APPENDING_NOTEBOOK, \
TEST_JOB_OUTPUT, TEST_HANDLER_BASE, COMPLETE_PYTHON_SCRIPT
TEST_JOB_OUTPUT, TEST_JOB_QUEUE, COMPLETE_PYTHON_SCRIPT
def failing_func():
@ -41,7 +41,10 @@ class MeowTests(unittest.TestCase):
# Test LocalPythonConductor executes valid python jobs
def testLocalPythonConductorValidPythonJob(self)->None:
lpc = LocalPythonConductor()
lpc = LocalPythonConductor(
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT
)
file_path = os.path.join(TEST_MONITOR_BASE, "test")
result_path = os.path.join(TEST_MONITOR_BASE, "output")
@ -82,13 +85,11 @@ class MeowTests(unittest.TestCase):
extras={
JOB_PARAMETERS:params_dict,
JOB_HASH: file_hash,
PYTHON_FUNC:python_job_func,
PYTHON_OUTPUT_DIR:TEST_JOB_OUTPUT,
PYTHON_EXECUTION_BASE:TEST_HANDLER_BASE
PYTHON_FUNC:python_job_func
}
)
job_dir = os.path.join(TEST_HANDLER_BASE, job_dict[JOB_ID])
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
make_dir(job_dir)
param_file = os.path.join(job_dir, PARAMS_FILE)
@ -100,14 +101,14 @@ class MeowTests(unittest.TestCase):
base_file = os.path.join(job_dir, get_base_file(JOB_TYPE_PYTHON))
write_file(lines_to_string(COMPLETE_PYTHON_SCRIPT), base_file)
lpc.execute(job_dict)
lpc.execute(job_dir)
self.assertFalse(os.path.exists(job_dir))
output_dir = os.path.join(TEST_JOB_OUTPUT, job_dict[JOB_ID])
self.assertTrue(os.path.exists(output_dir))
job_output_dir = os.path.join(TEST_JOB_OUTPUT, job_dict[JOB_ID])
self.assertTrue(os.path.exists(job_output_dir))
meta_path = os.path.join(output_dir, META_FILE)
meta_path = os.path.join(job_output_dir, META_FILE)
self.assertTrue(os.path.exists(meta_path))
status = read_yaml(meta_path)
self.assertIsInstance(status, Dict)
@ -116,18 +117,22 @@ class MeowTests(unittest.TestCase):
self.assertNotIn(JOB_ERROR, status)
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_base_file(JOB_TYPE_PYTHON))))
self.assertTrue(os.path.exists(os.path.join(output_dir, PARAMS_FILE)))
os.path.join(job_output_dir, get_base_file(JOB_TYPE_PYTHON))))
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_job_file(JOB_TYPE_PYTHON))))
os.path.join(job_output_dir, PARAMS_FILE)))
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_result_file(JOB_TYPE_PYTHON))))
os.path.join(job_output_dir, get_job_file(JOB_TYPE_PYTHON))))
self.assertTrue(os.path.exists(
os.path.join(job_output_dir, get_result_file(JOB_TYPE_PYTHON))))
self.assertTrue(os.path.exists(result_path))
# Test LocalPythonConductor executes valid papermill jobs
def testLocalPythonConductorValidPapermillJob(self)->None:
lpc = LocalPythonConductor()
lpc = LocalPythonConductor(
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT
)
file_path = os.path.join(TEST_MONITOR_BASE, "test")
result_path = os.path.join(TEST_MONITOR_BASE, "output", "test")
@ -168,13 +173,11 @@ class MeowTests(unittest.TestCase):
extras={
JOB_PARAMETERS:params_dict,
JOB_HASH: file_hash,
PYTHON_FUNC:papermill_job_func,
PYTHON_OUTPUT_DIR:TEST_JOB_OUTPUT,
PYTHON_EXECUTION_BASE:TEST_HANDLER_BASE
PYTHON_FUNC:papermill_job_func
}
)
job_dir = os.path.join(TEST_HANDLER_BASE, job_dict[JOB_ID])
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
make_dir(job_dir)
param_file = os.path.join(job_dir, PARAMS_FILE)
@ -186,16 +189,15 @@ class MeowTests(unittest.TestCase):
base_file = os.path.join(job_dir, get_base_file(JOB_TYPE_PAPERMILL))
write_notebook(APPENDING_NOTEBOOK, base_file)
lpc.execute(job_dict)
lpc.execute(job_dir)
job_dir = os.path.join(TEST_HANDLER_BASE, job_dict[JOB_ID])
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
self.assertFalse(os.path.exists(job_dir))
output_dir = os.path.join(TEST_JOB_OUTPUT, job_dict[JOB_ID])
self.assertTrue(os.path.exists(output_dir))
job_output_dir = os.path.join(TEST_JOB_OUTPUT, job_dict[JOB_ID])
self.assertTrue(os.path.exists(job_output_dir))
meta_path = os.path.join(output_dir, META_FILE)
meta_path = os.path.join(job_output_dir, META_FILE)
self.assertTrue(os.path.exists(meta_path))
status = read_yaml(meta_path)
self.assertIsInstance(status, Dict)
@ -203,18 +205,22 @@ class MeowTests(unittest.TestCase):
self.assertEqual(status[JOB_STATUS], STATUS_DONE)
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_base_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(os.path.join(output_dir, PARAMS_FILE)))
os.path.join(job_output_dir, get_base_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_job_file(JOB_TYPE_PAPERMILL))))
os.path.join(job_output_dir, PARAMS_FILE)))
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_result_file(JOB_TYPE_PAPERMILL))))
os.path.join(job_output_dir, get_job_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(
os.path.join(job_output_dir, get_result_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(result_path))
# Test LocalPythonConductor does not execute jobs with bad arguments
def testLocalPythonConductorBadArgs(self)->None:
lpc = LocalPythonConductor()
lpc = LocalPythonConductor(
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT
)
file_path = os.path.join(TEST_MONITOR_BASE, "test")
result_path = os.path.join(TEST_MONITOR_BASE, "output", "test")
@ -255,21 +261,34 @@ class MeowTests(unittest.TestCase):
extras={
JOB_PARAMETERS:params_dict,
JOB_HASH: file_hash,
PYTHON_FUNC:papermill_job_func,
}
)
job_dir = os.path.join(TEST_HANDLER_BASE, bad_job_dict[JOB_ID])
make_dir(job_dir)
bad_job_dir = os.path.join(TEST_JOB_QUEUE, bad_job_dict[JOB_ID])
make_dir(bad_job_dir)
param_file = os.path.join(job_dir, PARAMS_FILE)
write_yaml(params_dict, param_file)
bad_param_file = os.path.join(bad_job_dir, PARAMS_FILE)
write_yaml(params_dict, bad_param_file)
base_file = os.path.join(job_dir, get_base_file(JOB_TYPE_PAPERMILL))
write_notebook(APPENDING_NOTEBOOK, base_file)
bad_meta_path = os.path.join(bad_job_dir, META_FILE)
write_yaml(bad_job_dict, bad_meta_path)
with self.assertRaises(KeyError):
lpc.execute(bad_job_dict)
bad_base_file = os.path.join(bad_job_dir,
get_base_file(JOB_TYPE_PAPERMILL))
write_notebook(APPENDING_NOTEBOOK, bad_base_file)
lpc.execute(bad_job_dir)
bad_output_dir = os.path.join(TEST_JOB_OUTPUT, bad_job_dict[JOB_ID])
self.assertFalse(os.path.exists(bad_job_dir))
self.assertTrue(os.path.exists(bad_output_dir))
bad_meta_path = os.path.join(bad_output_dir, META_FILE)
self.assertTrue(os.path.exists(bad_meta_path))
bad_job = read_yaml(bad_meta_path)
self.assertIsInstance(bad_job, dict)
self.assertIn(JOB_ERROR, bad_job)
# Ensure execution can continue after one failed job
good_job_dict = create_job(
@ -283,42 +302,50 @@ class MeowTests(unittest.TestCase):
extras={
JOB_PARAMETERS:params_dict,
JOB_HASH: file_hash,
PYTHON_FUNC:papermill_job_func,
PYTHON_OUTPUT_DIR:TEST_JOB_OUTPUT,
PYTHON_EXECUTION_BASE:TEST_HANDLER_BASE
PYTHON_FUNC:papermill_job_func
}
)
job_dir = os.path.join(TEST_HANDLER_BASE, good_job_dict[JOB_ID])
make_dir(job_dir)
good_job_dir = os.path.join(TEST_JOB_QUEUE, good_job_dict[JOB_ID])
make_dir(good_job_dir)
param_file = os.path.join(job_dir, PARAMS_FILE)
write_yaml(params_dict, param_file)
good_param_file = os.path.join(good_job_dir, PARAMS_FILE)
write_yaml(params_dict, good_param_file)
base_file = os.path.join(job_dir, get_base_file(JOB_TYPE_PAPERMILL))
write_notebook(APPENDING_NOTEBOOK, base_file)
good_meta_path = os.path.join(good_job_dir, META_FILE)
write_yaml(good_job_dict, good_meta_path)
lpc.execute(good_job_dict)
good_base_file = os.path.join(good_job_dir,
get_base_file(JOB_TYPE_PAPERMILL))
write_notebook(APPENDING_NOTEBOOK, good_base_file)
job_dir = os.path.join(TEST_HANDLER_BASE, good_job_dict[JOB_ID])
self.assertFalse(os.path.exists(job_dir))
lpc.execute(good_job_dir)
output_dir = os.path.join(TEST_JOB_OUTPUT, good_job_dict[JOB_ID])
self.assertTrue(os.path.exists(output_dir))
self.assertTrue(os.path.exists(os.path.join(output_dir, META_FILE)))
good_job_dir = os.path.join(TEST_JOB_QUEUE, good_job_dict[JOB_ID])
self.assertFalse(os.path.exists(good_job_dir))
good_job_output_dir = os.path.join(TEST_JOB_OUTPUT, good_job_dict[JOB_ID])
self.assertTrue(os.path.exists(good_job_output_dir))
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_base_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(os.path.join(output_dir, PARAMS_FILE)))
os.path.join(good_job_output_dir, META_FILE)))
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_job_file(JOB_TYPE_PAPERMILL))))
os.path.join(good_job_output_dir, get_base_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(
os.path.join(output_dir, get_result_file(JOB_TYPE_PAPERMILL))))
os.path.join(good_job_output_dir, PARAMS_FILE)))
self.assertTrue(os.path.exists(
os.path.join(good_job_output_dir, get_job_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(
os.path.join(good_job_output_dir, get_result_file(JOB_TYPE_PAPERMILL))))
self.assertTrue(os.path.exists(result_path))
# Test LocalPythonConductor does not execute jobs with bad functions
def testLocalPythonConductorBadFunc(self)->None:
lpc = LocalPythonConductor()
# Test LocalPythonConductor does not execute jobs with missing metafile
def testLocalPythonConductorMissingMetafile(self)->None:
lpc = LocalPythonConductor(
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT
)
file_path = os.path.join(TEST_MONITOR_BASE, "test")
result_path = os.path.join(TEST_MONITOR_BASE, "output", "test")
@ -361,8 +388,83 @@ class MeowTests(unittest.TestCase):
}
)
with self.assertRaises(Exception):
lpc.execute(job_dict)
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
make_dir(job_dir)
with self.assertRaises(FileNotFoundError):
lpc.execute(job_dir)
# Test LocalPythonConductor does not execute jobs with bad functions
def testLocalPythonConductorBadFunc(self)->None:
lpc = LocalPythonConductor(
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT
)
file_path = os.path.join(TEST_MONITOR_BASE, "test")
result_path = os.path.join(TEST_MONITOR_BASE, "output", "test")
with open(file_path, "w") as f:
f.write("Data")
file_hash = get_file_hash(file_path, SHA256)
pattern = FileEventPattern(
"pattern",
file_path,
"recipe_one",
"infile",
parameters={
"extra":"A line from a test Pattern",
"outfile":result_path
})
recipe = JupyterNotebookRecipe(
"recipe_one", APPENDING_NOTEBOOK)
rule = create_rule(pattern, recipe)
params = {
"extra":"extra",
"infile":file_path,
"outfile":result_path
}
job_dict = create_job(
JOB_TYPE_PAPERMILL,
create_watchdog_event(
file_path,
rule,
TEST_MONITOR_BASE,
file_hash
),
extras={
JOB_PARAMETERS:params,
JOB_HASH: file_hash,
PYTHON_FUNC:failing_func,
}
)
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
make_dir(job_dir)
param_file = os.path.join(job_dir, PARAMS_FILE)
write_yaml(params, param_file)
meta_path = os.path.join(job_dir, META_FILE)
write_yaml(job_dict, meta_path)
lpc.execute(job_dir)
output_dir = os.path.join(TEST_JOB_OUTPUT, job_dict[JOB_ID])
self.assertFalse(os.path.exists(job_dir))
self.assertTrue(os.path.exists(output_dir))
meta_path = os.path.join(output_dir, META_FILE)
self.assertTrue(os.path.exists(meta_path))
job = read_yaml(meta_path)
self.assertIsInstance(job, dict)
self.assertIn(JOB_ERROR, job)
# TODO test job status funcs
# TODO test mangled status file reads

6
tests/test_functionality.py
View File

@ -10,7 +10,7 @@ from time import sleep
from core.correctness.vars import CHAR_LOWERCASE, CHAR_UPPERCASE, \
SHA256, EVENT_TYPE, EVENT_PATH, EVENT_TYPE_WATCHDOG, \
WATCHDOG_BASE, WATCHDOG_HASH, EVENT_RULE, JOB_PARAMETERS, JOB_HASH, \
PYTHON_FUNC, PYTHON_OUTPUT_DIR, PYTHON_EXECUTION_BASE, JOB_ID, JOB_EVENT, \
PYTHON_FUNC, JOB_ID, JOB_EVENT, \
JOB_TYPE, JOB_PATTERN, JOB_RECIPE, JOB_RULE, JOB_STATUS, JOB_CREATE_TIME, \
JOB_REQUIREMENTS, STATUS_QUEUED, JOB_TYPE_PAPERMILL
from core.functionality import generate_id, wait, get_file_hash, rmtree, \
@ -335,9 +335,7 @@ class CorrectnessTests(unittest.TestCase):
"outfile":"result_path"
},
JOB_HASH: "file_hash",
PYTHON_FUNC:max,
PYTHON_OUTPUT_DIR:"output",
PYTHON_EXECUTION_BASE:"execution"
PYTHON_FUNC:max
}
)

2
tests/test_meow.py
View File

@ -292,7 +292,7 @@ class MeowTests(unittest.TestCase):
TestConductor()
class FullTestConductor(BaseConductor):
def execute(self, job:Dict[str,Any])->None:
def execute(self, job_dir:str)->None:
pass
def valid_execute_criteria(self, job:Dict[str,Any]

69
tests/test_recipes.py
View File

@ -9,7 +9,7 @@ from typing import Dict
from core.correctness.vars import EVENT_TYPE, WATCHDOG_BASE, EVENT_RULE, \
EVENT_TYPE_WATCHDOG, EVENT_PATH, SHA256, WATCHDOG_HASH, JOB_ID, \
JOB_TYPE_PYTHON, JOB_PARAMETERS, JOB_HASH, PYTHON_FUNC, JOB_STATUS, \
PYTHON_OUTPUT_DIR, PYTHON_EXECUTION_BASE, META_FILE, JOB_ERROR, \
META_FILE, JOB_ERROR, \
PARAMS_FILE, SWEEP_STOP, SWEEP_JUMP, SWEEP_START, JOB_TYPE_PAPERMILL, \
get_base_file, get_job_file, get_result_file
from core.correctness.validation import valid_job
@ -23,7 +23,7 @@ from recipes.jupyter_notebook_recipe import JupyterNotebookRecipe, \
from recipes.python_recipe import PythonRecipe, PythonHandler, python_job_func
from rules import FileEventJupyterNotebookRule, FileEventPythonRule
from shared import setup, teardown, BAREBONES_PYTHON_SCRIPT, \
COMPLETE_PYTHON_SCRIPT, TEST_HANDLER_BASE, TEST_MONITOR_BASE, \
COMPLETE_PYTHON_SCRIPT, TEST_JOB_QUEUE, TEST_MONITOR_BASE, \
TEST_JOB_OUTPUT, BAREBONES_NOTEBOOK, APPENDING_NOTEBOOK, COMPLETE_NOTEBOOK
@ -112,18 +112,12 @@ class JupyterNotebookTests(unittest.TestCase):
# Test PapermillHandler can be created
def testPapermillHanderMinimum(self)->None:
PapermillHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
PapermillHandler(job_queue_dir=TEST_JOB_QUEUE)
# Test PapermillHandler will handle given events
def testPapermillHandlerHandling(self)->None:
from_handler_reader, from_handler_writer = Pipe()
ph = PapermillHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
ph = PapermillHandler(job_queue_dir=TEST_JOB_QUEUE)
ph.to_runner = from_handler_writer
with open(os.path.join(TEST_MONITOR_BASE, "A"), "w") as f:
@ -172,10 +166,7 @@ class JupyterNotebookTests(unittest.TestCase):
# Test PapermillHandler will create enough jobs from single sweep
def testPapermillHandlerHandlingSingleSweep(self)->None:
from_handler_reader, from_handler_writer = Pipe()
ph = PapermillHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
ph = PapermillHandler(job_queue_dir=TEST_JOB_QUEUE)
ph.to_runner = from_handler_writer
with open(os.path.join(TEST_MONITOR_BASE, "A"), "w") as f:
@ -240,10 +231,7 @@ class JupyterNotebookTests(unittest.TestCase):
# Test PapermillHandler will create enough jobs from multiple sweeps
def testPapermillHandlerHandlingMultipleSweep(self)->None:
from_handler_reader, from_handler_writer = Pipe()
ph = PapermillHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
ph = PapermillHandler(job_queue_dir=TEST_JOB_QUEUE)
ph.to_runner = from_handler_writer
with open(os.path.join(TEST_MONITOR_BASE, "A"), "w") as f:
@ -365,14 +353,11 @@ class JupyterNotebookTests(unittest.TestCase):
extras={
JOB_PARAMETERS:params_dict,
JOB_HASH: file_hash,
PYTHON_FUNC:papermill_job_func,
PYTHON_OUTPUT_DIR:TEST_JOB_OUTPUT,
PYTHON_EXECUTION_BASE:TEST_HANDLER_BASE
PYTHON_FUNC:papermill_job_func
}
)
job_dir = os.path.join(
job_dict[PYTHON_EXECUTION_BASE], job_dict[JOB_ID])
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
make_dir(job_dir)
meta_file = os.path.join(job_dir, META_FILE)
@ -384,9 +369,9 @@ class JupyterNotebookTests(unittest.TestCase):
base_file = os.path.join(job_dir, get_base_file(JOB_TYPE_PAPERMILL))
write_notebook(APPENDING_NOTEBOOK, base_file)
papermill_job_func(job_dict)
papermill_job_func(job_dir)
job_dir = os.path.join(TEST_HANDLER_BASE, job_dict[JOB_ID])
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
self.assertTrue(os.path.exists(job_dir))
meta_path = os.path.join(job_dir, META_FILE)
@ -413,7 +398,7 @@ class JupyterNotebookTests(unittest.TestCase):
except Exception:
pass
self.assertEqual(len(os.listdir(TEST_HANDLER_BASE)), 0)
self.assertEqual(len(os.listdir(TEST_JOB_QUEUE)), 0)
self.assertEqual(len(os.listdir(TEST_JOB_OUTPUT)), 0)
#TODO Test handling criteria function
@ -479,18 +464,12 @@ class PythonTests(unittest.TestCase):
# Test PythonHandler can be created
def testPythonHandlerMinimum(self)->None:
PythonHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
PythonHandler(job_queue_dir=TEST_JOB_QUEUE)
# Test PythonHandler will handle given events
def testPythonHandlerHandling(self)->None:
from_handler_reader, from_handler_writer = Pipe()
ph = PythonHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
ph = PythonHandler(job_queue_dir=TEST_JOB_QUEUE)
ph.to_runner = from_handler_writer
with open(os.path.join(TEST_MONITOR_BASE, "A"), "w") as f:
@ -539,10 +518,7 @@ class PythonTests(unittest.TestCase):
# Test PythonHandler will create enough jobs from single sweep
def testPythonHandlerHandlingSingleSweep(self)->None:
from_handler_reader, from_handler_writer = Pipe()
ph = PythonHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
ph = PythonHandler(job_queue_dir=TEST_JOB_QUEUE)
ph.to_runner = from_handler_writer
with open(os.path.join(TEST_MONITOR_BASE, "A"), "w") as f:
@ -607,10 +583,7 @@ class PythonTests(unittest.TestCase):
# Test PythonHandler will create enough jobs from multiple sweeps
def testPythonHandlerHandlingMultipleSweep(self)->None:
from_handler_reader, from_handler_writer = Pipe()
ph = PythonHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
)
ph = PythonHandler(job_queue_dir=TEST_JOB_QUEUE)
ph.to_runner = from_handler_writer
with open(os.path.join(TEST_MONITOR_BASE, "A"), "w") as f:
@ -732,14 +705,11 @@ class PythonTests(unittest.TestCase):
extras={
JOB_PARAMETERS:params_dict,
JOB_HASH: file_hash,
PYTHON_FUNC:python_job_func,
PYTHON_OUTPUT_DIR:TEST_JOB_OUTPUT,
PYTHON_EXECUTION_BASE:TEST_HANDLER_BASE
PYTHON_FUNC:python_job_func
}
)
job_dir = os.path.join(
job_dict[PYTHON_EXECUTION_BASE], job_dict[JOB_ID])
job_dir = os.path.join(TEST_JOB_QUEUE, job_dict[JOB_ID])
make_dir(job_dir)
meta_file = os.path.join(job_dir, META_FILE)
@ -752,9 +722,8 @@ class PythonTests(unittest.TestCase):
write_notebook(APPENDING_NOTEBOOK, base_file)
write_file(lines_to_string(COMPLETE_PYTHON_SCRIPT), base_file)
python_job_func(job_dict)
python_job_func(job_dir)
job_dir = os.path.join(TEST_HANDLER_BASE, job_dict[JOB_ID])
self.assertTrue(os.path.exists(job_dir))
meta_path = os.path.join(job_dir, META_FILE)
self.assertTrue(os.path.exists(meta_path))
@ -782,7 +751,7 @@ class PythonTests(unittest.TestCase):
except Exception:
pass
self.assertEqual(len(os.listdir(TEST_HANDLER_BASE)), 0)
self.assertEqual(len(os.listdir(TEST_JOB_QUEUE)), 0)
self.assertEqual(len(os.listdir(TEST_JOB_OUTPUT)), 0)
# TODO test default parameter function execution

75
tests/test_runner.py
View File

@ -16,8 +16,8 @@ from recipes.jupyter_notebook_recipe import PapermillHandler, \
JupyterNotebookRecipe
from recipes.python_recipe import PythonHandler, PythonRecipe
from shared import setup, teardown, \
TEST_HANDLER_BASE, TEST_JOB_OUTPUT, TEST_MONITOR_BASE, \
APPENDING_NOTEBOOK, COMPLETE_PYTHON_SCRIPT
TEST_JOB_QUEUE, TEST_JOB_OUTPUT, TEST_MONITOR_BASE, \
APPENDING_NOTEBOOK, COMPLETE_PYTHON_SCRIPT, TEST_DIR
class MeowTests(unittest.TestCase):
@ -31,13 +31,12 @@ class MeowTests(unittest.TestCase):
# Test MeowRunner creation
def testMeowRunnerSetup(self)->None:
monitor_one = WatchdogMonitor(TEST_MONITOR_BASE, {}, {})
monitor_two = WatchdogMonitor(TEST_MONITOR_BASE, {}, {})
monitors = [ monitor_one, monitor_two ]
handler_one = PapermillHandler(TEST_HANDLER_BASE, TEST_JOB_OUTPUT)
handler_two = PapermillHandler(TEST_HANDLER_BASE, TEST_JOB_OUTPUT)
handler_one = PapermillHandler()
handler_two = PapermillHandler()
handlers = [ handler_one, handler_two ]
conductor_one = LocalPythonConductor()
@ -118,6 +117,48 @@ class MeowTests(unittest.TestCase):
for conductor in runner.conductors:
self.assertIsInstance(conductor, BaseConductor)
# Test meow runner directory overrides
def testMeowRunnerDirOverridesSetup(self)->None:
monitor_one = WatchdogMonitor(TEST_MONITOR_BASE, {}, {})
original_queue_dir = os.path.join(TEST_DIR, "original_queue")
original_output_dir = os.path.join(TEST_DIR, "original_output")
overridden_queue_dir = os.path.join(TEST_DIR, "overridden_queue")
overridden_output_dir = os.path.join(TEST_DIR, "overridden_output")
handler_one = PapermillHandler(job_queue_dir=original_queue_dir)
conductor_one = LocalPythonConductor(
job_queue_dir=original_queue_dir,
job_output_dir=original_output_dir
)
self.assertTrue(os.path.exists(original_queue_dir))
self.assertTrue(os.path.exists(original_output_dir))
self.assertFalse(os.path.exists(overridden_queue_dir))
self.assertFalse(os.path.exists(overridden_output_dir))
self.assertEqual(handler_one.job_queue_dir, original_queue_dir)
self.assertEqual(conductor_one.job_queue_dir, original_queue_dir)
self.assertEqual(conductor_one.job_output_dir, original_output_dir)
MeowRunner(
monitor_one,
handler_one,
conductor_one,
job_queue_dir=overridden_queue_dir,
job_output_dir=overridden_output_dir
)
self.assertTrue(os.path.exists(original_queue_dir))
self.assertTrue(os.path.exists(original_output_dir))
self.assertTrue(os.path.exists(overridden_queue_dir))
self.assertTrue(os.path.exists(overridden_output_dir))
self.assertEqual(handler_one.job_queue_dir, overridden_queue_dir)
self.assertEqual(conductor_one.job_queue_dir, overridden_queue_dir)
self.assertEqual(conductor_one.job_output_dir, overridden_output_dir)
# Test single meow papermill job execution
def testMeowRunnerPapermillExecution(self)->None:
pattern_one = FileEventPattern(
@ -148,11 +189,10 @@ class MeowTests(unittest.TestCase):
recipes,
settletime=1
),
PapermillHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT,
),
PapermillHandler(),
LocalPythonConductor(),
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT,
print=runner_debug_stream,
logging=3
)
@ -243,10 +283,11 @@ class MeowTests(unittest.TestCase):
settletime=1
),
PapermillHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
job_queue_dir=TEST_JOB_QUEUE,
),
LocalPythonConductor(),
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT,
print=runner_debug_stream,
logging=3
)
@ -346,10 +387,11 @@ class MeowTests(unittest.TestCase):
settletime=1
),
PythonHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT,
job_queue_dir=TEST_JOB_QUEUE
),
LocalPythonConductor(),
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT,
print=runner_debug_stream,
logging=3
)
@ -448,10 +490,11 @@ class MeowTests(unittest.TestCase):
settletime=1
),
PythonHandler(
TEST_HANDLER_BASE,
TEST_JOB_OUTPUT
job_queue_dir=TEST_JOB_QUEUE
),
LocalPythonConductor(),
job_queue_dir=TEST_JOB_QUEUE,
job_output_dir=TEST_JOB_OUTPUT,
print=runner_debug_stream,
logging=3
)
@ -551,3 +594,5 @@ class MeowTests(unittest.TestCase):
# TODO test with several mismatched handlers
# TODO test with several matched conductors
# TODO test with several mismatched conductors
# TODO tests runner job queue dir
# TODO tests runner job output dir

13
tests/test_validation.py
View File

@ -8,7 +8,7 @@ from typing import Any, Union
from core.correctness.validation import check_type, check_implementation, \
valid_string, valid_dict, valid_list, valid_existing_file_path, \
valid_existing_dir_path, valid_non_existing_path, valid_event, valid_job, \
valid_dir_path, valid_non_existing_path, valid_event, valid_job, \
setup_debugging, valid_watchdog_event, check_callable
from core.correctness.vars import VALID_NAME_CHARS, SHA256, EVENT_TYPE, \
EVENT_PATH, JOB_TYPE, JOB_EVENT, JOB_ID, JOB_PATTERN, JOB_RECIPE, \
@ -215,21 +215,22 @@ class CorrectnessTests(unittest.TestCase):
with self.assertRaises(ValueError):
valid_existing_file_path(dir_path, SHA256)
# Test valid_existing_dir_path can find directories, or not
def testValidExistingDirPath(self)->None:
valid_existing_dir_path(TEST_MONITOR_BASE)
# Test valid_dir_path can find directories, or not
def testValidDirPath(self)->None:
valid_dir_path(TEST_MONITOR_BASE)
valid_dir_path(TEST_MONITOR_BASE, must_exist=False)
dir_path = os.path.join(TEST_MONITOR_BASE, "dir")
with self.assertRaises(FileNotFoundError):
valid_existing_dir_path("not_existing_"+dir_path, SHA256)
valid_dir_path("not_existing_"+dir_path, must_exist=True)
file_path = os.path.join(TEST_MONITOR_BASE, "file.txt")
with open(file_path, 'w') as hashed_file:
hashed_file.write("Some data\n")
with self.assertRaises(ValueError):
valid_existing_dir_path(file_path, SHA256)
valid_dir_path(file_path)
# Test valid_non_existing_path can find existing paths, or not
def testValidNonExistingPath(self)->None: