1 package org.opendaylight.controller.cluster.datastore;
3 import akka.actor.ActorRef;
4 import akka.actor.Props;
5 import akka.event.Logging;
6 import akka.testkit.JavaTestKit;
7 import junit.framework.Assert;
9 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
10 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
11 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
12 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
13 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
14 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
15 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
16 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
17 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
18 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
19 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
20 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
21 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeEvent;
22 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
23 import org.opendaylight.controller.protobuff.messages.transaction.ShardTransactionMessages.CreateTransactionReply;
24 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
25 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
27 import java.util.Collections;
28 import java.util.HashMap;
31 import static junit.framework.Assert.assertFalse;
32 import static org.junit.Assert.assertEquals;
33 import static org.junit.Assert.assertTrue;
35 public class ShardTest extends AbstractActorTest {
37 public void testOnReceiveCreateTransactionChain() throws Exception {
38 new JavaTestKit(getSystem()) {{
39 final ShardIdentifier identifier =
40 ShardIdentifier.builder().memberName("member-1")
41 .shardName("inventory").type("config").build();
43 final Props props = Shard.props(identifier, Collections.EMPTY_MAP, null);
44 final ActorRef subject =
45 getSystem().actorOf(props, "testCreateTransactionChain");
48 // Wait for a specific log message to show up
49 final boolean result =
50 new JavaTestKit.EventFilter<Boolean>(Logging.Info.class
52 protected Boolean run() {
55 }.from(subject.path().toString())
56 .message("Switching from state Candidate to Leader")
57 .occurrences(1).exec();
59 Assert.assertEquals(true, result);
61 new Within(duration("1 seconds")) {
62 protected void run() {
64 subject.tell(new CreateTransactionChain().toSerializable(), getRef());
66 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
67 // do not put code outside this method, will run afterwards
68 protected String match(Object in) {
69 if (in.getClass().equals(CreateTransactionChainReply.SERIALIZABLE_CLASS)){
70 CreateTransactionChainReply reply =
71 CreateTransactionChainReply.fromSerializable(getSystem(),in);
72 return reply.getTransactionChainPath()
78 }.get(); // this extracts the received message
80 assertEquals("Unexpected transaction path " + out,
81 "akka://test/user/testCreateTransactionChain/$a",
93 public void testOnReceiveRegisterListener() throws Exception {
94 new JavaTestKit(getSystem()) {{
95 final ShardIdentifier identifier =
96 ShardIdentifier.builder().memberName("member-1")
97 .shardName("inventory").type("config").build();
99 final Props props = Shard.props(identifier, Collections.EMPTY_MAP, null);
100 final ActorRef subject =
101 getSystem().actorOf(props, "testRegisterChangeListener");
103 new Within(duration("1 seconds")) {
104 protected void run() {
107 new UpdateSchemaContext(SchemaContextHelper.full()),
110 subject.tell(new RegisterChangeListener(TestModel.TEST_PATH,
111 getRef().path(), AsyncDataBroker.DataChangeScope.BASE),
114 final Boolean notificationEnabled = new ExpectMsg<Boolean>("enable notification") {
115 // do not put code outside this method, will run afterwards
116 protected Boolean match(Object in) {
117 if(in instanceof EnableNotification){
118 return ((EnableNotification) in).isEnabled();
123 }.get(); // this extracts the received message
125 assertFalse(notificationEnabled);
127 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
128 // do not put code outside this method, will run afterwards
129 protected String match(Object in) {
130 if (in.getClass().equals(RegisterChangeListenerReply.class)) {
131 RegisterChangeListenerReply reply =
132 (RegisterChangeListenerReply) in;
133 return reply.getListenerRegistrationPath()
139 }.get(); // this extracts the received message
141 assertTrue(out.matches(
142 "akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
151 public void testCreateTransaction(){
152 new JavaTestKit(getSystem()) {{
153 final ShardIdentifier identifier =
154 ShardIdentifier.builder().memberName("member-1")
155 .shardName("inventory").type("config").build();
157 final Props props = Shard.props(identifier, Collections.EMPTY_MAP, null);
158 final ActorRef subject =
159 getSystem().actorOf(props, "testCreateTransaction");
162 // Wait for a specific log message to show up
163 final boolean result =
164 new JavaTestKit.EventFilter<Boolean>(Logging.Info.class
166 protected Boolean run() {
169 }.from(subject.path().toString())
170 .message("Switching from state Candidate to Leader")
171 .occurrences(1).exec();
173 Assert.assertEquals(true, result);
175 new Within(duration("1 seconds")) {
176 protected void run() {
179 new UpdateSchemaContext(TestModel.createTestContext()),
182 subject.tell(new CreateTransaction("txn-1", TransactionProxy.TransactionType.READ_ONLY.ordinal() ).toSerializable(),
185 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
186 // do not put code outside this method, will run afterwards
187 protected String match(Object in) {
188 if (in instanceof CreateTransactionReply) {
189 CreateTransactionReply reply =
190 (CreateTransactionReply) in;
191 return reply.getTransactionActorPath()
197 }.get(); // this extracts the received message
199 assertTrue("Unexpected transaction path " + out,
200 out.contains("akka://test/user/testCreateTransaction/shard-txn-1"));
210 public void testPeerAddressResolved(){
211 new JavaTestKit(getSystem()) {{
212 Map<ShardIdentifier, String> peerAddresses = new HashMap<>();
214 final ShardIdentifier identifier =
215 ShardIdentifier.builder().memberName("member-1")
216 .shardName("inventory").type("config").build();
218 peerAddresses.put(identifier, null);
219 final Props props = Shard.props(identifier, peerAddresses, null);
220 final ActorRef subject =
221 getSystem().actorOf(props, "testPeerAddressResolved");
223 new Within(duration("1 seconds")) {
224 protected void run() {
227 new PeerAddressResolved(identifier, "akka://foobar"),
238 private AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> noOpDataChangeListener() {
239 return new AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>() {
241 public void onDataChanged(
242 AsyncDataChangeEvent<YangInstanceIdentifier, NormalizedNode<?, ?>> change) {