LineageOS/android_development
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge "Add test to ota_interface.JobInfo." am: ba7d668e84

Browse Source 
Original change: https://android-review.googlesource.com/c/platform/development/+/1793172

Change-Id: I8451f3fbd5fe0a52bf94fa9f1851b3f384306b4e
This commit is contained in:
Treehugger Robot
2021-08-12 14:11:24 +00:00
committed by Automerger Merge Worker
parent 303ca2baa8 ba7d668e84
commit 42f7ba7bc8
2 changed files with 202 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
tools/otagui/ota_interface.py
View File

@@ -31,6 +31,10 @@ class JobInfo:
isIncremental: bool = False
def __post_init__(self):
"""
If the output, stdout, stderr paths are not set, automatically use
the job id as the file name.
"""
if not self.output:
self.output = os.path.join('output', self.id + '.zip')
if not self.stdout:
@@ -48,6 +52,15 @@ class JobInfo:
self.isPartial = True
def to_sql_form_dict(self):
"""
Convert this instance to a dict, which can be later used to insert into
the SQL database.
Format:
id: string, target: string, incremental: string, verbose: int,
partial: string, output:string, status:string,
downgrade: bool, extra: string, stdout: string, stderr:string,
start_time:int, finish_time: int(not required)
"""
sql_form_dict = asdict(self)
sql_form_dict['partial'] = ','.join(sql_form_dict['partial'])
def bool_to_int(t): return 1 if t else 0
@@ -57,6 +70,9 @@ class JobInfo:
return sql_form_dict
def to_dict_basic(self):
"""
Convert the instance to a dict, which includes the file name of target.
"""
basic_info = asdict(self)
basic_info['target_name'] = self.target.split('/')[-1]
if self.isIncremental:
@@ -64,6 +80,10 @@ class JobInfo:
return basic_info
def to_dict_detail(self, target_lib, offset=0):
"""
Convert this instance into a dict, which includes some detailed information
of the target/source build, i.e. build version and file name.
"""
detail_info = asdict(self)
try:
with open(self.stdout, 'r') as fout:

182
tools/otagui/test_ota_interface.py Normal file
View File

@@ -0,0 +1,182 @@
import unittest
from ota_interface import JobInfo, ProcessesManagement
from unittest.mock import patch, mock_open, Mock, MagicMock
class TestJobInfo(unittest.TestCase):
def setUp(self):
self.test_id = '286feab0-f16b-11eb-a72a-f7b1de0921ef'
self.test_target = 'target/build.zip'
self.test_verbose = True
self.test_status = 'Running'
self.test_extra = '--downgrade'
self.test_start_time = 1628698830
self.test_finish_time = 1628698831
def setup_job(self, incremental = '', partial = [],
output = '', stdout = '', stderr = ''):
job_info = JobInfo(id=self.test_id,
target=self.test_target,
incremental=incremental,
verbose=self.test_verbose,
partial=partial,
output=output,
status=self.test_status,
extra=self.test_extra,
start_time=self.test_start_time,
finish_time=self.test_finish_time,
stderr=stderr,
stdout=stdout
)
return job_info
def test_init(self):
# No incremental, no output, no stdout/stderr set.
job_info1 = self.setup_job()
for key, value in self.__dict__.items():
if key.startswith('test_'):
self.assertEqual(job_info1.__dict__[key[5:]], value,
'The value of ' + key + 'is not initialized correctly'
)
self.assertEqual(job_info1.output, 'output/'+self.test_id+'.zip',
'Default output cannot be setup correctly'
)
self.assertEqual(job_info1.stderr, 'output/stderr.'+self.test_id,
'Default stderr cannot be setup correctly'
)
self.assertEqual(job_info1.stdout, 'output/stdout.'+self.test_id,
'Default stdout cannot be setup correctly'
)
# Test the incremental setup
job_info2 = self.setup_job(incremental='target/source.zip')
self.assertEqual(job_info2.incremental, 'target/source.zip',
'incremental source cannot be initialized correctly'
)
self.assertTrue(job_info2.isIncremental,
'incremental status cannot be initialized correctly'
)
# Test the stdout/stderr setup
job_info3 = self.setup_job(stderr='output/stderr',
stdout='output/stdout'
)
self.assertEqual(job_info3.stderr, 'output/stderr',
'the stderr cannot be setup manually'
)
self.assertEqual(job_info3.stdout, 'output/stdout',
'the stdout cannot be setup manually'
)
# Test the output setup
job_info4 = self.setup_job(output='output/output.zip')
self.assertEqual(job_info4.output, 'output/output.zip',
'output cannot be setup manually'
)
# Test the partial setup
job_info5 = self.setup_job(partial=['system', 'vendor'])
self.assertEqual(job_info5.partial, ['system', 'vendor'],
'partial list cannot be setup correctly'
)
def test_to_sql_form_dict(self):
partial_list = ['system', 'vendor']
job_info = self.setup_job(partial=partial_list)
sql_dict = job_info.to_sql_form_dict()
test_dict = {'id': self.test_id,
'target': self.test_target,
'incremental': '',
'verbose': int(self.test_verbose),
'partial': ','.join(partial_list),
'output': 'output/' + self.test_id + '.zip',
'status': self.test_status,
'extra': self.test_extra,
'stdout': 'output/stdout.' + self.test_id,
'stderr': 'output/stderr.' + self.test_id,
'start_time': self.test_start_time,
'finish_time': self.test_finish_time,
'isPartial': True,
'isIncremental': False
}
for key, value in test_dict.items():
self.assertEqual(value, sql_dict[key],
'the ' + key + ' is not converted to sql form dict correctly'
)
def test_to_dict_basic(self):
partial_list = ['system', 'vendor']
job_info = self.setup_job(partial=partial_list)
basic_dict = job_info.to_dict_basic()
test_dict = {'id': self.test_id,
'target': self.test_target,
'target_name': self.test_target.split('/')[-1],
'incremental': '',
'verbose': int(self.test_verbose),
'partial': partial_list,
'output': 'output/' + self.test_id + '.zip',
'status': self.test_status,
'extra': self.test_extra,
'stdout': 'output/stdout.' + self.test_id,
'stderr': 'output/stderr.' + self.test_id,
'start_time': self.test_start_time,
'finish_time': self.test_finish_time,
'isPartial': True,
'isIncremental': False
}
for key, value in test_dict.items():
self.assertEqual(value, basic_dict[key],
'the ' + key + ' is not converted to basic form dict correctly'
)
def test_to_dict_detail(self):
partial_list = ['system', 'vendor']
test_incremental = 'target/source.zip'
job_info = self.setup_job(partial=partial_list, incremental=test_incremental)
mock_target_lib = Mock()
mock_target_lib.get_build_by_path = Mock(
side_effect = [
Mock(file_name='build.zip', build_version=''),
Mock(file_name='source.zip', build_version='')
]
)
test_dict = {'id': self.test_id,
'target': self.test_target,
'incremental': 'target/source.zip',
'verbose': int(self.test_verbose),
'partial': partial_list,
'output': 'output/' + self.test_id + '.zip',
'status': self.test_status,
'extra': self.test_extra,
'stdout': 'NO STD OUTPUT IS FOUND',
'stderr': 'NO STD ERROR IS FOUND',
'start_time': self.test_start_time,
'finish_time': self.test_finish_time,
'isPartial': True,
'isIncremental': True
}
# Test with no stdout and stderr
dict_detail = job_info.to_dict_detail(mock_target_lib)
mock_target_lib.get_build_by_path.assert_any_call(self.test_target)
mock_target_lib.get_build_by_path.assert_any_call(test_incremental)
for key, value in test_dict.items():
self.assertEqual(value, dict_detail[key],
'the ' + key + ' is not converted to detailed dict correctly'
)
# Test with mocked stdout and stderr
mock_target_lib.get_build_by_path = Mock(
side_effect = [
Mock(file_name='build.zip', build_version=''),
Mock(file_name='source.zip', build_version='')
]
)
mock_file = mock_open(read_data="mock output")
with patch("builtins.open", mock_file):
dict_detail = job_info.to_dict_detail(mock_target_lib)
test_dict['stderr'] = 'mock output'
test_dict['stdout'] = 'mock output'
for key, value in test_dict.items():
self.assertEqual(value, dict_detail[key],
'the ' + key + ' is not converted to detailed dict correctly'
)
class TestProcessesManagement(unittest.TestCase):
pass
if __name__ == '__main__':
unittest.main()