package io.aiven.klaw.helpers.db.rdbms;

import static org.assertj.core.api.Assertions.assertThat;

import com.google.common.collect.Lists;
import io.aiven.klaw.UtilMethods;
import io.aiven.klaw.dao.Acl;
import io.aiven.klaw.dao.KafkaConnectorRequest;
import io.aiven.klaw.dao.Team;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.model.enums.RequestMode;
import io.aiven.klaw.model.enums.RequestOperationType;
import io.aiven.klaw.model.enums.RequestStatus;
import io.aiven.klaw.repository.AclRepo;
import io.aiven.klaw.repository.KwKafkaConnectorRequestsRepo;
import io.aiven.klaw.repository.TeamRepo;
import io.aiven.klaw.repository.UserInfoRepo;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest;
import org.springframework.boot.test.autoconfigure.orm.jpa.TestEntityManager;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.springframework.test.util.ReflectionTestUtils;

@ExtendWith(SpringExtension.class)
@DataJpaTest
public class KafkaConnectorsIntegrationTest {

  @Autowired private KwKafkaConnectorRequestsRepo repo;
  @Autowired private UserInfoRepo userInfoRepo;

  @Autowired private TeamRepo teamRepo;
  @Autowired private AclRepo aclRepo;

  @Autowired TestEntityManager entityManager;

  private SelectDataJdbc selectDataJdbc;

  private UtilMethods utilMethods;

  public void loadData() {
    generateData(10, 101, "firstconn", "dev", RequestOperationType.CLAIM, "created", 1, "103");
    generateData(10, 103, "firstconn", "dev", RequestOperationType.CREATE, "created", 11, null);
    generateData(10, 101, "secondconn", "dev", RequestOperationType.CREATE, "created", 21, null);
    generateData(10, 101, "secondconn", "test", RequestOperationType.DELETE, "declined", 31, null);
    generateData(1, 101, "secondconn", "test", RequestOperationType.UPDATE, "created", 41, null);
    UserInfo user = new UserInfo();
    user.setTenantId(101);
    user.setTeamId(101);
    user.setRole("USER");
    user.setUsername("James");
    UserInfo user2 = new UserInfo();
    user2.setTenantId(103);
    user2.setTeamId(103);
    user2.setRole("USER");
    user2.setUsername("John");
    UserInfo user3 = new UserInfo();
    user3.setTenantId(103);
    user3.setTeamId(103);
    user3.setRole("USER");
    user3.setUsername("Jackie");

    UserInfo user4 = new UserInfo();
    user4.setTenantId(104);
    user4.setTeamId(104);
    user4.setRole("USER");
    user4.setUsername("Joan");
    entityManager.persistAndFlush(user);
    entityManager.persistAndFlush(user2);
    entityManager.persistAndFlush(user3);
    entityManager.persistAndFlush(user4);

    Team t1 = new Team();
    t1.setTeamId(101);
    t1.setTenantId(101);
    t1.setTeamname("octopus");
    Team t2 = new Team();
    t2.setTeamname("pirates");
    t2.setTeamId(103);
    t2.setTenantId(103);

    entityManager.persistAndFlush(t1);
    entityManager.persistAndFlush(t2);

    Acl acl1 = new Acl();
    acl1.setReq_no(1);
    acl1.setTeamId(101);
    acl1.setTenantId(101);
    acl1.setConsumergroup("test");
    Acl acl2 = new Acl();
    acl2.setReq_no(2);
    acl2.setTeamId(101);
    acl2.setTenantId(101);
    acl2.setConsumergroup("test2");
    Acl acl3 = new Acl();
    acl3.setReq_no(3);
    acl3.setTeamId(103);
    acl3.setTenantId(101);
    acl3.setConsumergroup("test");
    entityManager.persistAndFlush(acl1);
    entityManager.persistAndFlush(acl2);
    entityManager.persistAndFlush(acl3);

    //
    generateData(
        5,
        104,
        "thirdtopic",
        "test",
        RequestOperationType.CREATE,
        RequestStatus.CREATED.value,
        50,
        null);
    generateData(
        1,
        104,
        "thirdtopic",
        "test",
        RequestOperationType.CLAIM,
        RequestStatus.DECLINED.value,
        55,
        "103");

    generateData(
        2,
        104,
        "thirdtopic",
        "test",
        RequestOperationType.UPDATE,
        RequestStatus.APPROVED.value,
        57,
        null);
    generateData(
        2,
        104,
        "thirdtopic",
        "test",
        RequestOperationType.DELETE,
        RequestStatus.CREATED.value,
        59,
        null);

    generateData(
        6,
        104,
        "thirdtopic",
        "test",
        RequestOperationType.CREATE,
        RequestStatus.DELETED.value,
        61,
        null);
  }

  @BeforeEach
  public void setUp() {
    selectDataJdbc = new SelectDataJdbc();
    utilMethods = new UtilMethods();
    ReflectionTestUtils.setField(selectDataJdbc, "kafkaConnectorRequestsRepo", repo);
    ReflectionTestUtils.setField(selectDataJdbc, "userInfoRepo", userInfoRepo);
    ReflectionTestUtils.setField(selectDataJdbc, "teamRepo", teamRepo);
    ReflectionTestUtils.setField(selectDataJdbc, "aclRepo", aclRepo);
    loadData();
  }

  @Test
  @Order(1)
  public void givenNewQueryMethodAssertThatDirectQueryAndNewQueriesReturnTheSameResultSet() {
    // Old way of getting all requests
    List<KafkaConnectorRequest> KafkaConnectorRequestList =
        Lists.newArrayList(repo.findAllByTenantId(101));

    List<KafkaConnectorRequest> results =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "John", "all", null, true, 101, null, null, false);

    assertThat(KafkaConnectorRequestList).hasSameSizeAs(results);

    assertThat(results).containsAll(KafkaConnectorRequestList);
  }

  @Test
  @Order(2)
  public void getallRequestsFilteredByTenantId() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", "all", null, false, 101, null, null, false);
    List<KafkaConnectorRequest> john =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", "all", null, true, 103, null, null, false);

    assertThat(james).hasSize(21);
    for (KafkaConnectorRequest req : james) {
      assertThat(req.getTenantId()).isEqualTo(101);
    }
    assertThat(john).hasSize(10);
    for (KafkaConnectorRequest req : john) {
      assertThat(req.getTenantId()).isEqualTo(103);
    }
  }

  @Test
  @Order(3)
  public void getallRequestsFilteredByTenantByStatus() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", RequestStatus.CREATED.value, null, false, 101, null, null, false);
    List<KafkaConnectorRequest> john =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "John", RequestStatus.DECLINED.value, null, false, 103, null, null, false);

    assertThat(james).hasSize(11);
    for (KafkaConnectorRequest req : james) {
      assertThat(req.getRequestStatus()).isEqualTo(RequestStatus.CREATED.value);
      assertThat(req.getTenantId()).isEqualTo(101);
    }
    assertThat(john).isEmpty();
  }

  @Test
  @Order(4)
  public void
      getallRequestsFilteredByAdditionalStatus_NotAllReqs_DoNotIgnoreStatus_ExpectTeamFiltering() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "James", "deleted", null, false, 103, null, null, false);
    List<KafkaConnectorRequest> james2 =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "James", "declined", null, true, 101, null, null, false);
    List<KafkaConnectorRequest> john =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "John", "created", null, true, 103, null, null, false);

    assertThat(james).isEmpty();

    assertThat(john).hasSize(10);
    assertThat(james2).hasSize(10);
    for (KafkaConnectorRequest req : james2) {
      assertThat(req.getTenantId()).isEqualTo(101);
    }

    for (KafkaConnectorRequest req : john) {
      assertThat(req.getTenantId()).isEqualTo(103);
    }
  }

  @Test
  @Order(5)
  public void getallRequestsFilteredByTeanantId() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", "all", null, false, 101, null, null, false);

    List<KafkaConnectorRequest> john =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "John", "all", null, false, 103, null, null, false);

    assertThat(james).hasSize(21);
    for (KafkaConnectorRequest req : james) {
      assertThat(req.getTenantId()).isEqualTo(101);
      assertThat(req.getTeamId()).isEqualTo(101);
    }

    assertThat(john).hasSize(10);
    for (KafkaConnectorRequest req : john) {
      assertThat(req.getTenantId()).isEqualTo(103);
      assertThat(req.getTeamId()).isEqualTo(103);
    }
  }

  @Test
  @Order(6)
  public void getallRequestsFilteredByMisSpeltStatus_ReturnNothing() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", "ALT", null, true, 101, null, null, false);
    assertThat(james).isEmpty();
  }

  @Test
  @Order(7)
  public void getNonApproversRequestsFilteredByTenantId_willIgnoreOtherPassedParameters() {

    List<KafkaConnectorRequest> tenant1 =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "James", null, null, true, 101, null, null, false);

    List<KafkaConnectorRequest> tenant2 =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "John", null, null, true, 103, null, null, false);

    assertThat(tenant1).hasSize(31);
    for (KafkaConnectorRequest req : tenant1) {
      assertThat(req.getTenantId()).isEqualTo(101);
    }

    assertThat(tenant2).hasSize(10);
    for (KafkaConnectorRequest req : tenant2) {
      assertThat(req.getTenantId()).isEqualTo(103);
    }
  }

  @Test
  @Order(8)
  public void getConnectors() {

    KafkaConnectorRequest tenant1 = selectDataJdbc.selectConnectorRequestsForConnector(1, 101);

    KafkaConnectorRequest tenant2 = selectDataJdbc.selectConnectorRequestsForConnector(11, 103);

    assertThat(tenant1.getTenantId()).isEqualTo(101);
    assertThat(tenant1.getConnectorName()).isEqualTo("firstconn");

    assertThat(tenant2.getTenantId()).isEqualTo(103);
    assertThat(tenant2.getConnectorName()).isEqualTo("firstconn");
  }

  @Test
  @Order(9)
  public void getAllRequestsAndFilterByApproversViewAndTestEnvReturnAll() {

    List<KafkaConnectorRequest> jackie =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", "all", null, true, 101, "test", null, false);

    assertThat(jackie).hasSize(11);

    for (KafkaConnectorRequest req : jackie) {
      assertThat(req.getEnvironment()).isEqualTo("test");
      assertThat(req.getRequestor()).isEqualTo("Jackie");
    }
  }

  @Test
  @Order(10)
  public void getAllRequestsAndFilterByApproversViewAndTestEnvFilterAllJackiesRequestsReturnNone() {

    List<KafkaConnectorRequest> jackie =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "Jackie", "all", null, true, 101, "test", null, false);

    assertThat(jackie).isEmpty();
  }

  @Test
  @Order(11)
  public void getNonApproversRequestsFilteredByStatusByWildcard_StatusIsNotIgnored() {

    List<KafkaConnectorRequest> tenant1 =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "James", "deleted", null, true, 101, null, "conn", false);

    List<KafkaConnectorRequest> tenant2 =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "John", "created", null, true, 103, null, "conn", false);

    assertThat(tenant1).isEmpty();

    assertThat(tenant2).hasSize(10);
  }

  @Test
  @Order(12)
  public void givennonExistentTenantId_ReturnNothing() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", "all", null, true, 99, null, null, false);

    assertThat(james).isEmpty();
  }

  @Test
  @Order(13)
  public void givenNewQueryMethodAssertThatOldAndNewQueriesReturnTheSameResultSet() {
    // Old way of getting all requests
    List<KafkaConnectorRequest> KafkaConnectorRequestList =
        Lists.newArrayList(repo.findAllByTenantId(101));

    List<KafkaConnectorRequest> results =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "James", null, null, true, 101, null, null, false);

    assertThat(KafkaConnectorRequestList).hasSameSizeAs(results);

    assertThat(results).containsAll(KafkaConnectorRequestList);
  }

  @Test
  @Order(14)
  public void getAllRequestsFilteredByTenantId() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", null, null, false, 101, null, null, false);
    List<KafkaConnectorRequest> john =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "John", null, null, false, 103, null, null, false);

    assertThat(james).hasSize(21);
    for (KafkaConnectorRequest req : james) {
      assertThat(req.getTenantId()).isEqualTo(101);
    }
    assertThat(john).hasSize(10);
    for (KafkaConnectorRequest req : john) {
      assertThat(req.getTenantId()).isEqualTo(103);
    }
  }

  @Test
  @Order(15)
  public void getAllRequestsFilteredByEnvironmentByStatus() {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", RequestStatus.CREATED.value, null, false, 101, "dev", null, false);
    List<KafkaConnectorRequest> john =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "John", RequestStatus.DECLINED.value, null, false, 103, "test", null, false);

    assertThat(james).hasSize(10);
    for (KafkaConnectorRequest req : james) {
      assertThat(req.getRequestStatus()).isEqualTo(RequestStatus.CREATED.value);
      assertThat(req.getTenantId()).isEqualTo(101);
      assertThat(req.getEnvironment()).isEqualTo("dev");
    }
    assertThat(john).isEmpty();
  }

  @Test
  @Order(16)
  public void getAllRequestsDontReturnOwnRequestsAsAllReqsIsTrue() {

    List<KafkaConnectorRequest> jackie =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "Jackie", "all", null, true, 101, null, null, false);

    assertThat(jackie).isEmpty();
  }

  @Test
  @Order(17)
  public void getAllRequestsAndReturnOwnRequestsAsAllReqsIsTrue() {

    List<KafkaConnectorRequest> jackie =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "Jackie", "all", null, true, 101, null, null, false);

    assertThat(jackie).hasSize(31);
  }

  @Test
  @Order(18)
  public void getAllCreatedStatusRequestsFromTenant() {

    List<KafkaConnectorRequest> joan =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "Joan", RequestStatus.CREATED.value, null, true, 104, null, null, false);

    assertThat(joan).hasSize(7);
    for (KafkaConnectorRequest req : joan) {
      assertThat(req.getRequestStatus()).isEqualTo(RequestStatus.CREATED.value);
      assertThat(req.getTenantId()).isEqualTo(104);
      assertThat(req.getRequestor()).isNotEqualTo("Joan");
    }
  }

  @Test
  @Order(19)
  public void getAllDELETEDStatusRequestsFromTenant() {

    List<KafkaConnectorRequest> joan =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "Joan", RequestStatus.DELETED.value, null, true, 104, null, null, false);

    assertThat(joan).hasSize(6);
    for (KafkaConnectorRequest req : joan) {
      assertThat(req.getRequestStatus()).isEqualTo(RequestStatus.DELETED.value);
      assertThat(req.getTenantId()).isEqualTo(104);
      assertThat(req.getRequestor()).isNotEqualTo("Joan");
    }
  }

  @Test
  @Order(20)
  public void getAllClaimStatusRequestsForApprovalTeam1() {

    List<KafkaConnectorRequest> resultSet =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", RequestStatus.ALL.value, null, false, 101, null, null, false);

    assertThat(resultSet).hasSize(21);
    for (KafkaConnectorRequest req : resultSet) {
      assertThat(req.getTenantId()).isEqualTo(101);
      assertThat(req.getRequestor()).isNotEqualTo("James");
      if (req.getRequestStatus().equals(RequestOperationType.CLAIM.value)) {
        assertThat(req.getTeamId()).isEqualTo(103);
        assertThat(req.getApprovingTeamId()).isEqualTo("101");
      } else {
        assertThat(req.getTeamId()).isEqualTo(101);
      }
    }
  }

  @Test
  @Order(21)
  public void getAllClaimStatusRequestsForMyRequestsViewCreatedByMyTeam() {

    List<KafkaConnectorRequest> resultSet =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "James", RequestStatus.ALL.value, null, false, 101, null, null, false);

    assertThat(resultSet).hasSize(31);
    for (KafkaConnectorRequest req : resultSet) {
      assertThat(req.getTenantId()).isEqualTo(101);
      assertThat(req.getRequestor()).isNotEqualTo("James");
      if (req.getRequestStatus().equals(RequestOperationType.CLAIM.value)) {
        assertThat(req.getApprovingTeamId()).isNotEqualTo("101");
      }
    }
  }

  @Test
  @Order(22)
  public void getAllClaimStatusRequestsForApprovalTeam2() {

    List<KafkaConnectorRequest> resultSet =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "John", RequestStatus.ALL.value, null, true, 101, null, null, false);

    assertThat(resultSet).hasSize(31);
    for (KafkaConnectorRequest req : resultSet) {
      assertThat(req.getTenantId()).isEqualTo(101);
      assertThat(req.getRequestor()).isNotEqualTo("James");
      if (req.getRequestStatus().equals(RequestOperationType.CLAIM.value)) {
        assertThat(req.getTeamId()).isEqualTo(101);
        assertThat(req.getApprovingTeamId()).isEqualTo("103");
      } else {
        assertThat(req.getTeamId()).isEqualTo(101);
      }
    }
  }

  @Test
  @Order(23)
  public void getAllClaimStatusRequestsForMyRequestsViewCreatedByMyTeamTeam2() {

    List<KafkaConnectorRequest> resultSet =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "John", RequestStatus.ALL.value, null, false, 103, null, null, false);
    assertThat(resultSet).hasSize(10);

    // MyTeamsRequests only
    for (KafkaConnectorRequest req : resultSet) {
      assertThat(req.getTenantId()).isEqualTo(103);
      assertThat(req.getRequestor()).isNotEqualTo("James");
      assertThat(req.getTeamId()).isEqualTo(103);
      if (req.getRequestStatus().equals(RequestOperationType.CLAIM.value)) {
        // only show my tems claim requests
        assertThat(req.getApprovingTeamId()).isNotEqualTo("103");
      }
    }
  }

  @Test
  @Order(24)
  public void getAllCRequestsForApprovalFilteredByWildcardSearch() {

    List<KafkaConnectorRequest> resultSet =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "John", RequestStatus.ALL.value, null, false, 101, null, "first", false);

    assertThat(resultSet).hasSize(10);

    // MyTeamsRequests only
    for (KafkaConnectorRequest req : resultSet) {
      assertThat(req.getTenantId()).isEqualTo(101);
      assertThat(req.getRequestor()).isNotEqualTo("James");
      assertThat(req.getTeamId()).isEqualTo(101);
      assertThat(req.getConnectorName()).isEqualTo("firstconn");
    }
  }

  @Test
  @Order(25)
  public void getAllCRequestsForApprovalFilteredByWildcardSearchNoMatching() {

    List<KafkaConnectorRequest> resultSet =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", RequestStatus.ALL.value, null, false, 103, null, "lots", false);
    assertThat(resultSet).isEmpty();
  }

  @Test
  @Order(26)
  public void getConnectorRequestsCountsForMyApprovals() {
    Map<String, Map<String, Long>> results =
        selectDataJdbc.getConnectorRequestsCounts(101, RequestMode.MY_APPROVALS, 101, "Jackie");

    Map<String, Long> statsCount = results.get("STATUS_COUNTS");
    Map<String, Long> operationTypeCount = results.get("OPERATION_TYPE_COUNTS");

    assertThat(results).hasSize(2);
    // Jackie created all the requests so we expect 0 for created status which is approval status
    assertThat(statsCount).containsEntry(RequestStatus.CREATED.value, 0L);
    assertThat(operationTypeCount)
        .containsEntry(RequestOperationType.CREATE.value, 10L)
        .containsEntry(RequestOperationType.CLAIM.value, 10L);
    assertThat(statsCount)
        .containsEntry(RequestStatus.APPROVED.value, 0L)
        .containsEntry(RequestStatus.DECLINED.value, 10L)
        .containsEntry(RequestStatus.DELETED.value, 0L);
    assertThat(operationTypeCount).containsEntry(RequestOperationType.UPDATE.value, 1L);
  }

  @Test
  @Order(27)
  public void getConnectorRequestsCountsForMyApprovalsJohnCreatedNone() {
    Map<String, Map<String, Long>> results =
        selectDataJdbc.getConnectorRequestsCounts(101, RequestMode.MY_APPROVALS, 101, "John");

    Map<String, Long> statsCount = results.get("STATUS_COUNTS");
    Map<String, Long> operationTypeCount = results.get("OPERATION_TYPE_COUNTS");

    assertThat(results).hasSize(2);
    assertThat(statsCount).containsEntry(RequestStatus.CREATED.value, 11L);
    assertThat(operationTypeCount)
        .containsEntry(RequestOperationType.CREATE.value, 10L)
        .containsEntry(RequestOperationType.CLAIM.value, 10L);
    assertThat(statsCount)
        .containsEntry(RequestStatus.APPROVED.value, 0L)
        .containsEntry(RequestStatus.DECLINED.value, 10L)
        .containsEntry(RequestStatus.DELETED.value, 0L);
    assertThat(operationTypeCount).containsEntry(RequestOperationType.UPDATE.value, 1L);
  }

  @Test
  @Order(28)
  public void getConnectorRequestsWhereRequestOperationIsCreate() {

    List<KafkaConnectorRequest> createRequests =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "James", null, RequestOperationType.CREATE, false, 101, null, null, false);

    assertThat(createRequests).hasSize(10);
  }

  @Order(26)
  @ParameterizedTest
  @CsvSource({
    "FIRSTTOPIC1,firsttopic1",
    "TOPIC1,firsttopic1",
    "FirstTopic1,firsttopic1",
    "firSToPic1,firsttopic1"
  })
  public void getConnectorRequestsForTeamViewFilteredbySearchTerm(
      String searchCriteria, String expectedTopicName) {

    List<KafkaConnectorRequest> john =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "John", null, null, false, 101, null, searchCriteria, false);

    for (KafkaConnectorRequest req : john) {
      assertThat(req.getConnectorName()).isEqualTo(expectedTopicName);
      assertThat(req.getTenantId()).isEqualTo(101);
    }
  }

  @Order(27)
  @ParameterizedTest
  @CsvSource({
    "Claim,Claim,10",
    "Delete,Delete,10",
    "Update,Update,1",
    "Promote,Promote,0",
    "Create,Create,10"
  })
  public void getConnectorRequestsForTeamViewFilteredbyRequestOperationType(
      String requestOperationType, String expectedTopicName, String number) {

    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false,
            "James",
            null,
            RequestOperationType.of(requestOperationType),
            false,
            101,
            null,
            null,
            false);

    for (KafkaConnectorRequest req : james) {
      assertThat(req.getRequestOperationType()).isEqualTo(expectedTopicName);
      assertThat(req.getTenantId()).isEqualTo(101);
    }

    assertThat(james).hasSize(Integer.parseInt(number));
  }

  @Order(28)
  @ParameterizedTest
  @CsvSource({"created,11", "deleted,0", "declined,10", "approved,0"})
  public void getConnectorRequestsForTeamViewFilteredbyRequestStatus(
      String requestStatus, String number) {
    // allreqs true so only requests from your team will be returned.
    List<KafkaConnectorRequest> james =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, "James", requestStatus, null, false, 101, null, null, false);

    for (KafkaConnectorRequest req : james) {
      assertThat(req.getRequestStatus()).isEqualTo(requestStatus);
      assertThat(req.getTenantId()).isEqualTo(101);
    }

    assertThat(james).hasSize(Integer.parseInt(number));
  }

  @Order(29)
  @ParameterizedTest
  @CsvSource({"James,0", "Jackie,0", "John,0"})
  public void getConnectorRequestsForApprovalIsMyRequestTrueNonReturned(
      String requestor, String number) {
    // allreqs true so only requests from your team will be returned.
    List<KafkaConnectorRequest> requests =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            true, requestor, null, null, false, 101, null, null, true);

    for (KafkaConnectorRequest req : requests) {
      assertThat(req.getRequestor()).isEqualTo(requestor);
      assertThat(req.getTenantId()).isEqualTo(101);
    }

    assertThat(requests).hasSize(Integer.parseInt(number));
  }

  @Order(30)
  @ParameterizedTest
  @CsvSource({"James,101,0", "Jackie,103,10", "John,103,0"})
  public void getConnectorRequestsForTeamViewIsMyRequestTrueRequestorsRequestsOnlyReturned(
      String requestor, String teamId, String number) {
    // allreqs true so only requests from your team will be returned.
    List<KafkaConnectorRequest> requests =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, requestor, null, null, false, 103, null, null, true);

    for (KafkaConnectorRequest req : requests) {
      assertThat(req.getRequestor()).isEqualTo(requestor);
      assertThat(req.getTeamId()).isEqualTo(Integer.valueOf(teamId));
    }

    assertThat(requests).hasSize(Integer.parseInt(number));
  }

  @Order(31)
  @ParameterizedTest
  @CsvSource({"James,0", "Jackie,31", "John,0"})
  public void getConnectorRequestsForTeamViewAllTeamsIsMyRequestTrueRequestorsRequestsOnlyReturned(
      String requestor, String number) {
    // allreqs true so only requests from your team will be returned.
    List<KafkaConnectorRequest> requests =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, requestor, null, null, true, 101, null, null, true);

    for (KafkaConnectorRequest req : requests) {
      assertThat(req.getRequestor()).isEqualTo(requestor);
    }

    assertThat(requests).hasSize(Integer.parseInt(number));
  }

  @Order(32)
  @ParameterizedTest
  @CsvSource({"James,31", "Jackie,31", "John,31"})
  public void getConnectorRequests_GetAllTeamsRequestsInTenantcy(String requestor, String number) {
    // allreqs true so only requests from your team will be returned.
    List<KafkaConnectorRequest> requests =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, requestor, null, null, true, 101, null, null, false);

    assertThat(requests).hasSize(Integer.parseInt(number));
  }

  @Order(33)
  @Test
  public void getConnectorRequests_GetAllTeamsRequestsInTenantcy() {
    // allreqs true so only requests from your team will be returned.
    List<KafkaConnectorRequest> requests =
        selectDataJdbc.selectFilteredKafkaConnectorRequests(
            false, "Jackie", null, null, true, 103, null, null, false);

    assertThat(requests).hasSize(10);
  }

  @Order(34)
  @Test
  public void testValidateIfConsumerGroupUsedByAnotherTeam() {
    assertThat(selectDataJdbc.validateIfConsumerGroupUsedByAnotherTeam(101, 101, "test")).isTrue();
    assertThat(selectDataJdbc.validateIfConsumerGroupUsedByAnotherTeam(101, 101, "test2"))
        .isFalse();
    assertThat(selectDataJdbc.validateIfConsumerGroupUsedByAnotherTeam(101, 101, "non existing"))
        .isFalse();
  }

  private void generateData(
      int number,
      int tenantId,
      String topicName,
      String env,
      RequestOperationType connectorType,
      String status,
      int id,
      String claimOwner) {

    for (int i = 0; i < number; i++) {
      KafkaConnectorRequest kc = new KafkaConnectorRequest();
      kc.setConnectorId(id++);
      kc.setTenantId(tenantId);
      kc.setTeamId(tenantId);
      kc.setRequestor("Jackie");
      kc.setConnectorName(topicName);
      kc.setEnvironment(env);
      kc.setRequestOperationType(connectorType.value);
      if (status != null) {
        kc.setRequestStatus(status);
      }

      if (connectorType.value.equals("Claim")) {
        kc.setApprovingTeamId(claimOwner);
      }

      entityManager.persistAndFlush(kc);
    }
  }
}
