2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.remote.rpc;
12 import akka.actor.ActorRef;
13 import akka.japi.Pair;
14 import akka.testkit.JavaTestKit;
16 import com.google.common.collect.Lists;
17 import com.google.common.util.concurrent.Futures;
18 import static org.junit.Assert.assertEquals;
19 import org.junit.Test;
20 import org.mockito.ArgumentCaptor;
21 import org.opendaylight.controller.remote.rpc.messages.ExecuteRpc;
22 import org.opendaylight.controller.remote.rpc.messages.InvokeRpc;
23 import org.opendaylight.controller.remote.rpc.messages.RpcResponse;
24 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
25 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRouters;
26 import org.opendaylight.controller.sal.connector.api.RpcRouter.RouteIdentifier;
27 import org.opendaylight.controller.xml.codec.XmlUtils;
28 import org.opendaylight.yangtools.yang.common.QName;
29 import org.opendaylight.yangtools.yang.common.RpcError;
30 import org.opendaylight.yangtools.yang.common.RpcError.ErrorSeverity;
31 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
32 import org.opendaylight.yangtools.yang.common.RpcResult;
33 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
34 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
35 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
37 import java.net.URISyntaxException;
38 import java.util.Arrays;
39 import java.util.Collections;
40 import java.util.List;
42 import static org.mockito.Mockito.when;
43 import static org.mockito.Mockito.eq;
44 import static org.mockito.Mockito.any;
46 public class RpcBrokerTest extends AbstractRpcTest {
49 public void testInvokeRpcWithNoRemoteActor() throws Exception {
50 new JavaTestKit(node1) {{
51 CompositeNode input = makeRPCInput("foo");
53 InvokeRpc invokeMsg = new InvokeRpc(TEST_RPC, null, input);
54 rpcBroker1.tell(invokeMsg, getRef());
56 probeReg1.expectMsgClass(duration("5 seconds"), RpcRegistry.Messages.FindRouters.class);
57 probeReg1.reply(new RpcRegistry.Messages.FindRoutersReply(
58 Collections.<Pair<ActorRef, Long>>emptyList()));
60 akka.actor.Status.Failure failure = expectMsgClass(duration("5 seconds"),
61 akka.actor.Status.Failure.class);
63 assertEquals("failure.cause()", RpcErrorsException.class, failure.cause().getClass());
69 * This test method invokes and executes the remote rpc
72 public void testInvokeRpc() throws URISyntaxException {
73 new JavaTestKit(node1) {{
74 QName instanceQName = new QName(new URI("ns"), "instance");
76 CompositeNode invokeRpcResult = makeRPCOutput("bar");
77 RpcResult<CompositeNode> rpcResult =
78 RpcResultBuilder.<CompositeNode>success(invokeRpcResult).build();
79 ArgumentCaptor<CompositeNode> inputCaptor = new ArgumentCaptor<>();
80 when(brokerSession.rpc(eq(TEST_RPC), inputCaptor.capture()))
81 .thenReturn(Futures.immediateFuture(rpcResult));
84 CompositeNode input = makeRPCInput("foo");
85 YangInstanceIdentifier instanceID = YangInstanceIdentifier.of(instanceQName);
86 InvokeRpc invokeMsg = new InvokeRpc(TEST_RPC, instanceID, input);
87 rpcBroker1.tell(invokeMsg, getRef());
89 FindRouters findRouters = probeReg1.expectMsgClass(RpcRegistry.Messages.FindRouters.class);
90 RouteIdentifier<?, ?, ?> routeIdentifier = findRouters.getRouteIdentifier();
91 assertEquals("getType", TEST_RPC, routeIdentifier.getType());
92 assertEquals("getRoute", instanceID, routeIdentifier.getRoute());
94 probeReg1.reply(new RpcRegistry.Messages.FindRoutersReply(
95 Arrays.asList(new Pair<ActorRef, Long>(rpcBroker2, 200L))));
97 RpcResponse rpcResponse = expectMsgClass(duration("5 seconds"), RpcResponse.class);
98 assertCompositeNodeEquals((CompositeNode)invokeRpcResult.getValue().get(0),
99 XmlUtils.xmlToCompositeNode(rpcResponse.getResultCompositeNode()));
100 assertCompositeNodeEquals(input, inputCaptor.getValue());
105 public void testInvokeRpcWithNoOutput() {
106 new JavaTestKit(node1) {{
108 RpcResult<CompositeNode> rpcResult = RpcResultBuilder.<CompositeNode>success().build();
109 when(brokerSession.rpc(eq(TEST_RPC), any(CompositeNode.class)))
110 .thenReturn(Futures.immediateFuture(rpcResult));
112 InvokeRpc invokeMsg = new InvokeRpc(TEST_RPC, null, makeRPCInput("foo"));
113 rpcBroker1.tell(invokeMsg, getRef());
115 probeReg1.expectMsgClass(RpcRegistry.Messages.FindRouters.class);
116 probeReg1.reply(new RpcRegistry.Messages.FindRoutersReply(
117 Arrays.asList(new Pair<ActorRef, Long>(rpcBroker2, 200L))));
119 RpcResponse rpcResponse = expectMsgClass(duration("5 seconds"), RpcResponse.class);
121 assertEquals("getResultCompositeNode", "", rpcResponse.getResultCompositeNode());
126 public void testInvokeRpcWithExecuteFailure() {
127 new JavaTestKit(node1) {{
129 RpcResult<CompositeNode> rpcResult = RpcResultBuilder.<CompositeNode>failed()
130 .withError(ErrorType.RPC, "tag", "error", "appTag", "info",
131 new Exception("mock"))
133 when(brokerSession.rpc(eq(TEST_RPC), any(CompositeNode.class)))
134 .thenReturn(Futures.immediateFuture(rpcResult));
136 InvokeRpc invokeMsg = new InvokeRpc(TEST_RPC, null, makeRPCInput("foo"));
137 rpcBroker1.tell(invokeMsg, getRef());
139 probeReg1.expectMsgClass(RpcRegistry.Messages.FindRouters.class);
140 probeReg1.reply(new RpcRegistry.Messages.FindRoutersReply(
141 Arrays.asList(new Pair<ActorRef, Long>(rpcBroker2, 200L))));
143 akka.actor.Status.Failure failure = expectMsgClass(duration("5 seconds"),
144 akka.actor.Status.Failure.class);
146 assertEquals("failure.cause()", RpcErrorsException.class, failure.cause().getClass());
148 RpcErrorsException errorsEx = (RpcErrorsException)failure.cause();
149 List<RpcError> rpcErrors = Lists.newArrayList(errorsEx.getRpcErrors());
150 assertEquals("RpcErrors count", 1, rpcErrors.size());
151 assertRpcErrorEquals(rpcErrors.get(0), ErrorSeverity.ERROR, ErrorType.RPC, "tag",
152 "error", "appTag", "info", "mock");
157 public void testInvokeRpcWithFindRoutersFailure() {
158 new JavaTestKit(node1) {{
160 InvokeRpc invokeMsg = new InvokeRpc(TEST_RPC, null, makeRPCInput("foo"));
161 rpcBroker1.tell(invokeMsg, getRef());
163 probeReg1.expectMsgClass(RpcRegistry.Messages.FindRouters.class);
164 probeReg1.reply(new akka.actor.Status.Failure(new TestException()));
166 akka.actor.Status.Failure failure = expectMsgClass(duration("5 seconds"),
167 akka.actor.Status.Failure.class);
169 assertEquals("failure.cause()", TestException.class, failure.cause().getClass());
174 public void testExecuteRpc() {
175 new JavaTestKit(node1) {{
177 String xml = "<input xmlns=\"urn:test\"><input-data>foo</input-data></input>";
179 CompositeNode invokeRpcResult = makeRPCOutput("bar");
180 RpcResult<CompositeNode> rpcResult =
181 RpcResultBuilder.<CompositeNode>success(invokeRpcResult).build();
182 ArgumentCaptor<CompositeNode> inputCaptor = new ArgumentCaptor<>();
183 when(brokerSession.rpc(eq(TEST_RPC), inputCaptor.capture()))
184 .thenReturn(Futures.immediateFuture(rpcResult));
186 ExecuteRpc executeMsg = new ExecuteRpc(xml, TEST_RPC);
188 rpcBroker1.tell(executeMsg, getRef());
190 RpcResponse rpcResponse = expectMsgClass(duration("5 seconds"), RpcResponse.class);
192 assertCompositeNodeEquals((CompositeNode)invokeRpcResult.getValue().get(0),
193 XmlUtils.xmlToCompositeNode(rpcResponse.getResultCompositeNode()));
198 public void testExecuteRpcFailureWithRpcErrors() {
199 new JavaTestKit(node1) {{
201 String xml = "<input xmlns=\"urn:test\"><input-data>foo</input-data></input>";
203 RpcResult<CompositeNode> rpcResult = RpcResultBuilder.<CompositeNode>failed()
204 .withError(ErrorType.RPC, "tag1", "error", "appTag1", "info1",
205 new Exception("mock"))
206 .withWarning(ErrorType.PROTOCOL, "tag2", "warning", "appTag2", "info2", null)
208 when(brokerSession.rpc(eq(TEST_RPC), any(CompositeNode.class)))
209 .thenReturn(Futures.immediateFuture(rpcResult));
211 ExecuteRpc executeMsg = new ExecuteRpc(xml, TEST_RPC);
213 rpcBroker1.tell(executeMsg, getRef());
215 akka.actor.Status.Failure failure = expectMsgClass(duration("5 seconds"),
216 akka.actor.Status.Failure.class);
218 assertEquals("failure.cause()", RpcErrorsException.class, failure.cause().getClass());
220 RpcErrorsException errorsEx = (RpcErrorsException)failure.cause();
221 List<RpcError> rpcErrors = Lists.newArrayList(errorsEx.getRpcErrors());
222 assertEquals("RpcErrors count", 2, rpcErrors.size());
223 assertRpcErrorEquals(rpcErrors.get(0), ErrorSeverity.ERROR, ErrorType.RPC, "tag1",
224 "error", "appTag1", "info1", "mock");
225 assertRpcErrorEquals(rpcErrors.get(1), ErrorSeverity.WARNING, ErrorType.PROTOCOL, "tag2",
226 "warning", "appTag2", "info2", null);
231 public void testExecuteRpcFailureWithNoRpcErrors() {
232 new JavaTestKit(node1) {{
234 String xml = "<input xmlns=\"urn:test\"><input-data>foo</input-data></input>";
236 RpcResult<CompositeNode> rpcResult = RpcResultBuilder.<CompositeNode>failed().build();
237 when(brokerSession.rpc(eq(TEST_RPC), any(CompositeNode.class)))
238 .thenReturn(Futures.immediateFuture(rpcResult));
240 ExecuteRpc executeMsg = new ExecuteRpc(xml, TEST_RPC);
242 rpcBroker1.tell(executeMsg, getRef());
244 akka.actor.Status.Failure failure = expectMsgClass(duration("5 seconds"),
245 akka.actor.Status.Failure.class);
247 assertEquals("failure.cause()", RpcErrorsException.class, failure.cause().getClass());
249 RpcErrorsException errorsEx = (RpcErrorsException)failure.cause();
250 List<RpcError> rpcErrors = Lists.newArrayList(errorsEx.getRpcErrors());
251 assertEquals("RpcErrors count", 1, rpcErrors.size());
252 assertRpcErrorEquals(rpcErrors.get(0), ErrorSeverity.ERROR, ErrorType.RPC,
253 "operation-failed", "failed", null, null, null);
258 public void testExecuteRpcFailureWithException() {
259 new JavaTestKit(node1) {{
261 String xml = "<input xmlns=\"urn:test\"><input-data>foo</input-data></input>";
263 when(brokerSession.rpc(eq(TEST_RPC), any(CompositeNode.class)))
264 .thenReturn(Futures.<RpcResult<CompositeNode>>immediateFailedFuture(
265 new TestException()));
267 ExecuteRpc executeMsg = new ExecuteRpc(xml, TEST_RPC);
269 rpcBroker1.tell(executeMsg, getRef());
271 akka.actor.Status.Failure failure = expectMsgClass(duration("5 seconds"),
272 akka.actor.Status.Failure.class);
274 assertEquals("failure.cause()", TestException.class, failure.cause().getClass());