ReducerFunctionalObjectSphinxTest.cpp

Go to the documentation of this file.

/*
 * ReducerFunctionalObjectSphinxTest.cpp
 *
 *  Created on: Oct 12, 2013
 *      Author: Steven
 */

#include <gtest/gtest.h>
#include "ReducerFunctionalObjectSphinx.hpp"
#include "SphinxResultSorterIf.hpp"
#include "SphinxResultData.hpp"
#include "EncodeDecodeBase64.hpp"
#include "ReducingInputMessage.hpp"
#include "SphinxResultCompareUtils.hpp"
#include <vector>
#include <Poco/Thread.h>

using namespace std;
using HCE::reduce::ReducerFunctionalObjectSphinx;
using HCE::sphinx::SphinxResultData;
using HCE::reduce_convertors::PocoJSONReducingInputMessageConvertor;
using HCE::reduce_types::ReducingInputMessage;
using namespace HCE::sphinx::reduce_task;

std::string buildJsonReducingInputMessage( int milliseconds_ttl ) {
    PocoJSONReducingInputMessageConvertor convertor;
    SphinxResultData result;
    std::string str_result = buildInputJSONFromSphinxResultData(result);
    
    Poco::SharedPtr<ReducingInputMessage> inputMessage( new ReducingInputMessage(1, milliseconds_ttl, str_result ) );
    
    return convertor.convertToJSONFrom( inputMessage );
}

class ReducerFunctionalObjectSphinxTest: public ::testing::Test {
protected:
    virtual void SetUp(){
      taskId = 1;
      std::string jsonReducingInputMessage = buildJsonReducingInputMessage( 200 );    
      obj.accumulate( jsonReducingInputMessage, taskId );
    }
    
    virtual void TearDown(){
    }
    
    void sleep_milliseconds( unsigned int milliseconds ) {
      Poco::Thread::sleep( milliseconds );
    }
    
    unsigned int milliseconds( int millis ) {
      return millis;
    }
    
    ReducerFunctionalObjectSphinx obj;
    unsigned long long taskId;
};


TEST_F(ReducerFunctionalObjectSphinxTest, taskCompleteByNodeCount ) {
    ASSERT_TRUE( obj.isCompleteTask( taskId, 1, milliseconds( 200 ) ) );
    ASSERT_TRUE( obj.isCompleteTask( taskId, 1 ) );
}

TEST_F(ReducerFunctionalObjectSphinxTest, taskCompleteByTTLParam ) {
    sleep_milliseconds( 10 );
    ASSERT_TRUE( obj.isCompleteTask( taskId, 2,  milliseconds( 10 ) ) );
}

TEST_F(ReducerFunctionalObjectSphinxTest, taskCompleteByTaskTTL )
{
    sleep_milliseconds( 200 );
    ASSERT_TRUE( obj.isCompleteTask( taskId, 2 ) );
}

TEST_F(ReducerFunctionalObjectSphinxTest, taskNotCompleteByTTLParam )
{    
    ASSERT_FALSE( obj.isCompleteTask( taskId, 2, milliseconds( 300 ) ) );
}

TEST_F(ReducerFunctionalObjectSphinxTest, taskNotCompleteByTaskTTL )
{
    ASSERT_FALSE( obj.isCompleteTask( taskId, 2 ) );
}

TEST_F(ReducerFunctionalObjectSphinxTest, cleanupByTTLParam )
{
    sleep_milliseconds( 10 );

    EXPECT_EQ( obj.cleanupExpiredTasksByTTL( 10, milliseconds( 10 ) ), 1 );        
}

TEST_F(ReducerFunctionalObjectSphinxTest, cleanupByTaskTTL )
{    
    sleep_milliseconds( 200 );

    EXPECT_EQ( obj.cleanupExpiredTasksByTTL( 10, 0 ), 1 );        
}

TEST_F(ReducerFunctionalObjectSphinxTest, NotCleanupByTaskParam )
{
    EXPECT_EQ( obj.cleanupExpiredTasksByTTL( 10, milliseconds( 100 ) ), 0 );        
}

TEST_F(ReducerFunctionalObjectSphinxTest, NotCleanupByTaskTTL )
{    
    sleep_milliseconds( 10 );

    EXPECT_EQ( obj.cleanupExpiredTasksByTTL( 10, 0 ), 0 );        
}

TEST_F(ReducerFunctionalObjectSphinxTest, ExceededByTTLParam )
{
    std::vector< unsigned long long > tasks;
    
    sleep_milliseconds( 10 );
     
    obj.getExceededTTLTasks( 10, tasks, milliseconds( 10 ) );
    
    ASSERT_TRUE( tasks.size() == 1 );
    
}

TEST_F(ReducerFunctionalObjectSphinxTest, exceededByTaskTTL )
{
    sleep_milliseconds( 200 );
    
    std::vector< unsigned long long > tasks;
    
    obj.getExceededTTLTasks( 10, tasks, 0 );
    
    ASSERT_TRUE( tasks.size() == 1 );
    
}


TEST_F(ReducerFunctionalObjectSphinxTest, NotExceededByTaskTTL )
{

    sleep_milliseconds( 100 );
    
    std::vector< unsigned long long > tasks;
    
    obj.getExceededTTLTasks( 10, tasks, 0 );
    
    ASSERT_TRUE( tasks.size() == 0 );
    
}

TEST_F(ReducerFunctionalObjectSphinxTest, NotExceededByTTLParam )
{
    sleep_milliseconds( 10 );
    
    std::vector< unsigned long long > tasks;
    
    obj.getExceededTTLTasks( 10, tasks, milliseconds( 50 ) );
    
    ASSERT_TRUE( tasks.size() == 0 );
    
}

TEST_F(ReducerFunctionalObjectSphinxTest, NoExpiredTasksInTTLQueue )
{    
    std::vector< unsigned long long > tasks;
    
    ASSERT_FALSE( obj.isCompleteTask( taskId, 2, milliseconds( 10 ) ) );
    obj.reduce( taskId );
    ASSERT_EQ( obj.cleanupExpiredTasksByTTLQueue( 10, milliseconds( 50 ) ), 0U );
}

TEST_F(ReducerFunctionalObjectSphinxTest, ExpiredTasksInTTLQueue )
{
    sleep_milliseconds( 201 );
    
    std::vector< unsigned long long > tasks;
    
    ASSERT_TRUE( obj.isCompleteTask( taskId, 2 ) ); 
    obj.reduce( taskId );
    ASSERT_EQ( obj.cleanupExpiredTasksByTTLQueue( 10, milliseconds( 10 ) ), 0U );
    sleep_milliseconds( 10 );
    ASSERT_EQ( obj.cleanupExpiredTasksByTTLQueue( 10, milliseconds( 10 ) ), 1U );
}

TEST_F(ReducerFunctionalObjectSphinxTest, RejectTerminateTaskAdd )
{
    sleep_milliseconds( 201 );
    
    ASSERT_TRUE( obj.isCompleteTask( taskId, 2, milliseconds( 10 ) ) );
    obj.reduce( taskId );
    EXPECT_EQ( obj.getRejectedMessages(), 0 );
    std::string jsonReducingInputMessage = buildJsonReducingInputMessage( 200 );    
    obj.accumulate( jsonReducingInputMessage, taskId );
    EXPECT_EQ( obj.getRejectedMessages(), 1 );    
}

TEST_F(ReducerFunctionalObjectSphinxTest, AddTaskAfterCleanupTTLQueue )
{
    sleep_milliseconds( 201 );
    
    ASSERT_TRUE( obj.isCompleteTask( taskId, 2, milliseconds( 1 ) ) );
    obj.reduce( taskId );
    EXPECT_EQ( obj.getRejectedMessages(), 0 );
    std::string jsonReducingInputMessage = buildJsonReducingInputMessage( 100 );    
    obj.accumulate( jsonReducingInputMessage, taskId );
    EXPECT_EQ( obj.getRejectedMessages(), 1 );
    
    sleep_milliseconds( 10 );
    EXPECT_EQ( obj.cleanupExpiredTasksByTTLQueue( 10, milliseconds( 10 ) ), 1U );
    obj.accumulate( jsonReducingInputMessage, taskId );
    EXPECT_EQ( obj.getRejectedMessages(), 1 );    
}