/*
 * Copyright Strimzi authors.
 * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
 */
package io.strimzi.operator.user.operator;

import io.strimzi.api.kafka.model.AclOperation;
import io.strimzi.api.kafka.model.AclResourcePatternType;
import io.strimzi.api.kafka.model.AclRuleType;
import io.strimzi.operator.user.model.acl.SimpleAclRule;
import io.strimzi.operator.user.model.acl.SimpleAclRuleResource;
import io.strimzi.operator.user.model.acl.SimpleAclRuleResourceType;
import io.vertx.core.Vertx;
import io.vertx.junit5.Checkpoint;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateAclsResult;
import org.apache.kafka.clients.admin.DeleteAclsResult;
import org.apache.kafka.clients.admin.DescribeAclsResult;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.acl.AccessControlEntry;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;

import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import static java.util.Arrays.asList;
import static java.util.Collections.emptyList;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.hasItem;
import static org.hamcrest.Matchers.hasItems;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@ExtendWith(VertxExtension.class)
public class SimpleAclOperatorTest {
    protected static Vertx vertx;

    @BeforeAll
    public static void before() {
        vertx = Vertx.vertx();
    }

    @AfterAll
    public static void after() {
        vertx.close();
    }

    @Test
    public void testGetUsersFromAcls(VertxTestContext context)  {
        Admin mockAdminClient = mock(AdminClient.class);
        SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient);

        ResourcePattern res1 = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL);
        ResourcePattern res2 = new ResourcePattern(ResourceType.GROUP, "my-group", PatternType.LITERAL);

        KafkaPrincipal foo = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=foo");
        AclBinding fooAclBinding = new AclBinding(res1, new AccessControlEntry(foo.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));
        KafkaPrincipal bar = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=bar");
        AclBinding barAclBinding = new AclBinding(res1, new AccessControlEntry(bar.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));
        KafkaPrincipal baz = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "baz");
        AclBinding bazAclBinding = new AclBinding(res2, new AccessControlEntry(baz.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));
        KafkaPrincipal all = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "*");
        AclBinding allAclBinding = new AclBinding(res1, new AccessControlEntry(all.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));
        KafkaPrincipal anonymous = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "ANONYMOUS");
        AclBinding anonymousAclBinding = new AclBinding(res2, new AccessControlEntry(anonymous.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));

        Collection<AclBinding> aclBindings =
                asList(fooAclBinding, barAclBinding, bazAclBinding, allAclBinding, anonymousAclBinding);

        assertDoesNotThrow(() -> mockDescribeAcls(mockAdminClient, AclBindingFilter.ANY, aclBindings));
        assertThat(aclOp.getUsersWithAcls(), is(new HashSet<>(asList("foo", "bar", "baz"))));
        context.completeNow();
    }

    @Test
    public void testReconcileInternalCreateAddsAclsToAuthorizer(VertxTestContext context) {
        Admin mockAdminClient = mock(AdminClient.class);
        SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient);

        ResourcePattern resource1 = new ResourcePattern(ResourceType.CLUSTER, "kafka-cluster", PatternType.LITERAL);
        ResourcePattern resource2 = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL);

        KafkaPrincipal foo = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=foo");
        AclBinding describeAclBinding = new AclBinding(resource1, new AccessControlEntry(foo.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.DESCRIBE, AclPermissionType.ALLOW));
        AclBinding readAclBinding = new AclBinding(resource2, new AccessControlEntry(foo.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));
        AclBinding writeAclBinding = new AclBinding(resource2, new AccessControlEntry(foo.toString(), "*",
                org.apache.kafka.common.acl.AclOperation.WRITE, AclPermissionType.ALLOW));

        SimpleAclRuleResource ruleResource1 = new SimpleAclRuleResource("kafka-cluster", SimpleAclRuleResourceType.CLUSTER, AclResourcePatternType.LITERAL);
        SimpleAclRuleResource ruleResource2 = new SimpleAclRuleResource("my-topic", SimpleAclRuleResourceType.TOPIC, AclResourcePatternType.LITERAL);
        SimpleAclRule resource1DescribeRule = new SimpleAclRule(AclRuleType.ALLOW, ruleResource1, "*", AclOperation.DESCRIBE);
        SimpleAclRule resource2ReadRule = new SimpleAclRule(AclRuleType.ALLOW, ruleResource2, "*", AclOperation.READ);
        SimpleAclRule resource2WriteRule = new SimpleAclRule(AclRuleType.ALLOW, ruleResource2, "*", AclOperation.WRITE);

        ArgumentCaptor<Collection<AclBinding>> aclBindingsCaptor = ArgumentCaptor.forClass(Collection.class);
        assertDoesNotThrow(() -> {
            mockDescribeAcls(mockAdminClient, null, emptyList());
            mockCreateAcls(mockAdminClient, aclBindingsCaptor);
        });

        Checkpoint async = context.checkpoint();
        aclOp.reconcile("CN=foo", new LinkedHashSet<>(asList(resource2ReadRule, resource2WriteRule, resource1DescribeRule)))
                .onComplete(context.succeeding(rr -> context.verify(() -> {
                    Collection<AclBinding> capturedAclBindings = aclBindingsCaptor.getValue();
                    assertThat(capturedAclBindings, hasSize(3));
                    assertThat(capturedAclBindings, hasItems(describeAclBinding, readAclBinding, writeAclBinding));

                    Set<ResourcePattern> capturedResourcePatterns =
                            capturedAclBindings.stream().map(AclBinding::pattern).collect(Collectors.toSet());
                    assertThat(capturedResourcePatterns, hasSize(2));
                    assertThat(capturedResourcePatterns, hasItems(resource1, resource2));

                    async.flag();
                })));
    }

    @Test
    public void testReconcileInternalUpdateCreatesNewAclsAndDeletesOldAcls(VertxTestContext context) {
        Admin mockAdminClient = mock(AdminClient.class);
        SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient);

        ResourcePattern resource1 = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL);
        ResourcePattern resource2 = new ResourcePattern(ResourceType.TOPIC, "my-topic2", PatternType.LITERAL);

        KafkaPrincipal foo = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "CN=foo");
        AclBinding readAclBinding = new AclBinding(resource1, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));
        AclBinding writeAclBinding = new AclBinding(resource2, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.WRITE, AclPermissionType.ALLOW));

        SimpleAclRuleResource resource = new SimpleAclRuleResource("my-topic2", SimpleAclRuleResourceType.TOPIC, AclResourcePatternType.LITERAL);
        SimpleAclRule rule1 = new SimpleAclRule(AclRuleType.ALLOW, resource, "*", AclOperation.WRITE);

        ArgumentCaptor<Collection<AclBinding>> aclBindingsCaptor = ArgumentCaptor.forClass(Collection.class);
        ArgumentCaptor<Collection<AclBindingFilter>> aclBindingFiltersCaptor = ArgumentCaptor.forClass(Collection.class);
        assertDoesNotThrow(() -> {
            mockDescribeAcls(mockAdminClient, null, Collections.singleton(readAclBinding));
            mockCreateAcls(mockAdminClient, aclBindingsCaptor);
            mockDeleteAcls(mockAdminClient, Collections.singleton(readAclBinding), aclBindingFiltersCaptor);
        });

        Checkpoint async = context.checkpoint();
        aclOp.reconcile("CN=foo", new LinkedHashSet(asList(rule1)))
                .onComplete(context.succeeding(rr -> context.verify(() -> {

                    // Create Write rule for resource 2
                    Collection<AclBinding> capturedAclBindings = aclBindingsCaptor.getValue();
                    assertThat(capturedAclBindings, hasSize(1));
                    assertThat(capturedAclBindings, hasItem(writeAclBinding));
                    Set<ResourcePattern> capturedResourcePatterns =
                            capturedAclBindings.stream().map(AclBinding::pattern).collect(Collectors.toSet());
                    assertThat(capturedResourcePatterns, hasSize(1));
                    assertThat(capturedResourcePatterns, hasItem(resource2));

                    // Delete read rule for resource 1
                    Collection<AclBindingFilter> capturedAclBindingFilters = aclBindingFiltersCaptor.getValue();
                    assertThat(capturedAclBindingFilters, hasSize(1));
                    assertThat(capturedAclBindingFilters, hasItem(readAclBinding.toFilter()));

                    Set<ResourcePatternFilter> capturedResourcePatternFilters =
                            capturedAclBindingFilters.stream().map(AclBindingFilter::patternFilter).collect(Collectors.toSet());
                    assertThat(capturedResourcePatternFilters, hasSize(1));
                    assertThat(capturedResourcePatternFilters, hasItem(resource1.toFilter()));

                    async.flag();
                })));
    }

    @Test
    public void testReconcileInternalDelete(VertxTestContext context) {
        Admin mockAdminClient = mock(AdminClient.class);
        SimpleAclOperator aclOp = new SimpleAclOperator(vertx, mockAdminClient);

        ResourcePattern resource = new ResourcePattern(ResourceType.TOPIC, "my-topic", PatternType.LITERAL);

        KafkaPrincipal foo = new KafkaPrincipal("User", "CN=foo");
        AclBinding readAclBinding = new AclBinding(resource, new AccessControlEntry(foo.toString(), "*", org.apache.kafka.common.acl.AclOperation.READ, AclPermissionType.ALLOW));

        ArgumentCaptor<Collection<AclBindingFilter>> aclBindingFiltersCaptor = ArgumentCaptor.forClass(Collection.class);
        assertDoesNotThrow(() -> {
            mockDescribeAcls(mockAdminClient, null, Collections.singleton(readAclBinding));
            mockDeleteAcls(mockAdminClient, Collections.singleton(readAclBinding), aclBindingFiltersCaptor);
        });

        Checkpoint async = context.checkpoint();
        aclOp.reconcile("CN=foo", null)
                .onComplete(context.succeeding(rr -> context.verify(() -> {

                    Collection<AclBindingFilter> capturedAclBindingFilters = aclBindingFiltersCaptor.getValue();
                    assertThat(capturedAclBindingFilters, hasSize(1));
                    assertThat(capturedAclBindingFilters, hasItem(readAclBinding.toFilter()));

                    Set<ResourcePatternFilter> capturedResourcePatternFilters =
                            capturedAclBindingFilters.stream().map(AclBindingFilter::patternFilter).collect(Collectors.toSet());
                    assertThat(capturedResourcePatternFilters, hasSize(1));
                    assertThat(capturedResourcePatternFilters, hasItem(resource.toFilter()));

                    async.flag();
                })));
    }

    private void mockDescribeAcls(Admin mockAdminClient, AclBindingFilter aclBindingFilter, Collection<AclBinding> aclBindings)
            throws InterruptedException, ExecutionException {
        DescribeAclsResult result = mock(DescribeAclsResult.class);
        KafkaFuture<Collection<AclBinding>> future = mock(KafkaFuture.class);
        when(future.get()).thenReturn(aclBindings);
        when(result.values()).thenReturn(future);
        when(mockAdminClient.describeAcls(aclBindingFilter != null ? aclBindingFilter : any())).thenReturn(result);
    }

    private void mockCreateAcls(Admin mockAdminClient, ArgumentCaptor<Collection<AclBinding>> aclBindingsCaptor)
            throws InterruptedException, ExecutionException {
        CreateAclsResult result = mock(CreateAclsResult.class);
        KafkaFuture<Void> future = mock(KafkaFuture.class);
        when(future.get()).thenReturn(null);
        when(result.all()).thenReturn(future);
        when(mockAdminClient.createAcls(aclBindingsCaptor.capture())).thenReturn(result);
    }

    private void mockDeleteAcls(Admin mockAdminClient, Collection<AclBinding> aclBindings, ArgumentCaptor<Collection<AclBindingFilter>> aclBindingFiltersCaptor)
            throws InterruptedException, ExecutionException {
        DeleteAclsResult result = mock(DeleteAclsResult.class);
        KafkaFuture<Collection<AclBinding>> future = mock(KafkaFuture.class);
        when(future.get()).thenReturn(aclBindings);
        when(result.all()).thenReturn(future);
        when(mockAdminClient.deleteAcls(aclBindingFiltersCaptor.capture())).thenReturn(result);
    }
}
