Repository: lanl/BEE (language: python)

Issue:
Slurm worker tests
I've updated the slurm worker tests. These will require a valid bee.conf and can be run with `pytest src/beeflow/tests/test_slurm_worker.py`.

I'm marking this as WIP for now since there is one test that is failing on Darwin for some reason. I also need to add some test cases for when slurmrestd fails to start.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/ci-workflow.yml b/.github/workflows/ci-workflow.yml
new file mode 100644
index 000000000..61ef37807
--- /dev/null
+++ b/.github/workflows/ci-workflow.yml
@@ -0,0 +1,33 @@
+name: Integration workflow
+
+on:
+  # For manual exec
+  # (https://github.blog/changelog/2020-07-06-github-actions-manual-triggers-with-workflow_dispatch/)
+  workflow_dispatch: {}
+  push:
+    branches: [main, develop]
+  pull_request:
+    types: [opened, synchronize, edited, labeled, unlabeled]
+    branches: [main, develop]
+
+env: {}
+
+jobs:
+  integration-test:
+    if: ${{ !contains(github.event.pull_request.labels.*.name, 'WIP') }}
+    name: BEE Integration Test
+    # Note: Needs to run on 22.04 or later since slurmrestd doesn't seem to be
+    # available on 20.04
+    runs-on: ubuntu-22.04
+    steps:
+      - uses: actions/checkout@v2
+      - name: Dependency Install
+        run: ./ci/deps_install.sh
+      - name: Slurm Setup and Install
+        run: ./ci/slurm_start.sh
+      - name: BEE Install
+        run: ./ci/bee_install.sh
+      - name: BEE Start
+        run: ./ci/bee_start.sh
+      - name: Integration Test
+        run: ./ci/integration_test.sh
diff --git a/.github/workflows/pylama.yml b/.github/workflows/pylama.yml
new file mode 100644
index 000000000..d06c65450
--- /dev/null
+++ b/.github/workflows/pylama.yml
@@ -0,0 +1,23 @@
+name: PyLama Lint
+
+on:
+  # For manual exec
+  # (https://github.blog/changelog/2020-07-06-github-actions-manual-triggers-with-workflow_dispatch/)
+  workflow_dispatch: {}
+  push:
+    branches: [main, develop]
+  pull_request:
+    types: [opened, synchronize, edited, labeled, unlabeled]
+    branches: [main, develop]
+
+jobs:
+  pylama:
+    if: ${{ !contains(github.event.pull_request.labels.*.name, 'WIP') }}
+    name: PyLama Lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Lint
+        run: |
+          pip install pylama 2>&1 >/dev/null
+          pylama src/
diff --git a/ci/bee_install.sh b/ci/bee_install.sh
new file mode 100755
index 000000000..d978ed381
--- /dev/null
+++ b/ci/bee_install.sh
@@ -0,0 +1,87 @@
+#!/bin/sh
+# Setup, install and run BEE.
+
+. ./ci/env.sh
+
+# BEE Containers
+printf "\n\n"
+printf "**Setting up BEE containers**\n"
+printf "\n\n"
+mkdir -p $HOME/img
+# Pull the neo4j container
+ch-image pull neo4j:3.5.22 || exit 1
+NEO4J_CONTAINER=$HOME/img/neo4j.tar.gz
+ch-convert -i ch-image -o tar neo4j:3.5.22 $NEO4J_CONTAINER || exit 1
+
+# BEE install
+printf "\n\n"
+printf "**Installing BEE**\n"
+printf "\n\n"
+$PYTHON -m venv venv
+. venv/bin/activate
+pip install --upgrade pip
+pip install poetry
+# Do a poetry install, making sure that all extras are added
+poetry install -E cloud_extras -E scheduler_extras || exit 1
+
+# BEE Configuration
+mkdir -p ~/.config/beeflow
+JOB_TEMPLATE=$HOME/.config/beeflow/submit.jinja
+cp src/beeflow/data/job_templates/slurm-submit.jinja $JOB_TEMPLATE
+cat >> ~/.config/beeflow/bee.conf <<EOF
+# BEE CONFIGURATION FILE #
+[DEFAULT]
+bee_workdir = $BEE_WORKDIR
+workload_scheduler = Slurm
+use_archive = False
+bee_dep_image = $NEO4J_CONTAINER
+
+[task_manager]
+listen_port = 8892
+container_runtime = Charliecloud
+job_template = $JOB_TEMPLATE
+runner_opts =
+
+[charliecloud]
+image_mntdir = /tmp
+chrun_opts = --cd $HOME
+setup =
+
+[graphdb]
+hostname = localhost
+dbpass = password
+bolt_port = 7687
+http_port = 7474
+https_port = 7473
+gdb_image_mntdir = /tmp
+sleep_time = 10
+
+[scheduler]
+log = $BEE_WORKDIR/logs/scheduler.log
+listen_port = 5100
+use_mars = False
+mars_model =
+mars_task_cnt = 3
+alloc_logfile = $BEE_WORKDIR/logs/scheduler_alloc.log
+algorithm = fcfs
+default_algorithm = fcfs
+workdir = $BEE_WORKDIR/scheduler
+
+[workflow_manager]
+listen_port = 7233
+
+[builder]
+deployed_image_root = /tmp
+container_output_path = /tmp
+container_type = charliecloud
+container_archive = $HOME/container_archive
+
+[slurmrestd]
+slurm_socket = /tmp/slurm.sock
+slurm_args = -s openapi/v0.0.35
+EOF
+printf "\n\n"
+printf "#### bee.conf ####\n"
+cat ~/.config/beeflow/bee.conf
+printf "#### bee.conf ####\n"
+printf "\n\n"
diff --git a/ci/bee_start.sh b/ci/bee_start.sh
new file mode 100755
index 000000000..7ae4f2ce6
--- /dev/null
+++ b/ci/bee_start.sh
@@ -0,0 +1,8 @@
+#!/bin/sh
+# Start BEE
+
+. ./ci/env.sh
+. venv/bin/activate
+# Slurmrestd will fail by default when running as `SlurmUser`
+SLURMRESTD_SECURITY=disable_user_check beeflow || exit 1
+sleep 4
diff --git a/ci/deps_install.sh b/ci/deps_install.sh
new file mode 100755
index 000000000..8efcdd5c7
--- /dev/null
+++ b/ci/deps_install.sh
@@ -0,0 +1,20 @@
+#!/bin/sh
+# Install all dependencies for BEE
+sudo apt-get update
+sudo apt-get install -y slurmctld slurmd slurmrestd munge python3 python3-venv \
+    curl build-essential zlib1g-dev libncurses5-dev libgdbm-dev libnss3-dev \
+    libssl-dev libsqlite3-dev libreadline-dev libffi-dev libbz2-dev
+
+# Install most recent Charliecloud
+curl -O -L https://github.com/hpc/charliecloud/releases/download/v0.27/charliecloud-0.27.tar.gz
+tar -xvf charliecloud-0.27.tar.gz
+(cd charliecloud-0.27
+ ./configure --prefix=/usr
+ make
+ sudo make install)
+
+# Use a PPA to install Python3.8
+sudo apt-get install -y software-properties-common
+sudo add-apt-repository -y ppa:deadsnakes/ppa
+sudo apt-get update
+sudo apt-get install -y python3.8 python3.8-dev python3.8-venv
diff --git a/ci/env.sh b/ci/env.sh
new file mode 100644
index 000000000..b51856c28
--- /dev/null
+++ b/ci/env.sh
@@ -0,0 +1,24 @@
+# Environment set up
+# Set up environment
+PYTHON=python3.8
+HOSTNAME=`python -c 'import socket; print(socket.gethostname())'`
+# Everything is in /tmp for right now
+SLURMCTLD_PID=/tmp/slurmctld.pid
+SLURMD_PID=/tmp/slurmd.pid
+SLURMD_SPOOL_DIR=/tmp/slurm_spool
+LOG_DIR=/tmp/slurm_log
+SLURM_STATE_SAVE_LOCATION=/tmp/slurm_state
+mkdir -p $SLURMD_SPOOL_DIR $SLURM_STATE_SAVE_LOCATION $LOG_DIR
+SLURMCTLD_LOG=$LOG_DIR/slurmctld.log
+SLURMD_LOG=$LOG_DIR/slurmd.log
+SLURM_USER=`whoami`
+MUNGE_SOCKET=/tmp/munge.sock
+MUNGE_LOG=/tmp/munge.log
+MUNGE_PID=/tmp/munge.pid
+mkdir -p /tmp/munge
+MUNGE_KEY=/tmp/munge/munge.key
+# Determine config of CI host
+NODE_CONFIG=`slurmd -C | head -n 1`
+BEE_WORKDIR=$HOME/.beeflow
+mkdir -p $BEE_WORKDIR
+export SLURM_CONF=~/slurm.conf
diff --git a/ci/slurm_start.sh b/ci/slurm_start.sh
new file mode 100755
index 000000000..6dfcc20e0
--- /dev/null
+++ b/ci/slurm_start.sh
@@ -0,0 +1,75 @@
+#!/bin/sh
+# Set up and start Slurm
+
+. ./ci/env.sh
+
+# Dump a new slurm.conf
+cat >> $SLURM_CONF <<EOF
+ClusterName=bee-ci
+# For some reason pmix is missing
+# MpiDefault=pmix
+MpiDefault=none
+ProctrackType=proctrack/pgid
+ReturnToService=2
+SlurmctldPort=7777
+SlurmdPort=8989
+
+SwitchType=switch/none
+
+TaskPlugin=task/affinity
+
+InactiveLimit=0
+KillWait=30
+MinJobAge=300
+Waittime=0
+
+SchedulerType=sched/backfill
+SelectType=select/cons_tres
+SelectTypeParameters=CR_Core
+
+AccountingStorageType=accounting_storage/none
+JobCompType=jobcomp/none
+JobAcctGatherType=jobacct_gather/none
+SlurmdDebug=info
+AuthType=auth/munge
+
+SlurmctldHost=$HOSTNAME
+SlurmctldPidFile=$SLURMCTLD_PID
+SlurmdPidFile=$SLURMD_PID
+StateSaveLocation=$SLURM_STATE_SAVE_LOCATION
+SlurmdSpoolDir=$SLURMD_SPOOL_DIR
+SlurmctldLogFile=$SLURMCTLD_LOG
+SlurmdLogFile=$SLURMD_LOG
+SlurmUser=$SLURM_USER
+SlurmdUser=$SLURM_USER
+AuthInfo=socket=$MUNGE_SOCKET
+
+$NODE_CONFIG
+PartitionName=debug Nodes=ALL Default=YES MaxTime=INFINITE State=UP
+EOF
+
+printf "\n\n"
+printf "#### slurm.conf ####\n"
+cat $SLURM_CONF
+printf "#### slurm.conf ####\n"
+printf "\n\n"
+
+# Start daemons
+printf "**Starting munged**\n"
+# Create the key
+head -c 1024 /dev/urandom > $MUNGE_KEY
+chmod 400 $MUNGE_KEY
+munged -f \
+    -S $MUNGE_SOCKET \
+    --log-file $MUNGE_LOG \
+    --pid-file $MUNGE_PID \
+    --key-file $MUNGE_KEY
+sleep 1
+printf "**Starting slurmctld**\n"
+slurmctld
+printf "**Starting slurmd**\n"
+slurmd
+
+printf "#### SUPPORTED MPI ####\n"
+srun --mpi=list
+printf "#######################\n"
diff --git a/example/clamr-ffmpeg-build/README.md b/example/clamr-ffmpeg-build/README.md
new file mode 100644
index 000000000..bea6e82fe
--- /dev/null
+++ b/example/clamr-ffmpeg-build/README.md
@@ -0,0 +1,13 @@
+# CLAMR - FFMPEG workflow using CWL
+
+This workflow uses the DockerRequirement to build the clamr and ffmpeg in a container.
+
+```
+clamr_wf.cwl - the main cwl  
+calmr_job.yml - yaml file for values used by the cwl files   
+clamr.cwl - cwl file for the clamr step
+ffmpeg.cwl - cwl file for the ffmpeg step
+```
+
+
+
diff --git a/example/clamr-ffmpeg-build/clamr.cwl b/example/clamr-ffmpeg-build/clamr.cwl
new file mode 100644
index 000000000..5175fbd45
--- /dev/null
+++ b/example/clamr-ffmpeg-build/clamr.cwl
@@ -0,0 +1,83 @@
+# -*- mode: YAML; -*-
+
+class: CommandLineTool
+cwlVersion: v1.0
+
+baseCommand: /CLAMR/clamr_cpuonly
+# This is the stdout field which makes all stdout be captured in this file
+# stderr is not currently implemented but it is also a thing
+stdout: clamr_stdout.txt
+# Arguments to the command
+inputs:
+  amr_type:
+    # ? means the argument is optional
+    # All of the ? here are legacy from the original CWL
+    type: string?
+    # Declare extra options
+    # We support prefix and position
+    inputBinding:
+      # Prefix is the flag for cli command
+      prefix: -A
+  grid_res:
+    type: int?
+    inputBinding:
+      prefix: -n
+  max_levels:
+    type: int?
+    inputBinding:
+      prefix: -l
+  time_steps:
+    type: int?
+    inputBinding:
+      prefix: -t
+  output_steps:
+    type: int?
+    inputBinding:
+      prefix: -i
+  graphic_steps:
+    type: int?
+    inputBinding:
+      prefix: -g
+  graphics_type:
+    type: string?
+    inputBinding:
+      prefix: -G
+  rollback_images:
+    type: int?
+    inputBinding:
+      prefix: -b
+  checkpoint_disk_interval:
+    type: int?
+    inputBinding:
+      prefix: -c
+  checkpoint_mem_interval:
+    type: int?
+    inputBinding:
+      prefix: -C
+  hash_method:
+    type: string?
+    inputBinding:
+      prefix: -e
+  
+outputs:
+  # Captures stdout. Name is arbitrary.
+  clamr_stdout:
+    # type is syntactic sugar to just grab the output file defined above
+    # stdout:
+    #     type: File
+    #     outputBinding: 
+    #       glob: clamr_stdout.txt
+    #     stdout is easy shorthand
+    type: stdout
+  outdir:
+    # directory is just another type. Scan the files for a directory with the name specified in glob
+    # If you add a wildcard, it'd do expansion
+    type: Directory
+    outputBinding:
+      # Glob can be either a constant string or have a wildcard 
+      # TODO verify CWLs glob support
+      glob: graphics_output/graph%05d.png
+  time_log:
+    type: File
+    outputBinding:
+      glob: total_execution_time.log
diff --git a/example/clamr-ffmpeg-build/clamr_job.json b/example/clamr-ffmpeg-build/clamr_job.json
new file mode 100644
index 000000000..ad88972f6
--- /dev/null
+++ b/example/clamr-ffmpeg-build/clamr_job.json
@@ -0,0 +1,13 @@
+{
+  "grid_resolution": 32,
+  "max_levels": 3,
+  "time_steps": 5000,
+  "steps_between_outputs": 10,
+  "steps_between_graphics": 25,
+  "graphics_type": "png",
+  "input_format": "image2",
+  "frame_rate": 12,
+  "frame_size": "800x800",
+  "pixel_format": "yuv420p",
+  "output_filename": "CLAMR_movie.mp4"
+}
diff --git a/example/clamr-ffmpeg-build/clamr_job.yml b/example/clamr-ffmpeg-build/clamr_job.yml
new file mode 100644
index 000000000..4051acc0a
--- /dev/null
+++ b/example/clamr-ffmpeg-build/clamr_job.yml
@@ -0,0 +1,18 @@
+# Inputs for CLAMR
+# /CLAMR/clamr_cpuonly -n 32 -l 3 -t 5000 -i 10 -g 25 -G png
+
+grid_resolution: 32
+max_levels: 3
+time_steps: 5000
+steps_between_outputs: 10
+steps_between_graphics: 25
+graphics_type: png
+
+# Inputs for FFMPEG
+#ffmpeg -f image2 -r 12 -s 800x800 -pix_fmt yuv420p $HOME/CLAMR_movie.mp4
+
+input_format: image2
+frame_rate: 12
+frame_size: 800x800
+pixel_format: yuv420p
+output_filename: CLAMR_movie.mp4
diff --git a/example/clamr-ffmpeg-build/clamr_wf.cwl b/example/clamr-ffmpeg-build/clamr_wf.cwl
new file mode 100644
index 000000000..a54c662ea
--- /dev/null
+++ b/example/clamr-ffmpeg-build/clamr_wf.cwl
@@ -0,0 +1,75 @@
+# -*- mode: YAML; -*-
+
+class: Workflow
+cwlVersion: v1.0
+
+# Main 3 components of workflow are inputs, outputs, and steps
+
+inputs:
+# All inputs go here for each step. No way to break them up.
+# We should talk to the CWL people about that. 
+##### CLAMR inputs #####
+# takes ID:Type syntax
+  grid_resolution: int
+  max_levels: int
+  time_steps: int
+  steps_between_outputs: int
+  steps_between_graphics: int
+  graphics_type: string
+##### FFMPEG inputs #####
+  input_format: string
+  frame_rate: int
+  frame_size: string
+  pixel_format: string
+  output_filename: string
+
+outputs:
+# Outputs for all the steps
+# Check where we copy these probably at CWD
+# Files need to exist at end of workflow, but we aren't currently checking
+# We have function to get workflow_outputs
+# TODO add step to workflow_manager to confirm that each of these outputs exist
+  clamr_stdout:
+    type: File
+    outputSource: clamr/clamr_stdout
+  clamr_time_log:
+    type: File
+    outputSource: clamr/time_log
+  clamr_movie:
+    type: File
+    outputSource: ffmpeg/movie
+
+steps:
+  clamr:
+    run: clamr.cwl
+    in:
+      grid_res: grid_resolution
+      max_levels: max_levels
+      time_steps: time_steps
+      output_steps: steps_between_outputs
+      graphic_steps: steps_between_graphics
+      graphics_type: graphics_type
+    out: [clamr_stdout, outdir, time_log]
+    hints:
+        DockerRequirement:
+            dockerFile: "src/beeflow/data/dockerfiles/Dockerfile.clamr-ffmpeg"
+            beeflow:containerName: "clamr-ffmpeg"
+
+  ffmpeg:
+    run: ffmpeg.cwl
+    in:
+      input_format: input_format
+      # input syntax is name: <step>/dependent_object
+      ffmpeg_input: clamr/outdir
+      frame_rate: frame_rate
+      frame_size: frame_size
+      pixel_format: pixel_format
+      # Setting output file with file_name
+      # output_filename set in wf inputs
+      output_file: output_filename
+    # Multiple outputs can be in array
+    out: [movie]
+    hints:
+        DockerRequirement:
+            dockerFile: "src/beeflow/data/dockerfiles/Dockerfile.clamr-ffmpeg"
+            beeflow:containerName: "clamr-ffmpeg"
diff --git a/example/clamr-ffmpeg-build/ffmpeg.cwl b/example/clamr-ffmpeg-build/ffmpeg.cwl
new file mode 100644
index 000000000..542195cc3
--- /dev/null
+++ b/example/clamr-ffmpeg-build/ffmpeg.cwl
@@ -0,0 +1,44 @@
+# -*- mode: YAML; -*-
+
+class: CommandLineTool
+cwlVersion: v1.0
+
+baseCommand: ffmpeg -y
+inputs:
+  input_format:
+    type: string?
+    inputBinding:
+      prefix: -f
+      position: 1
+  ffmpeg_input:
+    type: Directory
+    inputBinding:
+      prefix: -i
+      position: 2
+      valueFrom: $("/graph%05d.png")
+  frame_rate:
+    type: int?
+    inputBinding:
+      prefix: -r
+      position: 3
+  frame_size:
+    type: string?
+    inputBinding:
+      prefix: -s
+      position: 4
+  pixel_format:
+    type: string?
+    inputBinding:
+      prefix: -pix_fmt
+      position: 5
+  output_file:
+    type: string
+    inputBinding:
+      position: 6
+
+outputs:
+  movie:
+    type: File
+    outputBinding:
+      glob: $(inputs.output_file)
+      # glob: CLAMR_movie.mp4
diff --git a/example/clamr-ffmpeg-example.tgz b/example/clamr-ffmpeg-example.tgz
deleted file mode 100644
index 7e2b713e9..000000000
Binary files a/example/clamr-ffmpeg-example.tgz and /dev/null differ
diff --git a/pyproject.toml b/pyproject.toml
index 0cd0ef3db..4280604f9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -74,6 +74,7 @@ google-api-python-client = { version = "^1.12.8", optional = true }
 python-openstackclient = { version = "^5.5.0", optional = true }
 python-heatclient = { version = "^2.3.0", optional = true }
 typer = "^0.4.1"
+pyflakes = "2.4.0"
 
 [tool.poetry.extras]
 scheduler_extras = ["tensorflow", "matplotlib"]
diff --git a/setup.cfg b/setup.cfg
index 0481742af..bebb5e212 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,11 +1,7 @@
 [pylama]
 format = pylint
 linters = pycodestyle,pydocstyle,pyflakes,pylint
-ignore = D203,D213,D404,R0902,R0903,R0904,R0912,R0913,R0914,R0916,W0603,W1203
-# D203:  1 blank line required before class docstring (conflicts with D211)
-# D213:  multi-line docstring summary should start at the second line
-#        (conflicts with D212)
-# D404:  first word of the first line should not be "This"
+ignore = R0902,R0903,R0904,R0912,R0913,R0914,R0916,W0603,W1203,C0413
 # R0902: too many instance attributes (default: 8)
 # R0903: class has too few public methods (default: 2)
 # R0904: class has too many public methods (default: 20)
@@ -15,10 +11,18 @@ ignore = D203,D213,D404,R0902,R0903,R0904,R0912,R0913,R0914,R0916,W0603,W1203
 # R0916: if-statement contains too many boolean expressions (default: 5)
 # W0603: global statement used to update a global variable
 # W1203: use lazy % formatting in logging functions
+# C0413: requires imports at the top of the file
 
 [pylama:pylint]
 max_line_length = 99
-good_names = i,j,k,v,t,ex,id,tx,Run,_
+good_names = i,j,k,v,t,m,s,n,ex,fp,id,tx,fn,Run,_
 
 [pylama:pycodestyle]
 max_line_length = 99
+ignore = E402
+# E402: module level imports should be at the top of the file
+
+[pylama:pydocstyle]
+max_line_length = 99
+ignore = D203,D213,D404
+
diff --git a/src/beeflow/client/bee_client.py b/src/beeflow/client/bee_client.py
index 45b88ad47..a41d37169 100755
--- a/src/beeflow/client/bee_client.py
+++ b/src/beeflow/client/bee_client.py
@@ -11,24 +11,37 @@
 from beeflow.common.cli import NaturalOrderGroup
 
 
-bc.init()
-
 # Length of a shortened workflow ID
 short_id_len = 6
 
 # Maximum length of a workflow ID
 MAX_ID_LEN = 32
 
+# Global used to indicate whether this instance is interactive or not
+_interactive = False
+
 
 logging.basicConfig(level=logging.WARNING)
 workflow_manager = 'bee_wfm/v1/jobs'
 
 
+class ClientError(Exception):
+    """Client error class."""
+
+    def __init__(self, *args):
+        """Error constructor."""
+        self.args = args
+
+
 def error_exit(msg):
     caller_func = str.capitalize(inspect.stack()[1].function)
     msg = caller_func + ': ' + msg
-    typer.secho(msg, fg=typer.colors.RED)
-    exit(1)
+    if _interactive:
+        typer.secho(msg, fg=typer.colors.RED)
+        exit(1)
+    else:
+        # Raise an error so that client libraries can handle it
+        raise ClientError(msg) from None
 
 
 def _url():
@@ -143,6 +156,7 @@ def submit(wf_name: str = typer.Argument(..., help='The workflow name'),
     typer.secho("Workflow submitted! Your workflow id is "
                 f"{_short_id(wf_id)}.", fg=typer.colors.GREEN)
     logging.info('Sumit workflow: ' + resp.text)
+    return wf_id
 
 
 @app.command()
@@ -214,6 +228,7 @@ def query(wf_id: str = typer.Argument(..., callback=match_short_id)):
         typer.echo(tasks_status)
 
     logging.info('Query workflow: ' + resp.text)
+    return wf_status, tasks_status
 
 
 @app.command()
@@ -308,6 +323,9 @@ def reexecute(wf_name: str = typer.Argument(..., help='The workflow name'),
 
 
 def main():
+    global _interactive
+    _interactive = True
+    bc.init()
     app()
 
 
diff --git a/src/beeflow/common/build/README.md b/src/beeflow/common/build/README.md
index 1d625d2e5..defbe59ba 100644
--- a/src/beeflow/common/build/README.md
+++ b/src/beeflow/common/build/README.md
@@ -36,13 +36,21 @@ Each step in a workflow may include a reference to `DockerRequirement` in the CW
 1. `dockerPull:` Specify a container image to retrieve using the container runtime. This specifies the container name and tag. It does not include the path to the container. dockerPull should be used in combination with a container registery specified by dockerLoad, or a default will be assumed.
 2. `dockerLoad:` The HTTP URL associated with the container registry in use. The container runtime requested will be specified by the `<container>://` prefix. dockerLoad must be used in combination with dockerPull, such that dockerPull defines what the required container image is named.
 3. `dockerFile:` The path to a container definition file. The builder will fail if `dockerFile` is defined along with `dockerPull` or `dockerLoad`.
-4. `dockerImport:` Provide HTTP URL to download and gunzip a Docker image using `docker import`. This should be the path to a compressed image. 
-5. `containerName:` A reference to the image id that will be invoked by the container runtime's `run` or `exec` command. Note that this differs from `dockerPull` slightly, in that `dockerPull` is the image to be acquired. It is possible to pull an image in a workflow stage, and then run an entirely different image by specifying a different `containerName`. If `containerName` is not defined, assume `dockerPull` references the container to run. If `containerName` is specified and image does not exist, error.
+4. `dockerImport:` Provide HTTP URL to download and gunzip a Docker image using `docker import`. This should be the path to a compressed image.
+5. `beeflow:containerName:` A reference to the image id that will be invoked by the container runtime's `run` or `exec` command. Note that this differs from `dockerPull` slightly, in that `dockerPull` is the image to be acquired. It is possible to pull an image in a workflow stage, and then run an entirely different image by specifying a different `containerName`. If `containerName` is not defined, assume `dockerPull` references the container to run. If `containerName` is specified and image does not exist, error.
 6. `dockerOutputDirectory:` Set the designated output directory to a specific location inside the Docker container.
 
 
 A few examples to use for testing:
 ## CharliecloudBuildDriver Examples
+
+### Initialize BeeConfig for all the examples:
+
+```
+from beeflow.common.config_driver import BeeConfig as bc
+bc.init()
+```
+
 ### dockerPull
 ```
 from beeflow.common.build.container_drivers import CharliecloudBuildDriver
@@ -56,8 +64,8 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.dockerPull()
-a.dockerPull('git.lanl.gov:5050/trandles/baseimages/centos:7')
+a.process_docker_pull()
+a.process_docker_pull('git.lanl.gov:5050/trandles/baseimages/centos:7')
 
 task = Task(name='hi',base_command=['hi','hello'],
                  requirements={},
@@ -67,8 +75,8 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.dockerPull()
-a.dockerPull('git.lanl.gov:5050/qwofford/containerhub/lstopo')
+a.process_docker_pull()
+a.process_docker_pull('git.lanl.gov:5050/qwofford/containerhub/lstopo')
 
 task = Task(name='hi',base_command=['hi','hello'],
                  hints=None,
@@ -78,9 +86,9 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.dockerPull()
-a.dockerPull('git.lanl.gov:5050/qwofford/containerhub/lstopo')
-a.dockerPull('git.lanl.gov:5050/qwofford/containerhub/lstopo',force=True)
+a.process_docker_pull()
+a.process_docker_pull('git.lanl.gov:5050/qwofford/containerhub/lstopo')
+a.process_docker_pull('git.lanl.gov:5050/qwofford/containerhub/lstopo',force=True)
 ```
 ### dockerFile
 ```
@@ -88,15 +96,17 @@ from beeflow.common.build.container_drivers import CharliecloudBuildDriver
 from beeflow.common.wf_data import Task
 task = Task(name='hi',base_command=['hi','hello'],
                  requirements={'DockerRequirement':{'dockerFile':'src/beeflow/data/dockerfiles/Dockerfile.builder_demo',
-                                                    'containerName':'my_fun_container:sillytag'}},
+                                                    'beeflow:containerName':'my_fun_container:sillytag'}},
                  hints=None,
                  workflow_id=42,
                  stdout="output.txt",
                  inputs={},
                  outputs={})
 b = CharliecloudBuildDriver(task)
-b.containerName()
-b.dockerFile()
+b.process_docker_file()
+ERROR: dockerFile may not be specified without containerName
+b.process_container_name()
+b.process_docker_file()
 ```
 ### dockerImport
 ```
@@ -110,15 +120,7 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-# When a "builder" entry does not exist in bee.conf
-# >>> Build cache directory is: /yellow/users/qwofford/.beeflow/build_cache
-# >>> Config file is missing builder section.
-# >>> Assuming deployed image root is /var/tmp/qwofford/beeflow
-# >>> Wrote deployed image root to user BeeConfig file.
-# >>> Deployed image root directory is: /var/tmp/qwofford/beeflow
-# When a "builder entry does exist in bee.conf
-# >>> Build cache directory is: /yellow/users/qwofford/.beeflow/build_cache
-# >>> Deployed image root directory is: /var/tmp/qwofford/beeflow
+
 task = Task(name='hi',base_command=['hi','hello'],
                  requirements={'DockerRequirement':{'dockerImport':'/usr/projects/beedev/neo4j-3-5-17-ch.tar.gz'}},
                  hints=None,
@@ -127,7 +129,7 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.dockerImport()
+a.process_docker_import()
 
 task = Task(name='hi',base_command=['hi','hello'],
                  hints={'DockerRequirement':{'dockerImport':'/usr/projects/beedev/neo4j-3-5-17-ch.tar.gz'}},
@@ -137,9 +139,9 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.dockerImport()
+a.process_docker_import()
 ```
-### dockerOutputDirectory
+### dockerOutputDirectory   Needs work
 ```
 from beeflow.common.build.container_drivers import CharliecloudBuildDriver
 from beeflow.common.wf_data import Task
@@ -151,16 +153,15 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-# >>> Build cache directory is: /yellow/users/qwofford/.beeflow/build_cache
-# >>> Deployed image root directory is: /var/tmp/qwofford/beeflow
+
 # >>> Container-relative output path is: /
-# >>> a.dockerOutputDirectory()
+# >>> a.process_docker_output_directory()
 # >>> '/'
-a.dockerOutputDirectory(param_output_directory='/home/qwofford')
-# >>> '/home/qwofford'
+a.process_docker_output_directory(param_output_directory='/home/<username>')
+# >>> '/home/<username>'
 # Note: Changing the output directory by parameter changes the bc object, but it does NOT over-write the config file.
-a.dockerOutputDirectory()
-# >>> '/home/qwofford'
+a.process_docker_output_directory()
+# >>> '/home/<username>'
 ```
 ### dockerLoad
 ```
@@ -174,7 +175,7 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.dockerLoad()
+a.process_docker_load()
 # >>> Charliecloud does not have the concept of a layered image tarball.
 # >>> Did you mean to use dockerImport?
 # >>> 0
@@ -186,31 +187,29 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.dockerLoad()
+a.process_docker_load()
 # >>> Charliecloud does not have the concept of a layered image tarball.
 # >>> Did you mean to use dockerImport?
 # >>> ERROR: dockerLoad specified as requirement.
 # >>> 1
 ```
-### containerName
-Note that this is an extension to the CWL spec. CWL uses "dockerImageId as a container name, but this actually referes to the image ID hash, which cannot be produced until after a Docekrfile is built. To work around this problem, we created containerName.
+### beeflow:containerName
+
+Note: this is a BEE extension to the CWL spec. CWL uses "dockerImageId as a container name, but that actually refers to the image ID hash, which cannot be produced until after a Docekrfile is built. To work around this problem, we created beeflow:containerName.
 ```
 from beeflow.common.build.container_drivers import CharliecloudBuildDriver
 from beeflow.common.wf_data import Task
 task = Task(name='hi',base_command=['hi','hello'],
-                 hints={'DockerRequirement':{'containerName':'my_containerName'}},
+                 hints={'DockerRequirement':{'beeflow:containerName':'my_containerName'}},
                  requirements=None,
                  workflow_id=42,
                  stdout='output.txt',
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.containerName()
-# >>> 'my_containerName'
-a.containerName(param_containerName='another_containerName')
-# >>> 'another_containerName'
-a.containerName()
-# >>> 'my_containerName'
+a.process_container_name()
+# INFO: Setting container_name to: my_containerName
+# 0
 task = Task(name='hi',base_command=['hi','hello'],
                  hints=None,
                  requirements=None,
@@ -219,10 +218,9 @@ task = Task(name='hi',base_command=['hi','hello'],
                  inputs={},
                  outputs={})
 a = CharliecloudBuildDriver(task)
-a.containerName()
+a.process_container_name()
 # >>> 1
-a.containerName('another_containerName')
-# >>> 'another_containerName'
-a.containerName()
-# >>> 'another_containerName'
 ```
+
+### Add tests for beeflow:copyContainer and beeflow:useContainer
+
diff --git a/src/beeflow/common/build/build_driver.py b/src/beeflow/common/build/build_driver.py
index 18a41af4e..b76a64136 100644
--- a/src/beeflow/common/build/build_driver.py
+++ b/src/beeflow/common/build/build_driver.py
@@ -48,8 +48,8 @@ def __init__(self, task, kwargs):
         """
 
     @abstractmethod
-    def dockerPull(self, addr, force):
-        """CWL compliant dockerPull.
+    def process_docker_pull(self, addr, force):
+        """Get and process the CWL compliant dockerPull dockerRequirement.
 
         CWL spec 09-23-2020: Specify a Docker image to
         retrieve using docker pull. Can contain the immutable
@@ -57,32 +57,32 @@ def dockerPull(self, addr, force):
         """
 
     @abstractmethod
-    def dockerLoad(self):
-        """CWL compliant dockerLoad.
+    def process_docker_load(self):
+        """Get and process the CWL compliant dockerLoad dockerRequirement.
 
         CWL spec 09-23-2020: Specify a HTTP URL from which to
         download a Docker image using docker load.
         """
 
     @abstractmethod
-    def dockerFile(self, task_dockerfile, force):
-        """CWL compliant dockerFile.
+    def process_docker_file(self, task_dockerfile, force):
+        """Get and process the CWL compliant dockerFile dockerRequirement.
 
         CWL spec 09-23-2020: Supply the contents of a Dockerfile
         which will be built using docker build.
         """
 
     @abstractmethod
-    def dockerImport(self, param_import):
-        """CWL compliant dockerImport.
+    def process_docker_import(self, param_import):
+        """Get and process the CWL compliant dockerImport dockerRequirement.
 
         CWL spec 09-23-2020: Provide HTTP URL to download and
         gunzip a Docker images using docker import.
         """
 
     @abstractmethod
-    def dockerImageId(self, param_imageid):
-        """CWL compliant dockerImageId.
+    def process_docker_image_id(self, param_imageid):
+        """Get and process the CWL compliant dockerImageId dockerRequirement.
 
         CWL spec 09-23-2020: The image id that will be used for
         docker run. May be a human-readable image name or the
@@ -92,16 +92,18 @@ def dockerImageId(self, param_imageid):
         """
 
     @abstractmethod
-    def dockerOutputDirectory(self, param_output_directory):
-        """CWL compliant dockerOutputDirectory.
+    def process_docker_output_directory(self, param_output_directory):
+        """Get and process the CWL compliant dockerOutputDirectory dockerRequirement.
 
         CWL spec 09-23-2020: Set the designated output directory
         to a specific location inside the Docker container.
         """
 
     @abstractmethod
-    def copyContainer(self, force):
-        """CWL extension, copy an existing container into the build archive.
+    def process_copy_container(self, force):
+        """Get and process the BEE CWL extension copyContainer dockerRequirement.
+
+        This CWL extension will copy an existing container to the build archive.
 
         If you have a container tarball, and all you need to do is stage it,
         that is, all you need to do is copy it to a location that BEE knows,
@@ -109,8 +111,10 @@ def copyContainer(self, force):
         """
 
     @abstractmethod
-    def containerName(self):
-        """CWL extension, need a way to refer to containers human-readable name.
+    def process_container_name(self):
+        """Get and process BEE CWL extension for containerName dockerRequirement.
+
+        This is a BEE extension to CWL to refer to containers with human-readable name.
 
         The CWL spec currently uses dockerImageId to refer to the name of a container
         but this is explicitly not how Docker defines it. We need a way to name
@@ -129,16 +133,14 @@ def resolve_priority(self):
         failure will the builder build the docker file.
         """
         # cwl spec priority list consists of:
-        # (bound method, method name, priority, termainal case bool)
-        cwl_spec = [(self.dockerPull, 'dockerPull', 5, True),
-                    (self.dockerLoad, 'dockerLoad', 6, True),
-                    (self.dockerFile, 'dockerFile', 7, True),
-                    (self.dockerImport, 'dockerImport', 4, True),
-                    (self.copyContainer, 'beeflow:copyContainer', 3, True),
-                    (self.dockerImageId, 'dockerImageId', 1, False),
-                    (self.containerName, 'containerName', 2, False),
-                    (self.dockerOutputDirectory, 'dockerOutputDirectory', 0, False)]
+        # (bound method, method name, priority, terminal case bool)
+        cwl_spec = [(self.process_docker_pull, 'dockerPull', 5, True),
+                    (self.process_docker_load, 'dockerLoad', 6, True),
+                    (self.process_docker_file, 'dockerFile', 7, True),
+                    (self.process_docker_import, 'dockerImport', 4, True),
+                    (self.process_copy_container, 'beeflow:copyContainer', 3, True),
+                    (self.process_docker_image_id, 'dockerImageId', 1, False),
+                    (self.process_container_name, 'beeflow:containerName', 2, False),
+                    (self.process_docker_output_directory, 'dockerOutputDirectory', 0, False)]
         exec_list = sorted(cwl_spec, key=lambda x: x[2])
         return exec_list
-# Ignore snake_case requirement to enable CWL compliant names.
-# pylama:ignore=C0103
diff --git a/src/beeflow/common/build/container_drivers.py b/src/beeflow/common/build/container_drivers.py
index a20f3838f..cb844c0f1 100644
--- a/src/beeflow/common/build/container_drivers.py
+++ b/src/beeflow/common/build/container_drivers.py
@@ -6,7 +6,6 @@
 import os
 import shutil
 import subprocess
-import sys
 from beeflow.common.config_driver import BeeConfig as bc
 # from beeflow.common.crt.crt_drivers import CharliecloudDriver, SingularityDriver
 from beeflow.cli import log
@@ -30,7 +29,7 @@ class CharliecloudBuildDriver(ContainerBuildDriver):
     requests a RTE, and a method to return the requested RTE.
     """
 
-    def __init__(self, task, userconf_file=None):
+    def __init__(self, task):
         """Begin build request.
 
         Parse hints and requirements to determine target build
@@ -67,55 +66,64 @@ def __init__(self, task, userconf_file=None):
         self.task = task
         self.docker_image_id = None
         self.container_name = None
-        dockerRequirements = set()
+        docker_requirements = set()
         try:
-            requirement_DockerRequirements = self.task.requirements['DockerRequirement'].keys()
-            dockerRequirements = dockerRequirements.union(requirement_DockerRequirements)
-            req_string = (f'{set(requirement_DockerRequirements)}')
+            requirement_docker_requirements = self.task.requirements['DockerRequirement'].keys()
+            docker_requirements = docker_requirements.union(requirement_docker_requirements)
+            req_string = (f'{set(requirement_docker_requirements)}')
             log.info(f'task {self.task.id} requirement DockerRequirements: {req_string}')
         except (TypeError, KeyError):
             log.info(f'task {self.task.name} {self.task.id} no DockerRequirements in requirement')
         try:
-            hint_DockerRequirements = self.task.hints['DockerRequirement'].keys()
-            dockerRequirements = dockerRequirements.union(hint_DockerRequirements)
-            hint_str = (f'{set(hint_DockerRequirements)}')
+            hint_docker_requirements = self.task.hints['DockerRequirement'].keys()
+            docker_requirements = docker_requirements.union(hint_docker_requirements)
+            hint_str = (f'{set(hint_docker_requirements)}')
             log.info(f'task {self.task.name} {self.task.id} hint DockerRequirements: {hint_str}')
         except (TypeError, KeyError):
             log.info(f'task {self.task.name} {self.task.id} hints has no DockerRequirements')
-        log.info(f'task {self.task.id} union DockerRequirements : {dockerRequirements}')
+        log.info(f'task {self.task.id} union DockerRequirements : {docker_requirements}')
         exec_superset = self.resolve_priority()
-        self.exec_list = [i for i in exec_superset if i[1] in dockerRequirements]
+        self.exec_list = [i for i in exec_superset if i[1] in docker_requirements]
         log_exec_list = [i[1] for i in self.exec_list]
         log.info(f'task {self.task.id} DockerRequirement execution order will be: {log_exec_list}')
         log.info('Execution order pre-empts hint/requirement status.')
 
-    def dockerPull(self, addr=None, force=False):
-        """CWL compliant dockerPull.
+    def get_docker_req(self, docker_req_param):
+        """Get dockerRequirement, prioritizing requirements over hints.
 
-        CWL spec 09-23-2020: Specify a Docker image to
-        retrieve using docker pull. Can contain the immutable
-        digest to ensure an exact container is used.
+        :param docker_req_param: the dockerRequirement parameter (e.g. 'dockerFile')
+        :type docker_req_param: str
+
+        When requirements are specified hints will be ignored.
+        By default, tasks need not specify hints or requirements
         """
-        # By default, tasks need not specify hints or reqs
-        task_addr = None
-        try:
-            # Try to get Hints
-            hint_addr = self.task.hints['DockerRequirement']['dockerPull']
-        except (KeyError, TypeError):
-            # Task Hints are not mandatory. No dockerPull image specified in task hints.
-            hint_addr = None
+        task_docker_req = None
+        # Get value if specified in requirements
         try:
             # Try to get Requirements
-            req_addr = self.task.requirements['DockerRequirement']['dockerPull']
+            task_docker_req = self.task.requirements['DockerRequirement'][docker_req_param]
         except (KeyError, TypeError):
-            # Task Requirements are not mandatory. No dockerPull image specified in task reqs.
-            req_addr = None
+            # Task Requirements are not mandatory. No docker_req_param specified in task reqs.
+            task_docker_req = None
+        # Ignore hints if requirements available
+        if not task_docker_req:
+            # Get value if specified in hints
+            try:
+                # Try to get Hints
+                task_docker_req = self.task.hints['DockerRequirement'][docker_req_param]
+            except (KeyError, TypeError):
+                # Task Hints are not mandatory. No docker_req_param specified in task hints.
+                task_docker_req = None
+        return task_docker_req
 
-        # Prefer requirements over hints
-        if req_addr:
-            task_addr = req_addr
-        elif hint_addr:
-            task_addr = hint_addr
+    def process_docker_pull(self, addr=None, force=False):
+        """Get the CWL compliant dockerPull dockerRequirement.
+
+        CWL spec 09-23-2020: Specify a Docker image to
+        retrieve using docker pull. Can contain the immutable
+        digest to ensure an exact container is used.
+        """
+        task_addr = self.get_docker_req('dockerPull')
 
         # Use task specified image if image parameter empty
         if not addr:
@@ -126,7 +134,7 @@ def dockerPull(self, addr=None, force=False):
             log.error("dockerPull set but no image path specified.")
             return 1
         # If no image specified and no image required, nothing to do.
-        if not req_addr and not addr:
+        if not task_addr and not addr:
             log.info('No image specified and no image required, nothing to do.')
             return 0
 
@@ -158,20 +166,14 @@ def dockerPull(self, addr=None, force=False):
         return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               check=True, shell=True)
 
-    def dockerLoad(self):
-        """CWL compliant dockerLoad.
+    def process_docker_load(self):
+        """Get and process the CWL compliant dockerLoad dockerRequirment.
 
         CWL spec 09-23-2020: Specify a HTTP URL from which to
         download a Docker image using docker load.
         """
-        # Need to know if dockerLoad is a requirement in order to determine fail/success
-        try:
-            # Try to get Requirements
-            req_dockerload = self.task.requirements['DockerRequirement']['dockerLoad']
-        except (KeyError, TypeError):
-            # Task Requirements are not mandatory. No dockerload specified in task reqs.
-            req_dockerload = None
-
+        # Need to know if dockerLoad is specified in order to determine fail/success
+        req_dockerload = self.get_docker_req('dockerLoad')
         log.warning('Charliecloud does not have the concept of a layered image tarball.')
         log.warning('Did you mean to use dockerImport?')
         if req_dockerload:
@@ -179,40 +181,22 @@ def dockerLoad(self):
             return 1
         return 0
 
-    def dockerFile(self, task_dockerfile=None, force=False):
-        """CWL compliant dockerFile.
+    def process_docker_file(self, task_dockerfile=None, force=False):
+        """Get and process the CWL compliant dockerFile dockerRequirement.
 
         CWL spec 09-23-2020: Supply the contents of a Dockerfile
         which will be built using docker build. We have discussed implementing CWL
         change to expect a file handle instead of file contents, and use the file
         handle expectation here.
         """
-        # containerName is always processed before dockerFile, so safe to assume it exists
+        # beeflow:containerName is always processed before dockerFile, so safe to assume it exists
         # otherwise, raise an error.
         if self.container_name is None:
-            log.error("dockerFile may not be specified without containerName")
+            log.error("dockerFile may not be specified without beeflow:containerName")
             return 1
 
         # Need dockerfile in order to build, else fail
-        try:
-            # Try to get Hints
-            hint_dockerfile = self.task.hints['DockerRequirement']['dockerFile']
-        except (KeyError, TypeError):
-            # Task Hints are not mandatory. No dockerfile specified in task hints.
-            hint_dockerfile = None
-        try:
-            # Try to get Requirements
-            req_dockerfile = self.task.requirements['DockerRequirement']['dockerFile']
-        except (KeyError, TypeError):
-            # Task Requirements are not mandatory. No dockerfile specified in task reqs.
-            req_dockerfile = None
-
-        # Prefer requirements over hints
-        if req_dockerfile:
-            task_dockerfile = req_dockerfile
-        elif hint_dockerfile:
-            task_dockerfile = hint_dockerfile
-
+        task_dockerfile = self.get_docker_req('dockerFile')
         if not task_dockerfile:
             log.error("dockerFile not specified as task attribute or parameter.")
             return 1
@@ -251,8 +235,8 @@ def dockerFile(self, task_dockerfile=None, force=False):
         return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               check=True, shell=True)
 
-    def dockerImport(self, param_import=None):
-        """CWL compliant dockerImport.
+    def process_docker_import(self, param_import=None):
+        """Get and process the CWL compliant dockerImport dockerRequirement.
 
         CWL spec 09-23-2020: Provide HTTP URL to download and
         gunzip a Docker images using docker import. The param_import
@@ -263,25 +247,7 @@ def dockerImport(self, param_import=None):
             import_input_path = param_import
         else:
             # Get path for tarball to import
-            try:
-                # Try to get Hints
-                hint_import = self.task.hints['DockerRequirement']['dockerImport']
-            except (KeyError, TypeError):
-                # Task Hints are not mandatory. No import specified in task hints.
-                hint_import = None
-            try:
-                # Try to get Requirements
-                req_import = self.task.requirements['DockerRequirement']['dockerImport']
-            except (KeyError, TypeError):
-                # Task Requirements are not mandatory. No import specified in task reqs.
-                req_import = None
-            # Prefer requirements over hints
-            if req_import:
-                task_import = req_import
-            elif hint_import:
-                task_import = hint_import
-            # Set import
-            import_input_path = task_import
+            import_input_path = self.get_docker_req('dockerImport')
 
         # Pull the image.
         file_name = crt_driver.get_ccname(import_input_path)
@@ -290,12 +256,12 @@ def dockerImport(self, param_import=None):
         return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               check=True, shell=True)
 
-    def dockerImageId(self, param_imageid=None):
-        """CWL compliant dockerImageId.
+    def process_docker_image_id(self, param_imageid=None):
+        """Get and process the CWL compliant dockerImageId dockerRequirement.
 
         A divergence from the CWL spec. Docker image Id is defined by docker as a checksum
         on a container, not a human-readable name. The Docker image ID must be produced after
-        the container is build, and can not be used to tag the container for that reason.
+        the container is built, and can not be used to tag the container for that reason.
         The param_imageid may be used to override DockerRequirement specs.
         """
         # Parameter takes precedence
@@ -305,37 +271,19 @@ def dockerImageId(self, param_imageid=None):
         if self.docker_image_id:
             return 0
 
-        # Need imageid to know how dockerfile should be named, else fail
-        try:
-            # Try to get Hints
-            hint_imageid = self.task.hints['DockerRequirement']['dockerImageId']
-        except (KeyError, TypeError):
-            # Task Hints are not mandatory. No imageid specified in task hints.
-            hint_imageid = None
-        try:
-            # Try to get Requirements
-            req_imageid = self.task.requirements['DockerRequirement']['dockerImageId']
-        except (KeyError, TypeError):
-            # Task Requirements are not mandatory. No imageid specified in task reqs.
-            req_imageid = None
-
-        # Prefer requirements over hints
-        task_imageid = None
-        if req_imageid:
-            task_imageid = req_imageid
-        elif hint_imageid:
-            task_imageid = hint_imageid
+        # Need ImageId to know how dockerFile should be named, else fail
+        task_image_id = self.get_docker_req('dockerImageId')
 
         # Set imageid
-        self.docker_image_id = task_imageid
+        self.docker_image_id = task_image_id
 
-        # If task and parameter still doesn't specify image_id, consider this an error.
-        if self.docker_image_id:
+        # If task and parameter still doesn't specify ImageId, consider this an error.
+        if not self.docker_image_id:
             return 0
         return 1
 
-    def dockerOutputDirectory(self, param_output_directory=None):
-        """CWL compliant dockerOutputDirectory.
+    def process_docker_output_directory(self, param_output_directory=None):
+        """Get and process the CWL compliant dockerOutputDirectory dockerRequirement.
 
         CWL spec 09-23-2020: Set the designated output directory
         to a specific location inside the Docker container. The
@@ -347,33 +295,17 @@ def dockerOutputDirectory(self, param_output_directory=None):
             self.container_output_path = param_output_directory
         return 0
 
-    def copyContainer(self, force=False):
-        """CWL extension, copy an existing container into the build archive.
+    def process_copy_container(self, force=False):
+        """Get and process the BEE CWL extension copyContainer dockerRequirement.
+
+        This CWL extension will copy an existing container to the build archive.
 
         If you have a container tarball, and all you need to do is stage it,
         that is, all you need to do is copy it to a location that BEE knows,
         use this to put the container into the build archive.
         """
         # Need container_path to know how dockerfile should be named, else fail
-        try:
-            # Try to get Hints
-            hint_container_path = self.task.hints['DockerRequirement']['beeflow:copyContainer']
-        except (KeyError, TypeError):
-            # Task Hints are not mandatory. No container_path specified in task hints.
-            hint_container_path = None
-        try:
-            # Try to get Requirements
-            req_container_path = self.task.requirements['DockerRequirement']['beeflow:copyContainer']
-        except (KeyError, TypeError):
-            # Task Requirements are not mandatory. No container_path specified in task reqs.
-            req_container_path = None
-
-        # Prefer requirements over hints
-        if req_container_path:
-            task_container_path = req_container_path
-        elif hint_container_path:
-            task_container_path = hint_container_path
-
+        task_container_path = self.get_docker_req('beeflow:copyContainer')
         if not task_container_path:
             log.error("beeflow:copyContainer: You must specify the path to an existing container.")
             return 1
@@ -402,36 +334,21 @@ def copyContainer(self, force=False):
         return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               check=True, shell=True)
 
-    def containerName(self):
-        """CWL extension, need a way to refer to containers human-readable name.
+    def process_container_name(self):
+        """Get and process BEE CWL extension for containerName dockerRequirement.
+
+        This is a BEE extension to CWL to refer to containers with human-readable name.
 
         The CWL spec currently uses dockerImageId to refer to the name of a container
         but this is explicitly not how Docker defines it. We need a way to name
         containers in a human readable format.
         """
-        try:
-            # Try to get Hints
-            hint_container_name = self.task.hints['DockerRequirement']['containerName']
-        except (KeyError, TypeError):
-            # Task Hints are not mandatory. No container_name specified in task hints.
-            hint_container_name = None
-        try:
-            # Try to get Requirements
-            req_container_name = self.task.requirements['DockerRequirement']['containerName']
-        except (KeyError, TypeError):
-            # Task Requirements are not mandatory. No container_name specified in task reqs.
-            req_container_name = None
-
-        # Prefer requirements over hints
-        if req_container_name:
-            task_container_name = req_container_name
-        elif hint_container_name:
-            task_container_name = hint_container_name
-
+        task_container_name = self.get_docker_req('beeflow:containerName')
         if not task_container_name and self.docker_image_id is None:
-            log.error("containerName: You must specify the containerName or dockerImageId")
+            log.error("beeflow:containerName: You must specify the containerName or dockerImageId")
             return 1
         self.container_name = task_container_name
+        log.info(f'Setting container_name to: {self.container_name}')
         return 0
 
 
@@ -456,38 +373,38 @@ def __init__(self, task, userconf_file=None):
         :type kwargs: set of build system parameters
         """
 
-    def dockerPull(self, addr=None, force=False):
-        """CWL compliant dockerPull.
+    def process_docker_pull(self, addr=None, force=False):
+        """Get and process the CWL compliant dockerPull dockerRequirement.
 
         CWL spec 09-23-2020: Specify a Docker image to
         retrieve using docker pull. Can contain the immutable
         digest to ensure an exact container is used.
         """
 
-    def dockerLoad(self):
-        """CWL compliant dockerLoad.
+    def process_docker_load(self):
+        """Get and process the CWL compliant dockerLoad dockerRequirement.
 
         CWL spec 09-23-2020: Specify a HTTP URL from which to
         download a Docker image using docker load.
         """
 
-    def dockerFile(self, task_dockerfile=None, force=False):
-        """CWL compliant dockerFile.
+    def process_docker_file(self, task_dockerfile=None, force=False):
+        """Get and process the CWL compliant dockerFile dockerRequirement.
 
         CWL spec 09-23-2020: Supply the contents of a Dockerfile
         which will be built using docker build.
         """
 
-    def dockerImport(self, param_import=None):
-        """CWL compliant dockerImport.
+    def process_docker_import(self, param_import=None):
+        """Get and process the CWL compliant dockerImport dockerRequirement.
 
         CWL spec 09-23-2020: Provide HTTP URL to download and
         gunzip a Docker images using docker import. The param_import
         may be used to override DockerRequirement specs.
         """
 
-    def dockerImageId(self, param_imageid=None):
-        """CWL compliant dockerImageId.
+    def process_docker_image_id(self, param_imageid=None):
+        """Get and process the CWL compliant dockerImageId dockerRequirement.
 
         CWL spec 09-23-2020: The image id that will be used for
         docker run. May be a human-readable image name or the
@@ -497,17 +414,13 @@ def dockerImageId(self, param_imageid=None):
         specs.
         """
 
-    def dockerOutputDirectory(self, param_output_directory=None):
-        """CWL compliant dockerOutputDirectory.
+    def process_docker_output_directory(self, param_output_directory=None):
+        """Get and process the CWL compliant dockerOutputDirectory.
 
         CWL spec 09-23-2020: Set the designated output directory
         to a specific location inside the Docker container. The
         param_output_directory may be used to override DockerRequirement
         specs.
         """
-# Ignore snake_case requirement to enable CWL compliant names. (C0103)
-# Ignore "too many statements". Some of these methods are long, and that's ok (R0915)
 # Ignore W0231: linter doesn't know about abstract classes, it's ok to now call the parent __init__
-# Ignore W1202: Using fstrings does not cause us any issues in logging currently, improves
-#               readibility as is
-# pylama:ignore=C0103,R0915,W0231,W1202
+# pylama:ignore=W0231
diff --git a/src/beeflow/common/build_interfaces.py b/src/beeflow/common/build_interfaces.py
index d6c5a96b0..ffc787c58 100644
--- a/src/beeflow/common/build_interfaces.py
+++ b/src/beeflow/common/build_interfaces.py
@@ -18,47 +18,48 @@
 from beeflow.common.build.build_driver import arg2task
 
 
-def build_main(bc, task):
-    """Main build code."""
+def build_main(task):
+    """Process build instructions - main code."""
     # The build driver treats Hint and Requirement objects as Dicts.
-    local_task = task.copy()
-    local_task.hints = dict(local_task.hints)
-    local_task.requirements = dict(local_task.requirements)
-    builder = CharliecloudBuildDriver(local_task)
+    task_local = task.copy()
+    task_local.hints = dict(task_local.hints)
+    task_local.requirements = dict(task_local.requirements)
+    builder = CharliecloudBuildDriver(task_local)
     log.info('CharliecloudBuildDriver initialized')
 
     # Deque the next build instruction until empty or reach terminal case
     # ...catch the case where no build instructions are required.
+    op_keys = ["build_op", "op_name", "op_priority", "op_terminal"]
     try:
-        build_op, op_name, op_priority, op_terminal = builder.exec_list.pop(0)
+        op_values = builder.exec_list.pop(0)
     except IndexError:
         log.info('No build instructions provided. Assuming this is ok...')
-        build_op, op_name, op_priority, op_terminal = None, None, None, True
-
-    while build_op:
-        log.info('Executing build operation: "{}"'.format(op_name))
+        op_values = [None, None, None, True]
+    op_dict = dict(zip(op_keys, op_values))
+    while op_dict["build_op"]:
+        log.info(f'Executing build operation: {op_dict["op_name"]}')
         try:
-            return_obj = build_op()
+            return_obj = op_dict["build_op"]()
             # Return objects will be successful subprocess or return code.
             try:
-                RETURN_CODE = return_obj.returncode
+                return_code = return_obj.returncode
             except AttributeError:
-                RETURN_CODE = int(return_obj)
+                return_code = int(return_obj)
         except CalledProcessError:
-            RETURN_CODE = 1
-            log.warning('There was a problem executing {}, check relevant log for detail.'
-                        .format(op_name))
+            return_code = 1
+            log.warning(f'There was a problem executing {op_dict["op_name"]}!')
         # Case 1: Not the last operation spec'd, but is a terminal operation.
-        if op_terminal and RETURN_CODE == 0:
-            log.info('Reached terminal build case')
-            build_op, op_name, op_priority, op_terminal = None, None, None, True
+        if op_dict["op_terminal"] and return_code == 0:
+            op_values = [None, None, None, True]
+            op_dict = dict(zip(op_keys, op_values))
             continue
         # Case 2: Go to next, or last operation spec'd.
         try:
-            build_op, op_name, op_priority, op_terminal = builder.exec_list.pop(0)
+            op_values = builder.exec_list.pop(0)
         except IndexError:
-            build_op, op_name, op_priority, op_terminal = None, None, None, True
-    log.info('Out of build instructions. Build operations complete.')
+            op_values = [None, None, None, True]
+        finally:
+            op_dict = dict(zip(op_keys, op_values))
 
 
 if __name__ == '__main__':
@@ -66,20 +67,18 @@ def build_main(bc, task):
         userconfig = sys.argv[1]
         bc.init(userconfig=userconfig)
         my_args = sys.argv[2]
-    except IndexError:
-        raise IndexError('build_interface must execute with 2 arguments.')
+    except IndexError as exc:
+        raise IndexError('build_interface must execute with 2 arguments.') from exc
 
     bee_workdir = bc.get('DEFAULT', 'bee_workdir')
     handler = bee_logging.save_log(bee_workdir=bee_workdir, log=log, logfile='build_interface.log')
 
     try:
         local_task = arg2task(my_args)
-    except Exception as e:
-        log.info('{}'.format(e))
+    except Exception as err:
+        log.info(f'{err}')
 
-    build_main(bc, local_task)
+    build_main(local_task)
 
-# Ignore W0707: Re-raising with from keyword does not aid in readability or functionality
 # Ignore W0703: Catching generic exception isn't a problem if we just want a descriptive report
-# Ignore: W1202: fstring variable sub in logging as a pre-processing step is not causing problems.
-# pylama:ignore=W0707,W0703,W1202
+# pylama:ignore=W0703
diff --git a/src/beeflow/common/crt_drivers.py b/src/beeflow/common/crt_drivers.py
index acec7dd37..9dc8814b4 100644
--- a/src/beeflow/common/crt_drivers.py
+++ b/src/beeflow/common/crt_drivers.py
@@ -88,13 +88,13 @@ def run_text(self, task):
         requirements = dict(task.requirements)
         try:
             # Try to get Hints
-            hint_container_name = hints['DockerRequirement']['containerName']
+            hint_container_name = hints['DockerRequirement']['beeflow:containerName']
         except (KeyError, TypeError):
             # Task Hints are not mandatory. No container_name specified in task hints.
             hint_container_name = None
         try:
             # Try to get Requirements
-            req_container_name = requirements['DockerRequirement']['containerName']
+            req_container_name = requirements['DockerRequirement']['beeflow:containerName']
         except (KeyError, TypeError):
             # Task Requirements are not mandatory. No container_name specified in task reqs.
             req_container_name = None
@@ -105,9 +105,10 @@ def run_text(self, task):
         elif hint_container_name:
             task_container_name = hint_container_name
 
-        baremetal = True
+        baremetal = False 
         use_container = None
         if task_container_name is None:
+            baremetal = True
             log.info('No container name provided.')
             log.info('Assuming another DockerRequirement is runtime target.')
             runtime_target_list = []
@@ -167,7 +168,7 @@ def run_text(self, task):
                         'Too many container runtimes specified! Pick one per workflow step.'
                     )
             if len(runtime_target_list) == 0:
-                log.warning('No containerName specified.')
+                log.warning('No beeflow:containerName specified.')
                 log.warning('Cannot be inferred from other DockerRequirements.')
             else:
                 baremetal = False
@@ -188,6 +189,9 @@ def run_text(self, task):
                                           main_command=[str(arg) for arg in task.command],
                                           post_commands=[])
 
+        if task_container_name:
+            container_path = '/'.join([container_archive, task_container_name]) + '.tar.gz'
+
         # If use_container is specified, then no copying is done and the file
         # path is used directly
         if use_container is not None:
@@ -196,6 +200,7 @@ def run_text(self, task):
             task_container_name = os.path.splitext(tmp)[0]
         else:
             container_path = '/'.join([container_archive, task_container_name]) + '.tar.gz'
+
         log.info(f'Expecting container at {container_path}. Ready to deploy and run.')
 
         chrun_opts, cc_setup = self.get_cc_options()
diff --git a/src/beeflow/common/gdb/gdb_driver.py b/src/beeflow/common/gdb/gdb_driver.py
index bb154c1b3..96c50ac35 100644
--- a/src/beeflow/common/gdb/gdb_driver.py
+++ b/src/beeflow/common/gdb/gdb_driver.py
@@ -71,11 +71,11 @@ def initialize_ready_tasks(self):
 
         Runnable tasks are tasks with all input dependencies fulfilled.
         """
-    
+
     @abstractmethod
     def restart_task(self, old_task, new_task):
         """Restart a failed task.
-        
+
         Create a Task node for new_task with state 'RESTARTED' and an edge
         to indicate that it is the child of the Task node of old_task.
 
@@ -88,7 +88,7 @@ def restart_task(self, old_task, new_task):
     @abstractmethod
     def finalize_task(self, task):
         """Set task state to 'COMPLETED' and set inputs from source.
-        
+
         :param task: the task to finalize
         :type task: Task
         """
@@ -111,11 +111,19 @@ def get_workflow_description(self):
 
     @abstractmethod
     def get_workflow_state(self):
-        """Return the current state of the workflow
+        """Return the current state of the workflow.
 
         :rtype: str
         """
 
+    @abstractmethod
+    def set_workflow_state(self, state):
+        """Set the state of the workflow.
+
+        :param state: the new state of the workflow
+        :type state: str
+        """
+
     @abstractmethod
     def get_workflow_tasks(self):
         """Return a list of all workflow tasks from the graph database.
diff --git a/src/beeflow/common/gdb/neo4j_cypher.py b/src/beeflow/common/gdb/neo4j_cypher.py
index 4dbe527c0..4a1d038ed 100644
--- a/src/beeflow/common/gdb/neo4j_cypher.py
+++ b/src/beeflow/common/gdb/neo4j_cypher.py
@@ -51,7 +51,8 @@ def create_workflow_input_nodes(tx, inputs):
     """Create Input nodes for the workflow.
 
     :param inputs: the workflow inputs
-    :type inputs: list of InputParameter"""
+    :type inputs: list of InputParameter
+    """
     for input_ in inputs:
         input_query = ("MATCH (w:Workflow) CREATE (w)<-[:INPUT_OF]-(i:Input) "
                        "SET i.id = $input_id "
@@ -65,7 +66,8 @@ def create_workflow_output_nodes(tx, outputs):
     """Create Output nodes for the workflow.
 
     :param outputs: the workflow outputs
-    :type outputs: list of OutputParameter"""
+    :type outputs: list of OutputParameter
+    """
     for output in outputs:
         output_query = ("MATCH (w:Workflow) CREATE (w)<-[:OUTPUT_OF]-(o:Output) "
                         "SET o.id = $output_id "
@@ -127,7 +129,8 @@ def create_task_input_nodes(tx, task):
     """Create Input nodes for a task.
 
     :param task: the task whose inputs to add to the graph
-    :type task: Task"""
+    :type task: Task
+    """
     for input_ in task.inputs:
         input_query = ("MATCH (t:Task {id: $task_id}) "
                        "CREATE (t)<-[:INPUT_OF]-(i:Input) "
@@ -149,7 +152,8 @@ def create_task_output_nodes(tx, task):
     """Create Output nodes for a task.
 
     :param task: the task whose outputs to add to the graph
-    :type task: Task"""
+    :type task: Task
+    """
     for output in task.outputs:
         output_query = ("MATCH (t:Task {id: $task_id}) "
                         "CREATE (t)<-[:OUTPUT_OF]-(o:Output) "
@@ -352,7 +356,7 @@ def get_workflow_state(tx):
     :rtype: str
     """
     state_query = "MATCH (w:Workflow) RETURN w.state"
-    
+
     return tx.run(state_query).single().value()
 
 
@@ -360,10 +364,10 @@ def set_workflow_state(tx, state):
     """Get workflow state from the Neo4j database.
 
     :param state: the state the workflow will be set to
-    type state: str
+    :type state: str
     """
     state_query = "MATCH (w:Workflow) SET w.state = $state"
-    
+
     return tx.run(state_query, state=state)
 
 
@@ -440,7 +444,8 @@ def set_task_metadata(tx, task, metadata):
         # currently support parameter substitution for property keys
         if fullmatch(r"[A-Za-z][0-9A-Za-z_]*", k) is None:
             raise ValueError(f"invalid metadata key: {k}")
-        metadata_query = f"MATCH (m:Metadata)-[:DESCRIBES]->(:Task {{id: $task_id}}) SET m.{k} = $value"
+        metadata_query = ("MATCH (m:Metadata)-[:DESCRIBES]->(:Task {id: $task_id}) "
+                          f"SET m.{k} = $value")
 
         tx.run(metadata_query, task_id=task.id, value=v)
 
diff --git a/src/beeflow/common/gdb/neo4j_driver.py b/src/beeflow/common/gdb/neo4j_driver.py
index 7f92911d4..4141a7540 100644
--- a/src/beeflow/common/gdb/neo4j_driver.py
+++ b/src/beeflow/common/gdb/neo4j_driver.py
@@ -20,8 +20,10 @@
 DEFAULT_USER = "neo4j"
 DEFAULT_PASSWORD = "password"
 
+
 class Neo4JNotRunning(Exception):
-    pass
+    """Exception thrown when connection attempted while Neo4j is not running."""
+
 
 class Neo4jDriver(GraphDatabaseDriver):
     """The driver for a Neo4j Database.
@@ -48,8 +50,8 @@ def __init__(self, user=DEFAULT_USER, password=DEFAULT_PASSWORD, **kwargs):
         try:
             # Connect to the Neo4j database using the Neo4j proprietary driver
             self._driver = Neo4jDatabase.driver(uri, auth=(user, password))
-        except ServiceUnavailable:
-            raise Neo4JNotRunning("Neo4j database is unavailable")
+        except ServiceUnavailable as sue:
+            raise Neo4JNotRunning("Neo4j database is unavailable") from sue
 
     def initialize_workflow(self, workflow):
         """Begin construction of a workflow stored in Neo4j.
@@ -131,7 +133,7 @@ def initialize_ready_tasks(self):
 
     def restart_task(self, old_task, new_task):
         """Restart a failed task.
-        
+
         Create a Task node for new_task with 'RESTARTED_FROM' relationship to the
         Task node of old_task.
 
@@ -152,7 +154,7 @@ def restart_task(self, old_task, new_task):
 
     def finalize_task(self, task):
         """Set task state to 'COMPLETED' and set inputs from source.
-        
+
         :param task: the task to finalize
         :type task: Task
         """
@@ -183,16 +185,14 @@ def get_workflow_description(self):
 
     def get_workflow_state(self):
         """Return the current workflow state from the Neo4j database.
-        
+
         :rtype: str
         """
-        with self._driver.session() as session:
-           state = self._read_transaction(tx.get_workflow_state) 
-        return state
+        return self._read_transaction(tx.get_workflow_state)
 
     def set_workflow_state(self, state):
-        """Set the state of the workflow. 
-         
+        """Set the state of the workflow in the Neo4j database.
+
         :param state: the new state of the workflow
         :type state: str
         """
@@ -576,3 +576,5 @@ def _reconstruct_metadata(metadata_record):
     """
     rec = metadata_record["m"]
     return {key: val for key, val in rec.items() if key != "state"}
+
+# pylama:ignore=E1129
diff --git a/src/beeflow/common/gdb_interface.py b/src/beeflow/common/gdb_interface.py
index 0db10136b..8820dfb32 100644
--- a/src/beeflow/common/gdb_interface.py
+++ b/src/beeflow/common/gdb_interface.py
@@ -82,10 +82,10 @@ def load_task(self, task):
     def initialize_ready_tasks(self):
         """Set runnable tasks in a workflow to ready."""
         self._connection.initialize_ready_tasks()
-    
+
     def restart_task(self, old_task, new_task):
         """Create a new task from a failed task checkpoint restart enabled.
-        
+
         :param old_task: the failed task
         :type old_task: Task
         :param new_task: the new (restarted) task
@@ -126,8 +126,9 @@ def get_workflow_state(self):
 
     def set_workflow_state(self, state):
         """Return workflow's current state.
+
         :param state: the new state of the workflow
-        :type str
+        :type state: str
         """
         self._connection.set_workflow_state(state)
 
@@ -246,7 +247,7 @@ def set_task_output(self, task, output_id, value):
         """
         self._connection.set_task_output(task, output_id, value)
 
-    def evaluate_expression(self, task, id, output):
+    def evaluate_expression(self, task, id_, output):
         """Evaluate a task input/output expression.
 
         Expression can be a parameter substitution or a string concatenation in a StepInput
@@ -255,16 +256,16 @@ def evaluate_expression(self, task, id, output):
 
         :param task: the task whose expression to evaluate
         :type task: Task
-        :param id: the id of the step input/output
-        :type id: str
+        :param id_: the id of the step input/output
+        :type id_: str
         :param output: true if output glob expression being evaluated, else false
         :type output: bool
         """
         # Get ground-truth task from database before doing anything
-        task =  self._connection.get_task_by_id(task.id)
+        task = self._connection.get_task_by_id(task.id)
         input_pairs = {input.id: input.value for input in task.inputs}
         if output:
-            step_output = self._connection.get_task_output(task, id)
+            step_output = self._connection.get_task_output(task, id_)
             # Find all matches of string pattern $(inputs.foo), capturing value of foo
             expr_pattern = r"\$\(inputs\.(\w+)\)"
             match = re.findall(expr_pattern, step_output.glob)
@@ -282,9 +283,9 @@ def evaluate_expression(self, task, id, output):
                 eval_string = split[0]
                 for v, s in zip(values, split[1:]):
                     eval_string += v + s
-                self._connection.set_task_output_glob(task, id, eval_string)
+                self._connection.set_task_output_glob(task, id_, eval_string)
         else:
-            step_input = self._connection.get_task_input(task, id)
+            step_input = self._connection.get_task_input(task, id_)
             # Match any of the following patterns: self.path, inputs.foo, or "foo"
             # in an expression of the form $(A + B) in the valueFrom field
             expr_pattern = r'\$\((self\.path|inputs\.\w+|".+") \+ (self\.path|inputs\.\w+|".+")\)'
@@ -302,8 +303,8 @@ def evaluate_expression(self, task, id, output):
                     else:
                         # Pattern "foo" (strip quotes)
                         values.append(m[1:-1])
-                self._connection.set_task_input_type(task, id, "string")
-                self._connection.set_task_input(task, id, values[0] + values[1])
+                self._connection.set_task_input_type(task, id_, "string")
+                self._connection.set_task_input(task, id_, values[0] + values[1])
             else:
                 raise ValueError(f"unable to evaluate expression {step_input.value_from}")
 
diff --git a/src/beeflow/common/parser/dump_cwl.py b/src/beeflow/common/parser/dump_cwl.py
deleted file mode 100755
index 94324d62e..000000000
--- a/src/beeflow/common/parser/dump_cwl.py
+++ /dev/null
@@ -1,250 +0,0 @@
-#!/usr/bin/env python3
-
-# Argument parsing inspired by:
-# "Writing sustainable Python scripts"
-# https://vincent.bernat.ch/en/blog/2019-sustainable-python-script
-#
-# cwl-utils repo:
-# https://github.com/common-workflow-language/cwl-utils
-#
-# Yes, this is ugly and undocumented. Will be cleaned up a documented
-# when I move it into the main project structure.
-
-"""Dump contents a CWL file.
-
-This script will dump (display to terminal) the contents of a CWL file. The CWL
-file is parsed using parser_v1_0.py from the cwl-utils repository. That parser
-creates Python objects from the CWL file (as opposed to other parsing
-techniques that produce Python dictionaries.
-
-"""
-
-import sys
-import argparse
-import uuid
-import cwl_utils.parser.cwl_v1_0 as cwl
-
-
-class CWLCommandLineTool:
-    """Creates a shell command from CWL."""
-
-    def __init__(self, cwl_obj):
-        """Initialize the shell command.
-
-        param cwl_obj: must be a CWL CommandLineTool object
-        """
-        # processes CWL CommandInputParameter
-        def _add_input(obj):
-            arg = {}
-            arg['id'] = obj.id
-            arg['type'] = obj.type
-            arg['position'] = obj.inputBinding.position
-            self.inputs.append(arg)
-
-        # processes CWL CommandOutputParameter
-        def _add_output(obj):
-            arg = {}
-            arg['id'] = obj.id
-            arg['type'] = obj.type
-            self.outputs.append(arg)
-
-        self.inputs = []
-        self.outputs = []
-        self.uuid = uuid.uuid4()
-        self.base_command = cwl_obj.baseCommand
-        for i in cwl_obj.inputs:
-            _add_input(i)
-        for i in cwl_obj.outputs:
-            _add_output(i)
-
-    def __str__(self):
-        """Pretty print the shell command."""
-        # multi-line using parens: https://stackoverflow.com/a/10660477/227441
-        return(f"{self.uuid.hex}: CWLCommandLineTool\n"
-               f"    {self.base_command} param param")
-
-
-
-def cwl_dumper(obj):
-    cwl_dumper_dict = {
-        cwl.Workflow:                        dump_workflow,
-        cwl.InputParameter:                  dump_input_parameter,
-        cwl.WorkflowOutputParameter:         dump_output_parameter,
-        cwl.WorkflowStepInput:               dump_workflow_step_input,
-        cwl.CommandLineTool:                 dump_command_line_tool,
-        cwl.CommandInputParameter:           dump_command_input_parameter,
-        cwl.CommandOutputParameter:          dump_command_output_parameter,
-        cwl.CommandLineBinding:              dump_command_line_binding,
-        cwl.ScatterFeatureRequirement:       dump_requirement,
-        cwl.InlineJavascriptRequirement:     dump_requirement,
-        cwl.StepInputExpressionRequirement:  dump_requirement,
-        cwl.WorkflowStep:                    dump_step
-    }
-    # print(f" >>>>>>>>>>>>>>>>>>>> {type(obj)}")
-    func = cwl_dumper_dict.get(type(obj), lambda: "Invalid")
-    # print(f" >>>>>>>>>>>>>>>>>>>> {func}")
-    return func(obj)
-
-
-def dump_workflow(obj):
-    print(f"==== {type(obj)} ====")
-    # Workflow ID should be "" or some default like "main"
-    print(f"id:               #")
-    # print(f"id:               #{obj.id}")
-    # print(f"label:            {obj.label}")
-    # print(f"doc:              {obj.doc}")
-    # print(f"cwlVersion:       {obj.cwlVersion}")
-    # print(f"hints :           {obj.hints}")
-    # print(f"extension_fields: {obj.extension_fields}")
-    for i in obj.inputs:
-        cwl_dumper(i)
-    for i in obj.outputs:
-        cwl_dumper(i)
-    # for i in obj.requirements:
-    #     cwl_dumper(i)
-    for i in obj.steps:
-        cwl_dumper(i)
-
-
-def dump_input_parameter(obj):
-    print(f"---- {type(obj)} ----")
-    print(f"id:               #{obj.id.split('#')[1]}")
-    # print(f"label:            {obj.label}")
-    # print(f"doc:              {obj.doc}")
-    # print(f"secondaryFiles:   {obj.secondaryFiles}")
-    # print(f"streamable:       {obj.streamable}")
-    # print(f"format:           {obj.format}")
-    # print(f"inputBinding:     {obj.inputBinding}")
-    print(f"default:          {obj.default}")
-    print(f"type:             {obj.type}")
-    # print(f"extension_fields: {obj.extension_fields}")
-
-
-
-def dump_output_parameter(obj):
-    print(f"---- {type(obj)} ----")
-    print(f"id:               #{obj.id.split('#')[1]}")
-    # print(f"label:            {obj.label}")
-    # print(f"doc:              {obj.doc}")
-    # print(f"secondaryFiles:   {obj.secondaryFiles}")
-    # print(f"streamable:       {obj.streamable}")
-    # print(f"format:           {obj.format}")
-    # print(f"outputBinding:    {obj.outputBinding}")
-    print(f"outputSource:     #{obj.outputSource.split('#')[1]}")
-    # print(f"linkMerge:        {obj.linkMerge}")
-    print(f"type:             {obj.type}")
-    # print(f"extension_fields: {obj.extension_fields}")
-
-
-def dump_requirement(obj):
-    print(f"---- {type(obj)} ----")
-    print("    *** TBD")
-
-
-def dump_step(obj):
-    print(f"---- {type(obj)} ----")
-    print(f"id:               #{obj.id.split('#')[1]}")
-    # print(f"label:            {obj.label}")
-    # print(f"doc:              {obj.doc}")
-    # print(f"hints:            {obj.hints}")
-    # print(f"requirements:     {obj.requirements}")
-    # print(f"scatter:          {obj.scatter}")
-    # print(f"scatterMethod:    {obj.scatterMethod}")
-    # print(f"extension_fields: {obj.extension_fields}")
-    for i in obj.in_:
-        cwl_dumper(i)
-    for i in obj.out:
-        print(f"out:              #{i.split('#')[1]}")
-    cwl_dumper(obj.run)
-
-
-def dump_workflow_step_input(obj):
-    print(f"---- {type(obj)} ----")
-    print(f"id:               #{obj.id.split('#')[1]}")
-    print(f"source:           #{obj.source.split('#')[1]}")
-    # print(f"linkMerge:        {obj.linkMerge}")
-    print(f"default:          {obj.default}")
-    # print(f"valueFrom:        {obj.valueFrom}")
-    # print(f"extension_fields: {obj.extension_fields}")
-
-
-def dump_command_line_tool(obj):
-    print(f"---- {type(obj)} ----")
-    print(f"id:                 {obj.id}")
-    # print(f"label:              {obj.label}")
-    # print(f"doc:                {obj.doc}")
-    print(f"hints:              {obj.hints}")
-    # print(f"requirements:       {obj.requirements}")
-    print(f"baseCommand:        {obj.baseCommand}")
-    print(f"stdin:              {obj.stdin}")
-    print(f"stdout:             {obj.stdout}")
-    print(f"stderr:             {obj.stderr}")
-    # print(f"arguments:          {obj.arguments}")
-    # print(f"successCodes:       {obj.successCodes}")
-    # print(f"temporaryFailCodes: {obj.temporaryFailCodes}")
-    # print(f"permanentFailCodes: {obj.permanentFailCodes}")
-    # print(f"extension_fields:   {obj.extension_fields}")
-    for i in obj.inputs:
-        cwl_dumper(i)
-    for i in obj.outputs:
-        cwl_dumper(i)
-
-
-def dump_command_input_parameter(obj):
-    print(f"---- {type(obj)} ----")
-    print(f"id:               {obj.id}")
-    # print(f"label:            {obj.label}")
-    # print(f"doc:              {obj.doc}")
-    # print(f"secondaryFiles:   {obj.secondaryFiles}")
-    # print(f"streamable:       {obj.streamable}")
-    # print(f"format:           {obj.format}")
-    print(f"inputBinding:")
-    cwl_dumper(obj.inputBinding)
-    print(f"default:          {obj.default}")
-    print(f"type:             {obj.type}")
-    # print(f"extension_fields: {obj.extension_fields}")
-
-
-def dump_command_output_parameter(obj):
-    print(f"---- {type(obj)} ----")
-    print(f"id:               {obj.id}")
-    # print(f"label:            {obj.label}")
-    # print(f"doc:              {obj.doc}")
-    # print(f"secondaryFiles:   {obj.secondaryFiles}")
-    # print(f"streamable:       {obj.streamable}")
-    # print(f"format:           {obj.format}")
-    # print(f"outputBinding:")
-    print(f"type:             {obj.type}")
-
-
-def dump_command_line_binding(obj):
-    print(f"---- {type(obj)} ----")
-    # print(f"loadContents:     {obj.loadContents}")
-    print(f"position:         {obj.position}")
-    # print(f"prefix:           {obj.prefix}")
-    # print(f"sepatate:         {obj.separate}")
-    # print(f"itemSeparator:    {obj.itemSeparator}")
-    # print(f"shellQuote:       {obj.shellQuote}")
-    # print(f"extension_fields: {obj.extension_fields}")
-
-
-
-
-
-def parse_args(args=sys.argv[1:]):
-    """Parse arguments."""
-    parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
-
-    parser.add_argument("cwl_file", type=str, help="CWL file dump")
-    return parser.parse_args(args)
-
-
-def main():
-    args = parse_args()
-    top = cwl.load_document(args.cwl_file)
-    cwl_dumper(top)
-
-
-if __name__ == "__main__":
-    sys.exit(main())
-
diff --git a/src/beeflow/common/parser/parse_clamr.py b/src/beeflow/common/parser/parse_clamr.py
deleted file mode 100644
index f8047e8c7..000000000
--- a/src/beeflow/common/parser/parse_clamr.py
+++ /dev/null
@@ -1,242 +0,0 @@
-#!/usr/bin/env python3
-
-# Argument parsing inspired by:
-# "Writing sustainable Python scripts"
-# https://vincent.bernat.ch/en/blog/2019-sustainable-python-script
-#
-# cwl-utils repo:
-# https://github.com/common-workflow-language/cwl-utils
-
-"""Parse contents a CWL file.
-
-This script will parse (and build a workflow in the database) the
-contents of a CWL file. The CWL file is parsed using parser_v1_0.py
-from the cwl-utils repository. That parser creates Python objects from
-the CWL file (as opposed to other parsing techniques that produce
-Python dictionaries. This graph of Python objects is then traversed
-and loaded into the Neo4j database.
-
-"""
-
-import sys
-import argparse
-import uuid
-import cwl_utils.parser.cwl_v1_0 as cwl
-from beeflow.common.wf_interface import WorkflowInterface
-
-
-
-
-# Build a Neo4j graph (workflow) from Python object graph (produced by
-# prior loading of a CWL file).
-def create_workflow(obj, wfi):
-    print(f"==== {type(obj)} ====")
-
-    # Get workflow's required inputs and outputs.
-    ins = get_wf_inputs(obj.inputs)
-    outs = get_wf_outputs(obj.outputs)
-    
-    # Hack until we refactor databse to do dependencies
-    # correctly. Since our current workflow interface supports only
-    # single file dependencies, this intermediate file must be
-    # removed. This hack will be removed as we refactor the databse
-    # interface to correctly support multiple dependencies.
-    outs.discard("clamr/outfile")
-    
-    print(f"ins:  {ins}")
-    print(f"outs: {outs}")
-
-    # This his here for probable requirement to gather a Docker requirement.
-    # for i in obj.requirements:
-    #     pass
-
-    # Use workflow interface (note the passed in reference to a Neo4j
-    # instance) to instantiate a workflow.
-    workflow = wfi.initialize_workflow("Clamr Workflow", ins, outs)
-
-    # Now create (and store in the databse) all the workdlow's tasks.
-    for i in obj.steps:
-        create_task(i, wfi, workflow)
-
-
-
-# Create task based on the parsed CWL and load it into the Neo4j databse.
-def create_task(obj, wfi, workflow):
-    # Strip off leading garbage from the task name.
-    tname = obj.id.split('#')[1]
-
-    ins = set()
-    outs = set()
-
-    # Stip the garbase off the front of all task input names.
-    for i in obj.in_:
-        ins.add(i.source.split('#')[1])
-
-    # Hack until we refactor databse to do dependencies
-    # correctly. Since our current workflow interface supports only
-    # single file dependencies, this non-file input must be
-    # removed. This hack will be removed as we refactor the databse
-    # interface to correctly support multiple dependencies (and
-    # non-file dependencies).
-    #ins.discard("pattern")
-
-    # Stip the garbase off the front of all task otput names.
-    for i in obj.out:
-        outs.add(i.split('#')[1])
-
-    # Build the task's command (which will get sent to the task
-    # manager). This is a concatenation of the base command, any input
-    # parameters, an input file, and a redirection of stdout to an
-    # outpt file.
-    base = obj.run.baseCommand
-    params = get_task_params(obj.run.inputs)
-    # Need to sort the input parameters so they are in the correct
-    # order on the command line.
-    sorted_params = [value for (key, value) in sorted(params.items())]
-    sorted_params_str = ""
-    for p in sorted_params:
-        sorted_params_str = sorted_params_str + " " + p
-    redirect_out = obj.run.stdout
-    # Hack for CLAMR demo. Base command is the full command. Don't append bogus
-    # parameters. If time allows, do this right with real paraeters for the
-    # commands.
-    # cmd = base + " " + sorted_params_str + " > " + redirect_out
-    cmd = base
-
-    # Get any hints for the task. We only support DockerRequirement for now.
-    thints = []
-    if obj.run.hints is not None:
-        for i in obj.run.hints:
-            if "DockerRequirement" in i.values():
-                del i["class"]
-                thints.append(wfi.create_requirement("DockerRequirement",
-                                                     {key: value for key, value in i.items()}))
-
-    print(f"task:  {tname}")
-    print(f"  ins:      {ins}")
-    print(f"  outs:     {outs}")
-    print(f"  command:  {cmd}")
-    print(f"  hints:    {thints}")
-
-    # Using the BEE workflow interface (note the passed in reference
-    # to a Neo4j databse) to load the task nto the database.
-    wfi.add_task(name=tname, command=cmd, inputs=ins, outputs=outs, hints=thints)
-
-    
-
-# Get the workflow's inputs as parsed from the CWL file.
-def get_wf_inputs(objarray):
-    wf_inputs = set()
-    for i in objarray:
-        if i.type == "File":
-            wf_inputs.add(i.id.split("#")[1])
-    return wf_inputs
-
-
-
-# Get the workflow's outputs as parsed from the CWL file.
-def get_wf_outputs(objarray):
-    wf_outputs = set()
-    for i in objarray:
-        id = i.id.split("#")[1]
-        source = i.outputSource.split('#')[1]
-        wf_outputs.add(source.replace(id + "/", ""))
-    return wf_outputs
-
-
-
-# Get the task's command parameters as parsed from the CWL file.
-def get_task_params(objarray):
-    params = {}
-    for i in objarray:
-        params.update({i.inputBinding.position: i.default})
-    return params
-
-
-
-# Print array of tasks (as retreived frm Neo4j) to console.
-def dump_tasks(tarray, wfi):
-    print("-- Tasks: name, IDs (truncated), state, command")
-    for t in tarray:
-        print(f"{t.name:<12}{str(t.id):<6.5}{wfi.get_task_state(t):<10.9}{t.command}")
-        if t.hints is not None:
-            print("      hints:")
-            for h in t.hints:
-                req_class, params = h
-                print(f"        req_class: {req_class}   params: {params}")
-    print("\n-- Tasks: name, dependent tasks")
-    for t in tarray:
-        print(f"{t.name:<12}", end="")
-        for dt in wfi.get_dependent_tasks(t):
-            print(f"{dt.name:<12}", end="")
-        print("")
-
-
-
-# Simple test code to verify a Neo4j workflow.
-def verify_workflow(wfi):
-    print("\n\n==== Is the workflow loaded?")
-    print(wfi.workflow_loaded())
-
-    # Get all the tasks from Neo4j and print them to console.
-    (workflow, tasks) = wfi.get_workflow()
-    dump_tasks(tasks, wfi)
-    
-    # Fake a workflow manager loop. This little hack only works for workflows
-    # where each task has at most one dependent task. In other words,
-    # fan in and fan out workflows are not handled here.
-    wfi.execute_workflow()
-    tid = list(wfi.get_ready_tasks())[0].id
-    print(f"{wfi.get_task_by_id(tid).name}", end="")
-    while wfi.get_dependent_tasks(wfi.get_task_by_id(tid)):
-        print(" --> ", end="")
-        # HACK: doesn't handle set sizes greater than one (i.e. fan
-        # in, fan out))
-        dt = wfi.get_dependent_tasks(wfi.get_task_by_id(tid)).pop()
-        print(f"{dt.name}", end="")
-        tid = dt.id
-    print("")
-
-
-
-def parse_args(args=sys.argv[1:]):
-    """Parse arguments."""
-    parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
-
-    parser.add_argument("cwl_file", type=str, help="CWL file dump")
-    return parser.parse_args(args)
-
-
-
-def main():
-    args = parse_args()
-
-    # Open the database using default parameters. You'll obviously need one
-    # running at the default port. See the documentation on the internal BEE
-    # GitLab for instructions on how to o this:
-    #
-    # https://gitlab.lanl.gov/BEE/database/neo4j
-    wfi = WorkflowInterface()
-
-    # Load the file into the graph of Python objects representng the workflow.
-    top = cwl.load_document(args.cwl_file)
-
-    # Traverse the graph of Python objects and load them into the Neo4j database.
-    create_workflow(top, wfi)
-
-    # Run some simple test code to print the representation of the
-    # workflow as it exists in the database.
-    verify_workflow(wfi)
-    
-    # Clear the workflow that we just built from the database. Comment out if you
-    # want to take a look at the workflow in the database. Note that if this code
-    # dies with an error before it gets here you'll have to delete the databse via
-    # Cypher in the wen interface. This command will do the trick:
-    #
-    # MATCH(n) WITH n LIMIT 10000 DETACH DELETE n
-    #wfi.finalize_workflow()
-
-
-
-if __name__ == "__main__":
-    sys.exit(main())
diff --git a/src/beeflow/common/parser/parse_cwl.py b/src/beeflow/common/parser/parse_cwl.py
deleted file mode 100755
index c3781e106..000000000
--- a/src/beeflow/common/parser/parse_cwl.py
+++ /dev/null
@@ -1,236 +0,0 @@
-#!/usr/bin/env python3
-
-# Argument parsing inspired by:
-# "Writing sustainable Python scripts"
-# https://vincent.bernat.ch/en/blog/2019-sustainable-python-script
-#
-# cwl-utils repo:
-# https://github.com/common-workflow-language/cwl-utils
-
-"""Parse contents a CWL file.
-
-This script will parse (and build a workflow in the databse) the
-contents of a CWL file. The CWL file is parsed using parser_v1_0.py
-from the cwl-utils repository. That parser creates Python objects from
-the CWL file (as opposed to other parsing techniques that produce
-Python dictionaries. This graph of Python objects is then traversed
-and loaded into the Neo4j databse.
-
-"""
-
-import sys
-import argparse
-import uuid
-import cwl_utils.parser_v1_0 as cwl
-from beeflow.common.wf_interface import WorkflowInterface
-
-
-
-
-# Build a Neo4j graph (workflow) from Python object graph (produced by
-# prior loading of a CWL file).
-def create_workflow(obj, wfi):
-    print(f"==== {type(obj)} ====")
-
-    # Get workflow's required inputs and outputs.
-    ins = get_wf_inputs(obj.inputs)
-    outs = get_wf_outputs(obj.outputs)
-    
-    # Hack until we refactor databse to do dependencies
-    # correctly. Since our current workflow interface supports only
-    # single file dependencies, this intermediate file must be
-    # removed. This hack will be removed as we refactor the databse
-    # interface to correctly support multiple dependencies.
-    outs.discard("grep/outfile")
-    
-    print(f"ins:  {ins}")
-    print(f"outs: {outs}")
-
-    # This his here for probable requirement to gather a Docker requirement.
-    # for i in obj.requirements:
-    #     pass
-
-    # Use workflow interface (note the passed in reference to a Neo4j
-    # instance) to instantiate a workflow.
-    wfi.initialize_workflow(ins, outs)
-
-    # Now create (and store in the databse) all the workdlow's tasks.
-    for i in obj.steps:
-        create_task(i, wfi)
-
-
-
-# Create task based on the parsed CWL and load it into the Neo4j databse.
-def create_task(obj, wfi):
-    # Strip off leading garbage from the task name.
-    tname = obj.id.split('#')[1]
-
-    ins = set()
-    outs = set()
-
-    # Stip the garbase off the front of all task input names.
-    for i in obj.in_:
-        ins.add(i.source.split('#')[1])
-
-    # Hack until we refactor databse to do dependencies
-    # correctly. Since our current workflow interface supports only
-    # single file dependencies, this non-file input must be
-    # removed. This hack will be removed as we refactor the databse
-    # interface to correctly support multiple dependencies (and
-    # non-file dependencies).
-    ins.discard("pattern")
-
-    # Stip the garbase off the front of all task otput names.
-    for i in obj.out:
-        outs.add(i.split('#')[1])
-
-    # Build the task's command (which will get sent to the task
-    # manager). This is a concatenation of the base command, any input
-    # parameters, an input file, and a redirection of stdout to an
-    # outpt file.
-    base = obj.run.baseCommand
-    params = get_task_params(obj.run.inputs)
-    # Need to sort the input parameters so they are in the correct
-    # order on the command line.
-    sorted_params = [value for (key, value) in sorted(params.items())]
-    sorted_params_str = ""
-    for p in sorted_params:
-        sorted_params_str = sorted_params_str + " " + p
-    redirect_out = obj.run.stdout
-    cmd = base + " " + sorted_params_str + " > " + redirect_out
-
-    # Get any hints for the task. We only support DockerRequirement for now.
-    thints = set()
-    if obj.run.hints is not None:
-        for i in obj.run.hints:
-            for key, value in i.items():
-                thints.add(wfi.create_requirement(i["class"], key, value))
-
-    print(f"task:  {tname}")
-    print(f"  ins:      {ins}")
-    print(f"  outs:     {outs}")
-    print(f"  command:  {cmd}")
-    print(f"  hints:    {thints}")
-
-    # Using the BEE workflow interface (note the passed in reference
-    # to a Neo4j databse) to load the task nto the database.
-    wfi.add_task(name=tname, command=cmd, inputs=ins, outputs=outs, hints=thints)
-
-    
-
-# Get the workflow's inputs as parsed from the CWL file.
-def get_wf_inputs(objarray):
-    wf_inputs = set()
-    for i in objarray:
-        if i.type == "File":
-            wf_inputs.add(i.id.split("#")[1])
-    return wf_inputs
-
-
-
-# Get the workflow's inputs as parsed from the CWL file.
-def get_wf_outputs(objarray):
-    wf_outputs = set()
-    for i in objarray:
-        id = i.id.split("#")[1]
-        source = i.outputSource.split('#')[1]
-        wf_outputs.add(source.replace(id + "/", ""))
-    return wf_outputs
-
-
-
-# Get the task's command parameters as parsed from the CWL file.
-def get_task_params(objarray):
-    params = {}
-    for i in objarray:
-        params.update({i.inputBinding.position: i.default})
-    return params
-
-
-
-# Print array of tasks (as retreived frm Neo4j) to console.
-def dump_tasks(tarray, wfi):
-    print("-- Tasks: name, IDs (truncated), state, command")
-    for t in tarray:
-        print(f"{t.name:<12}{str(t.id):<6.5}{wfi.get_task_state(t):<10.9}{t.command}")
-        if t.hints is not None:
-            print("      hints:")
-            for h in t.hints:
-                req_class, key, value = h
-                print(f"        req_class: {req_class}   key: {key}   value: {value}")
-    print("\n-- Tasks: name, dependent tasks")
-    for t in tarray:
-        print(f"{t.name:<12}", end="")
-        for dt in wfi.get_dependent_tasks(t):
-            print(f"{dt.name:<12}", end="")
-        print("")
-
-
-
-# Simple test code to verify a Neo4j workflow.
-def verify_workflow(wfi):
-    print("\n\n==== Is the workflow loaded?")
-    print(wfi.workflow_loaded())
-
-    # Get all the tasks from Neo4j and print them to console.
-    (tasks, requirements, hints) = wfi.get_workflow()
-    dump_tasks(tasks, wfi)
-    
-    # Fake a workflow manager loop.  By convention, bee_init is ID 0,
-    # bee_exit is ID 1. This little hack only works for workflows
-    # where each task has at most one dependent task. In other words,
-    # fan in and fan out workflows are not handled here.
-    tid = 0
-    print(f"{wfi.get_task_by_id(tid).name}", end="")
-    while tid != 1:
-        print(" --> ", end="")
-        # HACK: doesn't handle set sizes greater than one (i.e. fan
-        # in, fan out))
-        dt = wfi.get_dependent_tasks(wfi.get_task_by_id(tid)).pop()
-        print(f"{dt.name}", end="")
-        tid = dt.id
-    print("")
-
-
-
-def parse_args(args=sys.argv[1:]):
-    """Parse arguments."""
-    parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
-
-    parser.add_argument("cwl_file", type=str, help="CWL file dump")
-    return parser.parse_args(args)
-
-
-
-def main():
-    args = parse_args()
-
-    # Open the database using default parameters. You'll obviously need one
-    # running at the default port. See the documentation on the internal BEE
-    # GitLab for instructions on how to o this:
-    #
-    # https://gitlab.lanl.gov/BEE/database/neo4j
-    wfi = WorkflowInterface()
-
-    # Load the file into the graph of Python objects representng the workflow.
-    top = cwl.load_document(args.cwl_file)
-
-    # Traverse the graph of Python objects and load them into the Neo4j database.
-    create_workflow(top, wfi)
-
-    # Run some simple test code to print the representation of the
-    # workflow as it exists in the database.
-    verify_workflow(wfi)
-    
-    # Clear the workflow that we just built from the database. Comment out if you
-    # want to take a look at the workflow in the database. Note that if this code
-    # dies with an error before it gets here you'll have to delete the databse via
-    # Cypher in the wen interface. This command will do the trick:
-    #
-    # MATCH(n) WITH n LIMIT 10000 DETACH DELETE n
-    wfi.finalize_workflow()
-
-
-
-if __name__ == "__main__":
-    sys.exit(main())
diff --git a/src/beeflow/common/parser/parser.py b/src/beeflow/common/parser/parser.py
index bf79e5a07..b23be2b6d 100644
--- a/src/beeflow/common/parser/parser.py
+++ b/src/beeflow/common/parser/parser.py
@@ -12,6 +12,7 @@
 import os
 import yaml
 import cwl_utils.parser.cwl_v1_2 as cwl_parser
+
 from beeflow.common.config_driver import BeeConfig as bc
 from beeflow.common.wf_data import (InputParameter,
                                     OutputParameter,
@@ -22,6 +23,10 @@
 from beeflow.common.wf_interface import WorkflowInterface
 
 
+if not bc.ready():
+    bc.init()
+
+
 # Map CWL types to Python types
 type_map = {
     "string": str,
@@ -76,6 +81,8 @@ def parse_workflow(self, cwl, job=None):
         if job:
             # Parse input job params into self.params
             self.parse_job(job)
+        else:
+            self.params = {}
 
         def resolve_input(input_, type_):
             """Resolve workflow input parameter from job file.
@@ -88,11 +95,12 @@ def resolve_input(input_, type_):
             """
             # Use parsed input parameter for input value if it exists
             input_id = _shortname(input_.id)
-            if job:
-                if not isinstance(self.params[input_id], type_map[type_]):
-                    raise ValueError("Input/param types do not match: "
-                                     f"{input_id}/{self.params[input_id]}")
-                return self.params[input_id]
+            if input_id not in self.params:
+                return None
+            if not isinstance(self.params[input_id], type_map[type_]):
+                raise ValueError("Input/param types do not match: "
+                                 f"{input_id}/{self.params[input_id]}")
+            return self.params[input_id]
 
         workflow_name = os.path.basename(cwl).split(".")[0]
         workflow_inputs = {InputParameter(_shortname(input_.id), input_.type,
@@ -105,7 +113,7 @@ def resolve_input(input_, type_):
         workflow_requirements = self.parse_requirements(self.cwl.requirements)
 
         self._wfi.initialize_workflow(workflow_name, workflow_inputs, workflow_outputs,
-                                                 workflow_requirements, workflow_hints)
+                                      workflow_requirements, workflow_hints)
         for step in self.cwl.steps:
             self.parse_step(step)
 
@@ -143,8 +151,8 @@ def parse_step(self, step):
         step_stdout = step_cwl.stdout
 
         self._wfi.add_task(step_name, base_command=step_command, inputs=step_inputs,
-                          outputs=step_outputs, requirements=step_requirements,
-                          hints=step_hints, stdout=step_stdout)
+                           outputs=step_outputs, requirements=step_requirements,
+                           hints=step_hints, stdout=step_stdout)
 
     def parse_job(self, job):
         """Parse a CWL input job file.
@@ -155,10 +163,10 @@ def parse_job(self, job):
         :type job: str
         """
         if job.endswith(".yml"):
-            with open(job) as fp:
+            with open(job, encoding="utf-8") as fp:
                 self.params = yaml.full_load(fp)
         elif job.endswith(".json"):
-            with open(job) as fp:
+            with open(job, encoding="utf-8") as fp:
                 self.params = json.load(fp)
         else:
             raise ValueError("Unsupported input job file extension")
@@ -186,11 +194,12 @@ def parse_step_inputs(cwl_in, step_inputs):
         for step_input in step_inputs:
             # If the input type is str, then the input is required
             # If it is a list containing 'null' and another type(s) then it is optional
+            # If the input is not in the source_map but has a default value then it will
+            # be set by the GDB later
             if _shortname(step_input.id) not in source_map.keys():
-                if isinstance(step_input.type, str):
+                if isinstance(step_input.type, str) and step_input.default is None:
                     raise ValueError(f"required input {_shortname(step_input.id)} not satisfied")
-                else:
-                    continue
+                continue
 
             if isinstance(step_input.type, str):
                 input_type = step_input.type
@@ -224,7 +233,10 @@ def parse_step_outputs(cwl_out, step_outputs, stdout):
         """
         out_short = list(map(_shortname, cwl_out))
         short_id = out_short[0].split("/")[0]
-        out_map = {short_id + "/" + _shortname(step_output.id): step_output
+        # Inline step outputs already have short_id+"/" prepended
+        out_map = {(_shortname(step_output.id)
+                   if _shortname(step_output.id).startswith(short_id + "/") else
+                   short_id + "/" + _shortname(step_output.id)): step_output
                    for step_output in step_outputs}
 
         outputs = []
@@ -298,12 +310,14 @@ def parse_args(args=None):
     parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
 
     parser.add_argument("wf_file", type=str, help="CWL workflow file")
-    parser.add_argument("-i", "--inputs", type=str, help="Workflow job inputs file", required=False)
+    parser.add_argument("-i", "--inputs", type=str, help="Workflow job inputs file",
+                        required=False)
 
     return parser.parse_args(args)
 
 
 def main():
+    """Run the parser on a CWL Workflow and job file directly."""
     parser = CwlParser()
     args = parse_args()
     parser.parse_workflow(args.wf_file, args.inputs)
diff --git a/src/beeflow/common/wf_data.py b/src/beeflow/common/wf_data.py
index cc8fdbd29..60f6fe98f 100644
--- a/src/beeflow/common/wf_data.py
+++ b/src/beeflow/common/wf_data.py
@@ -51,10 +51,10 @@ def __init__(self, name, hints, requirements, inputs, outputs, workflow_id=None)
             self.id = self.generate_workflow_id()
 
         self.state = "SUBMITTED"
-    
+
     def generate_workflow_id(self):
         """Generate a unique workflow ID.
-        
+
         :rtype: str
         """
         return uuid4().hex
@@ -68,9 +68,12 @@ def __eq__(self, other):
         :param other: the workflow with which to test equality
         :type other: Workflow
         """
-        if type(other) is not Workflow:
+        if not isinstance(other, Workflow):
             return False
-        id_sort = lambda i: i.id
+
+        def id_sort(i):
+            return i.id
+
         return bool(self.name == other.name and
                     sorted(self.hints) == sorted(other.hints) and
                     sorted(self.requirements) == sorted(other.requirements) and
@@ -137,14 +140,14 @@ def __init__(self, name, base_command, hints, requirements, inputs, outputs, std
 
     def generate_workflow_id(self):
         """Generate a unique workflow ID.
-        
+
         :rtype: str
         """
         return uuid4().hex
 
     def copy(self, new_id=False):
         """Make a copy of this task.
-        
+
         :param new_id: generate a new task ID
         :type new_id: bool
         :rtype: Task
@@ -163,9 +166,12 @@ def __eq__(self, other):
         :type other: Task
         :rtype: bool
         """
-        if type(other) is not Task:
+        if not isinstance(other, Task):
             return False
-        id_sort = lambda i: i.id
+
+        def id_sort(i):
+            return i.id
+
         return bool(self.name == other.name and
                     self.base_command == other.base_command and
                     sorted(self.hints) == sorted(other.hints) and
@@ -185,14 +191,14 @@ def __ne__(self, other):
 
     def __hash__(self):
         """Return the hash value for a task.
-        
+
         :rtype: int
         """
         return hash((self.id, self.workflow_id, self.name))
 
     def __repr__(self):
         """Construct a task's string representation.
-        
+
         :rtype: str
         """
         return (f"<Task id={self.id} name={self.name} base_command={self.base_command} "
@@ -231,7 +237,7 @@ def command(self):
             if input_.prefix is not None:
                 command.append(input_.prefix)
             command.append(str(input_.value))
-        
+
         # Append restart parameter and checkpoint file if CheckpointRequirement specified
         for hint in self.hints:
             if hint.class_ == "beeflow:CheckpointRequirement":
diff --git a/src/beeflow/common/wf_interface.py b/src/beeflow/common/wf_interface.py
index daa7663d1..b4f3e582a 100644
--- a/src/beeflow/common/wf_interface.py
+++ b/src/beeflow/common/wf_interface.py
@@ -33,9 +33,8 @@ def reconnect(self):
         """Reconnect to the graph database using stored credentials."""
         self._gdb_interface.connect(**self._kwargs)
         if self.workflow_loaded():
-            self._workflow_id = None
-            # This property automatically sets self._workflow_id when invoked
-            self.workflow_id
+            workflow, _ = self.get_workflow()
+            self._workflow_id = workflow.id
 
     def initialize_workflow(self, name, inputs, outputs, requirements=None, hints=None):
         """Begin construction of a BEE workflow.
@@ -127,10 +126,10 @@ def add_task(self, name, base_command, inputs, outputs, requirements=None, hints
         # Load the new task into the graph database
         self._gdb_interface.load_task(task)
         return task
-    
+
     def restart_task(self, task, checkpoint_file):
         """Restart a failed BEE workflow task.
-        
+
         The task must have a beeflow:CheckpointRequirement hint. If there are no
         remaining retry attemps (num_tries = 0) then the graph database is
         unmodified and this method returns None.
@@ -145,10 +144,9 @@ def restart_task(self, task, checkpoint_file):
                     hint.params["num_tries"] -= 1
                     hint.params["bee_checkpoint_file__"] = checkpoint_file
                     break
-                else:
-                    state = self.get_task_state(task)
-                    self.set_task_state(task, f"FAILED RESTART: {state}")
-                    return None
+                state = self.get_task_state(task)
+                self.set_task_state(task, f"FAILED RESTART: {state}")
+                return None
         else:
             raise ValueError("invalid task for checkpoint restart")
 
@@ -214,12 +212,20 @@ def get_workflow_outputs(self):
 
     def get_workflow_state(self):
         """Get the value of the workflow state.
-        
-        rtype: str
+
+        :rtype: str
         """
         state = self._gdb_interface.get_workflow_state()
         return state
 
+    def set_workflow_state(self, state):
+        """Return workflow's current state.
+
+        :param state: the new state of the workflow
+        :type state: str
+        """
+        self._gdb_interface.set_workflow_state(state)
+
     def get_ready_tasks(self):
         """Get ready tasks from a BEE workflow.
 
@@ -325,7 +331,7 @@ def set_task_output(self, task, output_id, value):
         """
         self._gdb_interface.set_task_output(task, output_id, value)
 
-    def evaluate_expression(self, task, id, output=False):
+    def evaluate_expression(self, task, id_, output=False):
         """Evaluate a task input/output expression.
 
         Expression can be either a string concatenation in a StepInput
@@ -335,12 +341,12 @@ def evaluate_expression(self, task, id, output=False):
 
         :param task: the task whose expression to evaluate
         :type task: Task
-        :param id: the id of the step input/output
-        :type id: str
+        :param id_: the id of the step input/output
+        :type id_: str
         :param output: true if output glob expression being evaluated, else false
         :type output: bool
         """
-        self._gdb_interface.evaluate_expression(task, id, output)
+        self._gdb_interface.evaluate_expression(task, id_, output)
 
     def workflow_completed(self):
         """Return true if all of a workflow's final tasks have completed, else false.
diff --git a/src/beeflow/common/worker/simple_worker.py b/src/beeflow/common/worker/simple_worker.py
index e55e4f155..9723a5e77 100644
--- a/src/beeflow/common/worker/simple_worker.py
+++ b/src/beeflow/common/worker/simple_worker.py
@@ -9,17 +9,6 @@
 import beeflow.common.log as bee_logging
 import os
 
-# Import all implemented container runtime drivers now
-# No error if they don't exist
-try:
-    from beeflow.common.crt_drivers import CharliecloudDriver
-except ModuleNotFoundError:
-    pass
-try:
-    from beeflow.common.crt_drivers import SingularityDriver
-except ModuleNotFoundError:
-    pass
-
 
 class SimpleWorker(Worker):
     """The Worker interface for no workload manager."""
diff --git a/src/beeflow/common/worker/slurm_worker.py b/src/beeflow/common/worker/slurm_worker.py
index a236ddbe5..c46486ca3 100644
--- a/src/beeflow/common/worker/slurm_worker.py
+++ b/src/beeflow/common/worker/slurm_worker.py
@@ -12,7 +12,7 @@
 import requests
 import getpass
 
-from beeflow.common.worker.worker import Worker
+from beeflow.common.worker.worker import (Worker, WorkerError)
 from beeflow.cli import log
 import beeflow.common.log as bee_logging
 
@@ -50,7 +50,7 @@ def query_job(job_id, session, slurm_url):
             resp = session.get(f'{slurm_url}/job/{job_id}')
 
             if resp.status_code != 200:
-                job_state = f"BAD_RESPONSE_{resp.status_code}"
+                raise WorkerError(f'Failed to query job {job_id}')
             else:
                 status = json.loads(resp.text)
                 job_state = status['job_state']
@@ -79,9 +79,12 @@ def query_task(self, job_id):
 
     def cancel_task(self, job_id):
         """Worker cancels job, returns job_state."""
-        resp = self.session.delete(f'{self.slurm_url}/job/{job_id}')
+        try:
+            resp = self.session.delete(f'{self.slurm_url}/job/{job_id}')
+        except requests.exceptions.ConnectionError:
+            return 'NOT_RESPONDING'
         if resp.status_code != 200:
-            raise Exception(f'Unable to cancel job id {job_id}!')
+            raise WorkerError(f'Unable to cancel job id {job_id}!')
         job_state = "CANCELLED"
         return job_state
 # Ignore R1732: Warnign about using open without "with' context. Seems like personal preference.
diff --git a/src/beeflow/common/worker/worker.py b/src/beeflow/common/worker/worker.py
index a622cabd9..774a844b5 100644
--- a/src/beeflow/common/worker/worker.py
+++ b/src/beeflow/common/worker/worker.py
@@ -19,6 +19,14 @@
     pass
 
 
+class WorkerError(Exception):
+    """Worker error class."""
+
+    def __init__(self, *args):
+        """Construct a new worker error."""
+        self.args = args
+
+
 class Worker(ABC):
     """Worker interface for a generic workload manager."""
 
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-ci/README.md b/src/beeflow/data/cwl/bee_workflows/clamr-ci/README.md
new file mode 100644
index 000000000..8cc2e9873
--- /dev/null
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-ci/README.md
@@ -0,0 +1,10 @@
+# CLAMR - FFMPEG workflow using CWL
+
+clamr_wf.cwl - the main cwl.
+calmr_job.yml - yaml file for values used by the cwl files.
+clamr.cwl - cwl file for the clamr step.
+ffmpeg.cwl - cwl file for the ffmpeg step.
+
+The values in these files run on fog a LANL cluster, using the container runtime Charliecloud. Fog uses slurm as the workload scheduler.
+
+
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr.cwl b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr.cwl
new file mode 100644
index 000000000..8c259ebfb
--- /dev/null
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr.cwl
@@ -0,0 +1,83 @@
+# -*- mode: YAML; -*-
+
+class: CommandLineTool
+cwlVersion: v1.0
+
+baseCommand: /CLAMR/clamr_cpuonly
+# This is the stdout field which makes all stdout be captured in this file
+# stderr is not currently implemented but it is also a thing
+stdout: clamr_stdout.txt
+# Arguments to the command
+inputs:
+  amr_type:
+    # ? means the argument is optional
+    # All of the ? here are legacy from the original CWL
+    type: string?
+    # Declare extra options
+    # We support prefix and position
+    inputBinding:
+      # Prefix is the flag for cli command
+      prefix: -A
+  grid_res:
+    type: int?
+    inputBinding:
+      prefix: -n
+  max_levels:
+    type: int?
+    inputBinding:
+      prefix: -l
+  time_steps:
+    type: int?
+    inputBinding:
+      prefix: -t
+  output_steps:
+    type: int?
+    inputBinding:
+      prefix: -i
+  graphic_steps:
+    type: int?
+    inputBinding:
+      prefix: -g
+  graphics_type:
+    type: string?
+    inputBinding:
+      prefix: -G
+  rollback_images:
+    type: int?
+    inputBinding:
+      prefix: -b
+  checkpoint_disk_interval:
+    type: int?
+    inputBinding:
+      prefix: -c
+  checkpoint_mem_interval:
+    type: int?
+    inputBinding:
+      prefix: -C
+  hash_method:
+    type: string?
+    inputBinding:
+      prefix: -e
+  
+outputs:
+  # Captures stdout. Name is arbitrary.
+  clamr_stdout:
+    # type is syntactic sugar to just grab the output file defined above
+    # stdout:
+    #     type: File
+    #     outputBinding: 
+    #       glob: clamr_stdout.txt
+    #     stdout is easy shorthand
+    type: stdout
+  outdir:
+    # directory is just another type. Scan the files for a directory with the name specified in glob
+    # If you add a wildcard, it'd do expansion
+    type: Directory
+    outputBinding:
+      # Glob can be either a constant string or have a wildcard 
+      # TODO verify CWLs glob support
+      glob: $HOME/graphics_output/graph%05d.png
+  time_log:
+    type: File
+    outputBinding:
+      glob: total_execution_time.log
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_job.json b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_job.json
new file mode 100644
index 000000000..ad88972f6
--- /dev/null
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_job.json
@@ -0,0 +1,13 @@
+{
+  "grid_resolution": 32,
+  "max_levels": 3,
+  "time_steps": 5000,
+  "steps_between_outputs": 10,
+  "steps_between_graphics": 25,
+  "graphics_type": "png",
+  "input_format": "image2",
+  "frame_rate": 12,
+  "frame_size": "800x800",
+  "pixel_format": "yuv420p",
+  "output_filename": "CLAMR_movie.mp4"
+}
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_job.yml b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_job.yml
new file mode 100644
index 000000000..87f71da34
--- /dev/null
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_job.yml
@@ -0,0 +1,18 @@
+# Inputs for CLAMR
+# /clamr/CLAMR-master/clamr_cpuonly -n 32 -l 3 -t 5000 -i 10 -g 25 -G png
+
+grid_resolution: 32
+max_levels: 3
+time_steps: 500
+steps_between_outputs: 10
+steps_between_graphics: 25
+graphics_type: png
+
+# Inputs for FFMPEG
+#ffmpeg -f image2 -r 12 -s 800x800 -pix_fmt yuv420p $HOME/CLAMR_movie.mp4
+
+input_format: image2
+frame_rate: 12
+frame_size: 800x800
+pixel_format: yuv420p
+output_filename: $HOME/CLAMR_movie.mp4
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_wf.cwl b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_wf.cwl
new file mode 100644
index 000000000..a5d50c4a8
--- /dev/null
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-ci/clamr_wf.cwl
@@ -0,0 +1,81 @@
+# -*- mode: YAML; -*-
+
+class: Workflow
+cwlVersion: v1.0
+
+# Main 3 components of workflow are inputs, outputs, and steps
+
+inputs:
+# All inputs go here for each step. No way to break them up.
+# We should talk to the CWL people about that. 
+##### CLAMR inputs #####
+# takes ID:Type syntax
+  grid_resolution: int
+  max_levels: int
+  time_steps: int
+  steps_between_outputs: int
+  steps_between_graphics: int
+  graphics_type: string
+##### FFMPEG inputs #####
+  input_format: string
+  frame_rate: int
+  frame_size: string
+  pixel_format: string
+  output_filename: string
+
+outputs:
+# Outputs for all the steps
+# Check where we copy these probably at CWD
+# Files need to exist at end of workflow, but we aren't currently checking
+# We have function to get workflow_outputs
+# TODO add step to workflow_manager to confirm that each of these outputs exist
+  clamr_stdout:
+    type: File
+    outputSource: clamr/clamr_stdout
+  clamr_time_log:
+    type: File
+    outputSource: clamr/time_log
+  clamr_movie:
+    type: File
+    outputSource: ffmpeg/movie
+
+steps:
+  clamr:
+    run: clamr.cwl
+    in:
+      grid_res: grid_resolution
+      max_levels: max_levels
+      time_steps: time_steps
+      output_steps: steps_between_outputs
+      graphic_steps: steps_between_graphics
+      graphics_type: graphics_type
+    out: [clamr_stdout, outdir, time_log]
+    hints:
+        DockerRequirement:
+            # TODO Sort this out
+            #dockerImport: clamr_img.tar.gz
+            #copyContainer: clamr
+            beeflow:copyContainer: "/tmp/clamr.tar.gz"
+
+  ffmpeg:
+    run: ffmpeg.cwl
+    in:
+      input_format: input_format
+      # input syntax is name: <step>/dependent_object
+      ffmpeg_input: clamr/outdir
+      frame_rate: frame_rate
+      frame_size: frame_size
+      pixel_format: pixel_format
+      # Setting output file with file_name
+      # output_filename set in wf inputs
+      output_file: output_filename
+    # Multiple outputs can be in array
+    out: [movie]
+    hints:
+        DockerRequirement:
+            # TODO Sort this out
+            #dockerImport: clamr_img.tar.gz
+            #copyContainer: clamr
+            beeflow:copyContainer: "/tmp/clamr.tar.gz"
+    requirements:
+        InlineJavascriptRequirement: {}
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-ci/ffmpeg.cwl b/src/beeflow/data/cwl/bee_workflows/clamr-ci/ffmpeg.cwl
new file mode 100644
index 000000000..8e10490a6
--- /dev/null
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-ci/ffmpeg.cwl
@@ -0,0 +1,44 @@
+# -*- mode: YAML; -*-
+
+class: CommandLineTool
+cwlVersion: v1.0
+
+baseCommand: ffmpeg -y
+inputs:
+  input_format:
+    type: string?
+    inputBinding:
+      prefix: -f
+      position: 1
+  ffmpeg_input:
+    type: Directory
+    inputBinding:
+      prefix: -i
+      position: 2
+      valueFrom: $(self.path + "/graph%05d.png")
+  frame_rate:
+    type: int?
+    inputBinding:
+      prefix: -r
+      position: 3
+  frame_size:
+    type: string?
+    inputBinding:
+      prefix: -s
+      position: 4
+  pixel_format:
+    type: string?
+    inputBinding:
+      prefix: -pix_fmt
+      position: 5
+  output_file:
+    type: string
+    inputBinding:
+      position: 6
+
+outputs:
+  movie:
+    type: File
+    outputBinding:
+      glob: $(inputs.output_file)
+      # glob: CLAMR_movie.mp4
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/copyContainer_containerName.cwl b/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/copyContainer_containerName.cwl
index 50c26580c..a8c59e900 100644
--- a/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/copyContainer_containerName.cwl
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/copyContainer_containerName.cwl
@@ -30,7 +30,7 @@ steps:
       hints:
         DockerRequirement:
           beeflow:copyContainer: "/usr/projects/beedev/clamr/clamr-toss.tar.gz"
-          containerName: "foo"
+          beeflow:containerName: "foo"
     in:
       infile: infile
     out: [outfile]
diff --git a/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/dockerFile_containerName.cwl b/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/dockerFile_containerName.cwl
index b38a9edf0..0cfdf3c9a 100644
--- a/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/dockerFile_containerName.cwl
+++ b/src/beeflow/data/cwl/bee_workflows/clamr-wf-noyaml/slurm-charliecloud/dockerFile_containerName.cwl
@@ -30,7 +30,7 @@ steps:
       hints:
         DockerRequirement:
           dockerFile: "src/beeflow/data/dockerfiles/Dockerfile.clamr-lanl-x86_64"
-          containerName: "foo2"
+          beeflow:containerName: "foo2"
     in:
       infile: infile
     out: [outfile]
diff --git a/src/beeflow/task_manager.py b/src/beeflow/task_manager.py
index 435a4b8b1..d2f783881 100755
--- a/src/beeflow/task_manager.py
+++ b/src/beeflow/task_manager.py
@@ -28,11 +28,11 @@
 else:
     bc.init()
 
-
 from beeflow.cli import log
 from beeflow.common.build_interfaces import build_main
 import beeflow.common.log as bee_logging
-
+from beeflow.common.worker_interface import WorkerInterface
+import beeflow.common.worker as worker_pkg
 
 sys.excepthook = bee_logging.catch_exception
 
@@ -112,7 +112,7 @@ def gen_task_metadata(task, job_id):
 
 def resolve_environment(task):
     """Use build interface to create a valid environment."""
-    build_main(bc, task)
+    build_main(task)
 
 
 def submit_jobs():
@@ -245,7 +245,6 @@ def process_queues():
 
 
 if "pytest" not in sys.modules:
-    # TODO Decide on the time interval for the scheduler
     scheduler = BackgroundScheduler({'apscheduler.timezone': 'UTC'})
     scheduler.add_job(func=process_queues, trigger="interval", seconds=8)
     scheduler.start()
@@ -304,14 +303,9 @@ def delete():
 @flask_app.route('/status')
 def get_status():
     """Report the current status of the Task Manager."""
-    # TODO: Report statistics about jobs, perhaps current system load, etc.
     return make_response(jsonify(status='up'), 200)
 
 
-# WorkerInterface needs to be placed here. Don't Move!
-from beeflow.common.worker_interface import WorkerInterface
-import beeflow.common.worker as worker_pkg
-
 WLS = bc.get('DEFAULT', 'workload_scheduler')
 worker_class = worker_pkg.find_worker(WLS)
 if worker_class is None:
@@ -325,7 +319,6 @@ def get_status():
     # extra options to be passed to the runner (i.e. srun [RUNNER_OPTS] ... for Slurm)
     'runner_opts': bc.get('task_manager', 'runner_opts'),
 }
-# TODO: Maybe this should be put into a sub class
 if WLS == 'Slurm':
     worker_kwargs['slurm_socket'] = bc.get('slurmrestd', 'slurm_socket')
 worker = WorkerInterface(worker_class, **worker_kwargs)
@@ -350,10 +343,6 @@ def get_status():
     # Flask logging
     flask_app.logger.addHandler(handler)
     flask_app.run(debug=False, port=str(tm_listen_port))
-# Ignore TODO comments
-# Ignoring "modules loaded below top of file" warning per Pat's comment
-# Ignoring flask.logger.AddHandler not found because logging is working...
-# Ignoring W1202: https://github.com/PyCQA/pylint/issues/2395
-# Ignoring W0703: Catching general exception is ok in our case.
-# Ignoring C0206: Iterating with .items() is not important for readability or functionality
-# pylama:ignore=W0511,E402,C0413,E1101,W1202,W0703,C0206
+# Ignoring CO413 beeflow modules must be loaded after bc.init()
+# Ignoring W0703: Catching general exception is ok for failed submit and cancel.
+# pylama:ignore=C0413,W0703


Test patch:
diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml
new file mode 100644
index 000000000..87529ec01
--- /dev/null
+++ b/.github/workflows/unit-tests.yml
@@ -0,0 +1,46 @@
+name: Unit tests
+
+on:
+  # For manual exec
+  # (https://github.blog/changelog/2020-07-06-github-actions-manual-triggers-with-workflow_dispatch/)
+  workflow_dispatch: {}
+  push:
+    branches: [main, develop]
+  pull_request:
+    types: [opened, synchronize, edited, labeled, unlabeled]
+    branches: [main, develop]
+
+env: {}
+
+jobs:
+  integration-test:
+    if: ${{ !contains(github.event.pull_request.labels.*.name, 'WIP') }}
+    name: BEE Unit Tests
+    # Note: Needs to run on 22.04 or later since slurmrestd doesn't seem to be
+    # available on 20.04
+    runs-on: ubuntu-22.04
+    steps:
+      - uses: actions/checkout@v2
+      - name: Dependency Install
+        run: ./ci/deps_install.sh
+      - name: Slurm Setup and Install
+        run: ./ci/slurm_start.sh
+      - name: BEE Install
+        run: ./ci/bee_install.sh
+      # TODO: Add the script for the unit tests here
+      - name: Unit tests
+        run: |
+          #!/bin/bash
+          . ./venv/bin/activate
+          . ./ci/env.sh
+          set +e
+          code=0
+          for test in src/beeflow/tests/test_*.py; do
+              pytest $test
+              result=$?
+              # Ensure the step fails if one test fails
+              if [ $result -ne 0 ]; then
+                  code=$result
+              fi
+          done
+          exit $code
diff --git a/ci/integration_test.py b/ci/integration_test.py
new file mode 100755
index 000000000..496cbd73a
--- /dev/null
+++ b/ci/integration_test.py
@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+"""CI workflow run script."""
+from beeflow.common.config_driver import BeeConfig as bc
+from beeflow.client import bee_client
+import subprocess
+import os
+import time
+import sys
+
+
+# Max time of each workflow
+TIMEOUT = 300
+
+
+class CIError(Exception):
+    """CI error class."""
+
+    def __init__(self, *args):
+        """Initialize the error with default args."""
+        self.args = args
+
+
+class Container:
+    """Container CI class for building BEE containers."""
+
+    def __init__(self, name, dockerfile, tarball):
+        """BEE CI container constructor."""
+        self.name = name
+        self.dockerfile = dockerfile
+        self.tarball = tarball
+        self.done = False
+
+    def build(self):
+        """Build the containers and save them in the proper location."""
+        if not self.done:
+            try:
+                subprocess.check_call(['ch-image', 'build', '-f', self.dockerfile, '-t', self.name,
+                                       os.path.dirname(self.dockerfile)])
+                subprocess.check_call(['ch-convert', '-i', 'ch-image', '-o', 'tar', self.name,
+                                       self.tarball])
+            except subprocess.CalledProcessError as error:
+                raise CIError(
+                    f'container build: {error}'
+                ) from None
+            self.done = True
+
+
+class Workflow:
+    """Workflow CI class for interacting with BEE."""
+
+    def __init__(self, name, path, main_cwl, job_file, check_fn, containers):
+        """Workflow constructor."""
+        self.name = name
+        self.path = path
+        self.main_cwl = main_cwl
+        self.job_file = job_file
+        self.check_fn = check_fn
+        self.containers = containers
+
+    def run(self):
+        """Attempt to submit, start and run the workflow to completion."""
+        # Build all the containers first
+        print('Building all containers')
+        for ctr in self.containers:
+            ctr.build()
+        out_tarball = f'{self.name}.tgz'
+        print('Creating the workflow tarball', out_tarball)
+        try:
+            subprocess.check_call(['tar', '-C', os.path.dirname(self.path), '-czf', out_tarball,
+                                   os.path.basename(self.path)])
+        except subprocess.CalledProcessError as error:
+            raise CIError(
+                f'workflow tar cmd: {error}'
+            ) from None
+        try:
+            wf_id = bee_client.submit(self.name, out_tarball, self.main_cwl, self.job_file)
+            bee_client.start(wf_id)
+            time.sleep(2)
+            t = 0
+            while bee_client.query(wf_id)[0] == 'Running' and t < TIMEOUT:
+                time.sleep(4)
+                t += 4
+        except bee_client.ClientError as error:
+            raise CIError(*error.args) from None
+        if t >= TIMEOUT:
+            raise CIError('timeout exceeded')
+        self.check_fn()
+
+
+def test_workflows(workflows):
+    """Test run all workflows."""
+    results = {}
+    for wfl in workflows:
+        try:
+            wfl.run()
+        except CIError as error:
+            results[wfl.name] = error
+        else:
+            results[wfl.name] = None
+
+    print('######## WORKFLOW RESULTS ########')
+    fails = sum(1 if results[name] is not None else 0 for name in results)
+    passes = len(results) - fails
+    print(f'{passes} passes, {fails} fails')
+    for name in results:
+        error = results[name]
+        if error is not None:
+            print(f'{name}: {error}')
+    print('##################################')
+    return 1 if fails > 0 else 0
+
+
+# Initialize BEE's config system
+bc.init()
+
+# Workflow check functions
+def check_clamr():
+    """Ensure that CLAMR was successful."""
+    path = os.path.expanduser('~/CLAMR_movie.mp4')
+    if not os.path.exists(path):
+        raise CIError(f'CLAMR: missing output file "{path}"')
+
+
+# Set up containers and workflows
+clamr_ctr = Container('clamr', 'src/beeflow/data/dockerfiles/Dockerfile.clamr-ffmpeg',
+                      '/tmp/clamr.tar.gz')
+workflows = [
+    Workflow('clamr', './src/beeflow/data/cwl/bee_workflows/clamr-ci', main_cwl='clamr_wf.cwl',
+             job_file='clamr_job.yml', check_fn=check_clamr, containers=[clamr_ctr]),
+]
+
+# Run the workflows and then show completion results
+sys.exit(test_workflows(workflows))
diff --git a/ci/integration_test.sh b/ci/integration_test.sh
new file mode 100755
index 000000000..4be42aec5
--- /dev/null
+++ b/ci/integration_test.sh
@@ -0,0 +1,20 @@
+#!/bin/sh
+
+. ./ci/env.sh
+. venv/bin/activate
+
+# Start the actual integration tests
+./ci/integration_test.py
+# Save the exit code for later
+EXIT_CODE=$?
+
+# Output the logs
+for log in $SLURMCTLD_LOG $SLURMD_LOG $MUNGE_LOG $BEE_WORKDIR/logs/*; do
+    printf "\n\n"
+    printf "#### $log ####\n"
+    cat $log
+    printf "#### $log ####\n"
+    printf "\n\n"
+done
+
+exit $EXIT_CODE
diff --git a/src/beeflow/tests/test_parser.py b/src/beeflow/tests/test_parser.py
index 99a3c3a43..a05718cee 100644
--- a/src/beeflow/tests/test_parser.py
+++ b/src/beeflow/tests/test_parser.py
@@ -33,8 +33,6 @@ def test_parse_workflow(self):
         wfi = self.parser.parse_workflow(cwl_wfi_file, cwl_job_yaml)
         self.assertTrue(wfi.workflow_loaded())
 
-        workflow = wfi.get_workflow()
-
         wfi.finalize_workflow()
         self.assertFalse(wfi.workflow_loaded())
 
@@ -42,8 +40,6 @@ def test_parse_workflow(self):
         wfi = self.parser.parse_workflow(cwl_wfi_file, cwl_job_json)
         self.assertTrue(wfi.workflow_loaded())
 
-        workflow = wfi.get_workflow()
-
     def test_parse_workflow_no_job(self):
         """Test parsing of a workflow without an input job file."""
         cwl_wfi_file = "cf.cwl"
diff --git a/src/beeflow/tests/test_slurm_worker.py b/src/beeflow/tests/test_slurm_worker.py
index 69c49b4dc..5a35bb1e8 100644
--- a/src/beeflow/tests/test_slurm_worker.py
+++ b/src/beeflow/tests/test_slurm_worker.py
@@ -1,129 +1,139 @@
-#! /usr/bin/env python3
-"""Unit test module for BEE slurm worker interface."""
-
-import unittest
+"""Tests of the Slurm worker."""
+import uuid
+import time
+import subprocess
 import os
-import string
+import pytest
+from beeflow.common.config_driver import BeeConfig as bc
+
+
+bc.init()
+
 
-from beeflow.common.data.wf_data import Task
-from beeflow.common.worker.worker_interface import WorkerInterface
+from beeflow.common.worker_interface import WorkerInterface
+from beeflow.common.worker.worker import WorkerError
 from beeflow.common.worker.slurm_worker import SlurmWorker
-from beeflow.common.worker.slurm_worker import submit_job
-
-
-class TestSlurmWorker(unittest.TestCase):
-    """Unit test case for worker interface."""
-
-    @classmethod
-    def setUpClass(cls):
-        """Initialize the Worker interface."""
-        cls.worker = WorkerInterface(SlurmWorker)
-        # for temporary scripts use template if exists
-        template_file = os.path.expanduser('~/.beeflow/worker/job.template')
-        job_template = ''
-        try:
-            template_f = open(template_file, 'r')
-            job_template = template_f.read()
-            template_f.close()
-        except OSError as err:
-            print("OS error: {0}".format(err))
-            print('No job_template: creating a simple job template!')
-            job_template = '#! /bin/bash\n#SBATCH\n'
-        template = string.Template(job_template)
-
-        # write good script
-        sub = {'name': 'good', 'id': 'job'}
-        command = 'echo "Good Job ran with job id:"; echo $SLURM_JOB_ID\n'
-        text = template.substitute(sub) + ''.join(command)
-        try:
-            script = open('good.slr', 'w')
-            script.write(text)
-            script.close()
-        except IOError as error:
-            print('Could not write good.slr!')
-            print('I/O error: {0}'.format(error))
-
-        # write bad script
-        sub['name'] = 'bad'
-        command = 'echo "Bad job should not run!"\n'
-        text = template.substitute(sub) + ''.join(command)
-        text = text.replace("#SBATCH", "#SBATCH BAD_DIRECTIVE", 1)
-        try:
-            script = open('bad.slr', 'w')
-            script.write(text)
-            script.close()
-        except IOError as error:
-            print('Could not write bad.slr!')
-            print('I/O error: {0}'.format(error))
-
-    @classmethod
-    def tearDownClass(cls):
-        """Delete temporary scripts created for tests."""
-        try:
-            os.remove('good.slr')
-        except IOError as error:
-            print('Could not remove good.slr!')
-            print('I/O error: {0}'.format(error))
-        try:
-            os.remove('bad.slr')
-        except IOError as error:
-            print('Could not remove bad.slr!')
-            print('I/O error: {0}'.format(error))
-
-    def test_submit_bad_job(self):
-        """Submit a job."""
-        job_info = submit_job('bad.slr')
-        self.assertEqual(job_info[0], -1)
-        self.assertIn('error', job_info[1])
-
-    def test_submit_good_job(self):
-        """Submit a job."""
-        job_info = submit_job('good.slr')
-        self.assertNotEqual(job_info[0], 1)
-        self.assertEqual('PENDING', job_info[1])
-
-    def test_submit_bad_task(self):
-        """Build and submit a bad task using  a bad directive."""
-        task = Task('bad', command=['#SBATCH BAD_DIRECTIVE'], hints={},
-                    subworkflow=None, inputs=None, outputs=None)
-        job_info = self.worker.submit_task(task)
-        self.assertEqual(job_info[0], -1)
-        self.assertIn('error', job_info[1])
-
-    def test_submit_good_task(self):
-        """Build and submit a good task, state should be 'PENDING' or 'RUNNING'."""
-        task = Task('good',
-                    command=['echo', ' "Good task ran with job id:"',
-                             ';', 'echo', ' $SLURM_JOB_ID'],
-                    hints={}, subworkflow=None, inputs=None, outputs=None)
-        job_info = self.worker.submit_task(task)
-        self.assertNotEqual(job_info[0], 1)
-        self.assertTrue(job_info[1] == 'PENDING' or job_info[1] == 'RUNNING')
-
-    def test_query_bad_job_id(self):
-        """Query a non-existent job."""
-        job_info = self.worker.query_job(888)
-        self.assertEqual(job_info[0], -1)
-        self.assertEqual('Invalid job id specified', job_info[1])
-
-    def test_query_good_job(self):
-        """Submit a job and query the state, should be 'PENDING' or 'RUNNING'."""
-        job_info = submit_job('good.slr')
-        job_id = job_info[0]
-        job_info = self.worker.query_job(job_id)
-        self.assertEqual(job_info[0], True)
-        self.assertTrue(job_info[1] == 'PENDING' or job_info[1] == 'RUNNING')
-
-    def test_cancel_good_job(self):
-        """Submit a job and cancel it."""
-        job_info = submit_job('good.slr')
-        job_id = job_info[0]
-        job_info = self.worker.cancel_job(job_id)
-        self.assertEqual(job_info[0], True)
-        self.assertTrue(job_info[1] == 'CANCELLED' or job_info[1] == 'CANCELLING')
-
-    def test_cancel_bad_job_id(self):
-        """Cancel a non-existent job."""
-        job_info = self.worker.cancel_job(888)
-        self.assertEqual(job_info[0], -1)
-        self.assertEqual('Invalid job id specified', job_info[1])
+from beeflow.common.wf_data import Task
+
+
+# Timeout (seconds) for waiting on tasks
+TIMEOUT = 150
+# Extra slurmrestd arguments. This may be something to take on the command line
+SLURMRESTD_ARGS = bc.get('slurmrestd', 'slurm_args')
+GOOD_TASK = Task(name='good-task', base_command=['sleep', '3'], hints=[],
+                 requirements=[], inputs=[], outputs=[], stdout='',
+                 workflow_id=uuid.uuid4().hex)
+BAD_TASK = Task(name='bad-task', base_command=['/this/is/not/a/command'], hints=[],
+                requirements=[], inputs=[], outputs=[], stdout='',
+                workflow_id=uuid.uuid4().hex)
+
+
+def wait_state(worker_iface, job_id, state):
+    """Wait for Slurm to switch the job to another state."""
+    time.sleep(1)
+    n = 1
+    last_state = worker_iface.query_task(job_id)
+    while last_state == state:
+        if n >= TIMEOUT:
+            raise RuntimeError(f'job timed out, still in state {state}')
+        time.sleep(1)
+        n += 1
+        last_state = worker_iface.query_task(job_id)
+    return last_state
+
+
+def setup_slurm_worker(fn):
+    """Add a decorator to set up the worker interface and slurmrestd."""
+
+    def decorator():
+        """Decorate the input function."""
+        slurm_socket = f'/tmp/{uuid.uuid4().hex}.sock'
+        bee_workdir = os.path.expanduser(f'~/{uuid.uuid4().hex}.tmp')
+        os.mkdir(bee_workdir)
+        proc = subprocess.Popen(f'slurmrestd {SLURMRESTD_ARGS} unix:{slurm_socket}', shell=True)
+        time.sleep(1)
+        worker_iface = WorkerInterface(worker=SlurmWorker, container_runtime='Charliecloud',
+                                       slurm_socket=slurm_socket, bee_workdir=bee_workdir,
+                                       job_template=bc.get('task_manager', 'job_template'))
+        fn(worker_iface)
+        time.sleep(1)
+        proc.kill()
+
+    return decorator
+
+
+def setup_worker_iface(fn):
+    """Add a decorator that creates the worker interface but not slurmrestd."""
+
+    def decorator():
+        """Decorate the input function."""
+        slurm_socket = f'/tmp/{uuid.uuid4().hex}.sock'
+        bee_workdir = os.path.expanduser(f'~/{uuid.uuid4().hex}.tmp')
+        os.mkdir(bee_workdir)
+        worker_iface = WorkerInterface(worker=SlurmWorker, container_runtime='Charliecloud',
+                                       slurm_socket=slurm_socket, bee_workdir=bee_workdir,
+                                       job_template=bc.get('task_manager', 'job_template'))
+        fn(worker_iface)
+
+    return decorator
+
+
+@setup_slurm_worker
+def test_good_task(worker_iface):
+    """Test submission of a good task."""
+    job_id, last_state = worker_iface.submit_task(GOOD_TASK)
+    assert last_state == 'PENDING'
+    last_state = wait_state(worker_iface, job_id, 'PENDING')
+    if last_state == 'RUNNING':
+        last_state = wait_state(worker_iface, job_id, 'RUNNING')
+    if last_state == 'COMPLETING':
+        last_state = wait_state(worker_iface, job_id, 'COMPLETING')
+    assert last_state == 'COMPLETED'
+
+
+@setup_slurm_worker
+def test_bad_task(worker_iface):
+    """Test submission of a bad task."""
+    job_id, last_state = worker_iface.submit_task(BAD_TASK)
+    assert last_state == 'PENDING'
+    last_state = wait_state(worker_iface, job_id, 'PENDING')
+    if last_state == 'RUNNING':
+        last_state = wait_state(worker_iface, job_id, 'RUNNING')
+    if last_state == 'COMPLETING':
+        last_state = wait_state(worker_iface, job_id, 'COMPLETING')
+    assert last_state == 'FAILED'
+
+
+@setup_slurm_worker
+def test_query_bad_job_id(worker_iface):
+    """Test querying a bad job ID."""
+    with pytest.raises(WorkerError):
+        worker_iface.query_task(888)
+
+
+@setup_slurm_worker
+def test_cancel_good_job(worker_iface):
+    """Cancel a good job."""
+    job_id, _ = worker_iface.submit_task(GOOD_TASK)
+    job_state = worker_iface.cancel_task(job_id)
+    assert job_state == 'CANCELLED'
+
+
+@setup_slurm_worker
+def test_cancel_bad_job_id(worker_iface):
+    """Cancel a non-existent job."""
+    with pytest.raises(WorkerError):
+        worker_iface.cancel_task(888)
+
+
+@setup_worker_iface
+def test_no_slurmrestd(worker_iface):
+    """Test without running slurmrestd."""
+    job_id, state = worker_iface.submit_task(GOOD_TASK)
+    assert state == 'NOT_RESPONDING'
+    assert worker_iface.query_task(job_id) == 'NOT_RESPONDING'
+    assert worker_iface.cancel_task(job_id) == 'NOT_RESPONDING'
+# Ignoring R1732: This is not what we need to do with the Popen of slurmrestd above;
+#                 using a with statement doesn't kill the process immediately but just
+#                 waits for it to complete and slurmrestd never will unless we kill it.
+# pylama:ignore=R1732
diff --git a/src/beeflow/tests/test_wf_interface.py b/src/beeflow/tests/test_wf_interface.py
index 937d79a48..7efb4546f 100644
--- a/src/beeflow/tests/test_wf_interface.py
+++ b/src/beeflow/tests/test_wf_interface.py
@@ -6,6 +6,7 @@
 
 import unittest
 
+from beeflow.common.config_driver import BeeConfig as bc
 from beeflow.common.wf_data import (Requirement, Hint, InputParameter, OutputParameter,
                                     StepInput, StepOutput)
 from beeflow.common.wf_interface import WorkflowInterface
@@ -17,7 +18,11 @@ class TestWorkflowInterface(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         """Initialize the Workflow interface."""
-        cls.wfi = WorkflowInterface()
+        bc.init()
+        cls.wfi = WorkflowInterface(user="neo4j",
+                                    bolt_port=bc.get("graphdb", "bolt_port"),
+                                    db_hostname=bc.get("graphdb", "hostname"),
+                                    password=bc.get("graphdb", "dbpass"))
 
     def tearDown(self):
         """Clear all data in the Neo4j database."""
@@ -26,10 +31,10 @@ def tearDown(self):
 
     def test_reconnect(self):
         """Test workflow reconnection."""
-        workflow = self.wfi.initialize_workflow("test_workflow",
-                                                [InputParameter("test_input", "File", "input.txt")],
-                                                [OutputParameter("test_output", "File",
-                                                                 "output.txt", "viz/output")])
+        workflow = self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
 
         # Destroy connection object
         self.wfi._gdb_interface._connection = None
@@ -57,62 +62,52 @@ def test_initialize_workflow(self):
 
         gdb_workflow, _ = self.wfi.get_workflow()
 
-        self.assertEqual(gdb_workflow, workflow)
-        self.assertEqual(gdb_workflow.id, workflow.id)
+        self.assertEqual(workflow, gdb_workflow)
+        self.assertEqual(workflow.id, gdb_workflow.id)
         self.assertIsNotNone(self.wfi._workflow_id)
+        self.assertEqual("SUBMITTED", self.wfi.get_workflow_state())
 
     def test_execute_workflow(self):
         """Test workflow execution initialization (set initial tasks' states to 'READY')."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
         tasks = self._create_test_tasks()
         self.wfi.execute_workflow()
 
-        self.assertEqual(self.wfi.get_task_state(tasks[0]), "READY")
+        self.assertEqual("READY", self.wfi.get_task_state(tasks[0]))
+        self.assertEqual("RUNNING", self.wfi.get_workflow_state())
 
     def test_pause_workflow(self):
         """Test workflow execution pausing (set running tasks' states to 'PAUSED')."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
-        tasks = self._create_test_tasks()
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
+        self._create_test_tasks()
 
-        # Set Compute tasks to RUNNING
-        for task in tasks[1:4]:
-            self.wfi.set_task_state(task, "RUNNING")
-        self.wfi.pause_workflow()
+        self.wfi.execute_workflow()
 
-        # Compute tasks should now be PAUSED
-        for task in tasks[1:4]:
-            self.assertEqual("PAUSED", self.wfi.get_task_state(task))
+        self.wfi.pause_workflow()
 
-        # No other tasks should be affected
-        self.assertEqual("WAITING", self.wfi.get_task_state(tasks[0]))
-        self.assertEqual("WAITING", self.wfi.get_task_state(tasks[4]))
+        # Workflow state should now be 'PAUSED'
+        self.assertEqual("PAUSED", self.wfi.get_workflow_state())
 
     def test_resume_workflow(self):
         """Test workflow execution resuming (set paused tasks' states to 'RUNNING')."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
-        tasks = self._create_test_tasks()
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
+        self._create_test_tasks()
 
-        # Set Compute tasks to PAUSED
-        for task in tasks[1:4]:
-            self.wfi.set_task_state(task, "PAUSED")
+        self.wfi.execute_workflow()
+        self.wfi.pause_workflow()
         self.wfi.resume_workflow()
 
-        # Compute tasks should now be PAUSED
-        for task in tasks[1:4]:
-            self.assertEqual("RUNNING", self.wfi.get_task_state(task))
-
-        # No other tasks should be affected
-        self.assertEqual("WAITING", self.wfi.get_task_state(tasks[0]))
-        self.assertEqual("WAITING", self.wfi.get_task_state(tasks[4]))
+        # Workflow state should now be 'RESUME'
+        self.assertEqual("RESUME", self.wfi.get_workflow_state())
 
     def test_reset_workflow(self):
         """Test workflow execution resetting (set all tasks to 'WAITING', delete metadata)."""
@@ -134,7 +129,7 @@ def test_reset_workflow(self):
 
         # States should be reset, metadata should be deleted
         for task in tasks:
-            self.assertDictEqual(dict(), self.wfi.get_task_metadata(task))
+            self.assertDictEqual({}, self.wfi.get_task_metadata(task))
             self.assertEqual("WAITING", self.wfi.get_task_state(task))
 
         # Workflow ID should be reset
@@ -147,10 +142,10 @@ def test_reset_workflow(self):
 
     def test_finalize_workflow(self):
         """Test workflow deletion from the graph database."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
         self.wfi.finalize_workflow()
 
         self.assertFalse(self.wfi.workflow_loaded())
@@ -168,10 +163,10 @@ def test_add_task(self):
                  Hint("NetworkAccess", {"networkAccess": True})]
         stdout = "output.txt"
 
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "test_input/test_task_done")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "test_input/test_task_done")])
         task = self.wfi.add_task(
             name=task_name,
             base_command=base_command,
@@ -196,7 +191,7 @@ def test_add_task(self):
         gdb_task = self.wfi.get_task_by_id(task.id)
         self.assertEqual(task, gdb_task)
         self.assertEqual(task.id, gdb_task.id)
-    
+
     def test_restart_task(self):
         """Test restart of failed task."""
         task_name = "test_task"
@@ -216,10 +211,10 @@ def test_restart_task(self):
         stdout = "output.txt"
         test_checkpoint_file = "backup0.crx"
 
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "test_input/test_task_done")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "test_input/test_task_done")])
         task = self.wfi.add_task(
             name=task_name,
             base_command=base_command,
@@ -228,7 +223,7 @@ def test_restart_task(self):
             requirements=requirements,
             hints=hints,
             stdout=stdout)
-        
+
         # Restart the task, should create a new Task
         new_task = self.wfi.restart_task(task, test_checkpoint_file)
 
@@ -245,8 +240,7 @@ def test_restart_task(self):
 
         # Check that task command includes checkpoint file
         self.assertListEqual(['ls', '-a', '-F', '-l', 'input.txt', '-R',
-                              'checkpoint_output/backup0.crx'],
-                              new_task.command)
+                              'checkpoint_output/backup0.crx'], new_task.command)
 
         # Restart once again
         newer_task = self.wfi.restart_task(new_task, test_checkpoint_file)
@@ -267,15 +261,14 @@ def test_restart_task(self):
 
         # Check that task command includes checkpoint file
         self.assertListEqual(['ls', '-a', '-F', '-l', 'input.txt', '-R',
-                              'checkpoint_output/backup0.crx'],
-                              newer_task.command)
+                              'checkpoint_output/backup0.crx'], newer_task.command)
 
     def test_finalize_task(self):
         """Test finalization of completed tasks."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
         tasks = self._create_test_tasks()
         self.wfi.execute_workflow()
         self.wfi.set_task_output(tasks[0], "prep/prep_output", "prep_output.txt")
@@ -304,10 +297,10 @@ def test_get_task_by_id(self):
                  Hint("NetworkAccess", {"networkAccess": True})]
         stdout = "output.txt"
 
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "test_task/test_task_done")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "test_task/test_task_done")])
         task = self.wfi.add_task(
             name=task_name,
             base_command=base_command,
@@ -323,10 +316,11 @@ def test_get_workflow(self):
         """Test obtaining the workflow from the graph database."""
         requirements = [Requirement("ResourceRequirement", {"ramMin": 1024})]
         hints = [Hint("NetworkAccess", {"networkAccess": True})]
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")], requirements, hints)
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")],
+            requirements, hints)
         tasks = self._create_test_tasks()
 
         (workflow, wf_tasks) = self.wfi.get_workflow()
@@ -340,17 +334,46 @@ def test_get_workflow_outputs(self):
         workflow = self.wfi.initialize_workflow(
             "test_workflow",
             [InputParameter("test_input", "File", "input.txt")],
-            [OutputParameter("test_output", "File", "output.txt",
-                             "test_task/test_task_done")])
+            [OutputParameter("test_output", "File", "output.txt", "test_task/test_task_done")])
 
         self.assertListEqual(workflow.outputs, self.wfi.get_workflow_outputs())
 
+    def test_get_workflow_state(self):
+        """Test obtaining the state of a workflow."""
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "test_task/test_task_done")])
+
+        # Initialized workflow state should be 'SUBMITTED'
+        self.assertEqual("SUBMITTED", self.wfi.get_workflow_state())
+
+        self.wfi.execute_workflow()
+
+        # Executed workflow state should be 'RUNNING'
+        self.assertEqual("RUNNING", self.wfi.get_workflow_state())
+
+    def test_set_workflow_state(self):
+        """Test setting the state of a workflow."""
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "test_task/test_task_done")])
+
+        # Initialized workflow state should be 'SUBMITTED'
+        self.assertEqual("SUBMITTED", self.wfi.get_workflow_state())
+
+        self.wfi.set_workflow_state("RUNNING")
+
+        # Workflow state should now be 'RUNNING'
+        self.assertEqual("RUNNING", self.wfi.get_workflow_state())
+
     def test_get_ready_tasks(self):
         """Test obtaining of ready workflow tasks."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
         tasks = self._create_test_tasks()
 
         # Should be no ready tasks
@@ -364,10 +387,10 @@ def test_get_ready_tasks(self):
 
     def test_get_dependent_tasks(self):
         """Test obtaining of dependent tasks."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
         tasks = self._create_test_tasks()
         # Get dependent tasks of Data Prep
         dependent_tasks = self.wfi.get_dependent_tasks(tasks[0])
@@ -443,7 +466,7 @@ def test_set_task_metadata(self):
                     "container_md5": "67df538c1b6893f4276d10b2af34ccfe", "job_id": 1337}
 
         # Metadata should be empty
-        self.assertDictEqual(dict(), self.wfi.get_task_metadata(task))
+        self.assertDictEqual({}, self.wfi.get_task_metadata(task))
 
         self.wfi.set_task_metadata(task, metadata)
 
@@ -451,6 +474,7 @@ def test_set_task_metadata(self):
         self.assertDictEqual(metadata, self.wfi.get_task_metadata(task))
 
     def test_get_task_input(self):
+        """Test the obtaining of a task input."""
         self.wfi.initialize_workflow(
             "test_workflow",
             [InputParameter("test_input", "File", "input.txt")],
@@ -465,6 +489,7 @@ def test_get_task_input(self):
         self.assertEqual(task.inputs[0], self.wfi.get_task_input(task, "test_input"))
 
     def test_set_task_input(self):
+        """Test the setting of a task input."""
         self.wfi.initialize_workflow(
             "test_workflow",
             [InputParameter("test_input", "File", "input.txt")],
@@ -481,6 +506,7 @@ def test_set_task_input(self):
         self.assertEqual(test_input, self.wfi.get_task_input(task, "test_input"))
 
     def test_get_task_output(self):
+        """Test the obtaining of a task output."""
         self.wfi.initialize_workflow(
             "test_workflow",
             [InputParameter("test_input", "File", "input.txt")],
@@ -495,6 +521,7 @@ def test_get_task_output(self):
         self.assertEqual(task.outputs[0], self.wfi.get_task_output(task, "test_task/output"))
 
     def test_set_task_output(self):
+        """Test the setting of a task output."""
         self.wfi.initialize_workflow(
             "test_workflow",
             [InputParameter("test_input", "File", "input.txt")],
@@ -511,6 +538,7 @@ def test_set_task_output(self):
         self.assertEqual(test_output, self.wfi.get_task_output(task, "test_task/output"))
 
     def test_evaluate_expression(self):
+        """Test the evaluation of an input/output expression."""
         self.wfi.initialize_workflow(
             "test_workflow",
             [InputParameter("test_input", "File", "input.txt")],
@@ -556,29 +584,29 @@ def test_workflow_completed(self):
 
     def test_workflow_initialized(self):
         """Test determining if a workflow is initialized."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
 
         # Workflow now initialized
         self.assertTrue(self.wfi.workflow_initialized())
 
     def test_workflow_loaded(self):
         """Test determining if a workflow is loaded."""
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
         self.wfi.finalize_workflow()
 
         # Workflow not loaded
         self.assertFalse(self.wfi.workflow_loaded())
 
-        self.wfi.initialize_workflow("test_workflow",
-                                     [InputParameter("test_input", "File", "input.txt")],
-                                     [OutputParameter("test_output", "File", "output.txt",
-                                                      "viz/output")])
+        self.wfi.initialize_workflow(
+            "test_workflow",
+            [InputParameter("test_input", "File", "input.txt")],
+            [OutputParameter("test_output", "File", "output.txt", "viz/output")])
 
         # Workflow now loaded
         self.assertTrue(self.wfi.workflow_loaded())