core/org.eclipse.smila.taskmanager.test/code/src/org/eclipse/smila/taskmanager/test/TestTask.java

/*******************************************************************************
 * Copyright (c) 2008, 2011 Attensity Europe GmbH and brox IT Solutions GmbH. All rights reserved. This program and the
 * accompanying materials are made available under the terms of the Eclipse Public License v1.0 which accompanies this
 * distribution, and is available at http://www.eclipse.org/legal/epl-v10.html
 * 
 * Contributors: Andreas Weber (Attensity Europe GmbH) - initial implementation
 **********************************************************************************************************************/

package org.eclipse.smila.taskmanager.test;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.UUID;

import junit.framework.TestCase;

import org.eclipse.smila.datamodel.Any;
import org.eclipse.smila.datamodel.AnyMap;
import org.eclipse.smila.datamodel.AnySeq;
import org.eclipse.smila.datamodel.DataFactory;

import org.eclipse.smila.taskmanager.BulkInfo;
import org.eclipse.smila.taskmanager.ResultDescription;
import org.eclipse.smila.taskmanager.Task;
import org.eclipse.smila.taskmanager.TaskCompletionStatus;

/**
 * Test for {@link org.eclipse.smila.taskmanager.Task} class.
 */
public class TestTask extends TestCase {

  /**
   * Test method for {@link org.eclipse.smila.taskmanager.Task#Task(java.lang.String, java.lang.String)}.
   */
  public void testTask() {
    final String taskId = UUID.randomUUID().toString();
    final String workerName = "worker";
    final Task task = new Task(taskId, workerName);

    assertEquals(taskId, task.getTaskId());
    assertEquals(workerName, task.getWorkerName());
    assertNull(task.getQualifier());
    assertTrue(task.getInputBulks().isEmpty());
    assertTrue(task.getOutputBulks().isEmpty());
    assertTrue(task.getParameters().isEmpty());
    assertTrue(task.getProperties().isEmpty());
  }

  /**
   * Test method for {@link org.eclipse.smila.taskmanager.Task#fromAny(org.eclipse.smila.datamodel.Any)} and
   * {@link org.eclipse.smila.taskmanager.Task#toAny()}.
   */
  public void testFromAny() throws Exception {
    final AnyMap taskInputAny = DataFactory.DEFAULT.createAnyMap();
    taskInputAny.put(Task.KEY_TASK_ID, UUID.randomUUID().toString());
    taskInputAny.put(Task.KEY_WORKER_NAME, "workerName");
    taskInputAny.put(Task.KEY_QUALIFIER, "qualifier/part");
    final AnyMap parametersAny = DataFactory.DEFAULT.createAnyMap();
    final String param1Key = "param1";
    final String param2Key = "param2";
    parametersAny.put(param1Key, "p1val");
    parametersAny.put(param2Key, "p2val");
    taskInputAny.put(Task.KEY_PARAMETERS, parametersAny);
    final AnyMap propertiesAny = DataFactory.DEFAULT.createAnyMap();
    propertiesAny.put(Task.PROPERTY_WORKFLOW_RUN_ID, UUID.randomUUID().toString());
    propertiesAny.put(Task.PROPERTY_JOB_NAME, "job");
    propertiesAny.put(Task.PROPERTY_JOB_RUN_ID, UUID.randomUUID().toString());
    taskInputAny.put(Task.KEY_PROPERTIES, propertiesAny);
    final AnyMap inputAny = DataFactory.DEFAULT.createAnyMap();
    final String input1 = "input1";
    final String input2 = "input2";
    final AnySeq inputSeq1Any = DataFactory.DEFAULT.createAnySeq();
    final AnySeq inputSeq2Any = DataFactory.DEFAULT.createAnySeq();
    inputSeq1Any.add((new BulkInfo("bucketName1", "storeName1", "objectName1")).toAny());
    inputSeq1Any.add((new BulkInfo("bucketName1", "storeName1", "objectName2")).toAny());
    inputSeq2Any.add((new BulkInfo("bucketName2", "storeName1", "objectName3")).toAny());
    inputSeq2Any.add((new BulkInfo("bucketName2", "storeName2", "objectName4")).toAny());
    inputAny.put(input1, inputSeq1Any);
    inputAny.put(input2, inputSeq2Any);
    taskInputAny.put(Task.KEY_INPUT, inputAny);
    final AnyMap outputAny = DataFactory.DEFAULT.createAnyMap();
    final String output1 = "output1";
    final String output2 = "output2";
    final AnySeq outputSeq1Any = DataFactory.DEFAULT.createAnySeq();
    final AnySeq outputSeq2Any = DataFactory.DEFAULT.createAnySeq();
    outputSeq1Any.add((new BulkInfo("bucketName1", "storeName1", "objectName5")).toAny());
    outputSeq1Any.add((new BulkInfo("bucketName1", "storeName1", "objectName6")).toAny());
    outputSeq2Any.add((new BulkInfo("bucketName2", "storeName1", "objectName7")).toAny());
    outputSeq2Any.add((new BulkInfo("bucketName2", "storeName2", "objectName8")).toAny());
    outputAny.put(output1, outputSeq1Any);
    outputAny.put(output2, outputSeq2Any);
    taskInputAny.put(Task.KEY_OUTPUT, outputAny);
    final ResultDescription resultDescr =
      new ResultDescription(TaskCompletionStatus.SUCCESSFUL, "errorCode", "errorMessage",
        new HashMap<String, Number>());
    taskInputAny.put(Task.KEY_RESULT_DESCRIPTION, resultDescr.toAny());

    final Task task = Task.fromAny(taskInputAny);

    assertEquals(taskInputAny.getValue(Task.KEY_TASK_ID).asString(), task.getTaskId());
    assertEquals(taskInputAny.getValue(Task.KEY_WORKER_NAME).asString(), task.getWorkerName());
    assertEquals(taskInputAny.getValue(Task.KEY_QUALIFIER).asString(), task.getQualifier());
    assertEquals(taskInputAny.getMap(Task.KEY_PROPERTIES).getValue(Task.PROPERTY_JOB_NAME).asString(), task
      .getProperties().get(Task.PROPERTY_JOB_NAME));
    assertEquals(taskInputAny.getMap(Task.KEY_PROPERTIES).getValue(Task.PROPERTY_JOB_RUN_ID).asString(), task
      .getProperties().get(Task.PROPERTY_JOB_RUN_ID));
    assertEquals(taskInputAny.getMap(Task.KEY_PROPERTIES).getValue(Task.PROPERTY_WORKFLOW_RUN_ID).asString(), task
      .getProperties().get(Task.PROPERTY_WORKFLOW_RUN_ID));
    assertEquals(taskInputAny.getMap(Task.KEY_PARAMETERS).getValue(param1Key).asString(),
      task.getParameters().getStringValue(param1Key));
    assertEquals(taskInputAny.getMap(Task.KEY_PARAMETERS).getValue(param2Key).asString(),
      task.getParameters().getStringValue(param2Key));
    assertNotNull(taskInputAny.get(Task.KEY_INPUT));
    final Iterator<String> inputIter = taskInputAny.getMap(Task.KEY_INPUT).keySet().iterator();
    final Set<BulkInfo> inputBulks = new HashSet<BulkInfo>();
    while (inputIter.hasNext()) {
      final String key = inputIter.next();
      final Iterator<Any> seqIter = taskInputAny.getMap(Task.KEY_INPUT).get(key).iterator();
      while (seqIter.hasNext()) {
        final BulkInfo bulkFromConvertedAny = BulkInfo.fromAny(seqIter.next());
        assertTrue(task.getInputBulks().get(key).contains(bulkFromConvertedAny));
        assertFalse(inputBulks.contains(bulkFromConvertedAny));
        inputBulks.add(bulkFromConvertedAny);
      }
    }
    assertEquals(4, inputBulks.size());
    assertNotNull(taskInputAny.get(Task.KEY_OUTPUT));
    final Set<BulkInfo> outputBulks = new HashSet<BulkInfo>();
    final Iterator<String> outputIter = taskInputAny.getMap(Task.KEY_OUTPUT).keySet().iterator();
    while (outputIter.hasNext()) {
      final String key = outputIter.next();
      final Iterator<Any> seqIter = taskInputAny.getMap(Task.KEY_OUTPUT).get(key).iterator();
      while (seqIter.hasNext()) {
        final BulkInfo bulkFromConvertedAny = BulkInfo.fromAny(seqIter.next());
        assertTrue(task.getOutputBulks().get(key).contains(bulkFromConvertedAny));
        assertFalse(outputBulks.contains(bulkFromConvertedAny));
        outputBulks.add(bulkFromConvertedAny);
      }
    }
    assertEquals(4, outputBulks.size());
    assertNotSame(inputBulks, outputBulks);
    // result description:
    assertNotNull(task.getResultDescription());
    assertEquals(TaskCompletionStatus.SUCCESSFUL, task.getResultDescription().getStatus());
    assertEquals("errorCode", task.getResultDescription().getErrorCode());
    assertEquals("errorMessage", task.getResultDescription().getErrorMessage());

    final Any taskOutputAny = task.toAny();
    assertEquals(taskInputAny, taskOutputAny);
  }
}