Fix job finish time and migrate job test to tck.

src/graph/executor/admin/SubmitJobExecutor.cpp

@@ -35,87 +35,97 @@ folly::Future<Status> SubmitJobExecutor::execute() {
           LOG(ERROR) << resp.status().toString();
           return std::move(resp).status();
         }
-        switch (jobOp) {
-          case meta::cpp2::AdminJobOp::ADD: {
-            nebula::DataSet v({"New Job Id"});
-            DCHECK(resp.value().job_id_ref().has_value());
-            if (!resp.value().job_id_ref().has_value()) {
-              return Status::Error("Response unexpected.");
-            }
-            v.emplace_back(nebula::Row({*resp.value().job_id_ref()}));
-            return finish(std::move(v));
-          }
-          case meta::cpp2::AdminJobOp::RECOVER: {
-            nebula::DataSet v({"Recovered job num"});
-            DCHECK(resp.value().recovered_job_num_ref().has_value());
-            if (!resp.value().recovered_job_num_ref().has_value()) {
-              return Status::Error("Response unexpected.");
-            }
-            v.emplace_back(nebula::Row({*resp.value().recovered_job_num_ref()}));
-            return finish(std::move(v));
-          }
-          case meta::cpp2::AdminJobOp::SHOW: {
-            nebula::DataSet v(
-                {"Job Id(TaskId)", "Command(Dest)", "Status", "Start Time", "Stop Time"});
-            DCHECK(resp.value().job_desc_ref().has_value());
-            if (!resp.value().job_desc_ref().has_value()) {
-              return Status::Error("Response unexpected.");
-            }
-            DCHECK(resp.value().task_desc_ref().has_value());
-            if (!resp.value().task_desc_ref().has_value()) {
-              return Status::Error("Response unexpected");
-            }
-            auto &jobDesc = *resp.value().job_desc_ref();
-            // job desc
-            v.emplace_back(nebula::Row({
-                jobDesc.front().get_id(),
-                apache::thrift::util::enumNameSafe(jobDesc.front().get_cmd()),
-                apache::thrift::util::enumNameSafe(jobDesc.front().get_status()),
-                time::TimeConversion::unixSecondsToDateTime(jobDesc.front().get_start_time()),
-                time::TimeConversion::unixSecondsToDateTime(jobDesc.front().get_stop_time()),
-            }));
-            // tasks desc
-            auto &tasksDesc = *resp.value().get_task_desc();
-            for (const auto &taskDesc : tasksDesc) {
-              v.emplace_back(nebula::Row({
-                  taskDesc.get_task_id(),
-                  taskDesc.get_host().host,
-                  apache::thrift::util::enumNameSafe(taskDesc.get_status()),
-                  time::TimeConversion::unixSecondsToDateTime(taskDesc.get_start_time()),
-                  time::TimeConversion::unixSecondsToDateTime(taskDesc.get_stop_time()),
-              }));
-            }
-            return finish(std::move(v));
-          }
-          case meta::cpp2::AdminJobOp::SHOW_All: {
-            nebula::DataSet v({"Job Id", "Command", "Status", "Start Time", "Stop Time"});
-            DCHECK(resp.value().job_desc_ref().has_value());
-            if (!resp.value().job_desc_ref().has_value()) {
-              return Status::Error("Response unexpected");
-            }
-            const auto &jobsDesc = *resp.value().job_desc_ref();
-            for (const auto &jobDesc : jobsDesc) {
-              v.emplace_back(nebula::Row({
-                  jobDesc.get_id(),
-                  apache::thrift::util::enumNameSafe(jobDesc.get_cmd()),
-                  apache::thrift::util::enumNameSafe(jobDesc.get_status()),
-                  time::TimeConversion::unixSecondsToDateTime(jobDesc.get_start_time()),
-                  time::TimeConversion::unixSecondsToDateTime(jobDesc.get_stop_time()),
-              }));
-            }
-            return finish(std::move(v));
-          }
-          case meta::cpp2::AdminJobOp::STOP: {
-            nebula::DataSet v({"Result"});
-            v.emplace_back(nebula::Row({"Job stopped"}));
-            return finish(std::move(v));
-          }
-            // no default so the compiler will warning when lack
-        }
-        DLOG(FATAL) << "Unknown job operation " << static_cast<int>(jobOp);
-        return Status::Error("Unknown job job operation %d.", static_cast<int>(jobOp));
+        auto status = buildResult(jobOp, std::move(resp).value());
+        NG_RETURN_IF_ERROR(status);
+        return finish(std::move(status).value());
       });
 }
 
+StatusOr<DataSet> SubmitJobExecutor::buildResult(meta::cpp2::AdminJobOp jobOp,
+                                                 meta::cpp2::AdminJobResult &&resp) {
+  switch (jobOp) {
+    case meta::cpp2::AdminJobOp::ADD: {
+      nebula::DataSet v({"New Job Id"});
+      DCHECK(resp.job_id_ref().has_value());
+      if (!resp.job_id_ref().has_value()) {
+        return Status::Error("Response unexpected.");
+      }
+      v.emplace_back(nebula::Row({*resp.job_id_ref()}));
+      return v;
+    }
+    case meta::cpp2::AdminJobOp::RECOVER: {
+      nebula::DataSet v({"Recovered job num"});
+      DCHECK(resp.recovered_job_num_ref().has_value());
+      if (!resp.recovered_job_num_ref().has_value()) {
+        return Status::Error("Response unexpected.");
+      }
+      v.emplace_back(nebula::Row({*resp.recovered_job_num_ref()}));
+      return v;
+    }
+    case meta::cpp2::AdminJobOp::SHOW: {
+      nebula::DataSet v({"Job Id(TaskId)", "Command(Dest)", "Status", "Start Time", "Stop Time"});
+      DCHECK(resp.job_desc_ref().has_value());
+      if (!resp.job_desc_ref().has_value()) {
+        return Status::Error("Response unexpected.");
+      }
+      DCHECK(resp.task_desc_ref().has_value());
+      if (!resp.task_desc_ref().has_value()) {
+        return Status::Error("Response unexpected");
+      }
+      auto &jobDesc = *resp.job_desc_ref();
+      // job desc
+      v.emplace_back(nebula::Row({
+          jobDesc.front().get_id(),
+          apache::thrift::util::enumNameSafe(jobDesc.front().get_cmd()),
+          apache::thrift::util::enumNameSafe(jobDesc.front().get_status()),
+          convertJobTimestampToDateTime(jobDesc.front().get_start_time()),
+          convertJobTimestampToDateTime(jobDesc.front().get_stop_time()),
+      }));
+      // tasks desc
+      auto &tasksDesc = *resp.get_task_desc();
+      for (const auto &taskDesc : tasksDesc) {
+        v.emplace_back(nebula::Row({
+            taskDesc.get_task_id(),
+            taskDesc.get_host().host,
+            apache::thrift::util::enumNameSafe(taskDesc.get_status()),
+            convertJobTimestampToDateTime(taskDesc.get_start_time()),
+            convertJobTimestampToDateTime(taskDesc.get_stop_time()),
+        }));
+      }
+      return v;
+    }
+    case meta::cpp2::AdminJobOp::SHOW_All: {
+      nebula::DataSet v({"Job Id", "Command", "Status", "Start Time", "Stop Time"});
+      DCHECK(resp.job_desc_ref().has_value());
+      if (!resp.job_desc_ref().has_value()) {
+        return Status::Error("Response unexpected");
+      }
+      const auto &jobsDesc = *resp.job_desc_ref();
+      for (const auto &jobDesc : jobsDesc) {
+        v.emplace_back(nebula::Row({
+            jobDesc.get_id(),
+            apache::thrift::util::enumNameSafe(jobDesc.get_cmd()),
+            apache::thrift::util::enumNameSafe(jobDesc.get_status()),
+            convertJobTimestampToDateTime(jobDesc.get_start_time()),
+            convertJobTimestampToDateTime(jobDesc.get_stop_time()),
+        }));
+      }
+      return v;
+    }
+    case meta::cpp2::AdminJobOp::STOP: {
+      nebula::DataSet v({"Result"});
+      v.emplace_back(nebula::Row({"Job stopped"}));
+      return v;
+    }
+      // no default so the compiler will warning when lack
+  }
+  DLOG(FATAL) << "Unknown job operation " << static_cast<int>(jobOp);
+  return Status::Error("Unknown job job operation %d.", static_cast<int>(jobOp));
+}
+
+Value SubmitJobExecutor::convertJobTimestampToDateTime(int64_t timestamp) {
+  return timestamp > 0 ? Value(time::TimeConversion::unixSecondsToDateTime(timestamp))
+                       : Value::kEmpty;
+}
 }  // namespace graph
 }  // namespace nebula

src/graph/executor/admin/SubmitJobExecutor.h

@@ -18,6 +18,11 @@ class SubmitJobExecutor final : public Executor {
       : Executor("SubmitJobExecutor", node, qctx) {}
 
   folly::Future<Status> execute() override;
+
+ private:
+  FRIEND_TEST(JobTest, JobFinishTime);
+  StatusOr<DataSet> buildResult(meta::cpp2::AdminJobOp jobOp, meta::cpp2::AdminJobResult &&resp);
+  Value convertJobTimestampToDateTime(int64_t timestamp);
 };
 
 }  // namespace graph

src/graph/executor/test/CMakeLists.txt

@@ -81,6 +81,7 @@ nebula_add_test(
         CartesianProductTest.cpp
         AssignTest.cpp
         ShowQueriesTest.cpp
+        JobTest.cpp
     OBJECTS
         ${EXEC_QUERY_TEST_OBJS}
     LIBRARIES

src/graph/executor/test/JobTest.cpp

@@ -0,0 +1,69 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include <gtest/gtest.h>
+
+#include "common/time/TimeUtils.h"
+#include "graph/context/QueryContext.h"
+#include "graph/executor/admin/SubmitJobExecutor.h"
+#include "graph/planner/plan/Admin.h"
+
+namespace nebula {
+namespace graph {
+class JobTest : public testing::Test {};
+
+TEST_F(JobTest, JobFinishTime) {
+  {
+    meta::cpp2::AdminJobResult resp;
+    resp.set_job_id(0);
+    meta::cpp2::JobDesc jobDesc;
+    jobDesc.set_id(0);
+    jobDesc.set_start_time(123);
+    jobDesc.set_stop_time(0);
+    resp.set_job_desc({std::move(jobDesc)});
+    meta::cpp2::TaskDesc taskDesc;
+    taskDesc.set_start_time(456);
+    taskDesc.set_stop_time(0);
+    resp.set_task_desc({std::move(taskDesc)});
+
+    auto qctx = std::make_unique<QueryContext>();
+    auto submitJob = SubmitJob::make(
+        qctx.get(), nullptr, meta::cpp2::AdminJobOp::SHOW, meta::cpp2::AdminCmd::UNKNOWN, {});
+    auto submitJobExe = std::make_unique<SubmitJobExecutor>(submitJob, qctx.get());
+
+    auto status = submitJobExe->buildResult(meta::cpp2::AdminJobOp::SHOW, std::move(resp));
+    EXPECT_TRUE(status.ok());
+    auto result = std::move(status).value();
+    EXPECT_EQ(result.rows.size(), 2);
+    EXPECT_EQ(result.rows[0][3], Value(time::TimeConversion::unixSecondsToDateTime(123)));
+    EXPECT_EQ(result.rows[0][4], Value::kEmpty);
+    EXPECT_EQ(result.rows[1][3], Value(time::TimeConversion::unixSecondsToDateTime(456)));
+    EXPECT_EQ(result.rows[1][4], Value::kEmpty);
+  }
+  {
+    meta::cpp2::AdminJobResult resp;
+    resp.set_job_id(0);
+    meta::cpp2::JobDesc jobDesc;
+    jobDesc.set_id(0);
+    jobDesc.set_start_time(123);
+    jobDesc.set_stop_time(0);
+    resp.set_job_desc({std::move(jobDesc)});
+
+    auto qctx = std::make_unique<QueryContext>();
+    auto submitJob = SubmitJob::make(
+        qctx.get(), nullptr, meta::cpp2::AdminJobOp::SHOW_All, meta::cpp2::AdminCmd::UNKNOWN, {});
+    auto submitJobExe = std::make_unique<SubmitJobExecutor>(submitJob, qctx.get());
+
+    auto status = submitJobExe->buildResult(meta::cpp2::AdminJobOp::SHOW_All, std::move(resp));
+    EXPECT_TRUE(status.ok());
+    auto result = std::move(status).value();
+    EXPECT_EQ(result.rows.size(), 1);
+    EXPECT_EQ(result.rows[0][3], Value(time::TimeConversion::unixSecondsToDateTime(123)));
+    EXPECT_EQ(result.rows[0][4], Value::kEmpty);
+  }
+}
+}  // namespace graph
+}  // namespace nebula

tests/Makefile

@@ -61,16 +61,16 @@ sess: currdir
 	python3 -m pytest -m "not skip" -k "not tck" job/test_session.py
 
 jobs: currdir
-	python3 -m pytest -m "not skip" -k "not tck" job/test_jobs.py
+	python3 -m pytest -m "not skip" tck/steps/test_jobs.py
 
-test: sess jobs
+test: sess
 	python3 -m pytest -n$(J) --dist=loadfile -m "not skip" -k "not tck" $(TEST_DIR)
 
 slow-query: currdir
 	python3 -m pytest -n$(J) -m "not skip" tck/steps/test_kill_slow_query_via_same_service.py && \
 	python3 -m pytest -n$(J) -m "not skip" tck/steps/test_kill_slow_query_via_different_service.py
 
-tck: slow-query
+tck: jobs slow-query
 	python3 -m pytest -n$(J) -m "not skip" tck/steps/test_tck.py
 
 fail: currdir

tests/common/comparator.py

@@ -29,11 +29,13 @@ def __init__(self,
                  strict=True,
                  order=False,
                  contains=CmpType.EQUAL,
+                 first_n_records=-1,
                  decode_type='utf-8',
                  vid_fn=None):
         self._strict = strict
         self._order = order
         self._contains = contains
+        self._first_n_records=first_n_records
         self._decode_type = decode_type
         self._vid_fn = vid_fn
 
@@ -65,13 +67,16 @@ def compare(self, resp: DataSet, expect: DataSet):
             if ln != self.bstr(rn):
                 return False, -2
         if self._order:
-            for i in range(0, len(expect.rows)):
-                cmp = self.compare_row(resp.rows[i], expect.rows[i])
-                if self._whether_return(cmp):
-                    return False, i
-            if self._contains == CmpType.CONTAINS:
+            if self._contains == CmpType.CONTAINS and self._first_n_records < 0:
+                for i in range(0, len(expect.rows)):
+                    cmp = self.compare_row(resp.rows[i], expect.rows[i])
+                    if self._whether_return(cmp):
+                        return False, i
                 return True, None
-            return len(resp.rows) == len(expect.rows), -1
+            elif self._contains == CmpType.CONTAINS and self._first_n_records > 0:
+                return self._compare_list(resp.rows[0:self._first_n_records], expect.rows, self.compare_row)
+            else:
+                return len(resp.rows) == len(expect.rows), -1
         return self._compare_list(resp.rows, expect.rows, self.compare_row,
                                   self._contains)