///*
// * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) 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.
// */
//
//package org.wso2.carbon.stream.processor.core;
//
//import org.apache.log4j.Level;
//import org.apache.log4j.Logger;
//import org.mockito.Mockito;
//import org.powermock.modules.testng.PowerMockTestCase;
//import org.testng.Assert;
//import org.testng.annotations.BeforeTest;
//import org.testng.annotations.Test;
//import org.wso2.carbon.stream.processor.core.ha.HACoordinationRecordTableHandler;
//import org.wso2.carbon.stream.processor.core.ha.HACoordinationRecordTableHandlerManager;
//import org.wso2.carbon.stream.processor.core.ha.RecordTableData;
//import org.wso2.siddhi.core.exception.ConnectionUnavailableException;
//import org.wso2.siddhi.core.table.record.RecordTableHandlerCallback;
//import org.wso2.siddhi.query.api.definition.TableDefinition;
//
//import java.util.ArrayList;
//import java.util.HashMap;
//import java.util.LinkedHashMap;
//
//import static org.powermock.api.mockito.PowerMockito.doNothing;
//import static org.powermock.api.mockito.PowerMockito.mock;
//import static org.powermock.api.mockito.PowerMockito.when;
//
//public class HACoordinationRecordTableHandlerTest extends PowerMockTestCase {
//
//    @BeforeTest
//    public void setDebugLogLevel() {
//        Logger.getLogger(HACoordinationRecordTableHandler.class.getName()).setLevel(Level.DEBUG);
//    }
//
//    @Test
//    public void testPassiveNodeEventQueue() throws ConnectionUnavailableException {
//        HACoordinationRecordTableHandlerManager recordTableHandlerManager = new HACoordinationRecordTableHandlerManager
//                (10);
//        HACoordinationRecordTableHandler recordTableHandler = (HACoordinationRecordTableHandler)
//                recordTableHandlerManager.generateRecordTableHandler();
//        recordTableHandler.init("element-id", TableDefinition.id("id"));
//
//        RecordTableHandlerCallback recordTableHandlerCallback = mock(RecordTableHandlerCallback.class);
//        doNothing().when(recordTableHandlerCallback).add(Mockito.any());
//        doNothing().when(recordTableHandlerCallback).delete(Mockito.any(), Mockito.any());
//        doNothing().when(recordTableHandlerCallback).update(Mockito.any(), Mockito.any(), Mockito.any(),
//                Mockito.any());
//        doNothing().when(recordTableHandlerCallback).updateOrAdd(Mockito.any(), Mockito.any(), Mockito.any(),
//                Mockito.any(), Mockito.any());
//        when(recordTableHandlerCallback.find(Mockito.any(), Mockito.any())).thenReturn(null);
//        when(recordTableHandlerCallback.contains(Mockito.any(), Mockito.any())).thenReturn(false);
//
//        //Events should be queued since passive node
//        recordTableHandler.contains(1L, new HashMap<>(), null, recordTableHandlerCallback);
//        recordTableHandler.find(2L, new HashMap<>(), null, recordTableHandlerCallback);
//        recordTableHandler.add(3L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.delete(4L, new ArrayList<>(), null, recordTableHandlerCallback);
//        recordTableHandler.update(5L, null, new ArrayList<>(), new LinkedHashMap<>(), new ArrayList<>(),
//                recordTableHandlerCallback);
//        recordTableHandler.updateOrAdd(6L, null, new ArrayList<>(), new LinkedHashMap<>(), new ArrayList<>(),
//                new ArrayList<>(), recordTableHandlerCallback);
//
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 4);
//        Assert.assertEquals(recordTableHandler.getEventQueue().peek().getTimestamp(), 3L);
//        Assert.assertEquals(recordTableHandler.getEventQueue().poll().getEventType(), RecordTableData.EventType.ADD);
//        Assert.assertEquals(recordTableHandler.getEventQueue().poll().getEventType(),
//                RecordTableData.EventType.DELETE);
//        Assert.assertEquals(recordTableHandler.getEventQueue().poll().getEventType(),
//                RecordTableData.EventType.UPDATE);
//        Assert.assertEquals(recordTableHandler.getEventQueue().poll().getEventType(), RecordTableData.EventType.
//                UPDATE_OR_ADD);
//    }
//
//    @Test
//    public void testActiveNodeEventQueue() throws ConnectionUnavailableException {
//        HACoordinationRecordTableHandlerManager recordTableHandlerManager = new HACoordinationRecordTableHandlerManager
//                (10);
//        HACoordinationRecordTableHandler recordTableHandler = (HACoordinationRecordTableHandler)
//                recordTableHandlerManager.generateRecordTableHandler();
//        recordTableHandler.init("element-id", TableDefinition.id("id"));
//
//        RecordTableHandlerCallback recordTableHandlerCallback = mock(RecordTableHandlerCallback.class);
//        doNothing().when(recordTableHandlerCallback).add(Mockito.any());
//        doNothing().when(recordTableHandlerCallback).delete(Mockito.any(), Mockito.any());
//        doNothing().when(recordTableHandlerCallback).update(Mockito.any(), Mockito.any(), Mockito.any(),
//                Mockito.any());
//        doNothing().when(recordTableHandlerCallback).updateOrAdd(Mockito.any(), Mockito.any(), Mockito.any(),
//                Mockito.any(), Mockito.any());
//        when(recordTableHandlerCallback.find(Mockito.any(), Mockito.any())).thenReturn(null);
//        when(recordTableHandlerCallback.contains(Mockito.any(), Mockito.any())).thenReturn(false);
//
//        //Events should be queued since passive node
//        recordTableHandler.add(1L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.delete(2L, new ArrayList<>(), null, recordTableHandlerCallback);
//        recordTableHandler.update(3L, null, new ArrayList<>(), new LinkedHashMap<>(), new ArrayList<>(),
//                recordTableHandlerCallback);
//        recordTableHandler.updateOrAdd(4L, null, new ArrayList<>(), new LinkedHashMap<>(), new ArrayList<>(),
//                new ArrayList<>(), recordTableHandlerCallback);
//
//        //Events older than this timestamp should be removed from queue
//        recordTableHandler.trimRecordTableEventQueue(2L);
//
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 2);
//        Assert.assertEquals(recordTableHandler.getEventQueue().peek().getEventType(),
//                RecordTableData.EventType.UPDATE);
//
//        recordTableHandler.add(5L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.delete(6L, new ArrayList<>(), null, recordTableHandlerCallback);
//        //All queued events should be handled
//        recordTableHandler.setAsActive();
//
//        //Timestamp of latest event should be updated
//        recordTableHandler.add(7L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.delete(8L, new ArrayList<>(), null, recordTableHandlerCallback);
//        recordTableHandler.update(9L, null, new ArrayList<>(), new LinkedHashMap<>(), new ArrayList<>(),
//                recordTableHandlerCallback);
//        recordTableHandler.updateOrAdd(10L, null, new ArrayList<>(), new LinkedHashMap<>(), new ArrayList<>(),
//                new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.find(11L, new HashMap<>(), null, recordTableHandlerCallback);
//        recordTableHandler.contains(12L, new HashMap<>(), null, recordTableHandlerCallback);
//
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 0);
//        Assert.assertEquals(recordTableHandler.getActiveNodeLastOperationTimestamp(), 12L);
//    }
//
//    @Test
//    public void testQueueCapacity() throws ConnectionUnavailableException {
//        HACoordinationRecordTableHandlerManager recordTableHandlerManager = new HACoordinationRecordTableHandlerManager
//                (5);
//        HACoordinationRecordTableHandler recordTableHandler = (HACoordinationRecordTableHandler)
//                recordTableHandlerManager.generateRecordTableHandler();
//        recordTableHandler.init("element-id", TableDefinition.id("id"));
//
//        RecordTableHandlerCallback recordTableHandlerCallback = mock(RecordTableHandlerCallback.class);
//        doNothing().when(recordTableHandlerCallback).add(Mockito.any());
//        doNothing().when(recordTableHandlerCallback).delete(Mockito.any(), Mockito.any());
//        doNothing().when(recordTableHandlerCallback).update(Mockito.any(), Mockito.any(), Mockito.any(),
//                Mockito.any());
//        doNothing().when(recordTableHandlerCallback).updateOrAdd(Mockito.any(), Mockito.any(), Mockito.any(),
//                Mockito.any(), Mockito.any());
//        when(recordTableHandlerCallback.find(Mockito.any(), Mockito.any())).thenReturn(null);
//        when(recordTableHandlerCallback.contains(Mockito.any(), Mockito.any())).thenReturn(false);
//
//        //Filling up queue
//        recordTableHandler.add(1L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.add(2L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.add(3L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.add(4L, new ArrayList<>(), recordTableHandlerCallback);
//        recordTableHandler.add(5L, new ArrayList<>(), recordTableHandlerCallback);
//
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 5);
//
//        recordTableHandler.add(1L, new ArrayList<>(), recordTableHandlerCallback);
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 5);
//
//        recordTableHandler.find(2L, new HashMap<>(), null, recordTableHandlerCallback);
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 5);
//
//        recordTableHandler.add(3L, new ArrayList<>(), recordTableHandlerCallback);
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 5);
//
//        recordTableHandler.delete(4L, new ArrayList<>(), null, recordTableHandlerCallback);
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 5);
//
//        recordTableHandler.update(5L, null, new ArrayList<>(), new LinkedHashMap<>(),
//                new ArrayList<>(), recordTableHandlerCallback);
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 5);
//
//        recordTableHandler.updateOrAdd(6L, null, new ArrayList<>(), new LinkedHashMap<>(),
//                new ArrayList<>(), new ArrayList<>(), recordTableHandlerCallback);
//        Assert.assertEquals(recordTableHandler.getEventQueue().size(), 5);
//    }
//}
