Skip to content

Commit 15d47e1

Browse files
MikeEdgarMikeEdgar
committed
Add test cases dealing with group configurations
Signed-off-by: Michael Edgar <medgar@redhat.com>
1 parent bf933aa commit 15d47e1

File tree

5 files changed

+244
-34
lines changed

5 files changed

+244
-34
lines changed

‎api/src/main/java/com/github/streamshub/console/api/model/Group.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,8 +94,8 @@ STATE, nullsLast(comparing(Group::state)),
9494
", " + STATE +
9595
", " + MEMBERS +
9696
", " + COORDINATOR +
97-
", " + OFFSETS
98-
+ ", " + SIMPLE_CONSUMER_GROUP;
97+
", " + OFFSETS +
98+
", " + SIMPLE_CONSUMER_GROUP;
9999

100100
private Fields() {
101101
// Prevent instances

‎api/src/main/java/com/github/streamshub/console/api/service/GroupService.java‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -915,27 +915,27 @@ private void addOffsets(Group group,
915915
}
916916

917917
CompletableFuture<Void> maybeDescribeConfigs(Admin adminClient, Map<String, Group> groups, List<String> fields) {
918-
if (fields.contains(Group.Fields.CONFIGS)) {
919-
List<ConfigResource> keys = groups.values()
920-
.stream()
921-
.map(Group::groupId)
922-
.filter(groupId -> {
923-
if (permissionService.permitted(ResourceTypes.Kafka.GROUP_CONFIGS, Privilege.GET, groupId)) {
924-
return true;
925-
}
926-
var error = permissionService.forbidden(ResourceTypes.Kafka.GROUP_CONFIGS, Privilege.GET, groupId);
927-
groups.get(groupId).addConfigs(Either.ofAlternate(error));
928-
return false;
929-
})
930-
.map(groupId -> new ConfigResource(ConfigResource.Type.GROUP, groupId))
931-
.toList();
932-
933-
return configService.describeConfigs(adminClient, keys)
934-
.thenAccept(configs ->
935-
configs.forEach((groupId, either) -> groups.get(groupId).addConfigs(either)))
936-
.toCompletableFuture();
918+
if (!fields.contains(Group.Fields.CONFIGS)) {
919+
return CompletableFuture.completedFuture(null);
937920
}
938921

939-
return CompletableFuture.completedFuture(null);
922+
List<ConfigResource> keys = groups.values()
923+
.stream()
924+
.map(Group::groupId)
925+
.filter(groupId -> {
926+
if (permissionService.permitted(ResourceTypes.Kafka.GROUP_CONFIGS, Privilege.GET, groupId)) {
927+
return true;
928+
}
929+
var error = permissionService.forbidden(ResourceTypes.Kafka.GROUP_CONFIGS, Privilege.GET, groupId);
930+
groups.get(groupId).addConfigs(Either.ofAlternate(error));
931+
return false;
932+
})
933+
.map(groupId -> new ConfigResource(ConfigResource.Type.GROUP, groupId))
934+
.toList();
935+
936+
return configService.describeConfigs(adminClient, keys)
937+
.thenAccept(configs ->
938+
configs.forEach((groupId, either) -> groups.get(groupId).addConfigs(either)))
939+
.toCompletableFuture();
940940
}
941941
}

‎api/src/test/java/com/github/streamshub/console/api/GroupsResourceIT.java‎

Lines changed: 137 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.kafka.clients.admin.ListOffsetsResult;
3030
import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
3131
import org.apache.kafka.clients.admin.ListShareGroupOffsetsResult;
32+
import org.apache.kafka.clients.admin.ListStreamsGroupOffsetsResult;
3233
import org.apache.kafka.clients.admin.OffsetSpec;
3334
import org.apache.kafka.clients.admin.SharePartitionOffsetInfo;
3435
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
@@ -53,12 +54,17 @@
5354
import org.skyscreamer.jsonassert.JSONAssert;
5455
import org.skyscreamer.jsonassert.JSONCompareMode;
5556

57+
import com.github.streamshub.console.api.model.Group;
5658
import com.github.streamshub.console.api.support.Holder;
5759
import com.github.streamshub.console.api.support.Promises;
5860
import com.github.streamshub.console.config.ConsoleConfig;
61+
import com.github.streamshub.console.config.security.GlobalSecurityConfigBuilder;
62+
import com.github.streamshub.console.config.security.KafkaSecurityConfigBuilder;
63+
import com.github.streamshub.console.config.security.Privilege;
5964
import com.github.streamshub.console.kafka.systemtest.TestPlainProfile;
6065
import com.github.streamshub.console.kafka.systemtest.utils.ConsumerUtils;
6166
import com.github.streamshub.console.kafka.systemtest.utils.ConsumerUtils.ConsumerType;
67+
import com.github.streamshub.console.kafka.systemtest.utils.TokenUtils;
6268
import com.github.streamshub.console.support.Identifiers;
6369
import com.github.streamshub.console.test.AdminClientSpy;
6470
import com.github.streamshub.console.test.TestHelper;
@@ -71,10 +77,12 @@
7177
import io.quarkus.test.junit.TestProfile;
7278
import io.strimzi.api.kafka.model.kafka.Kafka;
7379

80+
import static com.github.streamshub.console.test.EveryEntry.everyEntry;
7481
import static com.github.streamshub.console.test.TestHelper.whenRequesting;
7582
import static java.util.regex.Pattern.compile;
7683
import static org.awaitility.Awaitility.await;
7784
import static org.hamcrest.Matchers.allOf;
85+
import static org.hamcrest.Matchers.anEmptyMap;
7886
import static org.hamcrest.Matchers.contains;
7987
import static org.hamcrest.Matchers.containsInAnyOrder;
8088
import static org.hamcrest.Matchers.everyItem;
@@ -84,6 +92,7 @@
8492
import static org.hamcrest.Matchers.hasSize;
8593
import static org.hamcrest.Matchers.is;
8694
import static org.hamcrest.Matchers.matchesPattern;
95+
import static org.hamcrest.Matchers.not;
8796
import static org.hamcrest.Matchers.notNullValue;
8897
import static org.hamcrest.Matchers.nullValue;
8998
import static org.hamcrest.Matchers.startsWith;
@@ -92,6 +101,7 @@
92101
import static org.mockito.ArgumentMatchers.anyCollection;
93102
import static org.mockito.ArgumentMatchers.anyMap;
94103
import static org.mockito.Mockito.doAnswer;
104+
import static org.mockito.Mockito.when;
95105

96106
@QuarkusTest
97107
@TestHTTPEndpoint(GroupsResource.class)
@@ -451,7 +461,7 @@ void testListConsumerGroupsWithDescribeError() {
451461
"SHARE, ", // startOffset for share groups is not reliable so we just check for a numeric value
452462
"STREAMS, 5"
453463
})
454-
void testDescribeGroupDefault(ConsumerType consumerType, Integer expectedOffset) {
464+
void testDescribeGroupExtendedFields(ConsumerType consumerType, Integer expectedOffset) {
455465
String topic1 = "t1-" + consumerType.name() + "-" + UUID.randomUUID().toString();
456466
String group1 = "g1-" + consumerType.name() + "-" + UUID.randomUUID().toString();
457467

@@ -467,7 +477,9 @@ void testDescribeGroupDefault(ConsumerType consumerType, Integer expectedOffset)
467477
.autoClose(false)
468478
.consume()) {
469479

470-
whenRequesting(req -> req.get("{groupId}", clusterId1, Identifiers.encode(group1)))
480+
whenRequesting(req -> req
481+
.param("fields[groups]", Group.Fields.DESCRIBE_DEFAULT + "," + Group.Fields.CONFIGS)
482+
.get("{groupId}", clusterId1, Identifiers.encode(group1)))
471483
.assertThat()
472484
.statusCode(is(Status.OK.getStatusCode()))
473485
.body("data.attributes.groupId", is(group1))
@@ -485,7 +497,17 @@ void testDescribeGroupDefault(ConsumerType consumerType, Integer expectedOffset)
485497
.body("data.attributes.offsets[0].offset", offsetMatcher)
486498
.body("data.attributes.offsets[1].topicName", is(topic1))
487499
.body("data.attributes.offsets[1].partition", is(1))
488-
.body("data.attributes.offsets[1].offset", offsetMatcher);
500+
.body("data.attributes.offsets[1].offset", offsetMatcher)
501+
.body("data.attributes.configs", not(anEmptyMap()))
502+
.body("data.attributes.configs", everyEntry(
503+
Matchers.any(String.class), // any string key
504+
// all entries have same keys
505+
allOf(
506+
hasKey("value"),
507+
hasKey("source"),
508+
hasKey("sensitive"),
509+
hasKey("readOnly"),
510+
hasKey("type"))));
489511
}
490512
}
491513

@@ -585,6 +607,42 @@ void testDescribeShareGroupWithFetchGroupOffsetsError() throws Exception {
585607
}
586608
}
587609

610+
@Test
611+
void testDescribeStreamsGroupWithFetchTopicOffsetsError() throws Exception {
612+
String topic1 = "t1-" + UUID.randomUUID().toString();
613+
String group1 = "g1-" + UUID.randomUUID().toString();
614+
String group1Id = Identifiers.encode(group1);
615+
String client1 = "c1-" + UUID.randomUUID().toString();
616+
617+
Answer<ListStreamsGroupOffsetsResult> listOffsetsFailed = args -> {
618+
KafkaFutureImpl<Map<TopicPartition, OffsetAndMetadata>> failure = new KafkaFutureImpl<>();
619+
failure.completeExceptionally(new ApiException("EXPECTED TEST EXCEPTION"));
620+
621+
ListStreamsGroupOffsetsResult result = Mockito.mock(ListStreamsGroupOffsetsResult.class);
622+
when(result.partitionsToOffsetAndMetadata(group1)).thenReturn(failure);
623+
return result;
624+
};
625+
626+
AdminClientSpy.install(adminClient -> {
627+
// Mock listOffsets
628+
doAnswer(listOffsetsFailed)
629+
.when(adminClient)
630+
.listStreamsGroupOffsets(anyMap());
631+
});
632+
633+
try (var consumer = groupUtils.consume(ConsumerType.STREAMS, group1, topic1, client1, 2, false)) {
634+
whenRequesting(req -> req
635+
.param("fields[groups]", "offsets")
636+
.get("{groupId}", clusterId1, group1Id))
637+
.assertThat()
638+
.statusCode(is(Status.OK.getStatusCode()))
639+
.body("data.id", is(group1Id))
640+
.body("data.meta.errors", hasSize(1))
641+
.body("data.meta.errors.title", everyItem(startsWith("Unable to list group offsets")))
642+
.body("data.meta.errors.detail", everyItem(is("EXPECTED TEST EXCEPTION")));
643+
}
644+
}
645+
588646
@Test
589647
void testDescribeConsumerGroupWithFetchTopicOffsetsError() {
590648
Answer<ListOffsetsResult> listOffsetsFailed = args -> {
@@ -623,6 +681,82 @@ void testDescribeConsumerGroupWithFetchTopicOffsetsError() {
623681
}
624682
}
625683

684+
@Test
685+
void testDescribeGroupConfigsWithLimitedAuthorization() {
686+
utils.resetSecurity(consoleConfig, true);
687+
TokenUtils tokens = new TokenUtils(config);
688+
689+
utils.updateSecurity(consoleConfig.getSecurity(), new GlobalSecurityConfigBuilder()
690+
.addNewSubject()
691+
.withInclude("alice")
692+
.withRoleNames("limited-group-configs-role")
693+
.endSubject()
694+
.build());
695+
696+
String topic1 = "t1-" + UUID.randomUUID().toString();
697+
String group1 = "g1-" + UUID.randomUUID().toString();
698+
String client1 = "c1-" + UUID.randomUUID().toString();
699+
700+
String topic2 = "t2-" + UUID.randomUUID().toString();
701+
String group2 = "g2-" + UUID.randomUUID().toString();
702+
String client2 = "c2-" + UUID.randomUUID().toString();
703+
704+
consoleConfig.getKafka().getClusterById(clusterId1).ifPresent(cfg -> {
705+
cfg.setSecurity(new KafkaSecurityConfigBuilder()
706+
.addNewRole()
707+
.withName("limited-group-configs-role")
708+
.addNewRule()
709+
.withResources("groups")
710+
.withResourceNames("*")
711+
.withPrivileges(Privilege.GET)
712+
.endRule()
713+
.addNewRule()
714+
.withResources("groups/configs")
715+
.withResourceNames(group1)
716+
.withPrivileges(Privilege.GET)
717+
.endRule()
718+
// access to group2 configs is not granted
719+
.endRole()
720+
.build());
721+
});
722+
723+
try (var consumer1 = groupUtils.consume(group1, topic1, client1, 2, false);
724+
var consumer2 = groupUtils.consume(group2, topic2, client2, 2, false)) {
725+
whenRequesting(req -> req
726+
.auth()
727+
.oauth2(tokens.getToken("alice"))
728+
.param("fields[groups]", "groupId,configs")
729+
.get("{groupId}", clusterId1, Identifiers.encode(group1)))
730+
.assertThat()
731+
.statusCode(is(Status.OK.getStatusCode()))
732+
.body("data.attributes.groupId", is(group1))
733+
.body("data.attributes.configs", not(anEmptyMap()))
734+
.body("data.attributes.configs", everyEntry(
735+
Matchers.any(String.class), // any string key
736+
// all entries have same keys
737+
allOf(
738+
hasKey("value"),
739+
hasKey("source"),
740+
hasKey("sensitive"),
741+
hasKey("readOnly"),
742+
hasKey("type"))));
743+
744+
whenRequesting(req -> req
745+
.auth()
746+
.oauth2(tokens.getToken("alice"))
747+
.param("fields[groups]", "groupId,configs")
748+
.get("{groupId}", clusterId1, Identifiers.encode(group2)))
749+
.assertThat()
750+
.statusCode(is(Status.OK.getStatusCode()))
751+
.body("data.attributes.groupId", is(group2))
752+
.body("data.attributes.configs", not(anEmptyMap()))
753+
.body("data.attributes.configs", hasEntry(is("meta"), hasEntry("type", "error")))
754+
.body("data.attributes.configs", allOf(
755+
hasEntry("title", "Unable to describe group configs"),
756+
hasEntry(is("detail"), startsWith("Access denied:"))));
757+
}
758+
}
759+
626760
@Test
627761
void testDeleteConsumerGroupWithMembers() {
628762
String topic1 = "t1-" + UUID.randomUUID().toString();

‎api/src/test/java/com/github/streamshub/console/api/TopicsResourceIT.java‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363
import org.awaitility.core.EvaluatedCondition;
6464
import org.eclipse.microprofile.config.Config;
6565
import org.hamcrest.Description;
66+
import org.hamcrest.Matchers;
6667
import org.hamcrest.TypeSafeMatcher;
6768
import org.jboss.logging.Logger;
6869
import org.json.JSONException;
@@ -114,6 +115,7 @@
114115
import io.strimzi.api.kafka.model.topic.KafkaTopic;
115116
import io.strimzi.api.kafka.model.topic.KafkaTopicBuilder;
116117

118+
import static com.github.streamshub.console.test.EveryEntry.everyEntry;
117119
import static com.github.streamshub.console.test.TestHelper.whenRequesting;
118120
import static org.awaitility.Awaitility.await;
119121
import static org.hamcrest.MatcherAssert.assertThat;
@@ -278,15 +280,18 @@ void testListTopicsWithNameAndConfigsIncluded() {
278280
.assertThat()
279281
.statusCode(is(Status.OK.getStatusCode()))
280282
.body("data.size()", is(1))
281-
.body("data.attributes", contains(aMapWithSize(2)))
282-
.body("data.attributes.name", contains(topicName))
283-
.body("data.attributes.configs[0]", not(anEmptyMap()))
284-
.body("data.attributes.configs[0].findAll { it }.collect { it.value }",
285-
everyItem(allOf(
286-
hasKey("source"),
287-
hasKey("sensitive"),
288-
hasKey("readOnly"),
289-
hasKey("type"))));
283+
.body("data[0].attributes", is(aMapWithSize(2)))
284+
.body("data[0].attributes.name", is(topicName))
285+
.body("data[0].attributes.configs", not(anEmptyMap()))
286+
.body("data[0].attributes.configs", everyEntry(
287+
Matchers.any(String.class), // any string key
288+
// all entries have same keys
289+
allOf(
290+
hasKey("value"),
291+
hasKey("source"),
292+
hasKey("sensitive"),
293+
hasKey("readOnly"),
294+
hasKey("type"))));
290295
}
291296

292297
@Test

‎api/src/test/java/com/github/streamshub/console/test/EveryEntry.java‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
package com.github.streamshub.console.test;
2+
3+
import java.util.Map;
4+
import java.util.Map.Entry;
5+
6+
import org.hamcrest.Description;
7+
import org.hamcrest.Matcher;
8+
import org.hamcrest.TypeSafeMatcher;
9+
10+
import static org.hamcrest.core.IsAnything.anything;
11+
import static org.hamcrest.core.IsEqual.equalTo;
12+
13+
// XXX: consider contributing this to Hamcrest
14+
public class EveryEntry<K, V> extends TypeSafeMatcher<Map<? extends K, ? extends V>> {
15+
16+
private final Matcher<? super K> keyMatcher;
17+
private final Matcher<? super V> valueMatcher;
18+
19+
public EveryEntry(Matcher<? super K> keyMatcher, Matcher<? super V> valueMatcher) {
20+
this.keyMatcher = keyMatcher;
21+
this.valueMatcher = valueMatcher;
22+
}
23+
24+
@Override
25+
public boolean matchesSafely(Map<? extends K, ? extends V> map) {
26+
for (Entry<? extends K, ? extends V> entry : map.entrySet()) {
27+
if (!keyMatcher.matches(entry.getKey()) || !valueMatcher.matches(entry.getValue())) {
28+
return false;
29+
}
30+
}
31+
return true;
32+
}
33+
34+
@Override
35+
public void describeMismatchSafely(Map<? extends K, ? extends V> map, Description mismatchDescription) {
36+
mismatchDescription.appendText("map was ").appendValueList("[", ", ", "]", map.entrySet());
37+
}
38+
39+
@Override
40+
public void describeTo(Description description) {
41+
description.appendText("map with every entry [")
42+
.appendDescriptionOf(keyMatcher)
43+
.appendText("->")
44+
.appendDescriptionOf(valueMatcher)
45+
.appendText("]");
46+
}
47+
48+
public static <K, V> Matcher<Map<? extends K, ? extends V>> everyEntry(Matcher<? super K> keyMatcher, Matcher<? super V> valueMatcher) {
49+
return new EveryEntry<>(keyMatcher, valueMatcher);
50+
}
51+
52+
public static <K, V> Matcher<Map<? extends K, ? extends V>> everyEntry(K key, V value) {
53+
return new EveryEntry<>(equalTo(key), equalTo(value));
54+
}
55+
56+
public static <K> Matcher<Map<? extends K, ?>> everyKey(Matcher<? super K> keyMatcher) {
57+
return new EveryEntry<>(keyMatcher, anything());
58+
}
59+
60+
public static <K> Matcher<Map<? extends K, ?>> everyKey(K key) {
61+
return new EveryEntry<>(equalTo(key), anything());
62+
}
63+
64+
public static <V> Matcher<Map<?, ? extends V>> everyValue(Matcher<? super V> valueMatcher) {
65+
return new EveryEntry<>(anything(), valueMatcher);
66+
}
67+
68+
public static <V> Matcher<Map<?, ? extends V>> everyValue(V value) {
69+
return new EveryEntry<>(anything(), equalTo(value));
70+
}
71+
}

0 commit comments

Comments
 (0)