Quick Start

Installing Artemis

(myvirtualenv) > pip install openeye-orionplatform[dev]

Writing a Floe Test

To write floe tests you will need to define a new Test class from FloeTestCase.

from artemis.test import FloeTestCase
from artemis.decorators import package

class ExampleFloeTest(FloeTestCase):
    def test_example_floe(self):
        workfloe = WorkFloeWrapper.get_workfloe(
            os.path.join(FLOE_DIR, "classic_omega.py"),  # Path to the local workfloe
            run_timeout=1200,  # Max time to run
            queue_timeout=600,  # Max time to spend queued
        )

If the Floe package is installed locally, then it is possible to get the top level of directory dynamically using the following logic.

Note that use of the package decorator which takes the top level directory of the Floe Package to test as its inputs.

Note

You need to have a setup.py in the top level directory which Artemis uses to build the package for use in Orion via python setup.py sdist.

import os

# Python Floe Package pip installed with -e
import classic_floes

from artemis.test import FloeTestCase
from artemis.decorators import package
# Path to the top level directory
PACKAGE_DIR = os.path.dirname(os.path.dirname(classic_floes.__file__))

FILE_DIR = os.path.join(PACKAGE_DIR, "tests", "data")
FLOE_DIR = os.path.join(PACKAGE_DIR, "floes")


@package(PACKAGE_DIR)
class ExampleFloeTest(FloeTestCase):

Once the test is created, a wrapper around the Workfloe is instantiated that provides a simple API for running the Floe that can switch between running locally or in Orion.

    def test_example_floe(self):
        workfloe = WorkFloeWrapper.get_workfloe(
            os.path.join(FLOE_DIR, "classic_omega.py"),  # Path to the local workfloe
            run_timeout=1200,  # Max time to run
            queue_timeout=600,  # Max time to spend queued
        )

As Datasets and Files are handled differently whether they are local or in Orion, Artemis provides wrappers to these.

    def test_example_floe(self):
        workfloe = WorkFloeWrapper.get_workfloe(
            os.path.join(FLOE_DIR, "classic_omega.py"),  # Path to the local workfloe
            run_timeout=1200,  # Max time to run
            queue_timeout=600,  # Max time to spend queued
        )
        input_file = DatasetWrapper.from_file(os.path.join(FILE_DIR, "eMol_ran50.ism"))
        output_dataset = OutputDatasetWrapper()
        failure_dataset = OutputDatasetWrapper()
        workfloe.start(
            {

To run the Workfloe it is necessary to provide it with all of the parameters. The start() method takes the same parameters as what the Orion Job API takes. The basic structure is shown below.

{
  "promoted": {
    "promoted_name": "value"
  },
  "cube": {
    "cube_name": {
      "param_name": "value"
    }
  }
}

Pass the parameters directly to the start() method to run the job.

    def test_example_floe(self):
        workfloe = WorkFloeWrapper.get_workfloe(
            os.path.join(FLOE_DIR, "classic_omega.py"),  # Path to the local workfloe
            run_timeout=1200,  # Max time to run
            queue_timeout=600,  # Max time to spend queued
        )
        input_file = DatasetWrapper.from_file(os.path.join(FILE_DIR, "eMol_ran50.ism"))
        output_dataset = OutputDatasetWrapper()
        failure_dataset = OutputDatasetWrapper()
        workfloe.start(
            {
                "promoted": {
                    "Input Molecules": input_file.identifier,
                    "out": output_dataset.identifier,
                    "failed": failure_dataset.identifier,
                    "Maximum Conformers": 10,
                }
            }
        )
        self.assertWorkFloeComplete(workfloe)

        # Convert dataset's iterator of records to a list of records

Once it completes, the Wrapper objects provide access to the information from the floe necessary to write a test of the logic in the Floe.

        # Convert dataset's iterator of records to a list of records
        records = list(output_dataset.records())

        self.assertEqual(len(records), 38)
        expected_field = OEField("Conformer Torsion Profile", Types.String)
        for record in records:
            self.assertTrue(record.has_value(expected_field))

With that the test should be able to be run either locally or in Orion using the same test logic.

Download code

example-artemis-test.py

Running a Floe Test

To run the tests you specify whether to run them locally or whether to run them using Orion.

Locally

Locally run pytest to trigger the tests and the floes will be run locally.

pytest

Orion

To run the floes in Orion, add

pytest --orion

To control the Orion client credentials that are used, set the ORION_PROFILE environment variable to specify the profile

ORION_PROFILE=custom pytest --orion