1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
"""Unit tests for powercycle_sentinel.py."""
import unittest
from datetime import datetime, timezone, timedelta
from unittest.mock import Mock
from evergreen import EvergreenApi, Task
from buildscripts.powercycle_sentinel import watch_tasks, POWERCYCLE_TASK_EXEC_TIMEOUT_SECS
def make_task_mock(evg_api, task_id, start_time, finish_time):
return Task(
{
"task_id": task_id,
"start_time": start_time,
"finish_time": finish_time,
},
evg_api,
)
class TestWatchTasks(unittest.TestCase):
"""Test watch_tasks."""
def test_no_long_running_tasks(self):
evg_api = EvergreenApi()
task_ids = ["1", "2"]
now = datetime.now(timezone.utc).isoformat()
task_1 = make_task_mock(evg_api, task_ids[0], now, now)
task_2 = make_task_mock(evg_api, task_ids[1], now, now)
evg_api.task_by_id = Mock(
side_effect=(
lambda task_id: {
"1": task_1,
"2": task_2,
}[task_id]
)
)
long_running_task_ids = watch_tasks(task_ids, evg_api, 0)
self.assertEqual([], long_running_task_ids)
def test_found_long_running_tasks(self):
evg_api = EvergreenApi()
task_ids = ["1", "2"]
exec_timeout_seconds_ago = (
datetime.now(timezone.utc) - timedelta(hours=POWERCYCLE_TASK_EXEC_TIMEOUT_SECS)
).isoformat()
now = datetime.now(timezone.utc).isoformat()
task_1 = make_task_mock(evg_api, task_ids[0], exec_timeout_seconds_ago, now)
task_2 = make_task_mock(evg_api, task_ids[1], exec_timeout_seconds_ago, None)
evg_api.task_by_id = Mock(
side_effect=(
lambda task_id: {
"1": task_1,
"2": task_2,
}[task_id]
)
)
long_running_task_ids = watch_tasks(task_ids, evg_api, 0)
self.assertEqual([task_2.task_id], long_running_task_ids)
|
