hce-node application  1.4.3
HCE Hierarchical Cluster Engine node application
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros
DRCECleanupTest.cpp
Go to the documentation of this file.
1 #include <gtest/gtest.h>
2 #include <sstream>
3 #include <Poco/File.h>
4 
5 #include "DRCEFileStream.hpp"
6 #include "DRCENodeOptions.hpp"
7 #include "DRCEFunctionalObject.hpp"
8 #include "DRCETaskRequestSetExecute.hpp"
9 #include "DRCETaskRequestGetData.hpp"
10 #include "DRCETaskRequestDeleteData.hpp"
11 #include "DRCETaskRequestTerminate.hpp"
12 
13 #include "DRCETaskThreadingModeTest.hpp"
14 #include "DRCEPrintStatus.hpp"
15 #include "DRCECleanupTest.hpp"
16 #include "ProcExec.hpp"
17 #include "DRCEFileExtractor.hpp"
18 #include "CustomMessage.hpp"
19 #include "DRCEMessageConst.hpp"
20 
21 namespace HCE
22 {
23 namespace drce
24 {
25 namespace tests
26 {
27 //-----------------------------------------------------------------------------
28 DRCEResultDataItem DRCECleanupTest::extractResultData(unsigned int taskId, DRCENodeOptions& nodeOptions)
29 {
30  DRCEResultDataItem resultDataItem;
31  try
32  {
33  MessagesCollection messagesCollection(message_const::messages);
34  CustomMessage message(messagesCollection);
35  DataFileExtractor dataFileExctractor(message);
36  resultDataItem = dataFileExctractor.extract(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), taskId));
37  }
38  catch(Poco::Exception& e)
39  {
40  printError("Extract result data ("+e.message()+")");
41  }
42  return resultDataItem;
43 }
44 //-----------------------------------------------------------------------------
45 std::string DRCECleanupTest::getInputJson(unsigned int startTaskId,
46  SessionOptions::ThreadMode threadMode,
47  SessionOptions::CleanupFlag cleanupFlag,
48  const std::string& commandLine,
49  unsigned int timeMax)
50 {
51  DRCEInputJsonMessage inputJsonMessage = getInputJsonMessage(startTaskId, threadMode, cleanupFlag, commandLine, timeMax);
52 
53  const unsigned int SZ = 6;
54  DRCEInputJsonMessage subtasks[SZ];
55  for (unsigned int i=0;i<SZ;++i)
56  {
57  subtasks[i] = getInputJsonMessage(++startTaskId,
58  (!(i%2))?SessionOptions::ThreadMode::tmAsync:SessionOptions::ThreadMode::tmSync,
59  SessionOptions::CleanupFlag::cfNotDelete,
60  commandLine,
61  timeMax);
62  }
63 
64  subtasks[0].addSubtaskItem(subtasks[2]);
65  subtasks[0].addSubtaskItem(subtasks[3]);
66 
67  subtasks[1].addSubtaskItem(subtasks[4]);
68  subtasks[1].addSubtaskItem(subtasks[5]);
69 
70  inputJsonMessage.addSubtaskItem(subtasks[0]);
71  inputJsonMessage.addSubtaskItem(subtasks[1]);
72 
73  std::stringstream jsonStream;
74  jsonStream << inputJsonMessage;
75 
76  return jsonStream.str();
77 }
78 //-----------------------------------------------------------------------------
79 DRCEInputJsonMessage DRCECleanupTest::getInputJsonMessage(unsigned int taskId,
80  SessionOptions::ThreadMode threadMode,
81  SessionOptions::CleanupFlag cleanupFlag,
82  const std::string& commandLine,
83  unsigned int timeMax)
84 {
85  DRCETaskRequestSetExecute taskRequestSetExecute;
86  taskRequestSetExecute.setRequestType(DRCETaskRequest::RequestType::rtSetTaskExecute);
87  taskRequestSetExecute.setTaskId(taskId);
88 
89  if (commandLine.empty())
90  {
91  taskRequestSetExecute.setCommandLine("sort");
92  taskRequestSetExecute.setInputStream("def\nabc\n123\n");
93  }
94  else
95  taskRequestSetExecute.setCommandLine(commandLine);
96 
97  SessionOptions sessionOptions;
98  sessionOptions.tmode = threadMode;
99  sessionOptions.cleanup = cleanupFlag;
100  sessionOptions.timeMax = timeMax;
101 // sessionOptions.shellName = "hce-starter-bash.sh";
102  taskRequestSetExecute.setSessionOptions(std::forward<SessionOptions>(sessionOptions));
103 
104  std::stringstream jsonStream;
105 
106  jsonStream << taskRequestSetExecute;
107 
108  DRCEInputJsonMessage inputJsonMessage;
109  inputJsonMessage.setRequestId(taskId);
110  inputJsonMessage.setRequestType(DRCETaskRequest::RequestType::rtSetTaskExecute);
111  inputJsonMessage.setRequestData(jsonStream.str());
112 
113  return inputJsonMessage;
114 }
115 //-----------------------------------------------------------------------------
116 std::string DRCECleanupTest::getInputJson(unsigned int startTaskId,
117  SessionOptions::ThreadMode taskThreadMode,
118  SessionOptions::ThreadMode subtasksThreadMode)
119 {
120  DRCEInputJsonMessage inputJsonMessage = getInputJsonMessage(startTaskId, taskThreadMode, SessionOptions::CleanupFlag::cfNotDelete);
121 
122  const unsigned int SZ = 6;
123  DRCEInputJsonMessage subtasks[SZ];
124  for (unsigned int i=0;i<SZ;++i)
125  {
126  subtasks[i] = getInputJsonMessage(++startTaskId, subtasksThreadMode, SessionOptions::CleanupFlag::cfNotDelete);
127  }
128 
129  subtasks[0].addSubtaskItem(subtasks[2]);
130  subtasks[0].addSubtaskItem(subtasks[3]);
131 
132  subtasks[1].addSubtaskItem(subtasks[4]);
133  subtasks[1].addSubtaskItem(subtasks[5]);
134 
135  inputJsonMessage.addSubtaskItem(subtasks[0]);
136  inputJsonMessage.addSubtaskItem(subtasks[1]);
137 
138  std::stringstream jsonStream;
139  jsonStream << inputJsonMessage;
140 
141  return jsonStream.str();
142 }
143 //-----------------------------------------------------------------------------
144 std::string DRCECleanupTest::getInputJson(unsigned int startTaskId, SessionOptions::ThreadMode taskThreadMode)
145 {
146  DRCEInputJsonMessage inputJsonMessage = getInputJsonMessage(startTaskId, taskThreadMode, SessionOptions::CleanupFlag::cfNotDelete);
147 
148  const unsigned int SZ = 14;
149  DRCEInputJsonMessage subtasks[SZ];
150  for (unsigned int i=0;i<SZ;++i)
151  {
152  subtasks[i] = getInputJsonMessage(++startTaskId,
153  (!(i%2))?SessionOptions::ThreadMode::tmAsync:SessionOptions::ThreadMode::tmSync,
154  SessionOptions::CleanupFlag::cfNotDelete);
155  }
156 
157  subtasks[2].addSubtaskItem(subtasks[6]);
158  subtasks[2].addSubtaskItem(subtasks[7]);
159 
160  subtasks[3].addSubtaskItem(subtasks[8]);
161  subtasks[3].addSubtaskItem(subtasks[9]);
162 
163  subtasks[4].addSubtaskItem(subtasks[10]);
164  subtasks[4].addSubtaskItem(subtasks[11]);
165 
166  subtasks[5].addSubtaskItem(subtasks[12]);
167  subtasks[5].addSubtaskItem(subtasks[13]);
168 
169 
170  subtasks[0].addSubtaskItem(subtasks[2]);
171  subtasks[0].addSubtaskItem(subtasks[3]);
172 
173  subtasks[1].addSubtaskItem(subtasks[4]);
174  subtasks[1].addSubtaskItem(subtasks[5]);
175 
176  inputJsonMessage.addSubtaskItem(subtasks[0]);
177  inputJsonMessage.addSubtaskItem(subtasks[1]);
178 
179  std::stringstream jsonStream;
180  jsonStream << inputJsonMessage;
181 
182  return jsonStream.str();
183 }
184 //-----------------------------------------------------------------------------
185 std::string DRCECleanupTest::getInputJsonCleanup(unsigned int startTaskId, SessionOptions::ThreadMode taskThreadMode, const std::string& commandLine, unsigned int timeMax)
186 {
187  DRCEInputJsonMessage inputJsonMessage = getInputJsonMessage(startTaskId, taskThreadMode, SessionOptions::CleanupFlag::cfNotDelete, commandLine, timeMax);
188 
189  const unsigned int SZ = 14;
190  DRCEInputJsonMessage subtasks[SZ];
191  for (unsigned int i=0;i<SZ;++i)
192  {
193  subtasks[i] = getInputJsonMessage(++startTaskId,
194  (!(i%2))?SessionOptions::ThreadMode::tmAsync:SessionOptions::ThreadMode::tmSync,
195  (!(i%2))?SessionOptions::CleanupFlag::cfDelete:SessionOptions::CleanupFlag::cfNotDelete,
196  commandLine,
197  timeMax);
198  }
199 
200  subtasks[2].addSubtaskItem(subtasks[6]);
201  subtasks[2].addSubtaskItem(subtasks[7]);
202 
203  subtasks[3].addSubtaskItem(subtasks[8]);
204  subtasks[3].addSubtaskItem(subtasks[9]);
205 
206  subtasks[4].addSubtaskItem(subtasks[10]);
207  subtasks[4].addSubtaskItem(subtasks[11]);
208 
209  subtasks[5].addSubtaskItem(subtasks[12]);
210  subtasks[5].addSubtaskItem(subtasks[13]);
211 
212 
213  subtasks[0].addSubtaskItem(subtasks[2]);
214  subtasks[0].addSubtaskItem(subtasks[3]);
215 
216  subtasks[1].addSubtaskItem(subtasks[4]);
217  subtasks[1].addSubtaskItem(subtasks[5]);
218 
219  inputJsonMessage.addSubtaskItem(subtasks[0]);
220  inputJsonMessage.addSubtaskItem(subtasks[1]);
221 
222  std::stringstream jsonStream;
223  jsonStream << inputJsonMessage;
224 
225  return jsonStream.str();
226 }
227 //-----------------------------------------------------------------------------
228 std::string DRCECleanupTest::getInputJsonCleanup(unsigned int startTaskId, SessionOptions::ThreadMode taskThreadMode,
229  SessionOptions::CleanupFlag cleanupFlag, const std::string& commandLine)
230 {
231  DRCEInputJsonMessage inputJsonMessage = getInputJsonMessage(startTaskId, taskThreadMode, SessionOptions::CleanupFlag::cfNotDelete, commandLine);
232 
233  const unsigned int SZ = 14;
234  DRCEInputJsonMessage subtasks[SZ];
235  for (unsigned int i=0;i<SZ;++i)
236  {
237  subtasks[i] = getInputJsonMessage(++startTaskId,
238  (!(i%2))?SessionOptions::ThreadMode::tmAsync:SessionOptions::ThreadMode::tmSync, cleanupFlag, commandLine);
239  }
240 
241  subtasks[2].addSubtaskItem(subtasks[6]);
242  subtasks[2].addSubtaskItem(subtasks[7]);
243 
244  subtasks[3].addSubtaskItem(subtasks[8]);
245  subtasks[3].addSubtaskItem(subtasks[9]);
246 
247  subtasks[4].addSubtaskItem(subtasks[10]);
248  subtasks[4].addSubtaskItem(subtasks[11]);
249 
250  subtasks[5].addSubtaskItem(subtasks[12]);
251  subtasks[5].addSubtaskItem(subtasks[13]);
252 
253 
254  subtasks[0].addSubtaskItem(subtasks[2]);
255  subtasks[0].addSubtaskItem(subtasks[3]);
256 
257  subtasks[1].addSubtaskItem(subtasks[4]);
258  subtasks[1].addSubtaskItem(subtasks[5]);
259 
260  inputJsonMessage.addSubtaskItem(subtasks[0]);
261  inputJsonMessage.addSubtaskItem(subtasks[1]);
262 
263  std::stringstream jsonStream;
264  jsonStream << inputJsonMessage;
265 
266  return jsonStream.str();
267 }
268 //-----------------------------------------------------------------------------
269 bool DRCECleanupTest::isExistTaskData(unsigned int taskId, DRCENodeOptions& nodeOptions)
270 {
271  Poco::File dataFile(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), taskId));
272  Poco::File statusFile(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), taskId));
273  Poco::File requestFile(FileStream::getRequestFileName(nodeOptions.getTasksDataDir(), taskId));
274 
275  return (dataFile.exists() && statusFile.exists() && requestFile.exists());
276 }
277 //-----------------------------------------------------------------------------
278 bool DRCECleanupTest::isNotExistTaskData(unsigned int taskId, DRCENodeOptions& nodeOptions)
279 {
280  Poco::File dataFile(FileStream::getDataFileName(nodeOptions.getTasksDataDir(), taskId));
281  Poco::File statusFile(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), taskId));
282  Poco::File requestFile(FileStream::getRequestFileName(nodeOptions.getTasksDataDir(), taskId));
283 
284  return (!dataFile.exists() && !statusFile.exists() && !requestFile.exists());
285 }
286 //-----------------------------------------------------------------------------
287 void DRCECleanupTest::cleanupDir(const std::string& dirName)
288 {
289  Poco::File directory(dirName);
290  if (directory.exists())
291  {
292  std::vector<std::string> files;
293  directory.list(files);
294  for (const std::string fileName : files)
295  {
296  Poco::File file(fileName);
297  if (file.exists())
298  file.remove();
299  }
300  }
301 }
302 //-----------------------------------------------------------------------------
303 void DRCECleanupTest::cleanupTestData(DRCENodeOptions& nodeOptions)
304 {
305 #if CLEANUP_TEST == 1
306  cleanupDir(nodeOptions.getTasksDataDir());
307  cleanupDir(nodeOptions.getTasksStatusDir());
308 #endif
309 }
310 //-----------------------------------------------------------------------------
311 //-----------------------------------------------------------------------------
312 void DRCECleanupTest::testAsyncTaskAsyncSubtasks(void)
313 {
314  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
315  cleanupTestData(nodeOptions);
316  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
317  fObj.setNodeHost("NODE_HOST");
318  fObj.setNodePort("NODE_PORT");
319  fObj.setNodeName("NODE_NAME");
320 
321  unsigned int startTaskId = 600;
322 
323  std::string sourceJson = getInputJson(startTaskId, SessionOptions::ThreadMode::tmAsync, SessionOptions::ThreadMode::tmAsync);
324 
325 // std::cout << "sourceJson: " << sourceJson << std::endl;
326 
327  std::string resultJson = fObj.Process(sourceJson);
328 
329  ASSERT_FALSE(fObj.isError()) << printError(fObj);
330  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
331  ASSERT_TRUE(fObj.getErrorMsg().empty());
332  printSuccess("DRCEFunctionalObject Process");
333 
334 // std::cout << "result JSON: " << resultJson << std::endl;
335 
336  // check right result (unserialize and compare two objects)
337  DRCEOutputJsonMessage resultJsonMessage(resultJson);
338 
339  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
340  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
341  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
342  printSuccess("DRCEOutputJsonMessage unserialize");
343 
344  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
345  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
346  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR);
347  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), "NODE_NAME");
348  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), "NODE_HOST");
349  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), "NODE_PORT");
350  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
351  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
352  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::FINISHED);
353  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
354 
355  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
356  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
357  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), "NODE_NAME");
358  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), "NODE_HOST");
359  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), "NODE_PORT");
360  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
361  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
362  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::FINISHED ||
363  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
364  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
365 
366  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
367  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
368  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), "NODE_NAME");
369  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), "NODE_HOST");
370  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), "NODE_PORT");
371  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
372  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
373  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::FINISHED ||
374  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
375  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
376 
377  while (fObj.getAsyncTasksCount())
378  {
379  Poco::Thread::sleep(5000);
380  std::cout << "wait end task" << std::endl;
381  }
382  Poco::Thread::sleep(2000);
383 
384  DRCEResultDataItem resultDataItem = DRCECleanupTest::extractResultData(startTaskId, fObj);
385 
386  ASSERT_TRUE(resultDataItem.getStderrStream().empty());
387  ASSERT_EQ(resultDataItem.getErrorCode(), NO_ERROR);
388  ASSERT_EQ(resultDataItem.getNodeName(), fObj.getNodeName());
389  ASSERT_EQ(resultDataItem.getNodeHost(), fObj.getNodeHost());
390  ASSERT_EQ(resultDataItem.getNodePort(), fObj.getNodePort());
391  ASSERT_EQ(resultDataItem.getState(), DRCETaskRequest::TaskState::FINISHED);
392  ASSERT_EQ(resultDataItem.getSubtasksCount(), 2);
393 
394  ASSERT_TRUE(resultDataItem.getSubtaskItem(0).getStderrStream().empty());
395  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getErrorCode(), NO_ERROR);
396  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeName(), fObj.getNodeName());
397  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
398  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodePort(), fObj.getNodePort());
399  ASSERT_TRUE(resultDataItem.getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
400  resultDataItem.getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
401  resultDataItem.getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::FINISHED ||
402  resultDataItem.getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
403  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getSubtasksCount(), 2);
404 
405  ASSERT_TRUE(resultDataItem.getSubtaskItem(1).getStderrStream().empty());
406  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getErrorCode(), NO_ERROR);
407  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeName(), fObj.getNodeName());
408  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
409  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodePort(), fObj.getNodePort());
410  ASSERT_TRUE(resultDataItem.getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
411  resultDataItem.getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
412  resultDataItem.getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::FINISHED ||
413  resultDataItem.getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
414  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getSubtasksCount(), 2);
415 
416  for (unsigned int i=0;i<6;++i)
417  {
418  ASSERT_TRUE(isExistTaskData(startTaskId+i, nodeOptions));
419  }
420  cleanupTestData(nodeOptions);
421  printSuccess("!!! Test Async task and Async subtasks");
422 }
423 //-----------------------------------------------------------------------------
424 void DRCECleanupTest::testAsyncTaskSyncSubtasks(void)
425 {
426  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
427  cleanupTestData(nodeOptions);
428  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
429  fObj.setNodeHost("NODE_HOST");
430  fObj.setNodePort("NODE_PORT");
431  fObj.setNodeName("NODE_NAME");
432 
433  unsigned int startTaskId = 600;
434 
435  std::string sourceJson = getInputJson(startTaskId, SessionOptions::ThreadMode::tmAsync, SessionOptions::ThreadMode::tmSync);
436 
437 // std::cout << "sourceJson: " << sourceJson << std::endl;
438 
439  std::string resultJson = fObj.Process(sourceJson);
440 
441  ASSERT_FALSE(fObj.isError()) << printError(fObj);
442  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
443  ASSERT_TRUE(fObj.getErrorMsg().empty());
444  printSuccess("DRCEFunctionalObject Process");
445 
446 // std::cout << "result JSON: " << resultJson << std::endl;
447 
448  // check right result (unserialize and compare two objects)
449  DRCEOutputJsonMessage resultJsonMessage(resultJson);
450 
451  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
452  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
453  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
454  printSuccess("DRCEOutputJsonMessage unserialize");
455 
456  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
457  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
458  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR);
459  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
460  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
461  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
462  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
463  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
464  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::FINISHED)
465  << "State: " << resultJsonMessage.getResultData().getDataItem(0).getState();
466  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
467 
468  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
469  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
470  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
471  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
472  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
473  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
474  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
475  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::FINISHED ||
476  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
477  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
478 
479  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
480  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
481  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
482  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
483  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
484  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
485  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
486  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::FINISHED ||
487  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN)
488  << "State: " << resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState();
489  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
490 
491  while (fObj.getAsyncTasksCount())
492  {
493  Poco::Thread::sleep(5000);
494  std::cout << "wait end task" << std::endl;
495  }
496  Poco::Thread::sleep(2000);
497 
498  DRCEResultDataItem resultDataItem = DRCECleanupTest::extractResultData(startTaskId, fObj);
499 
500  ASSERT_TRUE(resultDataItem.getStderrStream().empty());
501  ASSERT_EQ(resultDataItem.getErrorCode(), NO_ERROR);
502  ASSERT_EQ(resultDataItem.getNodeName(), fObj.getNodeName());
503  ASSERT_EQ(resultDataItem.getNodeHost(), fObj.getNodeHost());
504  ASSERT_EQ(resultDataItem.getNodePort(), fObj.getNodePort());
505  ASSERT_EQ(resultDataItem.getState(), DRCETaskRequest::TaskState::FINISHED);
506  ASSERT_EQ(resultDataItem.getSubtasksCount(), 2);
507 
508  ASSERT_TRUE(resultDataItem.getSubtaskItem(0).getStderrStream().empty());
509  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getErrorCode(), NO_ERROR);
510  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeName(), fObj.getNodeName());
511  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
512  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodePort(), fObj.getNodePort());
513  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
514  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getSubtasksCount(), 2);
515 
516  ASSERT_TRUE(resultDataItem.getSubtaskItem(1).getStderrStream().empty());
517  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getErrorCode(), NO_ERROR);
518  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeName(), fObj.getNodeName());
519  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
520  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodePort(), fObj.getNodePort());
521  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::FINISHED);
522  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getSubtasksCount(), 2);
523 
524  for (unsigned int i=0;i<6;++i)
525  {
526  ASSERT_TRUE(isExistTaskData(startTaskId+i, nodeOptions));
527  }
528  cleanupTestData(nodeOptions);
529  printSuccess("!!! Test Async task and Sync subtasks");
530 }
531 //-----------------------------------------------------------------------------
532 void DRCECleanupTest::testSyncTaskAsyncSubtasks(void)
533 {
534  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
535  cleanupTestData(nodeOptions);
536  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
537  fObj.setNodeHost("NODE_HOST");
538  fObj.setNodePort("NODE_PORT");
539  fObj.setNodeName("NODE_NAME");
540 
541  unsigned int startTaskId = 600;
542 
543  std::string sourceJson = getInputJson(startTaskId, SessionOptions::ThreadMode::tmSync, SessionOptions::ThreadMode::tmAsync);
544 
545 // std::cout << "sourceJson: " << sourceJson << std::endl;
546 
547  std::string resultJson = fObj.Process(sourceJson);
548 
549  ASSERT_FALSE(fObj.isError()) << printError(fObj);
550  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
551  ASSERT_TRUE(fObj.getErrorMsg().empty());
552  printSuccess("DRCEFunctionalObject Process");
553 
554 // std::cout << "result JSON: " << resultJson << std::endl;
555 
556  // check right result (unserialize and compare two objects)
557  DRCEOutputJsonMessage resultJsonMessage(resultJson);
558 
559  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
560  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
561  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
562  printSuccess("DRCEOutputJsonMessage unserialize");
563 
564  while (fObj.getAsyncTasksCount())
565  {
566  Poco::Thread::sleep(5000);
567  std::cout << "wait end task" << std::endl;
568  }
569  Poco::Thread::sleep(2000);
570 
571  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
572  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
573  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR);
574  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
575  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
576  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
577  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
578  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
579 
580  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
581  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
582  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
583  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
584  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
585  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
586  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
587  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::FINISHED);
588  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
589 
590  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
591  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
592  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
593  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
594  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
595  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
596  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
597  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::FINISHED)
598  << "State: " << resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState();
599  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
600 
601  for (unsigned int i=0;i<6;++i)
602  {
603  ASSERT_TRUE(isExistTaskData(startTaskId+i, nodeOptions));
604  }
605  cleanupTestData(nodeOptions);
606  printSuccess("!!! Test Sync task and Async subtasks");
607 }
608 //-----------------------------------------------------------------------------
609 void DRCECleanupTest::testSyncTaskSyncSubtasks(void)
610 {
611  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
612  cleanupTestData(nodeOptions);
613  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
614  fObj.setNodeHost("NODE_HOST");
615  fObj.setNodePort("NODE_PORT");
616  fObj.setNodeName("NODE_NAME");
617 
618  unsigned int startTaskId = 600;
619 
620  std::string sourceJson = getInputJson(startTaskId, SessionOptions::ThreadMode::tmSync, SessionOptions::ThreadMode::tmSync);
621 
622 // std::cout << "sourceJson: " << sourceJson << std::endl;
623 
624  std::string resultJson = fObj.Process(sourceJson);
625 
626  ASSERT_FALSE(fObj.isError()) << printError(fObj);
627  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
628  ASSERT_TRUE(fObj.getErrorMsg().empty());
629  printSuccess("DRCEFunctionalObject Process");
630 
631 // std::cout << "result JSON: " << resultJson << std::endl;
632 
633  // check right result (unserialize and compare two objects)
634  DRCEOutputJsonMessage resultJsonMessage(resultJson);
635 
636  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
637  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
638  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
639  printSuccess("DRCEOutputJsonMessage unserialize");
640 
641  while (fObj.getAsyncTasksCount() || fObj.hasTask(startTaskId))
642  {
643  Poco::Thread::sleep(5000);
644  std::cout << "wait end task" << std::endl;
645  }
646  Poco::Thread::sleep(5000);
647 
648  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
649  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
650  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR);
651  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
652  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
653  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
654  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
655  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
656 
657  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
658  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
659  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
660  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
661  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
662  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
663  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
664 
665  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
666  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
667  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
668  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
669  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
670  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::FINISHED);
671  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
672 
673  for (unsigned int i=0;i<6;++i)
674  {
675  ASSERT_TRUE(isExistTaskData((startTaskId+i), nodeOptions)) << " Task ID: " << (startTaskId+i);
676  }
677  cleanupTestData(nodeOptions);
678  printSuccess("!!! Test Sync task and Sync subtasks");
679 }
680 //-----------------------------------------------------------------------------
681 void DRCECleanupTest::testAsyncTaskMixSubtasks(void)
682 {
683  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
684  cleanupTestData(nodeOptions);
685  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
686  fObj.setNodeHost("NODE_HOST");
687  fObj.setNodePort("NODE_PORT");
688  fObj.setNodeName("NODE_NAME");
689 
690  unsigned int startTaskId = 600;
691 
692  std::string sourceJson = getInputJson(startTaskId, SessionOptions::ThreadMode::tmAsync);
693 
694 // std::cout << "sourceJson: " << sourceJson << std::endl;
695 
696  std::string resultJson = fObj.Process(sourceJson);
697 
698  ASSERT_FALSE(fObj.isError()) << printError(fObj);
699  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
700  ASSERT_TRUE(fObj.getErrorMsg().empty());
701  printSuccess("DRCEFunctionalObject Process");
702 
703 // std::cout << "result JSON: " << resultJson << std::endl;
704 
705  // check right result (unserialize and compare two objects)
706  DRCEOutputJsonMessage resultJsonMessage(resultJson);
707 
708  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
709  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
710  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
711  printSuccess("DRCEOutputJsonMessage unserialize");
712 
713  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
714  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
715  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR);
716  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
717  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
718  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
719  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
720  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
721  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::FINISHED);
722  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
723 
724  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
725  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
726  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
727  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
728  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
729  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
730  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
731  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::FINISHED ||
732  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
733  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
734 
735  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
736  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
737  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
738  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
739  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
740  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
741  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
742  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::FINISHED ||
743  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
744  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
745 
746  while (fObj.getAsyncTasksCount())
747  {
748  Poco::Thread::sleep(5000);
749  std::cout << "wait end task" << std::endl;
750  }
751  Poco::Thread::sleep(2000);
752 
753  DRCEResultDataItem resultDataItem = DRCECleanupTest::extractResultData(startTaskId, fObj);
754 
755  ASSERT_TRUE(resultDataItem.getStderrStream().empty());
756  ASSERT_EQ(resultDataItem.getErrorCode(), NO_ERROR);
757  ASSERT_EQ(resultDataItem.getNodeName(), fObj.getNodeName());
758  ASSERT_EQ(resultDataItem.getNodeHost(), fObj.getNodeHost());
759  ASSERT_EQ(resultDataItem.getNodePort(), fObj.getNodePort());
760  ASSERT_EQ(resultDataItem.getState(), DRCETaskRequest::TaskState::FINISHED);
761  ASSERT_EQ(resultDataItem.getSubtasksCount(), 2);
762 
763  ASSERT_TRUE(resultDataItem.getSubtaskItem(0).getStderrStream().empty());
764  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getErrorCode(), NO_ERROR);
765  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeName(), fObj.getNodeName());
766  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
767  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodePort(), fObj.getNodePort());
768  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
769  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getSubtasksCount(), 2);
770 
771  ASSERT_TRUE(resultDataItem.getSubtaskItem(1).getStderrStream().empty());
772  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getErrorCode(), NO_ERROR);
773  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeName(), fObj.getNodeName());
774  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
775  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodePort(), fObj.getNodePort());
776  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::FINISHED);
777  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getSubtasksCount(), 2);
778 
779  for (unsigned int i=0;i<14;++i)
780  {
781  ASSERT_TRUE(isExistTaskData(startTaskId+i, nodeOptions));
782  }
783  cleanupTestData(nodeOptions);
784  printSuccess("!!! Test Async task and Mix subtasks");
785 }
786 //-----------------------------------------------------------------------------
787 void DRCECleanupTest::testSyncTaskMixSubtasks(void)
788 {
789  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
790  cleanupTestData(nodeOptions);
791  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
792  fObj.setNodeHost("NODE_HOST");
793  fObj.setNodePort("NODE_PORT");
794  fObj.setNodeName("NODE_NAME");
795 
796  unsigned int startTaskId = 600;
797 
798  std::string sourceJson = getInputJson(startTaskId, SessionOptions::ThreadMode::tmSync);
799 
800 // std::cout << "sourceJson: " << sourceJson << std::endl;
801 
802  std::string resultJson = fObj.Process(sourceJson);
803 
804  ASSERT_FALSE(fObj.isError()) << printError(fObj);
805  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
806  ASSERT_TRUE(fObj.getErrorMsg().empty());
807  printSuccess("DRCEFunctionalObject Process");
808 
809 // std::cout << "result JSON: " << resultJson << std::endl;
810 
811  // check right result (unserialize and compare two objects)
812  DRCEOutputJsonMessage resultJsonMessage(resultJson);
813 
814  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
815  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
816  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
817  printSuccess("DRCEOutputJsonMessage unserialize");
818 
819  while (fObj.getAsyncTasksCount())
820  {
821  Poco::Thread::sleep(5000);
822  std::cout << "wait end task" << std::endl;
823  }
824  Poco::Thread::sleep(5000);
825 
826  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
827  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
828  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR);
829  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
830  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
831  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
832  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
833  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
834 
835  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
836  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
837  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
838  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
839  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
840  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
841  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
842  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState() == DRCETaskRequest::TaskState::FINISHED);
843  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
844 
845  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
846  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
847  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
848  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
849  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
850  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
851  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
852  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState() == DRCETaskRequest::TaskState::FINISHED);
853  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
854 
855  for (unsigned int i=0;i<14;++i)
856  {
857  ASSERT_TRUE(isExistTaskData(startTaskId+i, nodeOptions));
858  }
859  cleanupTestData(nodeOptions);
860  printSuccess("!!! Test Sync task and Mix subtasks");
861 }
862 //-----------------------------------------------------------------------------
863 void DRCECleanupTest::testAsyncTaskMixSubtasksCleanup(void)
864 {
865  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
866  cleanupTestData(nodeOptions);
867  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
868  fObj.setNodeHost("NODE_HOST");
869  fObj.setNodePort("NODE_PORT");
870  fObj.setNodeName("NODE_NAME");
871 
872  unsigned int startTaskId = 600;
873 
874  std::string sourceJson = getInputJsonCleanup(startTaskId, SessionOptions::ThreadMode::tmAsync, "");
875 
876 // std::cout << "sourceJson: " << sourceJson << std::endl;
877 
878  std::string resultJson = fObj.Process(sourceJson);
879 
880  ASSERT_FALSE(fObj.isError()) << printError(fObj);
881  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
882  ASSERT_TRUE(fObj.getErrorMsg().empty());
883  printSuccess("DRCEFunctionalObject Process");
884 
885  std::cout << "result JSON: " << resultJson << std::endl;
886 
887  // check right result (unserialize and compare two objects)
888  DRCEOutputJsonMessage resultJsonMessage(resultJson);
889 
890  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
891  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
892  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
893  printSuccess("DRCEOutputJsonMessage unserialize");
894 
895  while (fObj.getAsyncTasksCount() || fObj.hasTask(startTaskId))
896  {
897  Poco::Thread::sleep(5000);
898  std::cout << "wait end task" << std::endl;
899  }
900  Poco::Thread::sleep(2000);
901 
902  DRCEResultDataItem resultDataItem = DRCECleanupTest::extractResultData(startTaskId, fObj);
903 
904  ASSERT_TRUE(resultDataItem.getStderrStream().empty());
905  ASSERT_EQ(resultDataItem.getErrorCode(), NO_ERROR);
906  ASSERT_EQ(resultDataItem.getNodeName(), fObj.getNodeName());
907  ASSERT_EQ(resultDataItem.getNodeHost(), fObj.getNodeHost());
908  ASSERT_EQ(resultDataItem.getNodePort(), fObj.getNodePort());
909  ASSERT_EQ(resultDataItem.getState(), DRCETaskRequest::TaskState::FINISHED);
910  ASSERT_EQ(resultDataItem.getSubtasksCount(), 2);
911 
912  ASSERT_TRUE(resultDataItem.getSubtaskItem(0).getStderrStream().empty());
913  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getErrorCode(), NO_ERROR);
914  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeName(), fObj.getNodeName());
915  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
916  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodePort(), fObj.getNodePort());
917  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
918  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getSubtasksCount(), 2);
919 
920  ASSERT_TRUE(resultDataItem.getSubtaskItem(1).getStderrStream().empty());
921  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getErrorCode(), NO_ERROR);
922  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeName(), fObj.getNodeName());
923  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
924  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodePort(), fObj.getNodePort());
925  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::FINISHED);
926  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getSubtasksCount(), 2);
927 
928  ASSERT_TRUE(isExistTaskData(startTaskId, nodeOptions));
929  ASSERT_TRUE(isExistTaskData(startTaskId+2, nodeOptions));
930  ASSERT_TRUE(isExistTaskData(startTaskId+6, nodeOptions));
931  ASSERT_TRUE(isExistTaskData(startTaskId+14, nodeOptions));
932 
933  cleanupTestData(nodeOptions);
934  printSuccess("!!! Test Async task and Mix subtasks Cleanup");
935 }
936 //-----------------------------------------------------------------------------
937 void DRCECleanupTest::testSyncTaskMixSubtasksCleanup(void)
938 {
939  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
940  cleanupTestData(nodeOptions);
941  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
942  fObj.setNodeHost("NODE_HOST");
943  fObj.setNodePort("NODE_PORT");
944  fObj.setNodeName("NODE_NAME");
945 
946  unsigned int startTaskId = 600;
947 
948  std::string sourceJson = getInputJsonCleanup(startTaskId, SessionOptions::ThreadMode::tmSync, "");
949 
950 // std::cout << "sourceJson: " << sourceJson << std::endl;
951 
952  std::string resultJson = fObj.Process(sourceJson);
953 
954  ASSERT_FALSE(fObj.isError()) << printError(fObj);
955  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
956  ASSERT_TRUE(fObj.getErrorMsg().empty());
957  printSuccess("DRCEFunctionalObject Process");
958 
959 // std::cout << "result JSON: " << resultJson << std::endl;
960 
961  // check right result (unserialize and compare two objects)
962  DRCEOutputJsonMessage resultJsonMessage(resultJson);
963 
964  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
965  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
966  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
967  printSuccess("DRCEOutputJsonMessage unserialize");
968 
969  while (fObj.getAsyncTasksCount())
970  {
971  Poco::Thread::sleep(5000);
972  std::cout << "wait end task" << std::endl;
973  }
974  Poco::Thread::sleep(2000);
975 
976  DRCEResultDataItem resultDataItem = DRCECleanupTest::extractResultData(startTaskId, fObj);
977 
978  ASSERT_TRUE(resultDataItem.getStderrStream().empty());
979  ASSERT_EQ(resultDataItem.getErrorCode(), NO_ERROR);
980  ASSERT_EQ(resultDataItem.getNodeName(), fObj.getNodeName());
981  ASSERT_EQ(resultDataItem.getNodeHost(), fObj.getNodeHost());
982  ASSERT_EQ(resultDataItem.getNodePort(), fObj.getNodePort());
983  ASSERT_EQ(resultDataItem.getState(), DRCETaskRequest::TaskState::FINISHED);
984  ASSERT_EQ(resultDataItem.getSubtasksCount(), 2);
985 
986  ASSERT_TRUE(resultDataItem.getSubtaskItem(0).getStderrStream().empty());
987  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getErrorCode(), NO_ERROR);
988  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeName(), fObj.getNodeName());
989  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
990  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getNodePort(), fObj.getNodePort());
991  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getState(), DRCETaskRequest::TaskState::FINISHED);
992  ASSERT_EQ(resultDataItem.getSubtaskItem(0).getSubtasksCount(), 2);
993 
994  ASSERT_TRUE(resultDataItem.getSubtaskItem(1).getStderrStream().empty());
995  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getErrorCode(), NO_ERROR);
996  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeName(), fObj.getNodeName());
997  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
998  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getNodePort(), fObj.getNodePort());
999  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::FINISHED);
1000  ASSERT_EQ(resultDataItem.getSubtaskItem(1).getSubtasksCount(), 2);
1001 
1002  ASSERT_TRUE(isExistTaskData(startTaskId, nodeOptions));
1003  ASSERT_TRUE(isExistTaskData(startTaskId+2, nodeOptions));
1004  ASSERT_TRUE(isExistTaskData(startTaskId+6, nodeOptions));
1005  ASSERT_TRUE(isExistTaskData(startTaskId+14, nodeOptions));
1006 
1007  cleanupTestData(nodeOptions);
1008  printSuccess("!!! Test Sync task and Mix subtasks Cleanup");
1009 }
1010 //-----------------------------------------------------------------------------
1011 void DRCECleanupTest::testRequestGetDataCleanup(void)
1012 {
1013  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
1014  cleanupTestData(nodeOptions);
1015  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
1016  fObj.setNodeHost("NODE_HOST");
1017  fObj.setNodePort("NODE_PORT");
1018  fObj.setNodeName("NODE_NAME");
1019 
1020  unsigned int startTaskId = 600;
1021 
1022  std::string sourceJson = getInputJsonCleanup(startTaskId, SessionOptions::ThreadMode::tmAsync, "");
1023 
1024 // std::cout << "sourceJson: " << sourceJson << std::endl;
1025 
1026  std::string resultJson = fObj.Process(sourceJson);
1027 
1028  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1029  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1030  ASSERT_TRUE(fObj.getErrorMsg().empty());
1031  printSuccess("DRCEFunctionalObject Process");
1032 
1033 // std::cout << "result JSON: " << resultJson << std::endl;
1034 
1035  // check right result (unserialize and compare two objects)
1036  DRCEOutputJsonMessage resultJsonMessage(resultJson);
1037 
1038  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1039  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
1040  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
1041  printSuccess("DRCEOutputJsonMessage unserialize");
1042 
1043  while (fObj.getAsyncTasksCount() || fObj.hasTask(startTaskId))
1044  {
1045  Poco::Thread::sleep(5000);
1046  std::cout << "wait end task" << std::endl;
1047  }
1048  Poco::Thread::sleep(5000);
1049 
1050  // make get data request
1051  std::string json;
1052  DRCETaskRequestGetData taskRequestGetData;
1053  taskRequestGetData.setFetchType(DRCETaskRequestGetData::FetchType::ftDeleteDataAfterFetch);
1054  taskRequestGetData.serialize(json);
1055  ASSERT_FALSE(taskRequestGetData.isError());
1056  printSuccess("DRCETaskRequestGetData serialize");
1057 
1058 // std::cout << "json: " << json << std::endl;
1059 
1060  // set request for execute task
1061  DRCEInputJsonMessage inputJsonMessage;
1062  inputJsonMessage.setRequestType(DRCETaskRequest::RequestType::rtGetTaskData);
1063  inputJsonMessage.setRequestData(json);
1064  inputJsonMessage.setRequestId(startTaskId);
1065 
1066  std::stringstream sourceJsonStream;
1067  sourceJsonStream << inputJsonMessage;
1068 
1069 // std::cout << "source JSON: " << sourceJson.str() << std::endl;
1070 
1071  ASSERT_FALSE(inputJsonMessage.isError()) << printError(inputJsonMessage);
1072  ASSERT_EQ(inputJsonMessage.getErrorCode(), NO_ERROR);
1073  ASSERT_TRUE(inputJsonMessage.getErrorMsg().empty());
1074  printSuccess("DRCEInputJsonMessage serialize");
1075 
1076  resultJson = fObj.Process(sourceJsonStream.str());
1077 
1078  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1079  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1080  ASSERT_TRUE(fObj.getErrorMsg().empty());
1081  printSuccess("DRCEFunctionalObject Process");
1082 
1083 // std::cout << "result JSON: " << resultJson << std::endl;
1084 
1085  // check right result (unserialize and compare two objects)
1086  resultJsonMessage.unserialize(resultJson);
1087 
1088  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1089  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
1090  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
1091  printSuccess("DRCEOutputJsonMessage unserialize");
1092 
1093  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
1094  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
1095  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR);
1096  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
1097  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
1098  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
1099  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::FINISHED ||
1100  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::IN_PROGRESS ||
1101  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::SET_AS_NEW ||
1102  resultJsonMessage.getResultData().getDataItem(0).getState() == DRCETaskRequest::TaskState::QUEUED_TO_RUN);
1103  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
1104 
1105  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
1106  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
1107  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
1108  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
1109  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
1110  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState()==DRCETaskRequest::TaskState::FINISHED ||
1111  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState()==DRCETaskRequest::TaskState::IN_PROGRESS ||
1112  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState()==DRCETaskRequest::TaskState::SET_AS_NEW ||
1113  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState()==DRCETaskRequest::TaskState::QUEUED_TO_RUN);
1114  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
1115 
1116  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
1117  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
1118  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
1119  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
1120  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
1121  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::FINISHED ||
1122  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState()==DRCETaskRequest::TaskState::IN_PROGRESS ||
1123  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState()==DRCETaskRequest::TaskState::SET_AS_NEW ||
1124  resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState()==DRCETaskRequest::TaskState::QUEUED_TO_RUN);
1125  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
1126 
1127  Poco::Thread::sleep(5000);
1128  for (unsigned int i=0;i<14;++i)
1129  {
1130  ASSERT_TRUE(isNotExistTaskData((startTaskId+i), nodeOptions)) << "Task ID: " << (startTaskId+i);
1131  }
1132 
1133  cleanupTestData(nodeOptions);
1134  printSuccess("!!! Test Request Get Data with Cleanup");
1135 }
1136 //-----------------------------------------------------------------------------
1137 void DRCECleanupTest::testRequestDeleteDataCleanup(void)
1138 {
1139  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
1140  cleanupTestData(nodeOptions);
1141  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
1142  fObj.setNodeHost("NODE_HOST");
1143  fObj.setNodePort("NODE_PORT");
1144  fObj.setNodeName("NODE_NAME");
1145 
1146  unsigned int startTaskId = 700;
1147 
1148  std::string sourceJson = getInputJsonCleanup(startTaskId, SessionOptions::ThreadMode::tmAsync, "echo \"Hello\" && sleep 5 && ps ", 100000);
1149 
1150 // std::cout << "sourceJson: " << sourceJson << std::endl;
1151 
1152  std::string resultJson = fObj.Process(sourceJson);
1153 
1154  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1155  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1156  ASSERT_TRUE(fObj.getErrorMsg().empty());
1157  printSuccess("DRCEFunctionalObject Process");
1158 
1159 // std::cout << "result JSON: " << resultJson << std::endl;
1160 
1161  // check right result (unserialize and compare two objects)
1162  DRCEOutputJsonMessage resultJsonMessage(resultJson);
1163 
1164  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1165  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
1166  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
1167  printSuccess("DRCEOutputJsonMessage unserialize");
1168 
1169  while (fObj.getAsyncTasksCount() || fObj.hasTask(startTaskId))
1170  {
1171  Poco::Thread::sleep(10000);
1172  std::cout << "wait end task" << std::endl;
1173  }
1174  Poco::Thread::sleep(10000);
1175 
1176  // make get data request
1177  std::string json;
1178  DRCETaskRequestDeleteData taskRequestDeleteData;
1179  taskRequestDeleteData.serialize(json);
1180  ASSERT_FALSE(taskRequestDeleteData.isError());
1181  printSuccess("DRCETaskRequestDeleteData serialize");
1182 
1183 // std::cout << "json: " << json << std::endl;
1184 
1185  // set request for execute task
1186  DRCEInputJsonMessage inputJsonMessage;
1187  inputJsonMessage.setRequestType(DRCETaskRequest::RequestType::rtDeleteTaskData);
1188  inputJsonMessage.setRequestData(json);
1189  inputJsonMessage.setRequestId(startTaskId);
1190 
1191  std::stringstream sourceJsonStream;
1192  sourceJsonStream << inputJsonMessage;
1193 
1194 // std::cout << "source JSON: " << sourceJson.str() << std::endl;
1195 
1196  ASSERT_FALSE(inputJsonMessage.isError()) << printError(inputJsonMessage);
1197  ASSERT_EQ(inputJsonMessage.getErrorCode(), NO_ERROR) << printError(inputJsonMessage);
1198  ASSERT_TRUE(inputJsonMessage.getErrorMsg().empty()) << printError(inputJsonMessage);
1199  printSuccess("DRCEInputJsonMessage serialize");
1200 
1201  resultJson = fObj.Process(sourceJsonStream.str());
1202 
1203  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1204  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1205  ASSERT_TRUE(fObj.getErrorMsg().empty());
1206  printSuccess("DRCEFunctionalObject Process");
1207 
1208  std::cout << "result JSON: " << resultJson << std::endl;
1209 
1210  // check right result (unserialize and compare two objects)
1211  resultJsonMessage.unserialize(resultJson);
1212 
1213  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1214  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR) << printError(resultJsonMessage);
1215  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty()) << printError(resultJsonMessage);
1216  printSuccess("DRCEOutputJsonMessage unserialize");
1217 
1218  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
1219  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
1220  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR) << resultJsonMessage.getResultData().getDataItem(0).getErrorCode();
1221  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
1222  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
1223  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
1224  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getState(), DRCETaskRequest::TaskState::DELETED);
1225  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
1226 
1227  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
1228  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR) << resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode();
1229  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
1230  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
1231  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
1232  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState(), DRCETaskRequest::TaskState::DELETED);
1233  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 2);
1234 
1235  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
1236  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR) << resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode();
1237  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
1238  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
1239  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
1240  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::DELETED);
1241  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 2);
1242 
1243  Poco::Thread::sleep(5000);
1244  for (unsigned int i=0;i<14;++i)
1245  {
1246  ASSERT_TRUE(isNotExistTaskData((startTaskId+i), nodeOptions)) << "Task ID: " << (startTaskId+i);
1247  }
1248 
1249  cleanupTestData(nodeOptions);
1250  printSuccess("!!! Test Request Delete Data with Cleanup");
1251 }
1252 //-----------------------------------------------------------------------------
1253 void DRCECleanupTest::testRequestTerminateCleanup(void)
1254 {
1255  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
1256  cleanupTestData(nodeOptions);
1257  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
1258  fObj.setNodeHost("NODE_HOST");
1259  fObj.setNodePort("NODE_PORT");
1260  fObj.setNodeName("NODE_NAME");
1261 
1262  unsigned int startTaskId(600), terminateTaskId(603);
1263  const std::string commandLine = "echo \"scale=400; 4*a(1)\" | bc -l && sleep 10 && ps xau";
1264 
1265  std::string sourceJson = getInputJsonCleanup(startTaskId, SessionOptions::ThreadMode::tmAsync, SessionOptions::CleanupFlag::cfNotDelete, commandLine);
1266 
1267 // std::cout << "sourceJson: " << sourceJson << std::endl;
1268 
1269  std::string resultJson = fObj.Process(sourceJson);
1270 
1271  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1272  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1273  ASSERT_TRUE(fObj.getErrorMsg().empty());
1274  printSuccess("DRCEFunctionalObject Process");
1275 
1276 // std::cout << "result JSON: " << resultJson << std::endl;
1277 
1278  // check right result (unserialize and compare two objects)
1279  DRCEOutputJsonMessage resultJsonMessage(resultJson);
1280 
1281  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1282  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
1283  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
1284  printSuccess("DRCEOutputJsonMessage unserialize");
1285 
1286  // wait timeout and make terminate request
1287  int tryCount = 10;
1288 
1289  Poco::File statusFile(FileStream::getStatusFileName(nodeOptions.getTasksStatusDir(), terminateTaskId));
1290  while (!statusFile.exists() || tryCount-->0)
1291  {
1292  Poco::Thread::sleep(2000);
1293  std::cout << "Wait appear task (" << terminateTaskId << ") in async queue ..." << std::endl;
1294  }
1295  Poco::Thread::sleep(2000);
1296 
1297  // make get data request
1298  std::string json;
1299  DRCETaskRequestTerminate taskRequestTerminate;
1300  taskRequestTerminate.setAlgorithmType(DRCETaskRequestTerminate::AlgorithmType::atDefault);
1301  taskRequestTerminate.setDelayValue(1000);
1302  taskRequestTerminate.setRepeatValue(3);
1303  taskRequestTerminate.setSignalValue(9);
1304  taskRequestTerminate.setCleanupFlag(DRCETaskRequestTerminate::CleanupFlag::cfDelete);
1305  taskRequestTerminate.serialize(json);
1306  ASSERT_FALSE(taskRequestTerminate.isError());
1307  printSuccess("DRCETaskRequestTerminate serialize");
1308 
1309 // std::cout << "json: " << json << std::endl;
1310 
1311  // set request for execute task
1312  DRCEInputJsonMessage inputJsonMessage;
1313  inputJsonMessage.setRequestType(DRCETaskRequest::RequestType::rtTerminateTask);
1314  inputJsonMessage.setRequestData(json);
1315  inputJsonMessage.setRequestId(terminateTaskId);
1316 
1317  std::stringstream sourceJsonStream;
1318  sourceJsonStream << inputJsonMessage;
1319 
1320 // std::cout << "source JSON: " << sourceJson.str() << std::endl;
1321 
1322  ASSERT_FALSE(inputJsonMessage.isError()) << printError(inputJsonMessage);
1323  ASSERT_EQ(inputJsonMessage.getErrorCode(), NO_ERROR);
1324  ASSERT_TRUE(inputJsonMessage.getErrorMsg().empty());
1325  printSuccess("DRCEInputJsonMessage serialize");
1326 
1327  resultJson = fObj.Process(sourceJsonStream.str());
1328 
1329  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1330  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1331  ASSERT_TRUE(fObj.getErrorMsg().empty());
1332  printSuccess("DRCEFunctionalObject Process");
1333 
1334  std::cout << "result JSON: " << resultJson << std::endl;
1335 
1336  // check right result (unserialize and compare two objects)
1337  resultJsonMessage.unserialize(resultJson);
1338 
1339  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1340  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
1341  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
1342  printSuccess("DRCEOutputJsonMessage unserialize");
1343 
1344  while (fObj.getAsyncTasksCount() || fObj.hasTask(startTaskId))
1345  {
1346  Poco::Thread::sleep(5000);
1347  std::cout << "wait end task" << std::endl;
1348  }
1349  Poco::Thread::sleep(5000);
1350 
1351  ASSERT_EQ(resultJsonMessage.getResultData().getItemsCount(), 1);
1352  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getRequestId(), 603);
1353  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getStderrStream().empty());
1354  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getErrorCode(), NO_ERROR); //??
1355  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeName(), fObj.getNodeName());
1356  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodeHost(), fObj.getNodeHost());
1357  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getNodePort(), fObj.getNodePort());
1358  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getState(), DRCETaskRequest::TaskState::TERMINATED);
1359  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtasksCount(), 2);
1360 
1361  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getStderrStream().empty());
1362  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getErrorCode(), NO_ERROR);
1363  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeName(), fObj.getNodeName());
1364  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodeHost(), fObj.getNodeHost());
1365  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getNodePort(), fObj.getNodePort());
1366  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getState(), DRCETaskRequest::TaskState::TERMINATED);
1367  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(0).getSubtasksCount(), 0);
1368 
1369  ASSERT_TRUE(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getStderrStream().empty());
1370  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getErrorCode(), NO_ERROR);
1371  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeName(), fObj.getNodeName());
1372  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodeHost(), fObj.getNodeHost());
1373  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getNodePort(), fObj.getNodePort());
1374  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getState(), DRCETaskRequest::TaskState::TERMINATED);
1375  ASSERT_EQ(resultJsonMessage.getResultData().getDataItem(0).getSubtaskItem(1).getSubtasksCount(), 0);
1376 
1377  Poco::Thread::sleep(10000);
1378  std::vector<unsigned int> cleanupTasks = {603, 607, 608};
1379  for (unsigned int i=0;i<cleanupTasks.size();++i)
1380  {
1381  ASSERT_TRUE(isNotExistTaskData(cleanupTasks[i], nodeOptions)) << "isNotExistTaskData cleanupTasks[" << i << "]: " << cleanupTasks[i];
1382  }
1383 
1384  for (unsigned int i=0;i<14;++i)
1385  {
1386  unsigned int taskId = startTaskId+i;
1387  bool isNotExist = false;
1388  for (unsigned int i=0;i<cleanupTasks.size();++i)
1389  {
1390  if (cleanupTasks[i]==taskId)
1391  {
1392  isNotExist = true;
1393  break;
1394  }
1395  }
1396  if (!isNotExist)
1397  {
1398  ASSERT_TRUE(isExistTaskData(taskId, nodeOptions)) << " isExistTaskData taskId: " << taskId;
1399  }
1400  }
1401 
1402  cleanupTestData(nodeOptions);
1403  printSuccess("!!! Test Request Terminate with Cleanup");
1404 }
1405 //-----------------------------------------------------------------------------
1406 void DRCECleanupTest::testRequestDeleteData(void)
1407 {
1408  DRCENodeOptions nodeOptions = DRCETaskThreadingModeTest::getNodeOptions();
1409  cleanupTestData(nodeOptions);
1410  DRCEFunctionalObject fObj(nodeOptions.getNodeName(), nodeOptions.getHomeDir(), nodeOptions.getTasksDataDir(), nodeOptions.getTasksStatusDir());
1411  fObj.setNodeHost("NODE_HOST");
1412  fObj.setNodePort("NODE_PORT");
1413  fObj.setNodeName("NODE_NAME");
1414 
1415  unsigned int startTaskId = 600;
1416 
1417  DRCEInputJsonMessage inputJsonMessage = getInputJsonMessage(startTaskId, SessionOptions::ThreadMode::tmAsync, SessionOptions::CleanupFlag::cfNotDelete,
1418  "echo \"Hello\" && sleep 5 && ps xau",
1419  300);
1420 
1421  std::stringstream jsonStream;
1422  jsonStream << inputJsonMessage;
1423 
1424  ASSERT_FALSE(inputJsonMessage.isError()) << printError(inputJsonMessage);
1425  ASSERT_EQ(inputJsonMessage.getErrorCode(), NO_ERROR) << printError(inputJsonMessage);
1426  ASSERT_TRUE(inputJsonMessage.getErrorMsg().empty()) << printError(inputJsonMessage);
1427 
1428 // std::cout << "sourceJson: " << sourceJson << std::endl;
1429 
1430  std::string resultJson = fObj.Process(jsonStream.str());
1431 
1432  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1433  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1434  ASSERT_TRUE(fObj.getErrorMsg().empty());
1435  printSuccess("DRCEFunctionalObject Process");
1436 
1437 // std::cout << "result JSON: " << resultJson << std::endl;
1438 
1439  // check right result (unserialize and compare two objects)
1440  DRCEOutputJsonMessage resultJsonMessage(resultJson);
1441 
1442  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1443  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR);
1444  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty());
1445  printSuccess("DRCEOutputJsonMessage unserialize");
1446 
1447  while (fObj.getAsyncTasksCount() || fObj.hasTask(startTaskId))
1448  {
1449  Poco::Thread::sleep(5000);
1450  std::cout << "wait end task" << std::endl;
1451  }
1452  Poco::Thread::sleep(5000);
1453 
1454  // make get data request
1455  std::string json;
1456  DRCETaskRequestDeleteData taskRequestDeleteData;
1457  taskRequestDeleteData.serialize(json);
1458  ASSERT_FALSE(taskRequestDeleteData.isError());
1459  printSuccess("DRCETaskRequestDeleteData serialize");
1460  // set request for execute task
1461  inputJsonMessage.clear();
1462  inputJsonMessage.setRequestType(DRCETaskRequest::RequestType::rtDeleteTaskData);
1463  inputJsonMessage.setRequestData(json);
1464  inputJsonMessage.setRequestId(startTaskId);
1465 
1466  std::stringstream sourceJsonStream;
1467  sourceJsonStream << inputJsonMessage;
1468 
1469  // std::cout << "source JSON: " << sourceJson.str() << std::endl;
1470 
1471  ASSERT_FALSE(inputJsonMessage.isError()) << printError(inputJsonMessage);
1472  ASSERT_EQ(inputJsonMessage.getErrorCode(), NO_ERROR) << printError(inputJsonMessage);
1473  ASSERT_TRUE(inputJsonMessage.getErrorMsg().empty()) << printError(inputJsonMessage);
1474  printSuccess("DRCEInputJsonMessage serialize");
1475 
1476  resultJson = fObj.Process(sourceJsonStream.str());
1477 
1478  ASSERT_FALSE(fObj.isError()) << printError(fObj);
1479  ASSERT_EQ(fObj.getErrorCode(), NO_ERROR);
1480  ASSERT_TRUE(fObj.getErrorMsg().empty());
1481  printSuccess("DRCEFunctionalObject Process");
1482 
1483 // std::cout << "result JSON: " << resultJson << std::endl;
1484 
1485  // check right result (unserialize and compare two objects)
1486  resultJsonMessage.unserialize(resultJson);
1487 
1488  ASSERT_FALSE(resultJsonMessage.isError()) << printError(resultJsonMessage);
1489  ASSERT_EQ(resultJsonMessage.getErrorCode(), NO_ERROR) << printError(resultJsonMessage);
1490  ASSERT_TRUE(resultJsonMessage.getErrorMsg().empty()) << printError(resultJsonMessage);
1491  printSuccess("DRCEOutputJsonMessage unserialize");
1492 }
1493 //-----------------------------------------------------------------------------
1494 //-----------------------------------------------------------------------------
1495 } // end namespace tests
1496 } // end namespace drce
1497 } // end namespace HCE