ksonnet utilities for testing. (kubeflow#271)

* ksonnet utilities for testing.

* Create a new module ks_util for common utilities related to ksonnet.

* Move the function get_ksonnet_cmd from run_e2e_workflow.py into ks_util

  * We want to reuse this function e.g. to submit TFJobs as part of E2E
    testing.

* Move setup_ks_app from
  https://github.com/kubeflow/tf-operator/blob/master/py/ks_util.py

  so it can be reused by other tests not just tf-operator.

* Fix.

* Update setup_ks_app to use the new command.

* Add missing yaml import.

* Fix lint.

py/kubeflow/testing/ks_util.py

@@ -0,0 +1,55 @@
+"""Utilities for working with ksonnet in the tests."""
+
+import filelock
+import logging
+import os
+import re
+import subprocess
+import yaml
+
+from kubeflow.testing import util
+
+def setup_ks_app(app_dir, env, namespace, component, params, ks_cmd=None):
+  """Setup the ksonnet app"""
+
+  if not ks_cmd:
+    ks_cmd = get_ksonnet_cmd(app_dir)
+
+  lock_file = os.path.join(app_dir, "app.lock")
+  logging.info("Acquiring lock on file: %s", lock_file)
+  lock = filelock.FileLock(lock_file, timeout=60)
+  with lock:
+    # Create a new environment for this run
+    try:
+      util.run([ks_cmd, "env", "add", env, "--namespace=" + namespace],
+                cwd=app_dir)
+    except subprocess.CalledProcessError as e:
+      if not re.search(".*environment.*already exists.*", e.output):
+        raise
+
+    for pair in params.split(","):
+      k, v = pair.split("=", 1)
+      util.run([ks_cmd, "param", "set", "--env=" + env, component, k, v],
+                cwd=app_dir)
+
+def get_ksonnet_cmd(app_dir):
+  """Get the ksonnet command based on the apiVersion in app.yaml.
+
+  Args:
+    app_dir: Directory of the ksonnet application.
+
+  Returns:
+    ks_cmd: Path to the ks binary to use.
+  """
+  app_yaml_file = app_dir + "/app.yaml"
+  with open(app_yaml_file) as app_yaml:
+    results = yaml.load(app_yaml)
+
+  if results["apiVersion"] == "0.1.0":
+    return "ks"
+
+  if results["apiVersion"] == "0.2.0":
+    return "ks-12"
+
+  # For compatibility reasons we'll keep the default cmd as "ks".
+  return "ks"

py/kubeflow/testing/run_e2e_workflow.py

@@ -45,6 +45,7 @@
 import os
 import tempfile
 from kubeflow.testing import argo_client
+from kubeflow.testing import ks_util
 from kubeflow.testing import prow_artifacts
 from kubeflow.testing import util
 import uuid
@@ -110,21 +111,6 @@ def generate_env_from_head(args):
       continue
     os.environ[k] = env_var.get(k)
 
-# Get the ksonnet cmd name based on apiVersion in app.yaml.
-def get_ksonnet_cmd(workflow):
-  app_yaml_file = workflow.app_dir + "/app.yaml"
-  with open(app_yaml_file) as app_yaml:
-    results = yaml.load(app_yaml)
-
-  if results["apiVersion"] == "0.1.0":
-    return "ks"
-
-  if results["apiVersion"] == "0.2.0":
-    return "ks-12"
-
-  # For compatibility reasons we'll keep the default cmd as "ks".
-  return "ks"
-
 def run(args, file_handler): # pylint: disable=too-many-statements,too-many-branches
   job_type = os.getenv("JOB_TYPE")
   repo_owner = os.getenv("REPO_OWNER")
@@ -175,7 +161,7 @@ def run(args, file_handler): # pylint: disable=too-many-statements,too-many-bran
     # Workflow name should not be more than 63 characters because its used
     # as a label on the pods.
     workflow_name = os.getenv("JOB_NAME") + "-" + w.name
-    ks_cmd = get_ksonnet_cmd(w)
+    ks_cmd = ks_util.get_ksonnet_cmd(w.app_dir)
 
     # Print ksonnet version
     util.run([ks_cmd, "version"])