Refactor ota_utils to reduce dependency on global OPTIONS

Move some depency on global OPTIONS to parameters. This makes it easier
for other modules to re-use these functions w/o having to modify OPTIONS
variables before/after calling.

Test: th
Bug: 227536004
Change-Id: I3bab292f65a4222f0c9502bcf04fa7a85f8124c6
This commit is contained in:
Kelvin Zhang
2022-08-30 18:25:43 +00:00
parent f2514a60cb
commit bf01f8b3f0
4 changed files with 97 additions and 65 deletions

View File

@@ -1030,7 +1030,7 @@ class AbOtaPropertyFilesTest(PropertyFilesTestCase):
0, proc.returncode,
'Failed to run brillo_update_payload:\n{}'.format(stdoutdata))
signed_metadata_sig_file = payload_signer.Sign(metadata_sig_file)
signed_metadata_sig_file = payload_signer.SignHashFile(metadata_sig_file)
# Finally we can compare the two signatures.
with open(signed_metadata_sig_file, 'rb') as verify_fp:
@@ -1170,7 +1170,7 @@ class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
def test_Sign(self):
payload_signer = PayloadSigner()
input_file = os.path.join(self.testdata_dir, self.SIGFILE)
signed_file = payload_signer.Sign(input_file)
signed_file = payload_signer.SignHashFile(input_file)
verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
self._assertFilesEqual(verify_file, signed_file)
@@ -1184,7 +1184,7 @@ class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
payload_signer = PayloadSigner(
OPTIONS.package_key, OPTIONS.private_key_suffix, payload_signer="openssl")
input_file = os.path.join(self.testdata_dir, self.SIGFILE)
signed_file = payload_signer.Sign(input_file)
signed_file = payload_signer.SignHashFile(input_file)
verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
self._assertFilesEqual(verify_file, signed_file)
@@ -1199,7 +1199,7 @@ class PayloadSignerTest(test_utils.ReleaseToolsTestCase):
payload_signer = PayloadSigner(
OPTIONS.package_key, OPTIONS.private_key_suffix, payload_signer=external_signer)
input_file = os.path.join(self.testdata_dir, self.SIGFILE)
signed_file = payload_signer.Sign(input_file)
signed_file = payload_signer.SignHashFile(input_file)
verify_file = os.path.join(self.testdata_dir, self.SIGNED_SIGFILE)
self._assertFilesEqual(verify_file, signed_file)
@@ -1222,7 +1222,7 @@ class PayloadTest(test_utils.ReleaseToolsTestCase):
@staticmethod
def _create_payload_full(secondary=False):
target_file = construct_target_files(secondary)
payload = PayloadGenerator(secondary)
payload = PayloadGenerator(secondary, OPTIONS.wipe_user_data)
payload.Generate(target_file)
return payload
@@ -1295,6 +1295,9 @@ class PayloadTest(test_utils.ReleaseToolsTestCase):
common.OPTIONS.wipe_user_data = True
payload = self._create_payload_full()
payload.Sign(PayloadSigner())
with tempfile.NamedTemporaryFile() as fp:
with zipfile.ZipFile(fp, "w") as zfp:
payload.WriteToZip(zfp)
with open(payload.payload_properties) as properties_fp:
self.assertIn("POWERWASH=1", properties_fp.read())
@@ -1303,6 +1306,9 @@ class PayloadTest(test_utils.ReleaseToolsTestCase):
def test_Sign_secondary(self):
payload = self._create_payload_full(secondary=True)
payload.Sign(PayloadSigner())
with tempfile.NamedTemporaryFile() as fp:
with zipfile.ZipFile(fp, "w") as zfp:
payload.WriteToZip(zfp)
with open(payload.payload_properties) as properties_fp:
self.assertIn("SWITCH_SLOT_ON_REBOOT=0", properties_fp.read())
@@ -1337,22 +1343,6 @@ class PayloadTest(test_utils.ReleaseToolsTestCase):
continue
self.assertEqual(zipfile.ZIP_STORED, entry_info.compress_type)
@test_utils.SkipIfExternalToolsUnavailable()
def test_WriteToZip_unsignedPayload(self):
"""Unsigned payloads should not be allowed to be written to zip."""
payload = self._create_payload_full()
output_file = common.MakeTempFile(suffix='.zip')
with zipfile.ZipFile(output_file, 'w', allowZip64=True) as output_zip:
self.assertRaises(AssertionError, payload.WriteToZip, output_zip)
# Also test with incremental payload.
payload = self._create_payload_incremental()
output_file = common.MakeTempFile(suffix='.zip')
with zipfile.ZipFile(output_file, 'w', allowZip64=True) as output_zip:
self.assertRaises(AssertionError, payload.WriteToZip, output_zip)
@test_utils.SkipIfExternalToolsUnavailable()
def test_WriteToZip_secondary(self):
payload = self._create_payload_full(secondary=True)