Kinesin/daggy
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fixing test cases on daggyr for new polling

Browse Source
This commit is contained in:
Ian Roddis
2022-01-04 17:03:11 -04:00
parent 5a4c6d0756
commit f1479a72d9
3 changed files with 52 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
daggyd/libdaggyd/src/Server.cpp
View File

@@ -180,7 +180,14 @@ namespace daggy::daggyd {
if (!handleAuth(request)) if (!handleAuth(request))
return; return;
auto dagSpec = dagFromJSON(request.body()); DAGSpec dagSpec;
try {
dagSpec = dagFromJSON(request.body());
}
catch (std::runtime_error &e) {
REQ_RESPONSE(Not_Acceptable, e.what());
}
dagSpec.tasks = dagSpec.tasks =
expandTaskSet(dagSpec.tasks, executor_, dagSpec.taskConfig.variables); expandTaskSet(dagSpec.tasks, executor_, dagSpec.taskConfig.variables);

75
daggyr/tests/unit_server.cpp
View File

@@ -36,12 +36,6 @@ TEST_CASE("rest_endpoint", "[server_basic]")
REQUIRE(response.code == HTTPCode::Ok); REQUIRE(response.code == HTTPCode::Ok);
} }
SECTION("Querying a non-existent task should yield a 404")
{
auto response = HTTP_REQUEST(baseURL + "/v1/task/100/sample_name");
REQUIRE(response.code == HTTPCode::Not_Found);
}
SECTION("Task Missing Cores should Fail") SECTION("Task Missing Cores should Fail")
{ {
std::string taskSpec = std::string taskSpec =
@@ -75,22 +69,24 @@ TEST_CASE("rest_endpoint", "[server_basic]")
} }
while (true) { while (true) {
auto [code, doc] = JSON_HTTP_REQUEST(baseURL + "/v1/task/0/sample_task"); std::this_thread::sleep_for(250ms);
REQUIRE(doc.IsObject()); auto [code, doc] = JSON_HTTP_REQUEST(baseURL + "/v1/poll");
REQUIRE(doc.HasMember("state")); REQUIRE(doc.IsArray());
if (doc.Size() == 0)
continue;
std::string state = doc["state"].GetString(); const auto &task = doc[0];
if (state != "COMPLETED") { REQUIRE(task.HasMember("state"));
std::this_thread::sleep_for(250ms); std::string state = task["state"].GetString();
} if (state != "COMPLETED")
else { continue;
REQUIRE(doc.HasMember("attempt"));
auto attempt = attemptRecordFromJSON(doc["attempt"]);
REQUIRE(attempt.rc == 0); REQUIRE(task.HasMember("attempt"));
REQUIRE(attempt.outputLog == "hello world\n"); auto attempt = attemptRecordFromJSON(task["attempt"]);
break;
} REQUIRE(attempt.rc == 0);
REQUIRE(attempt.outputLog == "hello world\n");
break;
} }
} }
@@ -131,10 +127,11 @@ TEST_CASE("rest_endpoint", "[server_basic]")
// Ensure the current job is running // Ensure the current job is running
{ {
auto [code, doc] = JSON_HTTP_REQUEST(baseURL + "/v1/task/0/sample_task"); auto [code, doc] = JSON_HTTP_REQUEST(baseURL + "/v1/poll");
REQUIRE(doc.IsObject()); REQUIRE(doc.IsArray());
REQUIRE(doc.HasMember("state")); REQUIRE(doc.Size() > 0);
REQUIRE(doc["state"] != "COMPLETED"); REQUIRE(doc[0].HasMember("state"));
REQUIRE(doc[0]["state"] != "COMPLETED");
} }
// Stop it // Stop it
@@ -146,25 +143,23 @@ TEST_CASE("rest_endpoint", "[server_basic]")
// Grab it and ensure it was killed // Grab it and ensure it was killed
while (true) { while (true) {
auto response = HTTP_REQUEST(baseURL + "/v1/task/0/sample_task"); auto [code, doc] = JSON_HTTP_REQUEST(baseURL + "/v1/poll");
REQUIRE(code == HTTPCode::Ok);
REQUIRE(response.code == HTTPCode::Ok); REQUIRE(doc.IsArray());
rj::Document doc; if (doc.Size() == 0)
daggy::checkRJParse(doc.Parse(response.body.c_str())); continue;
REQUIRE(doc.IsObject());
REQUIRE(doc.HasMember("state"));
std::string state = doc["state"].GetString(); const auto &task = doc[0];
if (state != "COMPLETED") { REQUIRE(task.HasMember("state"));
std::this_thread::sleep_for(250ms); std::string state = task["state"].GetString();
} if (state != "COMPLETED")
else { continue;
REQUIRE(doc.HasMember("attempt"));
auto attempt = attemptRecordFromJSON(doc["attempt"]);
REQUIRE(attempt.rc != 0); REQUIRE(task.HasMember("attempt"));
break; auto attempt = attemptRecordFromJSON(task["attempt"]);
} REQUIRE(attempt.rc != 0);
break;
} }
} }

50
examples/sample_dag.json
View File

@@ -1,43 +1,11 @@
{ {
"parameters": { "tag": "testdag",
"SOURCE": [ "tasks": {
"a", "A": { "job": { "command": [ "/usr/bin/echo", "hello" ], "cores": "1", "memoryMB": "100" }, "children": [ "b", "c", "d" ]},
"b", "b": { "job": { "command": [ "/usr/bin/sleep", "2" ], "cores": "1", "memoryMB": "100" }, "children": [ "d" ]},
"c" "c": { "job": { "command": [ "/usr/bin/echo", "hello" ], "cores": "1", "memoryMB": "100" }, "children": [ "e", "f" ]},
], "d": { "job": { "command": [ "/usr/bin/echo", "hello" ], "cores": "1", "memoryMB": "100" }, "children": [ "f" ]},
"DATE": [ "e": { "job": { "command": [ "/usr/bin/echo", "hello" ], "cores": "1", "memoryMB": "100" }, "children": [ "f" ]},
"2021-01-01", "f": { "job": { "command": [ "/usr/bin/echo", "hello" ], "cores": "1", "memoryMB": "100" }}
"2021-01-02" }
]
},
"tasks": [
{
"name": "pull_data_a",
"max_retries": 3,
"retry_interval_seconds": 600,
"if": "/path/to/should_pull.sh --date {{DATE}} --source {{SOURCE}}_A",
"command": "/path/to/pull.sh --date {{DATE}} --source {{SOURCE}}_A",
"verification_command": "/path/to/pull_verify.sh --date {{DATE}} --source {{SOURCE}}_A",
"timeout_seconds": 30,
"children": [
"merge_data"
]
},
{
"name": "pull_data_b",
"max_retries": 3,
"retry_interval_seconds": 600,
"if": "/path/to/should_pull.sh --date {{DATE}} --source {{SOURCE}}_B",
"command": "/path/to/pull.sh --date {{DATE}} --source {{SOURCE}}_B",
"verification_command": "/path/to/pull_verify.sh --date {{DATE}} --source {{SOURCE}}_B",
"timeout_seconds": 30,
"children": [
"merge_data"
]
},
{
"name": "merge_data",
"command": "/path/to/merge.sh --left {{SOURCE}}_A --right {{SOURCE}}_B"
}
]
} }