feat: Add an exec command to submit AP to a test workcell (#193)

This changes adds the ability for a user to submit autoprotocol to
a test instance of scle. This allows them to view a schedule against
a theoretical device set.

Example:
```
cat my_ap.json | transcriptic exec -a "my.scle.endpoint" -w "wc4"
```

Author: scottyantipa

CHANGELOG.rst

@@ -7,7 +7,7 @@ Unreleased
 
 Added
 ~~~~~
-
+- A new `exec` command to send autoprotocol to a test workcell
 - isort for automatic import sorting
 - Example initial tests for `commands` file using `responses` pattern, starting with
   `submit` and `projects`.

test/cli_test.py

@@ -2,49 +2,10 @@
 
 import pytest
 
-from click.testing import CliRunner
-from Crypto.PublicKey import RSA
 from transcriptic import commands
 from transcriptic.cli import cli
 
-
-@pytest.fixture(scope="session")
-def temp_tx_dotfile(tmpdir_factory):
-    path = tmpdir_factory.mktemp("config").join(".transcriptic")
-    config = {
-        "email": "somebody@transcriptic.com",
-        "token": "foobarinvalid",
-        "organization_id": "transcriptic",
-        "api_root": "http://foo:5555",
-        "analytics": True,
-        "user_id": "ufoo2",
-        "feature_groups": ["can_submit_autoprotocol", "can_upload_packages"],
-    }
-    with open(str(path), "w") as f:
-        json.dump(config, f)
-    return path
-
-
-@pytest.fixture()
-def temp_ssh_key(tmpdir_factory):
-    dir_path = tmpdir_factory.mktemp("ssh")
-    private_key_path = str(dir_path.join("private.pem"))
-    public_key_path = str(dir_path.join("public.pem"))
-    private_key = RSA.generate(2048)
-    with open(private_key_path, "wb") as private_key_file:
-        private_key_file.write(private_key.exportKey("PEM"))
-    pubkey = private_key.publickey()
-    with open(public_key_path, "wb") as public_key_file:
-        public_key_file.write(pubkey.exportKey("OpenSSH"))
-    yield private_key_path, public_key_path
-
-
-@pytest.fixture
-def cli_test_runner(temp_tx_dotfile):
-    runner = CliRunner(
-        env={"TRANSCRIPTIC_CONFIG": str(temp_tx_dotfile)}, mix_stderr=False
-    )
-    yield runner
+from .helpers.fixtures import *
 
 
 def test_kebab_case(cli_test_runner):

test/commands/exec_test.py

@@ -0,0 +1,103 @@
+import json
+
+import requests
+
+from transcriptic.cli import cli
+
+from ..helpers.fixtures import *
+from ..helpers.mockAPI import MockResponse
+
+
+# Structure of the response object from SCLE
+def bool_success_res():
+    return {"success": True}
+
+
+def mock_api_endpoint():
+    return "foo.bar.baz"
+
+
+@pytest.fixture
+def ap_file(tmpdir_factory):
+    """Make a temp autoprotocol file"""
+    path = tmpdir_factory.mktemp("foo").join("ap.json")
+    with open(str(path), "w") as f:
+        f.write("{}")  # any valid json works
+    return path
+
+
+def test_good_autoprotocol(cli_test_runner, monkeypatch, ap_file):
+    def mockpost(*args, **kwargs):
+        return MockResponse(0, bool_success_res(), json.dumps(bool_success_res()))
+
+    monkeypatch.setattr(requests, "post", mockpost)
+    result = cli_test_runner.invoke(
+        cli, ["exec", str(ap_file), "-a", mock_api_endpoint()]
+    )
+    assert result.exit_code == 0
+    assert (
+        f"Success. View {mock_api_endpoint()} to see the scheduling outcome."
+        in result.output
+    )
+
+
+def test_bad_autoprotocol(cli_test_runner):
+    result = cli_test_runner.invoke(
+        cli, ["exec", "bad-file-handle", "-a", mock_api_endpoint()]
+    )
+    assert result.exit_code != 0
+    assert "Invalid value for '[AUTOPROTOCOL]': Could not open file" in result.stderr
+
+
+def test_bad_deviceset(cli_test_runner, ap_file):
+    result = cli_test_runner.invoke(
+        cli,
+        [
+            "exec",
+            str(ap_file),
+            "--device-set",
+            "bad-file-handle",
+            "-a",
+            mock_api_endpoint(),
+        ],
+    )
+    assert result.exit_code != 0
+    assert (
+        "Invalid value for '--device-set' / '-d': Could not open file: bad-file-handle:"
+        in result.stderr
+    )
+
+
+def test_bad_api_response(cli_test_runner, monkeypatch, ap_file):
+    def mockpost(*args, **kwargs):
+        return MockResponse(0, "not-json", "not-json")
+
+    monkeypatch.setattr(requests, "post", mockpost)
+    result = cli_test_runner.invoke(
+        cli, ["exec", str(ap_file), "-a", mock_api_endpoint()]
+    )
+    assert result.exit_code == 0
+    assert "Error: " in result.stderr
+
+
+def test_good_workcell(cli_test_runner, monkeypatch, ap_file):
+    def mockpost(*args, **kwargs):
+        return MockResponse(0, bool_success_res(), json.dumps(bool_success_res()))
+
+    monkeypatch.setattr(requests, "post", mockpost)
+    result = cli_test_runner.invoke(
+        cli, ["exec", str(ap_file), "-a", mock_api_endpoint(), "-w", "wc3"]
+    )
+    assert result.exit_code == 0
+    assert (
+        f"Success. View {mock_api_endpoint()} to see the scheduling outcome."
+        in result.output
+    )
+
+
+def test_bad_workcell(cli_test_runner, ap_file):
+    result = cli_test_runner.invoke(
+        cli, ["exec", str(ap_file), "-a", mock_api_endpoint(), "-w", "bad-workcell-id"]
+    )
+    assert result.exit_code != 0
+    assert "Workcell id must be like wcN but was bad-workcell-id" in result.stderr

test/helpers/fixtures.py

@@ -0,0 +1,45 @@
+import json
+
+import pytest
+
+from click.testing import CliRunner
+from Crypto.PublicKey import RSA
+
+
+@pytest.fixture(scope="session")
+def temp_tx_dotfile(tmpdir_factory):
+    path = tmpdir_factory.mktemp("config").join(".transcriptic")
+    config = {
+        "email": "somebody@transcriptic.com",
+        "token": "foobarinvalid",
+        "organization_id": "transcriptic",
+        "api_root": "http://foo:5555",
+        "analytics": True,
+        "user_id": "ufoo2",
+        "feature_groups": ["can_submit_autoprotocol", "can_upload_packages"],
+    }
+    with open(str(path), "w") as f:
+        json.dump(config, f)
+    return path
+
+
+@pytest.fixture
+def cli_test_runner(temp_tx_dotfile):
+    runner = CliRunner(
+        env={"TRANSCRIPTIC_CONFIG": str(temp_tx_dotfile)}, mix_stderr=False
+    )
+    yield runner
+
+
+@pytest.fixture()
+def temp_ssh_key(tmpdir_factory):
+    dir_path = tmpdir_factory.mktemp("ssh")
+    private_key_path = str(dir_path.join("private.pem"))
+    public_key_path = str(dir_path.join("public.pem"))
+    private_key = RSA.generate(2048)
+    with open(private_key_path, "wb") as private_key_file:
+        private_key_file.write(private_key.exportKey("PEM"))
+    pubkey = private_key.publickey()
+    with open(public_key_path, "wb") as public_key_file:
+        public_key_file.write(pubkey.exportKey("OpenSSH"))
+    yield private_key_path, public_key_path

transcriptic/cli.py

@@ -628,3 +628,66 @@ def login_cmd(ctx, api_root=None, analytics=True, rsa_key=None):
 def format_cmd(manifest):
     """Check Autoprotocol format of manifest.json."""
     commands.format(manifest)
+
+
+@cli.command("exec")
+@click.argument("autoprotocol", type=click.File("r"), default=sys.stdin)
+@click.option(
+    "--api", "-a", help="The api endpoint of your scle test workcell instance."
+)
+@click.option(
+    "--workcell-id",
+    "-w",
+    help="The workcell id to use for the device set. This is not permitted along with the `deviceSet` option.",
+)
+@click.option(
+    "--device-set",
+    "-d",
+    type=click.File("r"),
+    help="A DeviceSet json file to use for scheduling. This is not permitted along with the `workcellId` option.",
+)
+@click.option(
+    "--time-limit",
+    "-t",
+    type=click.INT,
+    default=30,
+    help="The maximum time in seconds to spend scheduling. The scheduler will use all the time until an optimal solution is found.",
+)
+@click.option(
+    "--partition-group-size",
+    type=click.INT,
+    default=None,
+    help="The number of x_partition groups to be scheduled together.",
+)
+@click.option(
+    "--partition-horizon",
+    type=click.INT,
+    default=None,
+    help="The time in seconds to overlap partitions by.",
+)
+@click.option(
+    "--partitioning-swap-device-id",
+    default=None,
+    help="The device id to use as a swap space when partitioning.",
+)
+def execute(
+    autoprotocol,
+    api,
+    workcell_id,
+    device_set,
+    time_limit,
+    partition_group_size,
+    partition_horizon,
+    partitioning_swap_device_id,
+):
+    """Send autoprotocol to a test workcell (no hardware) for scheduling."""
+    commands.execute(
+        autoprotocol,
+        api,
+        workcell_id,
+        device_set,
+        time_limit,
+        partition_group_size,
+        partition_horizon,
+        partitioning_swap_device_id,
+    )

transcriptic/commands.py

@@ -10,6 +10,7 @@
 import json
 import locale
 import os
+import re
 import sys
 import time
 import warnings
@@ -22,6 +23,7 @@
 import click
 import requests
 
+from click.exceptions import BadParameter
 from jinja2 import Environment, PackageLoader
 from transcriptic import routes
 from transcriptic.auth import StrateosSign
@@ -1315,8 +1317,6 @@ def org_prompt(org_list):
             click.echo(f"{indx + 1}.  {o['name']} ({o['subdomain']})")
 
         def parse_valid_org(indx):
-            from click.exceptions import BadParameter
-
             try:
                 org_indx = int(indx) - 1
                 if org_indx < 0 or org_indx >= len(org_list):
@@ -1476,6 +1476,71 @@ def run_protocol(api, manifest, protocol, inputs, view=False, dye_test=False):
             return
 
 
+def execute(
+    autoprotocol,
+    api,
+    workcell_id,
+    device_set,
+    time_limit,
+    partition_group_size,
+    partition_horizon,
+    partitioning_swap_device_id,
+):
+    # Get the autoprotocol
+    autoprotocol_str = autoprotocol.read()
+    try:
+        autoprotocol_dict = json.loads(autoprotocol_str)
+    except json.decoder.JSONDecodeError as err:
+        click.echo(f"Error decoding autoprotocol json: {err}", err=True)
+        return
+
+    # Define the initial payload
+    payload = {"autoprotocol": autoprotocol_dict, "timeLimit": f"{time_limit}:second"}
+
+    if device_set:
+        device_str = device_set.read()
+        try:
+            device_json = json.loads(device_str)
+            payload["deviceSet"] = device_json
+        except json.decoder.JSONDecodeError as err:
+            click.echo(f"Error decoding device set json: {err}", err=True)
+            return
+    elif workcell_id:
+        if not re.search("^wc[a-z,0-9]+$", workcell_id):
+            raise BadParameter(f"Workcell id must be like wcN but was {workcell_id}")
+        payload["workcellIdForDeviceSet"] = f"{workcell_id}-mcx1"
+    else:
+        payload["workcellIdForDeviceSet"] = "wctest-mcx1"
+
+    if partition_group_size is not None:
+        payload["partitionGroupSize"] = partition_group_size
+
+    if partition_horizon is not None:
+        payload["partitionHorizon"] = f"{partition_horizon}:second"
+
+    if partitioning_swap_device_id is not None:
+        payload["partitioningSwapDeviceId"] = partitioning_swap_device_id
+
+    # Clean api end point
+    if api[-1] == "/":
+        clean_api = api[0:-1]  # remove trailing slash
+    else:
+        clean_api = api
+
+    # POST to workcell
+    test_run_endpoint = f"{clean_api}/testRun"
+    click.echo("Sending request...")
+    res = requests.post(test_run_endpoint, json=payload)
+    try:
+        res_json = json.loads(res.text)
+        if res_json["success"]:
+            click.echo(f"Success. View {clean_api} to see the scheduling outcome.")
+        else:
+            click.echo(f"Error: {res_json['message']}", err=True)
+    except json.decoder.JSONDecodeError:
+        click.echo(f"Error: {res.text}", err=True)
+
+
 def parse_json(json_file):
     try:
         return json.load(open(json_file))