+ assertTrue(compositeModification.getModifications().size() == 1);
+ assertEquals(modificationType, compositeModification.getModifications().get(0).getClass());
+ }};
+ }
+
+ @Test
+ public void testOnReceiveWriteData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testOnReceiveWriteData");
+
+ transaction.tell(new WriteData(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME)).toSerializable(
+ DataStoreVersions.HELIUM_2_VERSION), getRef());
+
+ expectMsgClass(duration("5 seconds"), ShardTransactionMessages.WriteDataReply.class);
+
+ assertModification(transaction, WriteModification.class);
+
+ // unserialized write
+ transaction.tell(new WriteData(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
+ getRef());
+
+ expectMsgClass(duration("5 seconds"), WriteDataReply.class);
+ }};
+ }
+
+ @Test
+ public void testOnReceiveHeliumR1WriteData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testOnReceiveHeliumR1WriteData", DataStoreVersions.HELIUM_1_VERSION);
+
+ Encoded encoded = new NormalizedNodeToNodeCodec(null).encode(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+ ShardTransactionMessages.WriteData serialized = ShardTransactionMessages.WriteData.newBuilder()
+ .setInstanceIdentifierPathArguments(encoded.getEncodedPath())
+ .setNormalizedNode(encoded.getEncodedNode().getNormalizedNode()).build();
+
+ transaction.tell(serialized, getRef());
+
+ expectMsgClass(duration("5 seconds"), ShardTransactionMessages.WriteDataReply.class);
+
+ assertModification(transaction, WriteModification.class);
+ }};
+ }
+
+ @Test
+ public void testOnReceiveMergeData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testMergeData");
+
+ transaction.tell(new MergeData(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME)).toSerializable(
+ DataStoreVersions.HELIUM_2_VERSION), getRef());
+
+ expectMsgClass(duration("5 seconds"), ShardTransactionMessages.MergeDataReply.class);
+
+ assertModification(transaction, MergeModification.class);
+
+ //unserialized merge
+ transaction.tell(new MergeData(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
+ getRef());
+
+ expectMsgClass(duration("5 seconds"), MergeDataReply.class);
+ }};
+ }
+
+ @Test
+ public void testOnReceiveHeliumR1MergeData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testOnReceiveHeliumR1MergeData", DataStoreVersions.HELIUM_1_VERSION);
+
+ Encoded encoded = new NormalizedNodeToNodeCodec(null).encode(TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+ ShardTransactionMessages.MergeData serialized = ShardTransactionMessages.MergeData.newBuilder()
+ .setInstanceIdentifierPathArguments(encoded.getEncodedPath())
+ .setNormalizedNode(encoded.getEncodedNode().getNormalizedNode()).build();
+
+ transaction.tell(serialized, getRef());
+
+ expectMsgClass(duration("5 seconds"), ShardTransactionMessages.MergeDataReply.class);
+
+ assertModification(transaction, MergeModification.class);
+ }};
+ }
+
+ @Test
+ public void testOnReceiveDeleteData() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testDeleteData");
+
+ transaction.tell(new DeleteData(TestModel.TEST_PATH).toSerializable(
+ DataStoreVersions.HELIUM_2_VERSION), getRef());
+
+ expectMsgClass(duration("5 seconds"), ShardTransactionMessages.DeleteDataReply.class);
+
+ assertModification(transaction, DeleteModification.class);
+
+ //unserialized
+ transaction.tell(new DeleteData(TestModel.TEST_PATH), getRef());
+
+ expectMsgClass(duration("5 seconds"), DeleteDataReply.class);
+ }};
+ }
+
+
+ @Test
+ public void testOnReceiveReadyTransaction() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testReadyTransaction");
+
+ watch(transaction);
+
+ transaction.tell(new ReadyTransaction().toSerializable(), getRef());
+
+ expectMsgAnyClassOf(duration("5 seconds"), ReadyTransactionReply.SERIALIZABLE_CLASS,
+ Terminated.class);
+ expectMsgAnyClassOf(duration("5 seconds"), ReadyTransactionReply.SERIALIZABLE_CLASS,
+ Terminated.class);
+ }};
+
+ // test
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testReadyTransaction2");
+
+ watch(transaction);
+
+ transaction.tell(new ReadyTransaction(), getRef());
+
+ expectMsgAnyClassOf(duration("5 seconds"), ReadyTransactionReply.class,
+ Terminated.class);
+ expectMsgAnyClassOf(duration("5 seconds"), ReadyTransactionReply.class,
+ Terminated.class);
+ }};
+ }
+
+ @Test
+ public void testOnReceiveCreateSnapshot() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ ShardTest.writeToStore(store, TestModel.TEST_PATH,
+ ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+ NormalizedNode<?,?> expectedRoot = ShardTest.readStore(store,
+ YangInstanceIdentifier.builder().build());
+
+ final ActorRef transaction = newTransactionActor(store.newReadOnlyTransaction(),
+ "testOnReceiveCreateSnapshot");
+
+ watch(transaction);
+
+ transaction.tell(CreateSnapshot.INSTANCE, getRef());
+
+ CaptureSnapshotReply reply = expectMsgClass(duration("3 seconds"), CaptureSnapshotReply.class);
+
+ assertNotNull("getSnapshot is null", reply.getSnapshot());
+
+ NormalizedNode<?,?> actualRoot = SerializationUtils.deserializeNormalizedNode(
+ reply.getSnapshot());
+
+ assertEquals("Root node", expectedRoot, actualRoot);
+
+ expectTerminated(duration("3 seconds"), transaction);
+ }};
+ }
+
+ @Test
+ public void testReadWriteTxOnReceiveCloseTransaction() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newReadWriteTransaction(),
+ "testReadWriteTxOnReceiveCloseTransaction");
+
+ watch(transaction);
+
+ transaction.tell(new CloseTransaction().toSerializable(), getRef());
+
+ expectMsgClass(duration("3 seconds"), CloseTransactionReply.SERIALIZABLE_CLASS);
+ expectTerminated(duration("3 seconds"), transaction);
+ }};
+ }
+
+ @Test
+ public void testWriteOnlyTxOnReceiveCloseTransaction() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newWriteOnlyTransaction(),
+ "testWriteTxOnReceiveCloseTransaction");
+
+ watch(transaction);
+
+ transaction.tell(new CloseTransaction().toSerializable(), getRef());
+
+ expectMsgClass(duration("3 seconds"), CloseTransactionReply.SERIALIZABLE_CLASS);
+ expectTerminated(duration("3 seconds"), transaction);
+ }};
+ }
+
+ @Test
+ public void testReadOnlyTxOnReceiveCloseTransaction() throws Exception {
+ new JavaTestKit(getSystem()) {{
+ final ActorRef transaction = newTransactionActor(store.newReadOnlyTransaction(),
+ "testReadOnlyTxOnReceiveCloseTransaction");
+
+ watch(transaction);
+
+ transaction.tell(new CloseTransaction().toSerializable(), getRef());
+
+ expectMsgClass(duration("3 seconds"), Terminated.class);
+ }};
+ }
+
+ @Test(expected=UnknownMessageException.class)
+ public void testNegativePerformingWriteOperationOnReadTransaction() throws Exception {
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn",
+ DataStoreVersions.CURRENT_VERSION);
+ final TestActorRef<ShardTransaction> transaction = TestActorRef.apply(props,getSystem());
+
+ transaction.receive(new DeleteData(TestModel.TEST_PATH).toSerializable(
+ DataStoreVersions.CURRENT_VERSION), ActorRef.noSender());
+ }