core/org.eclipse.smila.importing.state.objectstore.test/code/src/org/eclipse/smila/importing/state/objectstore/test/TestObjectStoreDeltaService.java

/*********************************************************************************************************************
 * Copyright (c) 2008, 2012 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
 **********************************************************************************************************************/
package org.eclipse.smila.importing.state.objectstore.test;

import java.util.Collection;
import java.util.Iterator;
import java.util.SortedSet;
import java.util.TreeSet;

import org.eclipse.smila.importing.DeltaService;
import org.eclipse.smila.importing.State;
import org.eclipse.smila.importing.state.objectstore.ObjectStoreDeltaService;
import org.eclipse.smila.test.DeclarativeServiceTestCase;
import org.eclipse.smila.utils.http.NotFoundHTTPResult;

/**
 * Unit test cases for the ObjectStore based DeltaService implementation.
 */
public class TestObjectStoreDeltaService extends DeclarativeServiceTestCase {
  /** the service under test. */
  private DeltaService _service;

  @Override
  protected void setUp() throws Exception {
    super.setUp();
    _service = getService(DeltaService.class);
    _service.clearAll();
  }

  /** check if service is activated and if we are testing the correct implementation. */
  public void testServiceActivated() throws Exception {
    assertNotNull(_service);
    assertTrue(_service instanceof ObjectStoreDeltaService);
  }

  /** test checking of new objects. */
  public void testCheckNewObject() throws Exception {
    final String recordId = "testCheckNewObject";
    assertEquals(State.NEW, _service.checkState("s", recordId, "j1", "h1"));
    // still new after repeats.
    assertEquals(State.NEW, _service.checkState("s", recordId, "j1", "h1"));
  }

  /** test adding of new objects. */
  public void testAddNewObject() throws Exception {
    final String recordId = "testAddNewObject";
    _service.markAsUpdated("s", recordId, "j1", "h1");
    assertEquals(State.UPTODATE, _service.checkState("s", recordId, "j1", "h1"));
  }

  /** test updating objects objects. */
  public void testUpdateObject() throws Exception {
    final String recordId = "testUpdateObject";
    _service.markAsUpdated("s", recordId, "j1", "h1");
    // next job run
    assertEquals(State.CHANGED, _service.checkState("s", recordId, "j2", "h2"));
    _service.markAsUpdated("s", recordId, "j2", "h2");
    assertEquals(State.UPTODATE, _service.checkState("s", recordId, "j2", "h2"));
  }

  /** test clearing a single source. */
  public void testClearSource() throws Exception {
    final String recordId = "testClearSource";
    _service.markAsUpdated("s1", recordId, "j1", "h1");
    _service.markAsUpdated("s2", recordId, "j1", "h1");
    _service.clearSource("s1");
    assertEquals(State.NEW, _service.checkState("s1", recordId, "j1", "h1"));
    assertEquals(State.UPTODATE, _service.checkState("s2", recordId, "j1", "h1"));
  }

  /** test clearing the complete delta service. */
  public void testClearAll() throws Exception {
    final String recordId = "testClearAll";
    _service.markAsUpdated("s1", recordId, "j1", "h1");
    _service.markAsUpdated("s2", recordId, "j1", "h1");
    _service.clearAll();
    assertEquals(State.NEW, _service.checkState("s1", recordId, "j1", "h1"));
    assertEquals(State.NEW, _service.checkState("s2", recordId, "j1", "h1"));
  }

  /** test getting source ids. */
  public void testGetSourceIdsInitial() throws Exception {
    final Collection<String> sourceIds = _service.getSourceIds();
    assertTrue(sourceIds.isEmpty());
  }

  /** test getting source ids. */
  public void testGetSourceIds() throws Exception {
    final String recordId = "testGetSourceIds";
    _service.markAsUpdated("s1", recordId, "j1", "h1");
    _service.markAsUpdated("s2", recordId, "j1", "h1");
    Collection<String> sourceIds = _service.getSourceIds();
    assertEquals(2, sourceIds.size());
    assertTrue(sourceIds.contains("s1"));
    assertTrue(sourceIds.contains("s2"));
    _service.clearSource("s1");
    sourceIds = _service.getSourceIds();
    assertEquals(1, sourceIds.size());
    assertTrue(sourceIds.contains("s2"));
    _service.clearSource("s2");
    sourceIds = _service.getSourceIds();
    assertTrue(sourceIds.isEmpty());
  }

  /** test counting entries. */
  public void testCountEntries() throws Exception {
    final String s1 = "source1";
    final String s2 = "source2";

    try {
      _service.countEntries(s1, true);
    } catch (final Exception e) {
      assertTrue(e instanceof NotFoundHTTPResult || e.getCause() instanceof NotFoundHTTPResult);
    }
    try {
      _service.countEntries(s2, true);
    } catch (final Exception e) {
      assertTrue(e instanceof NotFoundHTTPResult || e.getCause() instanceof NotFoundHTTPResult);
    }

    _service.markAsUpdated(s1, "recordId1", "jobrun", "hash1");
    assertEquals(1, _service.countEntries(s1, true));
    try {
      _service.countEntries(s2, true);
    } catch (final Exception e) {
      assertTrue(e instanceof NotFoundHTTPResult || e.getCause() instanceof NotFoundHTTPResult);
    }

    _service.markAsUpdated(s1, "recordId2", "jobrun", "hash2");
    _service.markAsUpdated(s2, "recordId3", "jobrun", "hash3");
    assertEquals(2, _service.countEntries(s1, true));
    assertEquals(1, _service.countEntries(s2, true));

    _service.markAsUpdated(s1, "recordId4", "jobrun", "hash4");
    assertEquals(3, _service.countEntries(s1, true));
    assertEquals(1, _service.countEntries(s2, true));

    _service.clearSource(s1);
    try {
      _service.countEntries(s1, true);
    } catch (final Exception e) {
      assertTrue(e instanceof NotFoundHTTPResult || e.getCause() instanceof NotFoundHTTPResult);
    }
    assertEquals(1, _service.countEntries(s2, true));

    try {
      _service.clearAll();
      assertEquals(0, _service.countEntries(s2, true));
    } catch (final Exception e) {
      assertTrue(e instanceof NotFoundHTTPResult || e.getCause() instanceof NotFoundHTTPResult);
    }
  }

  /** test {@link DeltaService#getShardPrefixes(String)} method. */
  public void testGetShardPrefixes() throws Exception {
    assertTrue(_service.getShardPrefixes("s").isEmpty());
    _service.markAsUpdated("s", "record1", "j1", "h1");
    assertEquals(1, _service.getShardPrefixes("s").size());
    _service.markAsUpdated("s", "record2", "j1", "h1");
    assertEquals(2, _service.getShardPrefixes("s").size());
    _service.markAsUpdated("s", "record3", "j1", "h1");
    assertEquals(3, _service.getShardPrefixes("s").size());
    _service.clearSource("s");
    assertTrue(_service.getShardPrefixes("s").isEmpty());
  }

  /** test {@link DeltaService#getUnvisitedRecordIds(String, String). */
  public void testGetUnvisitedRecordIds() throws Exception {
    _service.markAsUpdated("s", "record1", "j1", "h1");
    assertUnvisitedRecordIds("s", "j2", "record1");
    _service.markAsUpdated("s", "record2", "j2", "h1");
    assertUnvisitedRecordIds("s", "j3", "record1", "record2");
    _service.markAsUpdated("s", "record3", "j3", "h1");
    assertUnvisitedRecordIds("s", "j3", "record1", "record2");
  }

  /** test {@link DeltaService#getUnvisitedEntries(String, String) with invalid input value.*/
  public void testGetUnvisitedRecordIdsUndefinedShardPrefix() throws Exception {
    assertTrue(_service.getUnvisitedEntries("no/such/shard", "no/such/jobrun").isEmpty());
  }

  /** test {@link DeltaService#deleteEntry(String, String)}. */
  public void testDeleteEntry() throws Exception {
    _service.markAsUpdated("s", "record1", "j1", "h1");
    assertEquals(State.UPTODATE, _service.checkState("s", "record1", "j1", "h1"));
    _service.deleteEntry("s", new DeltaService.EntryId("record1"));
    assertEquals(State.NEW, _service.checkState("s", "record1", "j1", "h1"));
    // delete again: should not throw exception
    _service.deleteEntry("s", new DeltaService.EntryId("record1")); 
    assertEquals(State.NEW, _service.checkState("s", "record1", "j1", "h1"));
    // delete in undefined source: should not throw exception
    _service.deleteEntry("s1", new DeltaService.EntryId("record1")); 
    assertEquals(State.NEW, _service.checkState("s", "record1", "j1", "h1"));
  }
  
  /** test updating single compound element entries. */
  public void testCompoundHandling() throws Exception {
    _service.markAsUpdated("s", "compound", "j1", "h1");
    _service.markAsUpdated("s", "e1", "compound", "j1", "h1");
    _service.markAsUpdated("s", "e2", "compound", "j1", "h1");
    _service.markAsUpdated("s", "e3", "compound", "j1", "h1");
    assertUnvisitedRecordIds("s", "j2", "compound", "e1", "e2", "e3");
    _service.markAsUpdated("s", "compound", "j2", "h2");
    assertUnvisitedRecordIds("s", "j2", "e1", "e2", "e3");
    _service.markAsUpdated("s", "e2", "compound", "j2", "h2");
    assertUnvisitedRecordIds("s", "j2", "e1", "e3");
    _service.deleteEntry("s", new DeltaService.EntryId("e1", "compound"));
    assertUnvisitedRecordIds("s", "j2", "e3");
  }
  
  /** test handling of an unchanged compound. */
  public void testUnchangedCompoundHandling() throws Exception {
    _service.markAsUpdated("s", "compound", "j1", "h1");
    _service.markAsUpdated("s", "e1", "compound", "j1", "h1");
    _service.markAsUpdated("s", "e2", "compound", "j1", "h1");
    _service.markAsUpdated("s", "e3", "compound", "j1", "h1");
    assertUnvisitedRecordIds("s", "j2", "compound", "e1", "e2", "e3");
    _service.checkState("s", "compound", "j2", "h1");
    _service.markCompoundElementsVisited("s", "compound", "j2");
    assertUnvisitedRecordIds("s", "j2");
    assertEquals(State.UPTODATE, _service.checkState("s", "e1", "compound", "j2", "h1"));
    assertEquals(State.UPTODATE, _service.checkState("s", "e2", "compound", "j2", "h1"));
    assertEquals(State.UPTODATE, _service.checkState("s", "e3", "compound", "j2", "h1"));
  }

  
  private void assertUnvisitedRecordIds(final String source, final String jobRunId,
    final String... expectedRecordIds) throws Exception {
    final Collection<String> recordIds = getUnvisitedRecordIds(source, jobRunId);
    assertEquals(expectedRecordIds.length, recordIds.size());
    final Iterator<String> iter = recordIds.iterator();
    for (final String expectedRecordId : expectedRecordIds) {
      assertEquals(expectedRecordId, iter.next());
    }
  }
  
  private SortedSet<String> getUnvisitedRecordIds(final String source, final String jobRunId) throws Exception {
    final SortedSet<String> records = new TreeSet<String>();
    final Collection<String> shards = _service.getShardPrefixes(source);
    for (final String shard : shards) {
      for (DeltaService.EntryId entry : _service.getUnvisitedEntries(shard, jobRunId)) {
        records.add(entry.getRecordId());
      }
    }
    return records;
  }
}