package com.stratio.streaming.utils;

import static org.junit.Assert.*;

import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import com.stratio.streaming.commons.constants.STREAM_OPERATIONS;
import com.stratio.streaming.commons.messages.StratioStreamingMessage;
import com.stratio.streaming.service.StreamsHelper;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

/**
 * Created by aitor on 9/30/15.
 */
@RunWith(JUnit4.class)
public class ZKUtilsTestIT {

    private static Config conf;
    private static ZKUtils zkUtils;

    @BeforeClass
    public static void setUp() throws Exception {
        conf= ConfigFactory.load();
        zkUtils= ZKUtils.getZKUtils(conf.getString("zookeeper.hosts"));
        //LOGGER.debug("Zookeeper hosts: " + conf.getString("zookeeper.hosts"));
        //LOGGER.debug("Kafka hosts: " + conf.getString("kafka.hosts"));
    }

    @AfterClass
    public static void tearDown() throws Exception {
        ZKUtils.shutdownZKUtils();
    }

    @Test
    public void testCreateEphemeralZNode() throws Exception {
        String path= "/stratio/streaming/test-data";
        byte[] bytes= "test".getBytes();
        Exception ex= null;
        try {
            zkUtils.createEphemeralZNode(path, bytes);
        } catch (Exception e)   { ex= e; }

        assertNull("Not exception expected", ex);
    }

    @Test
    public void testCreateZNode() throws Exception {
        String path= "/stratio/streaming/myTestNode";
        byte[] bytes= "test".getBytes();

        Exception ex= null;
        try {
            zkUtils.createZNode(path, bytes);
        } catch (Exception e)   { ex= e; }

        assertNull("Not exception expected", ex);
        assertTrue("The Node has not been created", zkUtils.existZNode(path));
    }


    @Test
    public void testCreateZNodeJsonReply() throws Exception {
        String basePath= "/stratio/streaming/";
        String reply= "myValue";
        String operation= STREAM_OPERATIONS.ACTION.LISTEN;
        String requestId= "requestId";
        String path= basePath + operation.toLowerCase() + "/" + requestId;

        StratioStreamingMessage message= StreamsHelper.getSampleMessage();
        message.setOperation(operation);
        message.setRequest_id(requestId);

        Exception ex= null;
        try {
            zkUtils.createZNodeJsonReply(message, reply);
        } catch (Exception e)   { ex= e; }

        assertNull("Not exception expected", ex);
        assertTrue("The Node has not been created", zkUtils.existZNode(path));
        String result= new String(zkUtils.getZNode(path));//Base64.encodeBase64String(zkUtils.getZNode(path));
        assertEquals("Unexpected content from node", "\"" + reply + "\"", result);
    }


}
@@ -214,16 +214,74 @@
 JUnit4;%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A
 import c
@@ -338,16 +338,16 @@
 ATIONS;%0A
-
 import c
@@ -632,16 +632,92 @@
 stIT %7B%0A%0A
+    private static Logger LOGGER = LoggerFactory.getLogger(ZKUtils.class);%0A%0A
     priv
@@ -952,34 +952,32 @@
 sts%22));%0A        
-//
 LOGGER.debug(%22Zo
@@ -1035,81 +1035,8 @@
 ));%0A
-        //LOGGER.debug(%22Kafka hosts: %22 + conf.getString(%22kafka.hosts%22));%0A
     
@@ -1404,32 +1404,87 @@
 e(path, bytes);%0A
+            zkUtils.createEphemeralZNode(path, bytes);%0A
         %7D catch 
@@ -1807,32 +1807,78 @@
 e(path, bytes);%0A
+            zkUtils.createZNode(path, bytes);%0A
         %7D catch 
@@ -2548,32 +2548,32 @@
 ption ex= null;%0A
-
         try %7B%0A  
@@ -2628,16 +2628,74 @@
 reply);%0A
+            zkUtils.createZNodeJsonReply(message, reply);%0A
         

Add repeated calls to validate that class remove existing data from zookeeper, implementes DECISION-74