Merge pull request #2 from SimonJPegg/feature/testWorkflows

feature/testWorkflows: added basic workflow tester

build.sbt

@@ -1,3 +1,5 @@
+import ReleaseTransformations._
+
 name := "scoozie"
 
 organization := "org.antipathy"
@@ -8,7 +10,7 @@ scalaVersion := "2.12.8"
 
 crossScalaVersions := Seq("2.10.7","2.11.12", scalaVersion.value)
 
-scalacOptions ++= Seq("-encoding", "UTF-8", "-deprecation", "-unchecked")
+scalacOptions ++= Seq("-encoding", "UTF-8", "-deprecation", "-unchecked", "-feature")
 
 licenses := Seq((
   "Apache License, Version 2.0", 
@@ -61,8 +63,6 @@ publishMavenStyle := true
 
 publishArtifact in Test := false
 
-import ReleaseTransformations._
-
 releaseCrossBuild := true 
 
 releasePublishArtifactsAction := PgpKeys.publishSigned.value

src/main/scala/org/antipathy/scoozie/JobProperties.scala

@@ -0,0 +1,12 @@
+package org.antipathy.scoozie
+
+/**
+  * Base trait for obtaining job properties
+  */
+private[scoozie] trait JobProperties {
+
+  /**
+    * Get the job properties
+    */
+  def jobProperties: String
+}

src/main/scala/org/antipathy/scoozie/Nameable.scala

@@ -3,7 +3,7 @@ package org.antipathy.scoozie
 /**
   * base trait for namable objects
   */
-trait Nameable {
+private[scoozie] trait Nameable {
 
   /**
     * The name of the object

src/main/scala/org/antipathy/scoozie/Node.scala

@@ -5,32 +5,40 @@ import org.antipathy.scoozie.configuration.Credentials
 import org.antipathy.scoozie.control._
 import scala.xml.Elem
 import scala.collection.immutable.Map
+import org.antipathy.scoozie.exception.TransitionException
 
 /**
   * Wapper class for oozie actions, used to define transitions
   *
   * @param action the inner oozie action
-  * @param _transition the node to transition to on success
-  * @param _failure the node to tranistion to on failure
+  * @param successTransition the node to transition to on success
+  * @param failureTransition the node to tranistion to on failure
   * @param credentialsOption optional credentials for the nodes
   */
 private[scoozie] case class Node(
     action: Action,
-    _transition: Option[Node] = None,
-    _failure: Option[Node] = None
+    successTransition: Option[Node] = None,
+    failureTransition: Option[Node] = None
 )(implicit credentialsOption: Option[Credentials])
     extends XmlSerializable
-    with OozieProperties {
+    with OozieProperties
+    with Nameable {
 
   /**
     * The node to transition to on success
     */
-  def okTo(node: Node): Node = this.copy(_transition = Some(node))
+  def okTo(node: Node): Node = this.action match {
+    case _ @(_: Fork | _: Decision | _: End | _: Kill) => this
+    case _                                             => this.copy(successTransition = Some(node))
+  }
 
   /**
     * The node to transition to on failure
     */
-  def errorTo(node: Node): Node = this.copy(_failure = Some(node))
+  def errorTo(node: Node): Node = this.action match {
+    case _ @(_: Fork | _: Decision | _: End | _: Kill | _: Start) => this
+    case _                                                        => this.copy(failureTransition = Some(node))
+  }
 
   /**
     * The XML for this node
@@ -50,40 +58,41 @@ private[scoozie] case class Node(
     * Validate the start element has a transition and build it
     */
   private def buildStartXML: Elem =
-    if (_transition.isEmpty) {
-      throw new IllegalArgumentException(
-        "No node has been defined to start from"
-      )
+    if (successTransition.isEmpty) {
+      throw new TransitionException("No node has been defined to start from")
     } else {
-      <start to={_transition.get.action.name} />
+      <start to={successTransition.get.action.name} />
     }
 
   /**
     * validate a element has transitions and build it
     */
   private def buildActionXML: Elem = {
-    if (_transition.isEmpty) {
-      throw new IllegalArgumentException(
-        s"${action.name} does not have an okTo set"
-      )
+    if (successTransition.isEmpty) {
+      throw new TransitionException(s"${action.name} does not have an okTo set")
     }
 
-    if (_failure.isEmpty) {
-      throw new IllegalArgumentException(
+    if (failureTransition.isEmpty) {
+      throw new TransitionException(
         s"${action.name} does not have an errorTo set"
       )
     }
 
     <action name ={action.name}
       cred={credentialsOption.map(_.credential.name).orNull}>
       {action.toXML}
-      <ok to= {_transition.get.action.name}/>
-      <error to = {_failure.get.action.name}/>
+      <ok to= {successTransition.get.action.name}/>
+      <error to = {failureTransition.get.action.name}/>
     </action>
   }
 
   /**
     * Get the Oozie properties for this object
     */
   override def properties: Map[String, String] = action.properties
+
+  /**
+    * The name of the object
+    */
+  override def name: String = action.name
 }

src/main/scala/org/antipathy/scoozie/control/Decision.scala

@@ -22,6 +22,11 @@ final class Decision(override val name: String,
     */
   override val properties: Map[String, String] = Map()
 
+  /**
+    * The default path for this node
+    */
+  def defaultPath: Node = default
+
   /**
     * The nodes contained within this fork
     */

src/main/scala/org/antipathy/scoozie/control/Fork.scala

@@ -4,6 +4,7 @@ import org.antipathy.scoozie.Node
 import org.antipathy.scoozie.action.Action
 import scala.collection.immutable.Map
 import scala.xml.Elem
+import org.antipathy.scoozie.exception.TransitionException
 
 /**
   * Oozie Fork control node
@@ -32,7 +33,7 @@ final class Fork(override val name: String, nodes: Seq[Node]) extends Action {
     */
   override def toXML: Elem = {
     if (transitionPaths.length < 2) {
-      throw new IllegalArgumentException(
+      throw new TransitionException(
         s"Error in Fork($name): must have at least 2 actions"
       )
     }

src/main/scala/org/antipathy/scoozie/coordinator/CoOrdinator.scala

@@ -1,6 +1,7 @@
 package org.antipathy.scoozie.coordinator
 import org.antipathy.scoozie.{Nameable, XmlSerializable}
 import org.antipathy.scoozie.configuration.Configuration
+import org.antipathy.scoozie.workflow.Workflow
 
 import scala.xml.Elem
 
@@ -11,15 +12,15 @@ import scala.xml.Elem
   * @param start the CoOrdinator start time
   * @param end the CoOrdinator end time
   * @param timezone the CoOrdinator time-zone
-  * @param workflowPath the workflow application path
+  * @param workflow the workflow to run
   * @param configuration configuration for the workflow
   */
 case class CoOrdinator(override val name: String,
                        frequency: String,
                        start: String,
                        end: String,
                        timezone: String,
-                       workflowPath: String,
+                       workflow: Workflow,
                        configuration: Configuration)
     extends XmlSerializable
     with Nameable {
@@ -36,7 +37,7 @@ case class CoOrdinator(override val name: String,
                      xmlns="uri:oozie:coordinator:0.1">
       <action>
         <workflow>
-          <app-path>{workflowPath}</app-path>
+          <app-path>{workflow.path}</app-path>
           {if (configuration.configProperties.nonEmpty) {
               configuration.toXML
             }

src/main/scala/org/antipathy/scoozie/exception/LoopingException.scala

@@ -0,0 +1,7 @@
+package org.antipathy.scoozie.exception
+
+/**
+  * Thrown when a loop is detected in a workfloe
+  */
+class LoopingException(message: String, cause: Throwable = null)
+    extends RuntimeException(message, cause)

src/main/scala/org/antipathy/scoozie/exception/NoSchemaException.scala

@@ -0,0 +1,7 @@
+package org.antipathy.scoozie.exception
+
+/**
+  * Thrown when unable to find a schema for an oozie artifact
+  */
+class NoSchemaException(message: String, cause: Throwable = null)
+    extends RuntimeException(message, cause)

src/main/scala/org/antipathy/scoozie/exception/TransitionException.scala

@@ -0,0 +1,7 @@
+package org.antipathy.scoozie.exception
+
+/**
+  * Thrown when transition between oozie nodes is impossible
+  */
+class TransitionException(message: String, cause: Throwable = null)
+    extends RuntimeException(message, cause)

src/main/scala/org/antipathy/scoozie/testing/Visitor.scala

@@ -0,0 +1,15 @@
+package org.antipathy.scoozie.testing
+
+import org.antipathy.scoozie.Node
+import scala.collection.immutable.Seq
+
+/**
+  * Class for visiting oozie nodes
+  *
+  * @param visited list of oozie nodes
+  * @param failed true if the last node failed
+  * @param nextNodeOption the next node to visit
+  */
+private[testing] case class Visitor(visited: Seq[Seq[String]],
+                                    failed: Boolean = false,
+                                    nextNodeOption: Option[Node] = None)