/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.apache.hadoop.ha;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.ZooDefs.Ids;
import org.junit.Before;
import org.junit.Test;
import org.junit.Assert;
import org.mockito.Mockito;

import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.ha.ActiveStandbyElector.ActiveStandbyElectorCallback;
import org.apache.hadoop.ha.ActiveStandbyElector.ActiveNotFoundException;
import org.apache.hadoop.util.ZKUtil.ZKAuthInfo;
import org.apache.hadoop.test.GenericTestUtils;

public class TestActiveStandbyElector {

  private ZooKeeper mockZK;
  private int count;
  private ActiveStandbyElectorCallback mockApp;
  private final byte[] data = new byte[8];

  private ActiveStandbyElectorTester elector;

  class ActiveStandbyElectorTester extends ActiveStandbyElector {
    private int sleptFor = 0;
    
    ActiveStandbyElectorTester(String hostPort, int timeout, String parent,
        List<ACL> acl, ActiveStandbyElectorCallback app) throws IOException,
        KeeperException {
      super(hostPort, timeout, parent, acl, Collections
          .<ZKAuthInfo> emptyList(), app,
          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT);
    }

    @Override
    public ZooKeeper connectToZooKeeper() {
      ++count;
      return mockZK;
    }
    
    @Override
    protected void sleepFor(int ms) {
      // don't sleep in unit tests! Instead, just record the amount of
      // time slept
      LOG.info("Would have slept for " + ms + "ms");
      sleptFor += ms;
    }
  }

  private static final String ZK_PARENT_NAME = "/parent/node";
  private static final String ZK_LOCK_NAME = ZK_PARENT_NAME + "/" +
      ActiveStandbyElector.LOCK_FILENAME;
  private static final String ZK_BREADCRUMB_NAME = ZK_PARENT_NAME + "/" +
      ActiveStandbyElector.BREADCRUMB_FILENAME;

  @Before
  public void init() throws IOException, KeeperException {
    count = 0;
    mockZK = Mockito.mock(ZooKeeper.class);
    mockApp = Mockito.mock(ActiveStandbyElectorCallback.class);
    elector = new ActiveStandbyElectorTester("hostPort", 1000, ZK_PARENT_NAME,
        Ids.OPEN_ACL_UNSAFE, mockApp);
  }

  /**
   * Set up the mock ZK to return no info for a prior active in ZK.
   */
  private void mockNoPriorActive() throws Exception {
    Mockito.doThrow(new KeeperException.NoNodeException()).when(mockZK)
        .getData(Mockito.eq(ZK_BREADCRUMB_NAME), Mockito.anyBoolean(),
            Mockito.<Stat>any());
  }
  
  /**
   * Set up the mock to return info for some prior active node in ZK./
   */
  private void mockPriorActive(byte[] data) throws Exception {
    Mockito.doReturn(data).when(mockZK)
        .getData(Mockito.eq(ZK_BREADCRUMB_NAME), Mockito.anyBoolean(),
            Mockito.<Stat>any());
  }


  /**
   * verify that joinElection checks for null data
   */
  @Test(expected = HadoopIllegalArgumentException.class)
  public void testJoinElectionException() {
    elector.joinElection(null);
  }

  /**
   * verify that joinElection tries to create ephemeral lock znode
   */
  @Test
  public void testJoinElection() {
    elector.joinElection(data);
    Mockito.verify(mockZK, Mockito.times(1)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
  }

  /**
   * verify that successful znode create result becomes active and monitoring is
   * started
   */
  @Test
  public void testCreateNodeResultBecomeActive() throws Exception {
    mockNoPriorActive();
    
    elector.joinElection(data);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
    verifyExistCall(1);

    // monitor callback verifies the leader is ephemeral owner of lock but does
    // not call becomeActive since its already active
    Stat stat = new Stat();
    stat.setEphemeralOwner(1L);
    Mockito.when(mockZK.getSessionId()).thenReturn(1L);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    // should not call neutral mode/standby/active
    Mockito.verify(mockApp, Mockito.times(0)).enterNeutralMode();
    Mockito.verify(mockApp, Mockito.times(0)).becomeStandby();
    Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
    // another joinElection not called.
    Mockito.verify(mockZK, Mockito.times(1)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
    // no new monitor called
    verifyExistCall(1);
  }
  
  /**
   * Verify that, when the callback fails to enter active state,
   * the elector rejoins the election after sleeping for a short period.
   */
  @Test
  public void testFailToBecomeActive() throws Exception {
    mockNoPriorActive();
    elector.joinElection(data);
    Assert.assertEquals(0, elector.sleptFor);
    
    Mockito.doThrow(new ServiceFailedException("failed to become active"))
        .when(mockApp).becomeActive();
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    // Should have tried to become active
    Mockito.verify(mockApp).becomeActive();
    
    // should re-join
    Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
    Assert.assertEquals(2, count);
    Assert.assertTrue(elector.sleptFor > 0);
  }
  
  /**
   * Verify that, when the callback fails to enter active state, after
   * a ZK disconnect (i.e from the StatCallback), that the elector rejoins
   * the election after sleeping for a short period.
   */
  @Test
  public void testFailToBecomeActiveAfterZKDisconnect() throws Exception {
    mockNoPriorActive();
    elector.joinElection(data);
    Assert.assertEquals(0, elector.sleptFor);

    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);

    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    verifyExistCall(1);

    Stat stat = new Stat();
    stat.setEphemeralOwner(1L);
    Mockito.when(mockZK.getSessionId()).thenReturn(1L);

    // Fake failure to become active from within the stat callback
    Mockito.doThrow(new ServiceFailedException("fail to become active"))
        .when(mockApp).becomeActive();
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
    
    // should re-join
    Mockito.verify(mockZK, Mockito.times(3)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
    Assert.assertEquals(2, count);
    Assert.assertTrue(elector.sleptFor > 0);
  }

  
  /**
   * Verify that, if there is a record of a prior active node, the
   * elector asks the application to fence it before becoming active.
   */
  @Test
  public void testFencesOldActive() throws Exception {
    byte[] fakeOldActiveData = new byte[0];
    mockPriorActive(fakeOldActiveData);
    
    elector.joinElection(data);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    // Application fences active.
    Mockito.verify(mockApp, Mockito.times(1)).fenceOldActive(
        fakeOldActiveData);
    // Updates breadcrumb node to new data
    Mockito.verify(mockZK, Mockito.times(1)).setData(
        Mockito.eq(ZK_BREADCRUMB_NAME),
        Mockito.eq(data),
        Mockito.eq(0));
    // Then it becomes active itself
    Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
  }
  
  @Test
  public void testQuitElectionRemovesBreadcrumbNode() throws Exception {
    mockNoPriorActive();
    elector.joinElection(data);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    // Writes its own active info
    Mockito.verify(mockZK, Mockito.times(1)).create(
        Mockito.eq(ZK_BREADCRUMB_NAME), Mockito.eq(data),
        Mockito.eq(Ids.OPEN_ACL_UNSAFE),
        Mockito.eq(CreateMode.PERSISTENT));
    mockPriorActive(data);
    
    elector.quitElection(false);
    
    // Deletes its own active data
    Mockito.verify(mockZK, Mockito.times(1)).delete(
        Mockito.eq(ZK_BREADCRUMB_NAME), Mockito.eq(0));
  }

  /**
   * verify that znode create for existing node and no retry becomes standby and
   * monitoring is started
   */
  @Test
  public void testCreateNodeResultBecomeStandby() {
    elector.joinElection(data);

    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).becomeStandby();
    verifyExistCall(1);
  }

  /**
   * verify that znode create error result in fatal error
   */
  @Test
  public void testCreateNodeResultError() {
    elector.joinElection(data);

    elector.processResult(Code.APIERROR.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).notifyFatalError(
        "Received create error from Zookeeper. code:APIERROR " +
        "for path " + ZK_LOCK_NAME);
  }

  /**
   * verify that retry of network errors verifies master by session id and
   * becomes active if they match. monitoring is started.
   */
  @Test
  public void testCreateNodeResultRetryBecomeActive() throws Exception {
    mockNoPriorActive();
    
    elector.joinElection(data);

    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    // 4 errors results in fatalError
    Mockito
        .verify(mockApp, Mockito.times(1))
        .notifyFatalError(
            "Received create error from Zookeeper. code:CONNECTIONLOSS " +
            "for path " + ZK_LOCK_NAME + ". " +
            "Not retrying further znode create connection errors.");

    elector.joinElection(data);
    // recreate connection via getNewZooKeeper
    Assert.assertEquals(2, count);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    verifyExistCall(1);

    Stat stat = new Stat();
    stat.setEphemeralOwner(1L);
    Mockito.when(mockZK.getSessionId()).thenReturn(1L);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
    verifyExistCall(1);
    Mockito.verify(mockZK, Mockito.times(6)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
  }

  /**
   * verify that retry of network errors verifies active by session id and
   * becomes standby if they dont match. monitoring is started.
   */
  @Test
  public void testCreateNodeResultRetryBecomeStandby() {
    elector.joinElection(data);

    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    verifyExistCall(1);

    Stat stat = new Stat();
    stat.setEphemeralOwner(0);
    Mockito.when(mockZK.getSessionId()).thenReturn(1L);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Mockito.verify(mockApp, Mockito.times(1)).becomeStandby();
    verifyExistCall(1);
  }

  /**
   * verify that if create znode results in nodeexists and that znode is deleted
   * before exists() watch is set then the return of the exists() method results
   * in attempt to re-create the znode and become active
   */
  @Test
  public void testCreateNodeResultRetryNoNode() {
    elector.joinElection(data);

    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    verifyExistCall(1);

    elector.processResult(Code.NONODE.intValue(), ZK_LOCK_NAME, mockZK,
        (Stat) null);
    Mockito.verify(mockApp, Mockito.times(1)).enterNeutralMode();
    Mockito.verify(mockZK, Mockito.times(4)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
  }

  /**
   * verify that more than 3 network error retries result fatalError
   */
  @Test
  public void testStatNodeRetry() {
    elector.joinElection(data);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        (Stat) null);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        (Stat) null);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        (Stat) null);
    elector.processResult(Code.CONNECTIONLOSS.intValue(), ZK_LOCK_NAME, mockZK,
        (Stat) null);
    Mockito
        .verify(mockApp, Mockito.times(1))
        .notifyFatalError(
            "Received stat error from Zookeeper. code:CONNECTIONLOSS. "+
            "Not retrying further znode monitoring connection errors.");
  }

  /**
   * verify error in exists() callback results in fatal error
   */
  @Test
  public void testStatNodeError() {
    elector.joinElection(data);
    elector.processResult(Code.RUNTIMEINCONSISTENCY.intValue(), ZK_LOCK_NAME,
        mockZK, (Stat) null);
    Mockito.verify(mockApp, Mockito.times(0)).enterNeutralMode();
    Mockito.verify(mockApp, Mockito.times(1)).notifyFatalError(
        "Received stat error from Zookeeper. code:RUNTIMEINCONSISTENCY");
  }

  /**
   * verify behavior of watcher.process callback with non-node event
   */
  @Test
  public void testProcessCallbackEventNone() throws Exception {
    mockNoPriorActive();
    elector.joinElection(data);

    WatchedEvent mockEvent = Mockito.mock(WatchedEvent.class);
    Mockito.when(mockEvent.getType()).thenReturn(Event.EventType.None);

    // first SyncConnected should not do anything
    Mockito.when(mockEvent.getState()).thenReturn(
        Event.KeeperState.SyncConnected);
    elector.processWatchEvent(mockZK, mockEvent);
    Mockito.verify(mockZK, Mockito.times(0)).exists(Mockito.anyString(),
        Mockito.anyBoolean(), Mockito.<AsyncCallback.StatCallback> anyObject(),
        Mockito.<Object> anyObject());

    // disconnection should enter safe mode
    Mockito.when(mockEvent.getState()).thenReturn(
        Event.KeeperState.Disconnected);
    elector.processWatchEvent(mockZK, mockEvent);
    Mockito.verify(mockApp, Mockito.times(1)).enterNeutralMode();

    // re-connection should monitor master status
    Mockito.when(mockEvent.getState()).thenReturn(
        Event.KeeperState.SyncConnected);
    elector.processWatchEvent(mockZK, mockEvent);
    verifyExistCall(1);
    Assert.assertTrue(elector.isMonitorLockNodePending());
    elector.processResult(Code.SESSIONEXPIRED.intValue(), ZK_LOCK_NAME,
        mockZK, new Stat());
    Assert.assertFalse(elector.isMonitorLockNodePending());

    // session expired should enter safe mode and initiate re-election
    // re-election checked via checking re-creation of new zookeeper and
    // call to create lock znode
    Mockito.when(mockEvent.getState()).thenReturn(Event.KeeperState.Expired);
    elector.processWatchEvent(mockZK, mockEvent);
    // already in safe mode above. should not enter safe mode again
    Mockito.verify(mockApp, Mockito.times(1)).enterNeutralMode();
    // called getNewZooKeeper to create new session. first call was in
    // constructor
    Assert.assertEquals(2, count);
    // once in initial joinElection and one now
    Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);

    // create znode success. become master and monitor
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
    verifyExistCall(2);

    // error event results in fatal error
    Mockito.when(mockEvent.getState()).thenReturn(Event.KeeperState.AuthFailed);
    elector.processWatchEvent(mockZK, mockEvent);
    Mockito.verify(mockApp, Mockito.times(1)).notifyFatalError(
        "Unexpected Zookeeper watch event state: AuthFailed");
    // only 1 state change callback is called at a time
    Mockito.verify(mockApp, Mockito.times(1)).enterNeutralMode();
  }

  /**
   * verify behavior of watcher.process with node event
   */
  @Test
  public void testProcessCallbackEventNode() throws Exception {
    mockNoPriorActive();
    elector.joinElection(data);

    // make the object go into the monitoring state
    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).becomeStandby();
    verifyExistCall(1);
    Assert.assertTrue(elector.isMonitorLockNodePending());

    Stat stat = new Stat();
    stat.setEphemeralOwner(0L);
    Mockito.when(mockZK.getSessionId()).thenReturn(1L);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Assert.assertFalse(elector.isMonitorLockNodePending());

    WatchedEvent mockEvent = Mockito.mock(WatchedEvent.class);
    Mockito.when(mockEvent.getPath()).thenReturn(ZK_LOCK_NAME);

    // monitoring should be setup again after event is received
    Mockito.when(mockEvent.getType()).thenReturn(
        Event.EventType.NodeDataChanged);
    elector.processWatchEvent(mockZK, mockEvent);
    verifyExistCall(2);
    Assert.assertTrue(elector.isMonitorLockNodePending());
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Assert.assertFalse(elector.isMonitorLockNodePending());

    // monitoring should be setup again after event is received
    Mockito.when(mockEvent.getType()).thenReturn(
        Event.EventType.NodeChildrenChanged);
    elector.processWatchEvent(mockZK, mockEvent);
    verifyExistCall(3);
    Assert.assertTrue(elector.isMonitorLockNodePending());
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Assert.assertFalse(elector.isMonitorLockNodePending());

    // lock node deletion when in standby mode should create znode again
    // successful znode creation enters active state and sets monitor
    Mockito.when(mockEvent.getType()).thenReturn(Event.EventType.NodeDeleted);
    elector.processWatchEvent(mockZK, mockEvent);
    // enterNeutralMode not called when app is standby and leader is lost
    Mockito.verify(mockApp, Mockito.times(0)).enterNeutralMode();
    // once in initial joinElection() and one now
    Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).becomeActive();
    verifyExistCall(4);
    Assert.assertTrue(elector.isMonitorLockNodePending());
    stat.setEphemeralOwner(1L);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Assert.assertFalse(elector.isMonitorLockNodePending());

    // lock node deletion in active mode should enter neutral mode and create
    // znode again successful znode creation enters active state and sets
    // monitor
    Mockito.when(mockEvent.getType()).thenReturn(Event.EventType.NodeDeleted);
    elector.processWatchEvent(mockZK, mockEvent);
    Mockito.verify(mockApp, Mockito.times(1)).enterNeutralMode();
    // another joinElection called
    Mockito.verify(mockZK, Mockito.times(3)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(2)).becomeActive();
    verifyExistCall(5);
    Assert.assertTrue(elector.isMonitorLockNodePending());
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Assert.assertFalse(elector.isMonitorLockNodePending());

    // bad path name results in fatal error
    Mockito.when(mockEvent.getPath()).thenReturn(null);
    elector.processWatchEvent(mockZK, mockEvent);
    Mockito.verify(mockApp, Mockito.times(1)).notifyFatalError(
        "Unexpected watch error from Zookeeper");
    // fatal error means no new connection other than one from constructor
    Assert.assertEquals(1, count);
    // no new watches after fatal error
    verifyExistCall(5);

  }

  private void verifyExistCall(int times) {
    Mockito.verify(mockZK, Mockito.times(times)).exists(
        Mockito.eq(ZK_LOCK_NAME), Mockito.<Watcher>any(),
        Mockito.same(elector),
        Mockito.same(mockZK));
  }

  /**
   * verify becomeStandby is not called if already in standby
   */
  @Test
  public void testSuccessiveStandbyCalls() {
    elector.joinElection(data);

    // make the object go into the monitoring standby state
    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).becomeStandby();
    verifyExistCall(1);
    Assert.assertTrue(elector.isMonitorLockNodePending());

    Stat stat = new Stat();
    stat.setEphemeralOwner(0L);
    Mockito.when(mockZK.getSessionId()).thenReturn(1L);
    elector.processResult(Code.OK.intValue(), ZK_LOCK_NAME, mockZK, stat);
    Assert.assertFalse(elector.isMonitorLockNodePending());

    WatchedEvent mockEvent = Mockito.mock(WatchedEvent.class);
    Mockito.when(mockEvent.getPath()).thenReturn(ZK_LOCK_NAME);

    // notify node deletion
    // monitoring should be setup again after event is received
    Mockito.when(mockEvent.getType()).thenReturn(Event.EventType.NodeDeleted);
    elector.processWatchEvent(mockZK, mockEvent);
    // is standby. no need to notify anything now
    Mockito.verify(mockApp, Mockito.times(0)).enterNeutralMode();
    // another joinElection called.
    Mockito.verify(mockZK, Mockito.times(2)).create(ZK_LOCK_NAME, data,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
    // lost election
    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    // still standby. so no need to notify again
    Mockito.verify(mockApp, Mockito.times(1)).becomeStandby();
    // monitor is set again
    verifyExistCall(2);
  }

  /**
   * verify quit election terminates connection and there are no new watches.
   * next call to joinElection creates new connection and performs election
   */
  @Test
  public void testQuitElection() throws Exception {
    elector.joinElection(data);
    Mockito.verify(mockZK, Mockito.times(0)).close();
    elector.quitElection(true);
    Mockito.verify(mockZK, Mockito.times(1)).close();
    // no watches added
    verifyExistCall(0);

    byte[] data = new byte[8];
    elector.joinElection(data);
    // getNewZooKeeper called 2 times. once in constructor and once now
    Assert.assertEquals(2, count);
    elector.processResult(Code.NODEEXISTS.intValue(), ZK_LOCK_NAME, mockZK,
        ZK_LOCK_NAME);
    Mockito.verify(mockApp, Mockito.times(1)).becomeStandby();
    verifyExistCall(1);

  }

  /**
   * verify that receiveActiveData gives data when active exists, tells that
   * active does not exist and reports error in getting active information
   * 
   * @throws IOException
   * @throws InterruptedException
   * @throws KeeperException
   * @throws ActiveNotFoundException
   */
  @Test
  public void testGetActiveData() throws ActiveNotFoundException,
      KeeperException, InterruptedException, IOException {
    // get valid active data
    byte[] data = new byte[8];
    Mockito.when(
        mockZK.getData(Mockito.eq(ZK_LOCK_NAME), Mockito.eq(false),
            Mockito.<Stat> anyObject())).thenReturn(data);
    Assert.assertEquals(data, elector.getActiveData());
    Mockito.verify(mockZK, Mockito.times(1)).getData(
        Mockito.eq(ZK_LOCK_NAME), Mockito.eq(false),
        Mockito.<Stat> anyObject());

    // active does not exist
    Mockito.when(
        mockZK.getData(Mockito.eq(ZK_LOCK_NAME), Mockito.eq(false),
            Mockito.<Stat> anyObject())).thenThrow(
        new KeeperException.NoNodeException());
    try {
      elector.getActiveData();
      Assert.fail("ActiveNotFoundException expected");
    } catch(ActiveNotFoundException e) {
      Mockito.verify(mockZK, Mockito.times(2)).getData(
          Mockito.eq(ZK_LOCK_NAME), Mockito.eq(false),
          Mockito.<Stat> anyObject());
    }

    // error getting active data rethrows keeperexception
    try {
      Mockito.when(
          mockZK.getData(Mockito.eq(ZK_LOCK_NAME), Mockito.eq(false),
              Mockito.<Stat> anyObject())).thenThrow(
          new KeeperException.AuthFailedException());
      elector.getActiveData();
      Assert.fail("KeeperException.AuthFailedException expected");
    } catch(KeeperException.AuthFailedException ke) {
      Mockito.verify(mockZK, Mockito.times(3)).getData(
          Mockito.eq(ZK_LOCK_NAME), Mockito.eq(false),
          Mockito.<Stat> anyObject());
    }
  }

  /**
   * Test that ensureBaseNode() recursively creates the specified dir
   */
  @Test
  public void testEnsureBaseNode() throws Exception {
    elector.ensureParentZNode();
    StringBuilder prefix = new StringBuilder();
    for (String part : ZK_PARENT_NAME.split("/")) {
      if (part.isEmpty()) continue;
      prefix.append("/").append(part);
      if (!"/".equals(prefix.toString())) {
        Mockito.verify(mockZK).create(
            Mockito.eq(prefix.toString()), Mockito.<byte[]>any(),
            Mockito.eq(Ids.OPEN_ACL_UNSAFE), Mockito.eq(CreateMode.PERSISTENT));
      }
    }
  }

  /**
   * Test that ACLs are set on parent zNode even if the node already exists.
   */
  @Test
  public void testParentZNodeACLs() throws Exception {
    KeeperException ke = new KeeperException(Code.NODEEXISTS) {
      @Override
      public Code code() {
        return super.code();
      }
    };

    Mockito.when(mockZK.create(Mockito.anyString(), Mockito.eq(new byte[]{}),
        Mockito.anyListOf(ACL.class),
        Mockito.eq(CreateMode.PERSISTENT))).thenThrow(ke);

    elector.ensureParentZNode();

    StringBuilder prefix = new StringBuilder();
    for (String part : ZK_PARENT_NAME.split("/")) {
      if (part.isEmpty()) continue;
      prefix.append("/").append(part);
      if (!"/".equals(prefix.toString())) {
        Mockito.verify(mockZK).getACL(Mockito.eq(prefix.toString()),
            Mockito.eq(new Stat()));
        Mockito.verify(mockZK).setACL(Mockito.eq(prefix.toString()),
            Mockito.eq(Ids.OPEN_ACL_UNSAFE), Mockito.anyInt());
      }
    }
  }
  
  /**
   * Test for a bug encountered during development of HADOOP-8163:
   * ensureBaseNode() should throw an exception if it has to retry
   * more than 3 times to create any part of the path.
   */
  @Test
  public void testEnsureBaseNodeFails() throws Exception {
    Mockito.doThrow(new KeeperException.ConnectionLossException())
      .when(mockZK).create(
          Mockito.eq(ZK_PARENT_NAME), Mockito.<byte[]>any(),
          Mockito.eq(Ids.OPEN_ACL_UNSAFE), Mockito.eq(CreateMode.PERSISTENT));
    try {
      elector.ensureParentZNode();
      Assert.fail("Did not throw!");
    } catch (IOException ioe) {
      if (!(ioe.getCause() instanceof KeeperException.ConnectionLossException)) {
        throw ioe;
      }
    }
    // Should have tried three times
    Mockito.verify(mockZK, Mockito.times(3)).create(
        Mockito.eq(ZK_PARENT_NAME), Mockito.<byte[]>any(),
        Mockito.eq(Ids.OPEN_ACL_UNSAFE), Mockito.eq(CreateMode.PERSISTENT));
  }

  /**
   * verify the zookeeper connection establishment
   */
  @Test
  public void testWithoutZKServer() throws Exception {
    try {
      new ActiveStandbyElector("127.0.0.1", 2000, ZK_PARENT_NAME,
          Ids.OPEN_ACL_UNSAFE, Collections.<ZKAuthInfo> emptyList(), mockApp,
          CommonConfigurationKeys.HA_FC_ELECTOR_ZK_OP_RETRIES_DEFAULT) {

          @Override
          protected ZooKeeper createZooKeeper() throws IOException {
            return Mockito.mock(ZooKeeper.class);
          }
      };


      Assert.fail("Did not throw zookeeper connection loss exceptions!");
    } catch (KeeperException ke) {
      GenericTestUtils.assertExceptionContains( "ConnectionLoss", ke);
    }
  }

  /**
   * joinElection(..) should happen only after SERVICE_HEALTHY.
   */
  @Test
  public void testBecomeActiveBeforeServiceHealthy() throws Exception {
    mockNoPriorActive();
    WatchedEvent mockEvent = Mockito.mock(WatchedEvent.class);
    Mockito.when(mockEvent.getType()).thenReturn(Event.EventType.None);
    // session expired should enter safe mode
    // But for first time, before the SERVICE_HEALTY i.e. appData is set,
    // should not enter the election.
    Mockito.when(mockEvent.getState()).thenReturn(Event.KeeperState.Expired);
    elector.processWatchEvent(mockZK, mockEvent);
    // joinElection should not be called.
    Mockito.verify(mockZK, Mockito.times(0)).create(ZK_LOCK_NAME, null,
        Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, elector, mockZK);
  }
}