Repository: ExaWorks/psij-python (language: python)

Issue:
Saga connector
A prototype SAGA connector. There are a few issues in the code, marked by `# TODO:`. We should consider fixing those before merging.


Discussion / hints:
@hategan : I pushed a small iteration to the saga connector.  It seems to work now, in the sense that jobs get started and collected.  I did not work on the various comments from above.

One problem I stumbled over: I don't see the class destructor being called on termination.  The saga session needs an explicit `close()` for clean shutdown (and otherwise termination will hang on a thread lock).  Any insight on why the d'tor is not called, and what to do about it?
> One problem I stumbled over: I don't see the class destructor being called on termination.

There is a shutdown handler (using atexit) called at termination which calls close() on the job service. Is that not working?
> > One problem I stumbled over: I don't see the class destructor being called on termination.
> 
> There is a shutdown handler (using atexit) called at termination which calls close() on the job service. Is that not working?

Nope, it does not - the interpreter does not seem to go into exit on it's own.
> Nope, it does not - the interpreter does not seem to go into exit on it's own.

This would imply that the atexit module is broken. While possible, it seems unlikely. So maybe let's try to troubleshoot this and see what's happening.
> > Nope, it does not - the interpreter does not seem to go into exit on it's own.
> 
> This would imply that the atexit module is broken. While possible, it seems unlikely. So maybe let's try to troubleshoot this and see what's happening.

We clarified on a call that the issue is not that the atexit handler fails to be called, but that the process does not exit at all.

In this instance, there are two threads left alive after the main thread exists:
1. a shell_job._job_state_monitor
2. a threading.Timer started by ShellJobService.initialize

The former is a daemon thread. The timer is not.

The quickest solution would probably be to add, in ShellJobService.initialize:
```diff
self._ping = mt.Timer(_PING_DELAY, self._ping_cb)
+ self._ping.daemon = True
self._ping.start()
```

Then base the adaptor on an updated version of SAGA.

A hackish alternative would be to add, in the saga adapter:
```python
def _patched_initialize(self):
    self._old_initialize()
    self._ping.cancel()
    self._ping = mt.Timer(shell_job._PING_DELAY, self._ping_cb)
    self._ping.daemon = True
    self._ping.start()

ShellJobService._old_initialize = ShellJobService.initialize
ShellJobService.initialize = _patched_initialize
```

This comment is not meant to rule out the possibility of adding a `JobExecutor.close()` method in the spec, but to explore possible solutions.
The termination problem has been fixed with the SAGA patch suggested by @hategan.
@hategan , @SteVwonder : the SAGA executor should now be complete and ready for another review.  I expanded the tests to also cover the local and ssh backend for the SAGA executor (additional to the local executor).  The sshd installation in the VM is a bit clunky though, let me know if you have a better idea on how to handle that.

The ReadTheDocs build fails because of missing flux import.  Did we already find a way to handle that?  
@hategan : please press the merge button, if you don't mind...
> @hategan : please press the merge button, if you don't mind...

A conflict popped up in the tests. I'll need to fix that.

Fix patch:
diff --git a/Makefile b/Makefile
index eb2418d5..c5e654db 100644
--- a/Makefile
+++ b/Makefile
@@ -10,14 +10,18 @@ CWD = $(shell pwd)
 PYTHON = $(shell if python --version 2>&1 | egrep -q 'Python 3\..*' ; then echo "python"; else echo "python3"; fi)
 
 .PHONY: tests
-tests:
+tests: launcher-scripts
 	PYTHONPATH=$(CWD)/src:${PYTHONPATH} ${PYTHON} -m pytest -v $(TESTARGS)
 
 .PHONY: verbose-tests
-verbose-tests:
+verbose-tests: launcher-scripts
 	PYTHONPATH=$(CWD)/src:${PYTHONPATH} ${PYTHON} -m pytest -v --log-format="%(asctime)s %(levelname)s %(message)s" \
 		           --log-date-format="%Y-%m-%d %H:%M:%S" --log-cli-level=DEBUG
 
+.PHONY: ci-tests
+ci-tests:
+	${PYTHON} tests/ci_runner.py
+
 
 .PHONY: typecheck
 typecheck:
diff --git a/psij-ci-run b/psij-ci-run
new file mode 100755
index 00000000..d16bf041
--- /dev/null
+++ b/psij-ci-run
@@ -0,0 +1,53 @@
+#!/bin/bash
+
+usage() {
+    cat <<EOF
+Usage: ./psij-ci-run [--help] [--with-conda <conda_env_name>] 
+            [--with-venv <venv_name>]
+    --help          Displays this message
+    --with-conda    Specifies a conda environment to activate before running 
+                    the tests
+    --with-venv     Specifies a virtualenv environment to load before running
+                    the tests
+EOF
+}
+
+failifempty() {
+    if [ "$2" == "" ]; then
+        echo "Missing parameter for $1"
+        usage
+        exit 1
+    fi
+}
+
+while [ "$1" != "" ]; do
+    case "$1" in
+        --with-conda)
+            failifempty "$1" "$2"
+            conda activate "$2"
+            shift 2
+            ;;
+        --with-venv)
+            failifempty "$1" "$2"
+            source "$2/bin/activate"
+            shift 2
+            ;;
+        --help)
+            usage
+            exit 0
+            ;;
+        *)
+            echo "Unrecognized command line option: $1."
+            usage
+            exit 1
+    esac
+done
+
+
+if python --version 2>&1 | egrep -q 'Python 3\..*' >/dev/null 2>&1 ; then
+    PYTHON="python"
+else
+    PYTHON="python3"
+fi
+
+$PYTHON tests/ci_runner.py
diff --git a/psij-ci-setup b/psij-ci-setup
new file mode 100755
index 00000000..b6bdd39f
--- /dev/null
+++ b/psij-ci-setup
@@ -0,0 +1,96 @@
+#!/bin/bash
+
+if pip --version 2>&1 | egrep -q 'python 3\..*' >/dev/null 2>&1 ; then
+    PIP="pip"
+else
+    PIP="pip3"
+fi
+
+
+if [ "$VIRTUAL_ENV" != "" ]; then
+    VENV=1
+    NOUSER=1
+fi
+
+if [ "$CONDA_SHLVL" != "" ] && [ "$CONDA_SHLVL" != "0" ]; then
+    CONDA=1
+    NOUSER=1
+fi
+
+
+FORCE=0
+
+if [ "$1" == "-f" ]; then
+    FORCE=1
+    shift
+fi
+
+MYPATH="$( cd -- "$(dirname "$0")" >/dev/null 2>&1 ; pwd -P )"
+
+echo
+echo "================================================================"
+echo "This script will install requirements for the PSI/J CI tests and"
+echo "add a cron job to run the tests once a day at some random time. "
+echo "================================================================"
+echo
+
+RESPONSE=""
+
+while [ "$RESPONSE" != "C" ] && [ "$RESPONSE" != "X" ]; do
+
+    echo -n "Would you like to (C)ontinue or E(x)it? "
+    read -n1 RESPONSE
+    echo
+    RESPONSE=${RESPONSE^}
+    
+    if [ "$RESPONSE" == "X" ]; then
+        echo "Operation canceled"
+        exit 1
+    fi
+done
+
+cd "$MYPATH"
+
+
+if [ "$NOUSER" == "1" ]; then
+    $PIP install -r requirements-dev.txt
+else
+    $PIP install --user -r requirements-dev.txt
+fi
+
+
+HOUR=`echo $(($RANDOM % 24))`
+MINUTE=`echo $(($RANDOM % 60))`
+MYPATH="$( cd -- "$(dirname "$0")" >/dev/null 2>&1 ; pwd -P )"
+
+if crontab -l 2>/dev/null | grep "psij-ci-run" >/dev/null && [ "$FORCE" != "1" ]; then
+    EXISTING=`crontab -l 2>/dev/null | grep "psij-ci-run"`
+    echo
+    echo "================================================================"
+    echo "Error: a crontab for PSI/J tests already exists: "
+    echo ">>> $EXISTING"
+    echo
+    echo "You can edit your crontab with \"crontab -e\" and remove the    "
+    echo "existing entry, then re-run this tool. If you are certain that  "
+    echo "you want to install multiple entries, you can re-run this script"
+    echo "with the \"-f\" flag.                                           "
+    echo "================================================================"
+    exit 2
+else
+    LINE="$MINUTE $HOUR * * * cd \"$MYPATH\" && ./psij-ci-run"
+    if [ "$VENV" == "1" ]; then
+        LINE="$LINE --with-venv \"$VIRTUAL_ENV\""
+    fi
+    if [ "$CONDA" == "1" ]; then
+        LINE="$LINE --with-conda \"$CONDA_DEFAULT_ENV\""
+    fi
+    LINE="$LINE >> $MYPATH/testing.log 2>&1"
+    echo
+    echo "================================================================"
+    echo "The following line will be installed in your crontab:"
+    echo "$LINE"
+    { crontab -l & echo "$LINE"; } | crontab -
+    echo "Setup complete. If you have not already done so, please take    "
+    echo "some time to customize testing.conf.                            "
+    echo "================================================================"
+fi
diff --git a/requirements-connector-saga.txt b/requirements-connector-saga.txt
index 15be0f22..dbf595fa 100644
--- a/requirements-connector-saga.txt
+++ b/requirements-connector-saga.txt
@@ -1,3 +1,3 @@
 # For SAGA executor
-radical.saga
+radical.saga >= 1.8.0
 radical.utils
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 7097bca7..fd0ae378 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -1,3 +1,7 @@
+filelock
+psutil
+pystache==0.5.4
+
 six
 Sphinx
 sphinx_rtd_theme
@@ -8,3 +12,4 @@ mypy >=0.790
 pytest
 flake8
 autopep8
+types-requests
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index b82d809a..a884151d 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,3 @@
 filelock
 psutil
-pystache
\ No newline at end of file
+pystache>=0.5.4


Test patch:
diff --git a/README-testing.md b/README-testing.md
new file mode 100644
index 00000000..5489784e
--- /dev/null
+++ b/README-testing.md
@@ -0,0 +1,103 @@
+Introduction
+============
+
+PSI/J comes with a testing infrastructure that allows users to contribute
+tests of the PSI/J library on resources they have access to and where
+running such tests is appropriate.
+
+Contributed tests take the form of normal pytest runs for which results
+are uploaded to a test aggregation service. The compatibility of various
+PSI/J branches with various resources can then be assessed and is
+available at https://testing.exaworks.org
+
+How to run tests
+================
+
+There are a number of ways to run tests. Invoking `pytest` directly,
+running the integration tests and through `cron` (or a similar tool).
+
+Setting up a Cron testing job
+=============================
+
+This is the preferred way of running the tests since it allows the PSI/J
+team to keep a constant eye on the state of the library on various
+resources. To set up the Cron job, you can either use the provided
+script:
+
+```bash
+    ./psij-ci-setup
+```
+
+or manually set up the CI runner with Cron or your favorite scheduler.
+
+
+Testing with the CI runner
+==========================
+
+The CI runner is a convenience script which can clone the PSI/J
+repository and run tests on several git branches. To use the CI runner:
+
+1. Install dependencies:
+```bash
+    pip install --user -r requirements-dev.txt
+```
+
+2. Edit `testing.conf` and customize as needed.
+
+3. Run tests:
+
+```bash
+    ./psij-ci-run
+```
+
+or, alternatively, using  built-in `make` target:
+
+```bash
+    make ci-tests
+```
+
+
+Testing with pytest
+===================
+
+This is the most direct way to run the tests and it must be done from the
+main psi-j directory. Use the following steps:
+
+1. Install dependencies:
+```bash
+    pip install --user -r requirements-dev.txt
+```
+
+2. Run tests:
+
+```bash
+    PYTHONPATH=$PWD/src pytest tests
+```
+
+or, alternatively, using the built-in `make` target:
+
+```bash
+    make tests
+```
+
+A number of custom options are available for tests. They are described in
+detail in the `testing.conf` file. In order to use these options with
+`pytest`, they can be passed as double-dashed command line arguments with
+underscores converted to hyphens. For example, to run tests and
+automatically upload results to the default aggregation server, you can
+run:
+
+```bash
+    PYTHONPATH=$PWD/src pytest --upload-results tests
+```
+
+or, using `make`, prefix all options with a double dash ("--"):
+
+```bash
+    make tests -- --upload-results
+```
+
+Care must, however, be taken since it is impossible to preserve
+whitespace and certain special characters (such as the double quotes)
+when passing arguments through `make`.
+
diff --git a/testing.conf b/testing.conf
new file mode 100644
index 00000000..c9aaf0a0
--- /dev/null
+++ b/testing.conf
@@ -0,0 +1,139 @@
+#
+# An identifier for this site which will appear in the aggregated results which
+# are public. The options are as follows:
+#
+# hostname:
+#     The host name of the machine on which the tests are invoked. This is 
+#     useful in identifying the machine such that the community is aware of the
+#     status of PSI/J on it.
+#
+# random:
+#     A random string that is persisted on disk (in ~/.psij).
+#
+# "<string>":
+#     A user-specified string. Must be surrounded by double quotes.
+#
+# id = hostname | random | "<string>"
+
+id = hostname
+
+
+#
+# An email used to contact the test runner in the event that troubleshooting
+# from logs is unfeasable. This can be left empty if you do not wish to be
+# contacted with troubleshooting tasks. This email is not publicly visible
+# on the aggregation site.
+#
+# maintainer_email = <string>
+
+maintainer_email = 
+
+
+# 
+# The scope selects which branches get tested. This selection can potentially 
+# impact security. The options are:
+# 
+# stable: 
+#     Only test stable branches (currently the main branch). The stable 
+#     branches are branches in the core repository which are considered stable 
+#     and have been extensively reviewed and tested on other machines. They 
+#     carry the lowest risks.
+#
+# core:
+#     Test all core branches. This tests all stable branches, as above, and 
+#     all branches that are in the core namespace and are part of pull 
+#     requests. Since only core developers can create branches in the core
+#     namespace, this limits the tested code to code written by core 
+#     developers.
+# tagged:
+#     Test all core branches and all branches that are part of pull requests
+#     that have been explicitly tagged by core developers for testing. The 
+#     branches can exist in private namespaces, so, technically, anybody can
+#     submit such pull requests. However, a core developer needs to explicitly
+#     approve testing on such pull requests. This option carries the most
+#     risks.
+#
+# scope = stable | core | tagged
+
+scope = core
+
+
+#
+# Which executors to test. This is a comma separated list of 
+# <executor_name>[:<launcher_name>[:<url>]] items. The "auto_q" executor means 
+# that the testing infrastructure will attempt to determine which queuing system 
+# is present on the test machine and use the corresponding executor for testing.
+# Similar to "auto_q", "auto_l" means that the testing infrastructure will 
+# determine an appropriate launcher for the system. The launcher name can be
+# empty if a particular executor does not use launchers but uses URLs 
+# (e.g., saga::localhost). The special value "auto" (without quotes) means that
+# the executors to be tested should be determined automatically based on a 
+# survey of the runtime environment.
+#
+# executors = local:single, local:multiple, \
+#            batch-test:single, batch-test:multiple, \
+#            saga::fork://localhost/, \
+#            auto_q:single, auto_q:multiple, auto_q:auto_l
+
+executors = auto
+
+
+
+
+
+################################################################################
+#                                                                              #
+#        The settings below are advanced and unlikely to need updating         #
+#                                                                              #
+################################################################################
+
+#
+# The location of the server that receives and aggregates test results.
+#
+# server_url = <url>
+
+server_url = https://testing.exaworks.org
+
+
+#
+# A key authenticates a test ID such that it reduces the odds that two machines
+# compete to overwrite each other's results. The machines are identified by their
+# IDs. However, without additional steps, any other test deployment can set its
+# ID to an existing one. The "key" setting allows a secret to be used to 
+# authenticate requests. There are two valid values:
+# 
+# random:
+#     Generates a random key on the first upload attempt. The server will then
+#     only accept subsequent requests from the same ID if the key matches. There
+#     is a 7 day expiration of a pairing which allows for a new key to be used 
+#     if an old key has not been used for 7 or more days. Random keys are 
+#     persisted locally in ~/.psij
+#
+# "<string>":
+#     A custom string to be used as a key.
+#
+# key = random | "<string>"
+
+key = random
+
+
+#
+# The maximum age, in hours, for saved test results to be kept on disk. When 
+# tests are run with the "--save-results", test results are saved on disk. 
+# They can be later uploaded to the aggregation server using 
+# "make upload-results". In order to prevent a very large number of test 
+# results from occupying disk space indefinitely, the various tools involved 
+# purge old results from disk. This parameter sets a limit on the age of 
+# results that should be kept on disk.
+#
+# max_age = 48
+
+max_age = 48
+
+
+#
+# The repository where the code is located. This is needed to clone the code
+# and to retrieve the list of PRs and their comments.
+#
+
+repository = ExaWorks/psi-j-python
diff --git a/tests/ci_runner.py b/tests/ci_runner.py
new file mode 100755
index 00000000..427d4e5c
--- /dev/null
+++ b/tests/ci_runner.py
@@ -0,0 +1,158 @@
+#!/usr/bin/python
+import datetime
+import os
+import secrets
+import subprocess
+import sys
+from contextlib import contextmanager
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Dict, TextIO, List, Optional, Generator
+
+import requests
+
+
+STABLE_BRANCHES = ['main']
+FAKE_BRANCHES = ['main', 'feature_1', 'feature_x']
+GITHUB_API_ROOT = 'https://api.github.com'
+
+
+def read_line(f: TextIO) -> Optional[str]:
+    line = f.readline()
+    if line:
+        line = line.strip()
+        if len(line) > 0 and line[-1] == '\\':
+            next_line = read_line(f)
+            if not next_line:
+                raise ValueError('Premature end of file')
+            line = line[:-1] + next_line
+        return line
+    else:
+        return None
+
+
+def read_conf(fname: str) -> Dict[str, str]:
+    conf = {}
+    with open(fname, 'r') as f:
+        line = read_line(f)
+        while line is not None:
+            if len(line) == 0:
+                pass
+            elif line[0] == '#':
+                pass
+            else:
+                kv = line.split('=', 2)
+                if len(kv) != 2:
+                    raise ValueError('Invalid line in configuration file: "%s"' % line)
+                conf[kv[0].strip()] = kv[1].strip()
+            line = read_line(f)
+    return conf
+
+
+def get_conf(conf: Dict[str, str], name: str) -> str:
+    if name not in conf:
+        raise KeyError('Missing configuration option "%s"' % name)
+    return conf[name]
+
+
+def get_core_pr_branches(conf: Dict[str, str]) -> List[str]:
+    repo = get_conf(conf, 'repository')
+    resp = requests.get('%s/repos/%s/pulls?state=open&per_page=100' % (GITHUB_API_ROOT, repo))
+    resp.raise_for_status()
+    branches = []
+    for pr in resp.json():
+        if pr['head']['repo']['full_name'] == repo:
+            branches.append(pr['head']['ref'])
+    return branches
+
+
+def do_clone(conf: Dict[str, str], dir: Path) -> None:
+    repo = get_conf(conf, 'repository')
+    subprocess.run(['git', 'clone', 'https://github.com/%s.git' % repo, 'code'], check=True,
+                   cwd=str(dir))
+
+
+def checkout(branch: str, dir: Path) -> None:
+    subprocess.run(['git', 'checkout', branch], check=True, cwd=str(dir / 'code'))
+
+
+def run_branch_tests(conf: Dict[str, str], dir: Path, run_id: str, clone: bool = True,
+                     site_id: Optional[str] = None, fake_branch_name: Optional[str] = None) -> None:
+    # it would be nice if this could be run through make; however, gnu make cannot preserve
+    # spaces or quotes in arguments
+    args = [sys.executable, '-m', 'pytest', '-v', '--upload-results', '--run-id', run_id]
+    if site_id:
+        args.append('--id')
+        args.append(site_id)
+    else:
+        args.append('--id')
+        args.append(get_conf(conf, 'id'))
+    if fake_branch_name:
+        args.append('--branch-name-override')
+        args.append(fake_branch_name)
+    for opt in ['maintainer_email', 'executors', 'server_url', 'key', 'max_age']:
+        args.append('--' + opt.replace('_', '-'))
+        args.append(get_conf(conf, opt))
+    cwd = (dir / 'code') if clone else Path('.')
+    env = dict(os.environ)
+    env['PYTHONPATH'] = str(cwd.absolute() / 'src') + \
+        (':' + env['PYTHONPATH'] if 'PYTHONPATH' in env else '')
+    subprocess.run([sys.executable, 'setup.py', 'launcher_scripts'], cwd=cwd.absolute(), check=True)
+    subprocess.run(args, cwd=cwd.absolute(), env=env)
+
+
+def get_run_id() -> str:
+    now = datetime.datetime.now()
+    return '%04d%02d%02d%02d-%s' % (now.year, now.month, now.day, now.hour, secrets.token_hex())
+
+
+def run_tests(conf: Dict[str, str], site_ids: List[str], branches: List[str], clone: bool) -> None:
+    run_id = get_run_id()
+    with TemporaryDirectory() as tmp:
+        tmpp = Path(tmp)
+        if clone:
+            with info('Cloning repository'):
+                do_clone(conf, tmpp)
+        for site_id in site_ids:
+            for branch in branches:
+                if clone:
+                    checkout(branch, tmpp)
+                with info('Testing branch "%s"' % branch):
+                    run_branch_tests(conf, tmpp, run_id, clone, site_id, branch)
+
+
+@contextmanager
+def info(msg: str) -> Generator[bool, None, None]:
+    print(msg + '... ', end='')
+    sys.stdout.flush()
+    try:
+        yield True
+        print('OK')
+    except Exception:
+        print('FAILED')
+        raise
+
+
+if __name__ == '__main__':
+    with info('Reading configuration'):
+        conf = read_conf('testing.conf')
+    scope = get_conf(conf, 'scope')
+    clone = True
+    site_ids = [get_conf(conf, 'id')]
+    if scope == 'stable':
+        branches = STABLE_BRANCHES
+    elif scope == 'core':
+        branches = get_core_pr_branches(conf)
+    elif scope == 'fake':
+        site_ids = ['"mira.alcf.anl.gov"', '"bw.ncsa.illinois.edu"', '"frontera.tacc.utexas.edu"']
+        branches = FAKE_BRANCHES
+        clone = False
+    elif scope == 'local':
+        branches = ['main']
+        clone = False
+    elif scope == 'tagged':
+        raise NotImplementedError('Tagged scope not implemented yet.')
+    else:
+        raise ValueError('Unrecognized value for scope: "%s"' % scope)
+
+    run_tests(conf, site_ids, branches, clone)
diff --git a/tests/conftest.py b/tests/conftest.py
index f29f1d6c..df242ab4 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,11 +1,451 @@
 #!/usr/bin/python3
 # type: ignore
+import datetime
+import io
+import json
+import logging
+import os
+import re
+import secrets
+import shutil
+import socket
+import subprocess
+import threading
+from functools import partial
+from pathlib import Path
+from typing import Dict, List
+
+import requests
+from _pytest._io import TerminalWriter
+from ci_runner import get_run_id
+from filelock import FileLock
+
+from executor_test_params import ExecutorTestParams
+
+import pytest
+
+
+logger = logging.getLogger(__name__)
+
+
+LOG_FORMATTER = logging.Formatter(fmt='%(asctime)s %(levelname)s %(message)s\n',
+                                  datefmt='%Y-%m-%d %H:%M:%S')
+SETTINGS_DIR = Path.home() / '.psij'
+KEY_FILE = SETTINGS_DIR / '.key'
+ID_FILE = SETTINGS_DIR / '.id'
+RESULTS_ROOT = Path('tests') / 'results'
+
 
 def pytest_addoption(parser):
-    parser.addoption('--executors', action='append', default=['local', 'batch-test', 'slurm'],
-                     help='A set of executors to run the tests on.')
+    parser.addoption('--executors', action='store', default='auto',
+                     help='A set of executors (or executor:launcher pairs) to run the tests on. '
+                          'The special value "auto" means all executors that can be determined to '
+                          'be accessible.')
+    parser.addoption('--save-results', action='store_true', default=False,
+                     help='If specified, save test results in a way that allows them to be '
+                          'uplaoded to a results server later.')
+    parser.addoption('--upload-results', action='store_true', default=False,
+                     help='If specified, upload results to a test result server specified in'
+                          'the testing.conf file.')
+    parser.addoption('--max-age', action='store', default='48',
+                     help='Maximum age, in hours, of tests to keep saved.')
+    parser.addoption('--id', action='store', default='hostname',
+                     help='An identifier for this site to attach to tests when aggregated.')
+    parser.addoption('--server-url', action='store', default='https://testing.exaworks.org',
+                     help='The base URL of the test aggregation server.')
+    parser.addoption('--key', action='store', default='random',
+                     help='A secret to use when communicating to the aggregation server.')
+    parser.addoption('--maintainer-email', action='store', default=None,
+                     help='An optional email for the test maintainer.')
+    parser.addoption('--run-id', action='store', default=None,
+                     help='Externally supplied run ID.')
+    parser.addoption('--branch-name-override', action='store', default=None,
+                     help='Pretend that the current git branch is this value.')
+
+
+def _get_executors(config: Dict[str, str]) -> List[str]:
+    execs_str = config.getoption('executors')
+    execs = execs_str.split(',')
+    processed = []
+    for exec_str in execs:
+        exec_str = exec_str.strip()
+        comps = re.split(':', exec_str, maxsplit=2)
+        execs_t = _translate_executor(config, comps[0])
+        if len(execs_t) > 0:
+            if len(comps) == 1:
+                # executor(s) only; this allows for "auto" which means all local and auto_q:auto_l
+                processed.extend(execs_t)
+            elif len(comps) in [2, 3]:
+                # executor and launcher
+                assert len(execs_t) == 1
+                executor = execs_t[0]
+                launcher = _translate_launcher(config, executor, comps[1])
+                if len(comps) == 2:
+                    processed.append(executor + ':' + launcher)
+                else:
+                    processed.append(executor + ':' + launcher + ':' + comps[2])
+
+    return processed
+
+
+def _translate_executor(config: Dict[str, str], executor: str) -> List[str]:
+    if executor == 'auto':
+        execs = ['local:single', 'local:multiple', 'batch-test:single', 'batch-test:multiple']
+        if config.option.environment.get('has_saga'):
+            execs.append('saga::fork://localhost/')
+            if config.option.environment.get('can_ssh_to_localhost'):
+                execs.append('saga::ssh://localhost/')
+        queue_execs = _translate_executor(config, 'auto_q')
+        assert len(queue_execs) in [0, 1]
+        queue_exec = None
+        if len(queue_execs) == 1:
+            queue_exec = queue_execs[0]
+            queue_launcher = _translate_launcher(config, queue_exec, 'auto_l')
+        if config.option.environment.get('has_mpirun'):
+            execs.extend(['local:mpirun', 'batch-test:mpirun'])
+        if queue_exec:
+            execs.extend([queue_exec + ':single', queue_exec + ':multiple'])
+            if queue_launcher:
+                execs.append(queue_exec + ':' + queue_launcher)
+            else:
+                execs.append(queue_exec)
+        return execs
+    if executor == 'auto_q':
+        if config.option.environment.get('has_slurm'):
+            return ['slurm']
+        else:
+            # nothing yet
+            return []
+    return [executor]
+
+
+def _translate_launcher(config: Dict[str, str], exec: str, launcher: str) -> str:
+    if launcher == 'auto_l':
+        if exec == 'slurm':
+            return 'srun'
+        else:
+            raise ValueError('Don\'t know how to get launcher for executor "' + exec + '"')
+    else:
+        return launcher
 
 
 def pytest_generate_tests(metafunc):
-    if 'executor' in metafunc.fixturenames:
-        metafunc.parametrize('executor', metafunc.config.getoption('executors'))
+    if 'execparams' in metafunc.fixturenames:
+        metafunc.parametrize('execparams',
+                             [ExecutorTestParams(x) for x in _get_executors(metafunc.config)],
+                             ids=lambda x: str(x))
+
+
+def _set_root_logger(logger):
+    logging.root = logger
+    logging.Logger.root = logger
+
+
+_capture_lock = threading.RLock()
+
+
+def _capture_log(fn):
+    # need to serialize this since there's only one root logger
+    with _capture_lock:
+        orig_logger = logging.getLogger()
+        logger = logging.getLogger('capture')
+        logger.setLevel(logging.DEBUG)
+        logger.propagate = True
+        buffer = io.StringIO()
+        logger.addHandler(logging.StreamHandler(buffer))
+        _set_root_logger(logger)
+        try:
+            result = fn()
+        finally:
+            _set_root_logger(orig_logger)
+            log = buffer.getvalue()
+            buffer.close()
+
+        return result, log
+
+
+def _purge_old_results(config):
+    pass
+
+
+def pytest_configure(config):
+    logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
+                        datefmt='%Y-%m-%d %H:%M:%S', level=logging.DEBUG)
+    logging.root.setLevel(logging.DEBUG)
+
+    _purge_old_results(config)
+    start_time = _now()
+    (env, key), log = _capture_log(partial(_discover_environment, config))
+    save = config.getoption('save_results')
+    upload = config.getoption('upload_results')
+    config.option.key = key
+    end_time = _now()
+    data = {}
+    if save or upload:
+        data.update({
+            'module': '_conftest',
+            'cls': None,
+            'function': '_discover_environment',
+            'test_name': '_discover_environment',
+            'log': log,
+            'results': {},
+            'extras': env,
+            'test_start_time': start_time,
+            'test_end_time': end_time,
+            'run_id': env['run_id'],
+            'branch': env['git_branch']
+        })
+        _save_or_upload(config, data)
+
+
+def _get_config_env(config, name):
+    if hasattr(config.option, 'environment'):
+        return config.option.environment[name]
+    else:
+        return None
+
+
+def pytest_unconfigure(config):
+    data = {
+        'module': '_conftest',
+        'cls': None,
+        'function': '_end',
+        'test_name': '_end',
+        'results': {},
+        'extras': None,
+        'test_start_time': _now(),
+        'test_end_time': _now(),
+        'run_id': _get_config_env(config, 'run_id'),
+        'branch': _get_config_env(config, 'git_branch')
+    }
+    if hasattr(config.option, 'environment'):
+        # only upload if we were able to get a basic environment
+        _save_or_upload(config, data)
+
+
+def _cache(file_path, fn):
+    lock = FileLock(file_path.with_suffix('.lock'))
+    with lock:
+        if not os.path.exists(file_path):
+            value = fn()
+            with open(file_path, 'w') as f:
+                f.write(value)
+        else:
+            with open(file_path, 'r') as f:
+                value = f.read().strip()
+    return value
+
+
+def _get_key(config):
+    key = config.getoption('key')
+    if key == 'random':
+        return _cache(KEY_FILE, secrets.token_hex)
+    elif key[0] == '"' and key[-1] == '"':
+        return key[1:-1]
+    else:
+        raise ValueError('Invalid value for --key argument: "%s"' % key)
+
+
+def _get_id(config):
+    id = config.getoption('id')
+    if id == 'hostname':
+        return socket.gethostname()
+    elif id == 'random':
+        return _cache(ID_FILE, secrets.token_hex)
+    elif id[0] == '"' and id[-1] == '"':
+        return id[1:-1]
+    else:
+        raise ValueError('Invalid value for --id argument: "%s"' % id)
+
+
+def _run(*args) -> str:
+    process = subprocess.run(args, check=True, text=True,
+                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    return process.stdout.strip()
+
+
+def _get_git_branch(config) -> str:
+    conf_branch = config.getoption('branch_name_override')
+    if conf_branch:
+        return conf_branch
+    else:
+        return _run('git', 'rev-parse', '--abbrev-ref', 'HEAD')
+
+
+def _get_last_commit() -> str:
+    return _run('git', 'log', '-n', '1', '--pretty=format:"%H"')
+
+
+def _get_commit_diff():
+    result = _run('git', 'rev-list', '--left-right', '--count', 'origin...HEAD')
+    lr = result.split()
+    assert len(lr) == 2
+    return int(lr[0]), int(lr[1])
+
+
+def _get_git_diff_stat():
+    return _run('git', 'diff', '--stat')
+
+
+def _get_run_id(config):
+    run_id = config.getoption('run_id')
+    if run_id is None:
+        run_id = get_run_id()
+    return run_id
+
+
+def _has_saga():
+    # try to instantiate SAGA executor
+    try:
+        from psij.executors.saga import SagaExecutor
+        SagaExecutor()
+        return True
+    except Exception as ex:
+        logger.info(ex)
+        return False
+
+
+def _try_run_command(args, timeout=None):
+    try:
+        process = subprocess.run(args, timeout=timeout, text=True,
+                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        logger.debug('Output from {}: {} {}'.format(args, process.stdout, process.stderr))
+        return process.returncode == 0
+    except subprocess.TimeoutExpired:
+        return False
+
+
+def _get_env(name: str) -> str:
+    if name in os.environ:
+        return os.environ[name]
+    else:
+        return ''
+
+
+def _discover_environment(config):
+    SETTINGS_DIR.mkdir(parents=True, exist_ok=True)
+    RESULTS_ROOT.mkdir(parents=True, exist_ok=True)
+    env = {}
+    conf = {}
+    env['config'] = conf
+    conf['id'] = _get_id(config)
+    conf['executors'] = config.getoption('executors')
+    conf['maintainer_email'] = config.getoption('maintainer_email')
+    key = _get_key(config)
+    env['start_time'] = _now()
+    env['run_id'] = _get_run_id(config)
+    env['in_conda'] = _get_env('CONDA_SHLVL') != '' and _get_env('CONDA_SHLVL') != '0'
+    env['in_venv'] = _get_env('VIRTUAL_ENV') != ''
+    try:
+        env['has_slurm'] = shutil.which('sbatch') is not None
+        env['has_mpirun'] = shutil.which('mpirun') is not None
+        env['has_saga'] = _has_saga()
+        env['can_ssh_to_localhost'] = _try_run_command(['ssh', 'localhost', 'true'], timeout=5)
+        env['git_branch'] = _get_git_branch(config)
+        env['git_last_commit'] = _get_last_commit()
+        ahead, behind = _get_commit_diff()
+        env['git_ahead_remote_commit_count'] = ahead
+        env['git_behind_remote_commit_count'] = behind
+        env['git_local_change_summary'] = _get_git_diff_stat()
+        env['git_has_local_changes'] = (env['git_local_change_summary'] != '')
+    except Exception as ex:
+        env['error'] = str(ex)
+    config.option.environment = env
+    env['computed_executors'] = _get_executors(config)
+    return env, key
+
+
+def _now():
+    return datetime.datetime.now(tz=datetime.timezone.utc).isoformat(' ')
+
+
+@pytest.hookimpl(tryfirst=True, hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    outcome = yield
+    report = outcome.get_result()
+
+    if hasattr(item, 'result'):
+        d = getattr(item, 'result')
+    else:
+        d = {}
+        setattr(item, 'result', d)
+    buf = io.StringIO()
+
+    buf.isatty = lambda: True
+    tw = TerminalWriter(file=buf)
+    report.toterminal(tw)
+    report_text = buf.getvalue()
+    d[report.when] = {'status': report.outcome, 'exception': outcome.excinfo, 'report': report_text}
+
+
+@pytest.fixture(autouse=True)
+def report(capsys, caplog, request, pytestconfig):
+    save = pytestconfig.getoption('save_results')
+    upload = pytestconfig.getoption('upload_results')
+    if save or upload:
+        caplog.set_level(logging.DEBUG)
+        start = _now()
+        yield
+        captured = capsys.readouterr()
+        log = ''
+        for step in ['setup', 'call', 'teardown']:
+            if step in request.node.result:
+                records = caplog.get_records(step)
+                for record in records:
+                    log += LOG_FORMATTER.format(record)
+        data = {
+            'stdout': captured.out,
+            'stderr': captured.err,
+            'log': log,
+            'module': request.module.__name__,
+            'cls': request.cls.__name__ if request.cls else None,
+            'function': request.function.__name__,
+            'test_name': request.node.name if request.node.name else request.function.__name__,
+            'test_start_time': start,
+            'test_end_time': _now(),
+            'run_id': pytestconfig.option.environment['run_id'],
+            'branch': pytestconfig.option.environment['git_branch'],
+            'results': {},
+            'extras': None
+        }
+        for step in ['setup', 'call', 'teardown']:
+            if step in request.node.result:
+                data['results'][step] = {'passed': request.node.result[step]['status'] == 'passed',
+                                         'status': request.node.result[step]['status'],
+                                         'exception': request.node.result[step]['exception'],
+                                         'report': request.node.result[step]['report']}
+
+        _save_or_upload(pytestconfig, data)
+    else:
+        yield
+
+
+def _mk_test_fname(data):
+    cls = data['cls']
+    return '%s-%s-%s.json' % (data['module'], cls if cls is not None else '_', data['function'])
+
+
+def _save_or_upload(config, data):
+    save = config.getoption('save_results')
+    upload = config.getoption('upload_results')
+    if save:
+        _save_report(config, data)
+    if upload:
+        _upload_report(config, data)
+
+
+def _save_report(config, data):
+    env = config.option.environment
+    path = RESULTS_ROOT / env['run_id'] / env['git_branch'] / _mk_test_fname(data)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, 'w') as f:
+        json.dump(data, f, indent=4)
+
+
+def _upload_report(config, data):
+    env = config.option.environment
+
+    url = config.getoption('server_url')
+    resp = requests.post('%s/result' % url, json={'id': env['config']['id'],
+                                                  'key': config.option.key, 'data': data})
+    resp.raise_for_status()
diff --git a/tests/executor_test_params.py b/tests/executor_test_params.py
new file mode 100644
index 00000000..ecc1e776
--- /dev/null
+++ b/tests/executor_test_params.py
@@ -0,0 +1,40 @@
+import re
+from typing import Optional
+
+
+class ExecutorTestParams:
+    """A class holding executor, launcher, url pairs."""
+
+    def __init__(self, spec: str) -> None:
+        """
+        Construct a new instance.
+
+        Parameters
+        ----------
+        spec:
+            A string in the format "<executor>[:<launcher>[:<url>]]". If only the executor and the
+            url are specified, the string should be formatted as "<executor>::<url>".
+        """
+        spec_l = re.split(':', spec, maxsplit=2)
+        self.executor = spec_l[0]
+        if len(spec_l) > 1:
+            self.launcher = spec_l[1]  # type: Optional[str]
+        else:
+            self.launcher = None
+        if len(spec_l) == 3:
+            self.url = spec_l[2]  # type: Optional[str]
+        else:
+            self.url = None
+
+    def __repr__(self) -> str:
+        """Returns a string representation of this object."""
+        if self.launcher is not None:
+            if self.url is not None:
+                return '{}:{}:{}'.format(self.executor, self.launcher, self.url)
+            else:
+                return '{}:{}'.format(self.executor, self.launcher)
+        else:
+            if self.url is not None:
+                return '{}::{}'.format(self.executor, self.url)
+            else:
+                return '{}'.format(self.executor)
diff --git a/tests/run_job.py b/tests/run_job.py
index 2d074eaf..f117679d 100755
--- a/tests/run_job.py
+++ b/tests/run_job.py
@@ -19,8 +19,8 @@
 
     jobs = list()
     job = Job(JobSpec(executable='/bin/date'))
-    exec = JobExecutor.get_instance(name=name, url=url)
-    exec.submit(job)
+    ex = JobExecutor.get_instance(name=name, url=url)
+    ex.submit(job)
     jobs.append(job)
     print('Job submitted')
     status = job.wait()
diff --git a/tests/test_executor.py b/tests/test_executor.py
index bc3e7815..1288c5f2 100644
--- a/tests/test_executor.py
+++ b/tests/test_executor.py
@@ -1,57 +1,62 @@
-import pytest
-
-from typing import Optional
 from pathlib import Path
+from psij import SubmitException, Job, JobExecutor, JobSpec, JobState
 from tempfile import TemporaryDirectory
 
-from psij import SubmitException, Job, JobExecutor, JobSpec, JobState
+from executor_test_params import ExecutorTestParams
+
+
+def assert_completed(job: Job) -> None:
+    if job.status.state == JobState.FAILED:
+        raise RuntimeError('Job (native_id: {}) failed: {}'.format(job.native_id,
+                                                                   job.status.message))
+    elif job.status.state == JobState.COMPLETED:
+        return
+    else:
+        raise RuntimeError('Unexpected job state: {}'.format(job.status.state))
+
 
-tests = [['local', None],
-         ['saga', 'fork://localhost/'],
-         ['saga', 'ssh://localhost/']]
+def _get_executor_instance(ep: ExecutorTestParams, job: Job) -> JobExecutor:
+    assert job.spec is not None
+    job.spec.launcher = ep.launcher
+    return JobExecutor.get_instance(ep.executor, url=ep.url)
 
 
-@pytest.mark.parametrize('name,url', tests)
-def test_simple_job(name: str, url: Optional[str]) -> None:
-    job = Job(JobSpec(executable='/bin/date'))
-    jex = JobExecutor.get_instance(name=name, url=url)
-    jex.submit(job)
+def test_simple_job(execparams: ExecutorTestParams) -> None:
+    job = Job(JobSpec(executable='/bin/date', launcher=execparams.launcher))
+    ex = _get_executor_instance(execparams, job)
+    ex.submit(job)
     job.wait()
 
 
-@pytest.mark.parametrize('name,url', tests)
-def test_simple_job_redirect(name: str, url: Optional[str]) -> None:
-    with TemporaryDirectory() as td:
+def test_simple_job_redirect(execparams: ExecutorTestParams) -> None:
+    with TemporaryDirectory(dir=Path.home() / '.psij' / 'work') as td:
         outp = Path(td, 'stdout.txt')
-        job = Job(JobSpec(executable='/bin/echo', arguments=['-n', '_x_'],
-                          stdout_path=outp))
-        jex = JobExecutor.get_instance(name=name, url=url)
-        jex.submit(job)
+        job = Job(JobSpec(executable='/bin/echo', arguments=['-n', '_x_'], stdout_path=outp))
+        ex = _get_executor_instance(execparams, job)
+        ex.submit(job)
         job.wait()
         f = outp.open("r")
         contents = f.read()
         assert contents == '_x_'
 
 
-@pytest.mark.parametrize('name,url', tests)
-def test_attach(name: str, url: Optional[str]) -> None:
+def test_attach(execparams: ExecutorTestParams) -> None:
     job = Job(JobSpec(executable='/bin/sleep', arguments=['1']))
-    jex = JobExecutor.get_instance(name=name, url=url)
-    jex.submit(job)
-    job.wait(target_states=[JobState.ACTIVE])
+    ex = _get_executor_instance(execparams, job)
+    ex.submit(job)
+    job.wait(target_states=[JobState.ACTIVE, JobState.COMPLETED])
     native_id = job.native_id
 
     assert native_id is not None
     job2 = Job()
-    jex.attach(job2, native_id)
+    ex.attach(job2, native_id)
     job2.wait()
 
 
-@pytest.mark.parametrize('name,url', tests)
-def test_cancel(name: str, url: Optional[str]) -> None:
-    job = Job(JobSpec(executable='/bin/sleep', arguments=['1']))
-    jex = JobExecutor.get_instance(name=name, url=url)
-    jex.submit(job)
+def test_cancel(execparams: ExecutorTestParams) -> None:
+    job = Job(JobSpec(executable='/bin/sleep', arguments=['60']))
+    ex = _get_executor_instance(execparams, job)
+    ex.submit(job)
     job.wait(target_states=[JobState.ACTIVE])
     job.cancel()
     status = job.wait()
@@ -59,11 +64,10 @@ def test_cancel(name: str, url: Optional[str]) -> None:
     assert status.state == JobState.CANCELED
 
 
-@pytest.mark.parametrize('name,url', tests)
-def test_failing_job(name: str, url: Optional[str]) -> None:
+def test_failing_job(execparams: ExecutorTestParams) -> None:
     job = Job(JobSpec(executable='/bin/false'))
-    jex = JobExecutor.get_instance(name=name, url=url)
-    jex.submit(job)
+    ex = _get_executor_instance(execparams, job)
+    ex.submit(job)
     status = job.wait()
     assert status is not None
     assert status.state == JobState.FAILED
@@ -71,14 +75,13 @@ def test_failing_job(name: str, url: Optional[str]) -> None:
     assert status.exit_code != 0
 
 
-@pytest.mark.parametrize('name,url', tests)
-def test_missing_executable(name: str, url: Optional[str]) -> None:
+def test_missing_executable(execparams: ExecutorTestParams) -> None:
     job = Job(JobSpec(executable='/bin/no_such_file_or_directory'))
-    jex = JobExecutor.get_instance(name=name, url=url)
+    ex = _get_executor_instance(execparams, job)
     # we don't know if this will fail with an exception or JobState.FAILED,
     # so handle both
     try:
-        jex.submit(job)
+        ex.submit(job)
         status = job.wait()
         assert status is not None
         assert status.state == JobState.FAILED
@@ -86,3 +89,16 @@ def test_missing_executable(name: str, url: Optional[str]) -> None:
         assert status.exit_code != 0
     except SubmitException:
         pass
+
+
+def test_parallel_jobs(execparams: ExecutorTestParams) -> None:
+    spec = JobSpec(executable='/bin/sleep', arguments=['5'])
+    job1 = Job(spec)
+    job2 = Job(spec)
+    ex = _get_executor_instance(execparams, job1)
+    ex.submit(job1)
+    ex.submit(job2)
+    job1.wait()
+    job2.wait()
+    assert_completed(job1)
+    assert_completed(job2)
diff --git a/tests/test_executor_versions.py b/tests/test_executor_versions.py
index bc50f45a..a1fdee1c 100755
--- a/tests/test_executor_versions.py
+++ b/tests/test_executor_versions.py
@@ -8,5 +8,5 @@
 
 
 def test_executor_version(name: str = 'local') -> None:
-    exec = JobExecutor.get_instance(name)
-    assert isinstance(exec.version, Version)
+    ex = JobExecutor.get_instance(name)
+    assert isinstance(ex.version, Version)
diff --git a/tests/test_infrastructure.py b/tests/test_infrastructure.py
new file mode 100644
index 00000000..b2bd7ade
--- /dev/null
+++ b/tests/test_infrastructure.py
@@ -0,0 +1,26 @@
+import logging
+import sys
+
+from _pytest.capture import CaptureFixture
+from _pytest.logging import LogCaptureFixture
+
+logger = logging.getLogger('test')
+
+
+def test_capture(capsys: CaptureFixture[str], caplog: LogCaptureFixture) -> None:
+    caplog.set_level(logging.DEBUG)
+    print('out')
+    print('err', file=sys.stderr)
+    magic = 'zzyggdr'
+    logger.info(magic)
+
+    streams = capsys.readouterr()
+    assert streams.out == 'out\n'
+    assert streams.err == 'err\n'
+    assert magic in caplog.text
+
+
+def test_capture_2() -> None:
+    print('out')
+    print('err', file=sys.stderr)
+    logger.info('info')