"""Tests for job_directories."""
|
|
from __future__ import absolute_import
|
from __future__ import division
|
from __future__ import print_function
|
|
import contextlib
|
import datetime
|
import mox
|
import os
|
import shutil
|
import tempfile
|
import unittest
|
|
import common
|
from autotest_lib.site_utils import job_directories
|
from autotest_lib.client.common_lib import time_utils
|
|
|
class SwarmingJobDirectoryTestCase(unittest.TestCase):
|
"""Tests SwarmingJobDirectory."""
|
|
def test_get_job_directories_legacy(self):
|
with _change_to_tempdir():
|
os.makedirs("swarming-3e4391423c3a4311/b")
|
os.mkdir("not-a-swarming-dir")
|
results = job_directories.SwarmingJobDirectory.get_job_directories()
|
self.assertEqual(set(results), {"swarming-3e4391423c3a4311"})
|
|
def test_get_job_directories(self):
|
with _change_to_tempdir():
|
os.makedirs("swarming-3e4391423c3a4310/1")
|
os.makedirs("swarming-3e4391423c3a4310/0")
|
os.makedirs("swarming-3e4391423c3a4310/a")
|
os.mkdir("not-a-swarming-dir")
|
results = job_directories.SwarmingJobDirectory.get_job_directories()
|
self.assertEqual(set(results),
|
{"swarming-3e4391423c3a4310/1",
|
"swarming-3e4391423c3a4310/a"})
|
|
|
class GetJobIDOrTaskID(unittest.TestCase):
|
"""Tests get_job_id_or_task_id."""
|
|
def test_legacy_swarming_path(self):
|
self.assertEqual(
|
"3e4391423c3a4311",
|
job_directories.get_job_id_or_task_id(
|
"/autotest/results/swarming-3e4391423c3a4311"),
|
)
|
self.assertEqual(
|
"3e4391423c3a4311",
|
job_directories.get_job_id_or_task_id(
|
"swarming-3e4391423c3a4311"),
|
)
|
|
def test_swarming_path(self):
|
self.assertEqual(
|
"3e4391423c3a4311",
|
job_directories.get_job_id_or_task_id(
|
"/autotest/results/swarming-3e4391423c3a4310/1"),
|
)
|
self.assertEqual(
|
"3e4391423c3a431f",
|
job_directories.get_job_id_or_task_id(
|
"swarming-3e4391423c3a4310/f"),
|
)
|
|
|
|
class JobDirectorySubclassTests(mox.MoxTestBase):
|
"""Test specific to RegularJobDirectory and SpecialJobDirectory.
|
|
This provides coverage for the implementation in both
|
RegularJobDirectory and SpecialJobDirectory.
|
|
"""
|
|
def setUp(self):
|
super(JobDirectorySubclassTests, self).setUp()
|
self.mox.StubOutWithMock(job_directories, '_AFE')
|
|
|
def test_regular_job_fields(self):
|
"""Test the constructor for `RegularJobDirectory`.
|
|
Construct a regular job, and assert that the `dirname`
|
and `_id` attributes are set as expected.
|
|
"""
|
resultsdir = '118-fubar'
|
job = job_directories.RegularJobDirectory(resultsdir)
|
self.assertEqual(job.dirname, resultsdir)
|
self.assertEqual(job._id, '118')
|
|
|
def test_special_job_fields(self):
|
"""Test the constructor for `SpecialJobDirectory`.
|
|
Construct a special job, and assert that the `dirname`
|
and `_id` attributes are set as expected.
|
|
"""
|
destdir = 'hosts/host1'
|
resultsdir = destdir + '/118-reset'
|
job = job_directories.SpecialJobDirectory(resultsdir)
|
self.assertEqual(job.dirname, resultsdir)
|
self.assertEqual(job._id, '118')
|
|
|
def _check_finished_job(self, jobtime, hqetimes, expected):
|
"""Mock and test behavior of a finished job.
|
|
Initialize the mocks for a call to
|
`get_timestamp_if_finished()`, then simulate one call.
|
Assert that the returned timestamp matches the passed
|
in expected value.
|
|
@param jobtime Time used to construct a _MockJob object.
|
@param hqetimes List of times used to construct
|
_MockHostQueueEntry objects.
|
@param expected Expected time to be returned by
|
get_timestamp_if_finished
|
|
"""
|
job = job_directories.RegularJobDirectory('118-fubar')
|
job_directories._AFE.get_jobs(
|
id=job._id, finished=True).AndReturn(
|
[_MockJob(jobtime)])
|
job_directories._AFE.get_host_queue_entries(
|
finished_on__isnull=False,
|
job_id=job._id).AndReturn(
|
[_MockHostQueueEntry(t) for t in hqetimes])
|
self.mox.ReplayAll()
|
self.assertEqual(expected, job.get_timestamp_if_finished())
|
self.mox.VerifyAll()
|
|
|
def test_finished_regular_job(self):
|
"""Test getting the timestamp for a finished regular job.
|
|
Tests the return value for
|
`RegularJobDirectory.get_timestamp_if_finished()` when
|
the AFE indicates the job is finished.
|
|
"""
|
created_timestamp = make_timestamp(1, True)
|
hqe_timestamp = make_timestamp(0, True)
|
self._check_finished_job(created_timestamp,
|
[hqe_timestamp],
|
hqe_timestamp)
|
|
|
def test_finished_regular_job_multiple_hqes(self):
|
"""Test getting the timestamp for a regular job with multiple hqes.
|
|
Tests the return value for
|
`RegularJobDirectory.get_timestamp_if_finished()` when
|
the AFE indicates the job is finished and the job has multiple host
|
queue entries.
|
|
Tests that the returned timestamp is the latest timestamp in
|
the list of HQEs, regardless of the returned order.
|
|
"""
|
created_timestamp = make_timestamp(2, True)
|
older_hqe_timestamp = make_timestamp(1, True)
|
newer_hqe_timestamp = make_timestamp(0, True)
|
hqe_list = [older_hqe_timestamp,
|
newer_hqe_timestamp]
|
self._check_finished_job(created_timestamp,
|
hqe_list,
|
newer_hqe_timestamp)
|
self.mox.ResetAll()
|
hqe_list.reverse()
|
self._check_finished_job(created_timestamp,
|
hqe_list,
|
newer_hqe_timestamp)
|
|
|
def test_finished_regular_job_null_finished_times(self):
|
"""Test getting the timestamp for an aborted regular job.
|
|
Tests the return value for
|
`RegularJobDirectory.get_timestamp_if_finished()` when
|
the AFE indicates the job is finished and the job has aborted host
|
queue entries.
|
|
"""
|
timestamp = make_timestamp(0, True)
|
self._check_finished_job(timestamp, [], timestamp)
|
|
|
def test_unfinished_regular_job(self):
|
"""Test getting the timestamp for an unfinished regular job.
|
|
Tests the return value for
|
`RegularJobDirectory.get_timestamp_if_finished()` when
|
the AFE indicates the job is not finished.
|
|
"""
|
job = job_directories.RegularJobDirectory('118-fubar')
|
job_directories._AFE.get_jobs(
|
id=job._id, finished=True).AndReturn([])
|
self.mox.ReplayAll()
|
self.assertIsNone(job.get_timestamp_if_finished())
|
self.mox.VerifyAll()
|
|
|
def test_finished_special_job(self):
|
"""Test getting the timestamp for a finished special job.
|
|
Tests the return value for
|
`SpecialJobDirectory.get_timestamp_if_finished()` when
|
the AFE indicates the job is finished.
|
|
"""
|
job = job_directories.SpecialJobDirectory(
|
'hosts/host1/118-reset')
|
timestamp = make_timestamp(0, True)
|
job_directories._AFE.get_special_tasks(
|
id=job._id, is_complete=True).AndReturn(
|
[_MockSpecialTask(timestamp)])
|
self.mox.ReplayAll()
|
self.assertEqual(timestamp,
|
job.get_timestamp_if_finished())
|
self.mox.VerifyAll()
|
|
|
def test_unfinished_special_job(self):
|
"""Test getting the timestamp for an unfinished special job.
|
|
Tests the return value for
|
`SpecialJobDirectory.get_timestamp_if_finished()` when
|
the AFE indicates the job is not finished.
|
|
"""
|
job = job_directories.SpecialJobDirectory(
|
'hosts/host1/118-reset')
|
job_directories._AFE.get_special_tasks(
|
id=job._id, is_complete=True).AndReturn([])
|
self.mox.ReplayAll()
|
self.assertIsNone(job.get_timestamp_if_finished())
|
self.mox.VerifyAll()
|
|
|
class JobExpirationTests(unittest.TestCase):
|
"""Tests to exercise `job_directories.is_job_expired()`."""
|
|
def test_expired(self):
|
"""Test detection of an expired job."""
|
timestamp = make_timestamp(_TEST_EXPIRATION_AGE, True)
|
self.assertTrue(
|
job_directories.is_job_expired(
|
_TEST_EXPIRATION_AGE, timestamp))
|
|
|
def test_alive(self):
|
"""Test detection of a job that's not expired."""
|
# N.B. This test may fail if its run time exceeds more than
|
# about _MARGIN_SECS seconds.
|
timestamp = make_timestamp(_TEST_EXPIRATION_AGE, False)
|
self.assertFalse(
|
job_directories.is_job_expired(
|
_TEST_EXPIRATION_AGE, timestamp))
|
|
|
# When constructing sample time values for testing expiration,
|
# allow this many seconds between the expiration time and the
|
# current time.
|
_MARGIN_SECS = 10.0
|
# Test value to use for `days_old`, if nothing else is required.
|
_TEST_EXPIRATION_AGE = 7
|
|
|
class _MockJob(object):
|
"""Class to mock the return value of `AFE.get_jobs()`."""
|
def __init__(self, created):
|
self.created_on = created
|
|
|
class _MockHostQueueEntry(object):
|
"""Class to mock the return value of `AFE.get_host_queue_entries()`."""
|
def __init__(self, finished):
|
self.finished_on = finished
|
|
|
class _MockSpecialTask(object):
|
"""Class to mock the return value of `AFE.get_special_tasks()`."""
|
def __init__(self, finished):
|
self.time_finished = finished
|
|
|
@contextlib.contextmanager
|
def _change_to_tempdir():
|
old_dir = os.getcwd()
|
tempdir = tempfile.mkdtemp('job_directories_unittest')
|
try:
|
os.chdir(tempdir)
|
yield
|
finally:
|
os.chdir(old_dir)
|
shutil.rmtree(tempdir)
|
|
|
def make_timestamp(age_limit, is_expired):
|
"""Create a timestamp for use by `job_directories.is_job_expired()`.
|
|
The timestamp will meet the syntactic requirements for
|
timestamps used as input to `is_job_expired()`. If
|
`is_expired` is true, the timestamp will be older than
|
`age_limit` days before the current time; otherwise, the
|
date will be younger.
|
|
@param age_limit The number of days before expiration of the
|
target timestamp.
|
@param is_expired Whether the timestamp should be expired
|
relative to `age_limit`.
|
|
"""
|
seconds = -_MARGIN_SECS
|
if is_expired:
|
seconds = -seconds
|
delta = datetime.timedelta(days=age_limit, seconds=seconds)
|
reference_time = datetime.datetime.now() - delta
|
return reference_time.strftime(time_utils.TIME_FMT)
|
|
|
if __name__ == '__main__':
|
unittest.main()
|
