1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
/*-
* ============LICENSE_START=======================================================
* ONAP
* ================================================================================
* Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
* ============LICENSE_END=========================================================
*/
package org.onap.policy.controlloop.actorserviceprovider.topic;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import java.util.Arrays;
import java.util.Map;
import java.util.function.BiConsumer;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.common.utils.coder.CoderException;
import org.onap.policy.common.utils.coder.StandardCoder;
import org.onap.policy.common.utils.coder.StandardCoderObject;
@RunWith(MockitoJUnitRunner.class)
public class TopicListenerImplTest {
private static final StandardCoder coder = new StandardCoder();
private static final CommInfrastructure INFRA = CommInfrastructure.NOOP;
private static final String MY_TOPIC = "my-topic";
private static final String KEY1 = "requestId";
private static final String KEY2 = "container";
private static final String SUBKEY = "subRequestId";
private static final String VALUEA_REQID = "hello";
private static final String VALUEA_SUBREQID = "world";
private static final String VALUEB_REQID = "bye";
private Forwarder forwarder1;
private Forwarder forwarder2;
private TopicListenerImpl topic;
@Mock
private BiConsumer<String, StandardCoderObject> listener1;
@Mock
private BiConsumer<String, StandardCoderObject> listener1b;
@Mock
private BiConsumer<String, StandardCoderObject> listener2;
/**
* Sets up.
*/
@Before
public void setUp() {
topic = new TopicListenerImpl();
forwarder1 = topic.addForwarder(new SelectorKey(KEY1));
forwarder2 = topic.addForwarder(new SelectorKey(KEY1), new SelectorKey(KEY2, SUBKEY));
assertNotNull(forwarder1);
assertNotNull(forwarder2);
assertNotSame(forwarder1, forwarder2);
forwarder1.register(Arrays.asList(VALUEA_REQID), listener1);
forwarder1.register(Arrays.asList(VALUEB_REQID), listener1b);
forwarder2.register(Arrays.asList(VALUEA_REQID, VALUEA_SUBREQID), listener2);
}
@Test
public void testShutdown() {
// shut it down, which should clear all forwarders
topic.shutdown();
// should get a new forwarder now
Forwarder forwarder = topic.addForwarder(new SelectorKey(KEY1));
assertNotSame(forwarder1, forwarder);
assertNotSame(forwarder2, forwarder);
// new forwarder should be unchanged
assertSame(forwarder, topic.addForwarder(new SelectorKey(KEY1)));
}
@Test
public void testAddForwarder() {
assertSame(forwarder1, topic.addForwarder(new SelectorKey(KEY1)));
assertSame(forwarder2, topic.addForwarder(new SelectorKey(KEY1), new SelectorKey(KEY2, SUBKEY)));
}
@Test
public void testOnTopicEvent() {
/*
* send a message that should go to listener1 on forwarder1 and listener2 on
* forwarder2
*/
String msg = makeMessage(Map.of(KEY1, VALUEA_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
topic.onTopicEvent(INFRA, MY_TOPIC, msg);
verify(listener1).accept(eq(msg), any());
verify(listener2).accept(eq(msg), any());
// not to listener1b
verify(listener1b, never()).accept(any(), any());
/*
* now send a message that should only go to listener1b on forwarder1
*/
msg = makeMessage(Map.of(KEY1, VALUEB_REQID, KEY2, Map.of(SUBKEY, VALUEA_SUBREQID)));
topic.onTopicEvent(INFRA, MY_TOPIC, msg);
// should route to listener1 on forwarder1 and listener2 on forwarder2
verify(listener1b).accept(eq(msg), any());
// try one where the coder throws an exception
topic.onTopicEvent(INFRA, MY_TOPIC, "{invalid-json");
// no extra invocations
verify(listener1).accept(any(), any());
verify(listener1b).accept(any(), any());
verify(listener2).accept(any(), any());
}
/**
* Makes a message from a map.
*/
private String makeMessage(Map<String, Object> map) {
try {
return coder.encode(map);
} catch (CoderException e) {
throw new IllegalArgumentException(e);
}
}
}
|
