+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.verify;
+
+public class ShardTest extends AbstractActorTest {
+
+ private static final DatastoreContext DATA_STORE_CONTEXT =
+ new DatastoreContext("", null, Duration.create(10, TimeUnit.MINUTES), 5, 3, 5000, 500);
+
+ private static final SchemaContext SCHEMA_CONTEXT = TestModel.createTestContext();
+
+ private static final ShardIdentifier IDENTIFIER = ShardIdentifier.builder().memberName("member-1")
+ .shardName("inventory").type("config").build();
+
+ @Before
+ public void setUp() {
+ System.setProperty("shard.persistent", "false");
+
+ InMemorySnapshotStore.clear();
+ InMemoryJournal.clear();
+ }
+
+ @After
+ public void tearDown() {
+ InMemorySnapshotStore.clear();
+ InMemoryJournal.clear();
+ }
+
+ private Props newShardProps() {
+ return Shard.props(IDENTIFIER, Collections.<ShardIdentifier,String>emptyMap(),
+ DATA_STORE_CONTEXT, SCHEMA_CONTEXT);
+ }
+
+ @Test
+ public void testOnReceiveRegisterListener() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ ActorRef subject = getSystem().actorOf(newShardProps(), "testRegisterChangeListener");
+
+ subject.tell(new UpdateSchemaContext(SchemaContextHelper.full()), getRef());
+
+ subject.tell(new RegisterChangeListener(TestModel.TEST_PATH,
+ getRef().path(), AsyncDataBroker.DataChangeScope.BASE), getRef());
+
+ EnableNotification enable = expectMsgClass(duration("3 seconds"), EnableNotification.class);
+ assertEquals("isEnabled", false, enable.isEnabled());
+
+ RegisterChangeListenerReply reply = expectMsgClass(duration("3 seconds"),
+ RegisterChangeListenerReply.class);
+ assertTrue(reply.getListenerRegistrationPath().toString().matches(
+ "akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
+ }};
+ }
+
+ @Test
+ public void testCreateTransaction(){
+ new ShardTestKit(getSystem()) {{
+ ActorRef subject = getSystem().actorOf(newShardProps(), "testCreateTransaction");
+
+ waitUntilLeader(subject);
+
+ subject.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
+
+ subject.tell(new CreateTransaction("txn-1",
+ TransactionProxy.TransactionType.READ_ONLY.ordinal() ).toSerializable(), getRef());
+
+ CreateTransactionReply reply = expectMsgClass(duration("3 seconds"),
+ CreateTransactionReply.class);
+
+ String path = reply.getTransactionActorPath().toString();
+ assertTrue("Unexpected transaction path " + path,
+ path.contains("akka://test/user/testCreateTransaction/shard-txn-1"));
+ expectNoMsg();
+ }};
+ }
+
+ @Test
+ public void testCreateTransactionOnChain(){
+ new ShardTestKit(getSystem()) {{
+ final ActorRef subject = getSystem().actorOf(newShardProps(), "testCreateTransactionOnChain");
+
+ waitUntilLeader(subject);
+
+ subject.tell(new CreateTransaction("txn-1",
+ TransactionProxy.TransactionType.READ_ONLY.ordinal() , "foobar").toSerializable(),
+ getRef());
+
+ CreateTransactionReply reply = expectMsgClass(duration("3 seconds"),
+ CreateTransactionReply.class);
+
+ String path = reply.getTransactionActorPath().toString();
+ assertTrue("Unexpected transaction path " + path,
+ path.contains("akka://test/user/testCreateTransactionOnChain/shard-txn-1"));
+ expectNoMsg();
+ }};
+ }
+
+ @Test
+ public void testPeerAddressResolved(){
+ new JavaTestKit(getSystem()) {{
+ final ShardIdentifier identifier =
+ ShardIdentifier.builder().memberName("member-1")
+ .shardName("inventory").type("config").build();
+
+ Props props = Shard.props(identifier,
+ Collections.<ShardIdentifier, String>singletonMap(identifier, null),
+ DATA_STORE_CONTEXT, SCHEMA_CONTEXT);
+ final ActorRef subject = getSystem().actorOf(props, "testPeerAddressResolved");
+
+ new Within(duration("3 seconds")) {
+ @Override
+ protected void run() {
+
+ subject.tell(
+ new PeerAddressResolved(identifier, "akka://foobar"),
+ getRef());
+
+ expectNoMsg();
+ }
+ };
+ }};
+ }
+
+ @Test
+ public void testApplySnapshot() throws ExecutionException, InterruptedException {
+ TestActorRef<Shard> ref = TestActorRef.create(getSystem(), newShardProps());
+
+ NormalizedNodeToNodeCodec codec =
+ new NormalizedNodeToNodeCodec(SCHEMA_CONTEXT);
+
+ ref.underlyingActor().writeToStore(TestModel.TEST_PATH, ImmutableNodes.containerNode(
+ TestModel.TEST_QNAME));
+
+ YangInstanceIdentifier root = YangInstanceIdentifier.builder().build();
+ NormalizedNode<?,?> expected = ref.underlyingActor().readStore(root);
+
+ NormalizedNodeMessages.Container encode = codec.encode(root, expected);
+
+ ApplySnapshot applySnapshot = new ApplySnapshot(Snapshot.create(
+ encode.getNormalizedNode().toByteString().toByteArray(),
+ Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4));
+
+ ref.underlyingActor().onReceiveCommand(applySnapshot);
+
+ NormalizedNode<?,?> actual = ref.underlyingActor().readStore(root);
+
+ assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testApplyState() throws Exception {
+
+ TestActorRef<Shard> shard = TestActorRef.create(getSystem(), newShardProps());
+
+ NormalizedNode<?, ?> node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
+
+ MutableCompositeModification compMod = new MutableCompositeModification();
+ compMod.addModification(new WriteModification(TestModel.TEST_PATH, node, SCHEMA_CONTEXT));
+ Payload payload = new CompositeModificationPayload(compMod.toSerializable());
+ ApplyState applyState = new ApplyState(null, "test",
+ new ReplicatedLogImplEntry(1, 2, payload));
+
+ shard.underlyingActor().onReceiveCommand(applyState);
+
+ NormalizedNode<?,?> actual = shard.underlyingActor().readStore(TestModel.TEST_PATH);
+ assertEquals("Applied state", node, actual);
+ }
+
+ @SuppressWarnings("serial")
+ @Test
+ public void testRecovery() throws Exception {
+
+ // Set up the InMemorySnapshotStore.
+
+ InMemoryDOMDataStore testStore = InMemoryDOMDataStoreFactory.create("Test", null, null);
+ testStore.onGlobalContextUpdated(SCHEMA_CONTEXT);
+
+ DOMStoreWriteTransaction writeTx = testStore.newWriteOnlyTransaction();
+ writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+ DOMStoreThreePhaseCommitCohort commitCohort = writeTx.ready();
+ commitCohort.preCommit().get();
+ commitCohort.commit().get();
+
+ DOMStoreReadTransaction readTx = testStore.newReadOnlyTransaction();
+ NormalizedNode<?, ?> root = readTx.read(YangInstanceIdentifier.builder().build()).get().get();
+
+ InMemorySnapshotStore.addSnapshot(IDENTIFIER.toString(), Snapshot.create(
+ new NormalizedNodeToNodeCodec(SCHEMA_CONTEXT).encode(
+ YangInstanceIdentifier.builder().build(), root).
+ getNormalizedNode().toByteString().toByteArray(),
+ Collections.<ReplicatedLogEntry>emptyList(), 0, 1, -1, -1));
+
+ // Set up the InMemoryJournal.