| #!/usr/bin/python |
| # |
| # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit testing checker.py.""" |
| |
| import array |
| import collections |
| import cStringIO |
| import hashlib |
| import itertools |
| import os |
| import unittest |
| |
| # Pylint cannot find mox. |
| # pylint: disable=F0401 |
| import mox |
| |
| import checker |
| import common |
| import payload as update_payload # avoid name conflicts later. |
| import test_utils |
| import update_metadata_pb2 |
| |
| |
| _PRIVKEY_FILE_NAME = 'payload-test-key.pem' |
| _PUBKEY_FILE_NAME = 'payload-test-key.pub' |
| |
| |
| def _OpTypeByName(op_name): |
| op_name_to_type = { |
| 'REPLACE': common.OpType.REPLACE, |
| 'REPLACE_BZ': common.OpType.REPLACE_BZ, |
| 'MOVE': common.OpType.MOVE, |
| 'BSDIFF': common.OpType.BSDIFF, |
| } |
| return op_name_to_type[op_name] |
| |
| |
| def _KiB(count): |
| """Return the byte size of a given number of (binary) kilobytes.""" |
| return count << 10 |
| |
| |
| def _MiB(count): |
| """Return the byte size of a given number of (binary) megabytes.""" |
| return count << 20 |
| |
| |
| def _GiB(count): |
| """Return the byte size of a given number of (binary) gigabytes.""" |
| return count << 30 |
| |
| |
| def _GetPayloadChecker(payload_gen_write_to_file_func, *largs, **dargs): |
| """Returns a payload checker from a given payload generator.""" |
| payload_file = cStringIO.StringIO() |
| payload_gen_write_to_file_func(payload_file, *largs, **dargs) |
| payload_file.seek(0) |
| payload = update_payload.Payload(payload_file) |
| payload.Init() |
| return checker.PayloadChecker(payload) |
| |
| |
| def _GetPayloadCheckerWithData(payload_gen): |
| """Returns a payload checker from a given payload generator.""" |
| payload_file = cStringIO.StringIO() |
| payload_gen.WriteToFile(payload_file) |
| payload_file.seek(0) |
| payload = update_payload.Payload(payload_file) |
| payload.Init() |
| return checker.PayloadChecker(payload) |
| |
| |
| # (i) this class doesn't need an __init__(); (ii) unit testing is all about |
| # running protected methods; (iii) don't bark about missing members of classes |
| # you cannot import. |
| # pylint: disable=W0232 |
| # pylint: disable=W0212 |
| # pylint: disable=E1101 |
| class PayloadCheckerTest(mox.MoxTestBase): |
| """Tests the PayloadChecker class. |
| |
| In addition to ordinary testFoo() methods, which are automatically invoked by |
| the unittest framework, in this class we make use of DoBarTest() calls that |
| implement parametric tests of certain features. In order to invoke each test, |
| which embodies a unique combination of parameter values, as a complete unit |
| test, we perform explicit enumeration of the parameter space and create |
| individual invocation contexts for each, which are then bound as |
| testBar__param1=val1__param2=val2(). The enumeration of parameter spaces for |
| all such tests is done in AddAllParametricTests(). |
| |
| """ |
| |
| def MockPayload(self): |
| """Create a mock payload object, complete with a mock menifest.""" |
| payload = self.mox.CreateMock(update_payload.Payload) |
| payload.is_init = True |
| payload.manifest = self.mox.CreateMock( |
| update_metadata_pb2.DeltaArchiveManifest) |
| return payload |
| |
| @staticmethod |
| def NewExtent(start_block, num_blocks): |
| """Returns an Extent message. |
| |
| Each of the provided fields is set iff it is >= 0; otherwise, it's left at |
| its default state. |
| |
| Args: |
| start_block: the starting block of the extent |
| num_blocks: the number of blocks in the extent |
| Returns: |
| An Extent message. |
| |
| """ |
| ex = update_metadata_pb2.Extent() |
| if start_block >= 0: |
| ex.start_block = start_block |
| if num_blocks >= 0: |
| ex.num_blocks = num_blocks |
| return ex |
| |
| @staticmethod |
| def NewExtentList(*args): |
| """Returns an list of extents. |
| |
| Args: |
| *args: (start_block, num_blocks) pairs defining the extents |
| Returns: |
| A list of Extent objects. |
| |
| """ |
| ex_list = [] |
| for start_block, num_blocks in args: |
| ex_list.append(PayloadCheckerTest.NewExtent(start_block, num_blocks)) |
| return ex_list |
| |
| @staticmethod |
| def AddToMessage(repeated_field, field_vals): |
| for field_val in field_vals: |
| new_field = repeated_field.add() |
| new_field.CopyFrom(field_val) |
| |
| def assertIsNone(self, val): |
| """Asserts that val is None (TODO remove once we upgrade to Python 2.7). |
| |
| Note that we're using assertEqual so as for it to show us the actual |
| non-None value. |
| |
| Args: |
| val: value/object to be equated to None |
| |
| """ |
| self.assertEqual(val, None) |
| |
| def SetupAddElemTest(self, is_present, is_submsg, convert=str, |
| linebreak=False, indent=0): |
| """Setup for testing of _CheckElem() and its derivatives. |
| |
| Args: |
| is_present: whether or not the element is found in the message |
| is_submsg: whether the element is a sub-message itself |
| convert: a representation conversion function |
| linebreak: whether or not a linebreak is to be used in the report |
| indent: indentation used for the report |
| Returns: |
| msg: a mock message object |
| report: a mock report object |
| subreport: a mock sub-report object |
| name: an element name to check |
| val: expected element value |
| |
| """ |
| name = 'foo' |
| val = 'fake submsg' if is_submsg else 'fake field' |
| subreport = 'fake subreport' |
| |
| # Create a mock message. |
| msg = self.mox.CreateMock(update_metadata_pb2.message.Message) |
| msg.HasField(name).AndReturn(is_present) |
| setattr(msg, name, val) |
| |
| # Create a mock report. |
| report = self.mox.CreateMock(checker._PayloadReport) |
| if is_present: |
| if is_submsg: |
| report.AddSubReport(name).AndReturn(subreport) |
| else: |
| report.AddField(name, convert(val), linebreak=linebreak, indent=indent) |
| |
| self.mox.ReplayAll() |
| return (msg, report, subreport, name, val) |
| |
| def DoAddElemTest(self, is_present, is_mandatory, is_submsg, convert, |
| linebreak, indent): |
| """Parametric testing of _CheckElem(). |
| |
| Args: |
| is_present: whether or not the element is found in the message |
| is_mandatory: whether or not it's a mandatory element |
| is_submsg: whether the element is a sub-message itself |
| convert: a representation conversion function |
| linebreak: whether or not a linebreak is to be used in the report |
| indent: indentation used for the report |
| |
| """ |
| msg, report, subreport, name, val = self.SetupAddElemTest( |
| is_present, is_submsg, convert, linebreak, indent) |
| |
| largs = [msg, name, report, is_mandatory, is_submsg] |
| dargs = {'convert': convert, 'linebreak': linebreak, 'indent': indent} |
| if is_mandatory and not is_present: |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckElem, *largs, **dargs) |
| else: |
| ret_val, ret_subreport = checker.PayloadChecker._CheckElem(*largs, |
| **dargs) |
| self.assertEquals(ret_val, val if is_present else None) |
| self.assertEquals(ret_subreport, |
| subreport if is_present and is_submsg else None) |
| |
| def DoAddFieldTest(self, is_mandatory, is_present, convert, linebreak, |
| indent): |
| """Parametric testing of _Check{Mandatory,Optional}Field(). |
| |
| Args: |
| is_mandatory: whether we're testing a mandatory call |
| is_present: whether or not the element is found in the message |
| convert: a representation conversion function |
| linebreak: whether or not a linebreak is to be used in the report |
| indent: indentation used for the report |
| |
| """ |
| msg, report, _, name, val = self.SetupAddElemTest( |
| is_present, False, convert, linebreak, indent) |
| |
| # Prepare for invocation of the tested method. |
| largs = [msg, name, report] |
| dargs = {'convert': convert, 'linebreak': linebreak, 'indent': indent} |
| if is_mandatory: |
| largs.append('bar') |
| tested_func = checker.PayloadChecker._CheckMandatoryField |
| else: |
| tested_func = checker.PayloadChecker._CheckOptionalField |
| |
| # Test the method call. |
| if is_mandatory and not is_present: |
| self.assertRaises(update_payload.PayloadError, tested_func, *largs, |
| **dargs) |
| else: |
| ret_val = tested_func(*largs, **dargs) |
| self.assertEquals(ret_val, val if is_present else None) |
| |
| def DoAddSubMsgTest(self, is_mandatory, is_present): |
| """Parametrized testing of _Check{Mandatory,Optional}SubMsg(). |
| |
| Args: |
| is_mandatory: whether we're testing a mandatory call |
| is_present: whether or not the element is found in the message |
| |
| """ |
| msg, report, subreport, name, val = self.SetupAddElemTest(is_present, True) |
| |
| # Prepare for invocation of the tested method. |
| largs = [msg, name, report] |
| if is_mandatory: |
| largs.append('bar') |
| tested_func = checker.PayloadChecker._CheckMandatorySubMsg |
| else: |
| tested_func = checker.PayloadChecker._CheckOptionalSubMsg |
| |
| # Test the method call. |
| if is_mandatory and not is_present: |
| self.assertRaises(update_payload.PayloadError, tested_func, *largs) |
| else: |
| ret_val, ret_subreport = tested_func(*largs) |
| self.assertEquals(ret_val, val if is_present else None) |
| self.assertEquals(ret_subreport, subreport if is_present else None) |
| |
| def testCheckPresentIff(self): |
| """Tests _CheckPresentIff().""" |
| self.assertIsNone(checker.PayloadChecker._CheckPresentIff( |
| None, None, 'foo', 'bar', 'baz')) |
| self.assertIsNone(checker.PayloadChecker._CheckPresentIff( |
| 'a', 'b', 'foo', 'bar', 'baz')) |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckPresentIff, |
| 'a', None, 'foo', 'bar', 'baz') |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckPresentIff, |
| None, 'b', 'foo', 'bar', 'baz') |
| |
| def DoCheckSha256SignatureTest(self, expect_pass, expect_subprocess_call, |
| sig_data, sig_asn1_header, |
| returned_signed_hash, expected_signed_hash): |
| """Parametric testing of _CheckSha256SignatureTest(). |
| |
| Args: |
| expect_pass: whether or not it should pass |
| expect_subprocess_call: whether to expect the openssl call to happen |
| sig_data: the signature raw data |
| sig_asn1_header: the ASN1 header |
| returned_signed_hash: the signed hash data retuned by openssl |
| expected_signed_hash: the signed hash data to compare against |
| |
| """ |
| # Stub out the subprocess invocation. |
| self.mox.StubOutWithMock(checker.PayloadChecker, '_Run') |
| if expect_subprocess_call: |
| checker.PayloadChecker._Run(mox.IsA(list), send_data=sig_data).AndReturn( |
| (sig_asn1_header + returned_signed_hash, None)) |
| |
| self.mox.ReplayAll() |
| if expect_pass: |
| self.assertIsNone(checker.PayloadChecker._CheckSha256Signature( |
| sig_data, 'foo', expected_signed_hash, 'bar')) |
| else: |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckSha256Signature, |
| sig_data, 'foo', expected_signed_hash, 'bar') |
| |
| self.mox.UnsetStubs() |
| |
| def testCheckSha256Signature_Pass(self): |
| """Tests _CheckSha256Signature(); pass case.""" |
| sig_data = 'fake-signature'.ljust(256) |
| signed_hash = hashlib.sha256('fake-data').digest() |
| self.DoCheckSha256SignatureTest(True, True, sig_data, |
| common.SIG_ASN1_HEADER, signed_hash, |
| signed_hash) |
| |
| def testCheckSha256Signature_FailBadSignature(self): |
| """Tests _CheckSha256Signature(); fails due to malformed signature.""" |
| sig_data = 'fake-signature' # malformed (not 256 bytes in length) |
| signed_hash = hashlib.sha256('fake-data').digest() |
| self.DoCheckSha256SignatureTest(False, False, sig_data, |
| common.SIG_ASN1_HEADER, signed_hash, |
| signed_hash) |
| |
| def testCheckSha256Signature_FailBadOutputLength(self): |
| """Tests _CheckSha256Signature(); fails due to unexpected output length.""" |
| sig_data = 'fake-signature'.ljust(256) |
| signed_hash = 'fake-hash' # malformed (not 32 bytes in length) |
| self.DoCheckSha256SignatureTest(False, True, sig_data, |
| common.SIG_ASN1_HEADER, signed_hash, |
| signed_hash) |
| |
| def testCheckSha256Signature_FailBadAsnHeader(self): |
| """Tests _CheckSha256Signature(); fails due to bad ASN1 header.""" |
| sig_data = 'fake-signature'.ljust(256) |
| signed_hash = hashlib.sha256('fake-data').digest() |
| bad_asn1_header = 'bad-asn-header'.ljust(len(common.SIG_ASN1_HEADER)) |
| self.DoCheckSha256SignatureTest(False, True, sig_data, bad_asn1_header, |
| signed_hash, signed_hash) |
| |
| def testCheckSha256Signature_FailBadHash(self): |
| """Tests _CheckSha256Signature(); fails due to bad hash returned.""" |
| sig_data = 'fake-signature'.ljust(256) |
| expected_signed_hash = hashlib.sha256('fake-data').digest() |
| returned_signed_hash = hashlib.sha256('bad-fake-data').digest() |
| self.DoCheckSha256SignatureTest(False, True, sig_data, |
| common.SIG_ASN1_HEADER, |
| expected_signed_hash, returned_signed_hash) |
| |
| def testCheckBlocksFitLength_Pass(self): |
| """Tests _CheckBlocksFitLength(); pass case.""" |
| self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( |
| 64, 4, 16, 'foo')) |
| self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( |
| 60, 4, 16, 'foo')) |
| self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( |
| 49, 4, 16, 'foo')) |
| self.assertIsNone(checker.PayloadChecker._CheckBlocksFitLength( |
| 48, 3, 16, 'foo')) |
| |
| def testCheckBlocksFitLength_TooManyBlocks(self): |
| """Tests _CheckBlocksFitLength(); fails due to excess blocks.""" |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 64, 5, 16, 'foo') |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 60, 5, 16, 'foo') |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 49, 5, 16, 'foo') |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 48, 4, 16, 'foo') |
| |
| def testCheckBlocksFitLength_TooFewBlocks(self): |
| """Tests _CheckBlocksFitLength(); fails due to insufficient blocks.""" |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 64, 3, 16, 'foo') |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 60, 3, 16, 'foo') |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 49, 3, 16, 'foo') |
| self.assertRaises(update_payload.PayloadError, |
| checker.PayloadChecker._CheckBlocksFitLength, |
| 48, 2, 16, 'foo') |
| |
| def DoCheckManifestTest(self, fail_mismatched_block_size, fail_bad_sigs, |
| fail_mismatched_oki_ori, fail_bad_oki, fail_bad_ori, |
| fail_bad_nki, fail_bad_nri, fail_missing_ops): |
| """Parametric testing of _CheckManifest(). |
| |
| Args: |
| fail_mismatched_block_size: simulate a missing block_size field |
| fail_bad_sigs: make signatures descriptor inconsistent |
| fail_mismatched_oki_ori: make old rootfs/kernel info partially present |
| fail_bad_oki: tamper with old kernel info |
| fail_bad_ori: tamper with old rootfs info |
| fail_bad_nki: tamper with new kernel info |
| fail_bad_nri: tamper with new rootfs info |
| fail_missing_ops: simulate a manifest without any operations |
| |
| """ |
| # Generate a test payload. For this test, we only care about the manifest |
| # and don't need any data blobs, hence we can use a plain paylaod generator |
| # (which also gives us more control on things that can be screwed up). |
| payload_gen = test_utils.PayloadGenerator() |
| |
| # Tamper with block size, if required. |
| if fail_mismatched_block_size: |
| payload_gen.SetBlockSize(_KiB(1)) |
| else: |
| payload_gen.SetBlockSize(_KiB(4)) |
| |
| # Add some operations. |
| if not fail_missing_ops: |
| payload_gen.AddOperation(False, common.OpType.MOVE, |
| src_extents=[(0, 16), (16, 497)], |
| dst_extents=[(16, 496), (0, 16)]) |
| payload_gen.AddOperation(True, common.OpType.MOVE, |
| src_extents=[(0, 8), (8, 8)], |
| dst_extents=[(8, 8), (0, 8)]) |
| |
| # Set an invalid signatures block (offset but no size), if required. |
| if fail_bad_sigs: |
| payload_gen.SetSignatures(32, None) |
| |
| # Add old kernel/rootfs partition info, as required. |
| if fail_mismatched_oki_ori or fail_bad_oki: |
| oki_hash = (None if fail_bad_oki |
| else hashlib.sha256('fake-oki-content').digest()) |
| payload_gen.SetPartInfo(True, False, _KiB(512), oki_hash) |
| if not fail_mismatched_oki_ori and fail_bad_ori: |
| payload_gen.SetPartInfo(False, False, _MiB(8), None) |
| |
| # Add new kernel/rootfs partition info. |
| payload_gen.SetPartInfo( |
| True, True, _KiB(512), |
| None if fail_bad_nki else hashlib.sha256('fake-nki-content').digest()) |
| payload_gen.SetPartInfo( |
| False, True, _MiB(8), |
| None if fail_bad_nri else hashlib.sha256('fake-nri-content').digest()) |
| |
| # Create the test object. |
| payload_checker = _GetPayloadChecker(payload_gen.WriteToFile) |
| report = checker._PayloadReport() |
| |
| should_fail = (fail_mismatched_block_size or fail_bad_sigs or |
| fail_mismatched_oki_ori or fail_bad_oki or fail_bad_ori or |
| fail_bad_nki or fail_bad_nri or fail_missing_ops) |
| if should_fail: |
| self.assertRaises(update_payload.PayloadError, |
| payload_checker._CheckManifest, report) |
| else: |
| self.assertIsNone(payload_checker._CheckManifest(report)) |
| |
| def testCheckLength(self): |
| """Tests _CheckLength().""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| block_size = payload_checker.block_size |
| |
| # Passes. |
| self.assertIsNone(payload_checker._CheckLength( |
| int(3.5 * block_size), 4, 'foo', 'bar')) |
| # Fails, too few blocks. |
| self.assertRaises(update_payload.PayloadError, |
| payload_checker._CheckLength, |
| int(3.5 * block_size), 3, 'foo', 'bar') |
| # Fails, too many blocks. |
| self.assertRaises(update_payload.PayloadError, |
| payload_checker._CheckLength, |
| int(3.5 * block_size), 5, 'foo', 'bar') |
| |
| def testCheckExtents(self): |
| """Tests _CheckExtents().""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| block_size = payload_checker.block_size |
| |
| # Passes w/ all real extents. |
| extents = self.NewExtentList((0, 4), (8, 3), (1024, 16)) |
| self.assertEquals( |
| payload_checker._CheckExtents(extents, (1024 + 16) * block_size, |
| collections.defaultdict(int), 'foo'), |
| 23) |
| |
| # Passes w/ pseudo-extents (aka sparse holes). |
| extents = self.NewExtentList((0, 4), (common.PSEUDO_EXTENT_MARKER, 5), |
| (8, 3)) |
| self.assertEquals( |
| payload_checker._CheckExtents(extents, (1024 + 16) * block_size, |
| collections.defaultdict(int), 'foo', |
| allow_pseudo=True), |
| 12) |
| |
| # Passes w/ pseudo-extent due to a signature. |
| extents = self.NewExtentList((common.PSEUDO_EXTENT_MARKER, 2)) |
| self.assertEquals( |
| payload_checker._CheckExtents(extents, (1024 + 16) * block_size, |
| collections.defaultdict(int), 'foo', |
| allow_signature=True), |
| 2) |
| |
| # Fails, extent missing a start block. |
| extents = self.NewExtentList((-1, 4), (8, 3), (1024, 16)) |
| self.assertRaises( |
| update_payload.PayloadError, payload_checker._CheckExtents, |
| extents, (1024 + 16) * block_size, collections.defaultdict(int), |
| 'foo') |
| |
| # Fails, extent missing block count. |
| extents = self.NewExtentList((0, -1), (8, 3), (1024, 16)) |
| self.assertRaises( |
| update_payload.PayloadError, payload_checker._CheckExtents, |
| extents, (1024 + 16) * block_size, collections.defaultdict(int), |
| 'foo') |
| |
| # Fails, extent has zero blocks. |
| extents = self.NewExtentList((0, 4), (8, 3), (1024, 0)) |
| self.assertRaises( |
| update_payload.PayloadError, payload_checker._CheckExtents, |
| extents, (1024 + 16) * block_size, collections.defaultdict(int), |
| 'foo') |
| |
| # Fails, extent exceeds partition boundaries. |
| extents = self.NewExtentList((0, 4), (8, 3), (1024, 16)) |
| self.assertRaises( |
| update_payload.PayloadError, payload_checker._CheckExtents, |
| extents, (1024 + 15) * block_size, collections.defaultdict(int), |
| 'foo') |
| |
| def testCheckReplaceOperation(self): |
| """Tests _CheckReplaceOperation() where op.type == REPLACE.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| block_size = payload_checker.block_size |
| data_length = 10000 |
| |
| op = self.mox.CreateMock( |
| update_metadata_pb2.DeltaArchiveManifest.InstallOperation) |
| op.type = common.OpType.REPLACE |
| |
| # Pass. |
| op.src_extents = [] |
| self.assertIsNone( |
| payload_checker._CheckReplaceOperation( |
| op, data_length, (data_length + block_size - 1) / block_size, |
| 'foo')) |
| |
| # Fail, src extents founds. |
| op.src_extents = ['bar'] |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckReplaceOperation, |
| op, data_length, (data_length + block_size - 1) / block_size, 'foo') |
| |
| # Fail, missing data. |
| op.src_extents = [] |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckReplaceOperation, |
| op, None, (data_length + block_size - 1) / block_size, 'foo') |
| |
| # Fail, length / block number mismatch. |
| op.src_extents = ['bar'] |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckReplaceOperation, |
| op, data_length, (data_length + block_size - 1) / block_size + 1, 'foo') |
| |
| def testCheckReplaceBzOperation(self): |
| """Tests _CheckReplaceOperation() where op.type == REPLACE_BZ.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| block_size = payload_checker.block_size |
| data_length = block_size * 3 |
| |
| op = self.mox.CreateMock( |
| update_metadata_pb2.DeltaArchiveManifest.InstallOperation) |
| op.type = common.OpType.REPLACE_BZ |
| |
| # Pass. |
| op.src_extents = [] |
| self.assertIsNone( |
| payload_checker._CheckReplaceOperation( |
| op, data_length, (data_length + block_size - 1) / block_size + 5, |
| 'foo')) |
| |
| # Fail, src extents founds. |
| op.src_extents = ['bar'] |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckReplaceOperation, |
| op, data_length, (data_length + block_size - 1) / block_size + 5, 'foo') |
| |
| # Fail, missing data. |
| op.src_extents = [] |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckReplaceOperation, |
| op, None, (data_length + block_size - 1) / block_size, 'foo') |
| |
| # Fail, too few blocks to justify BZ. |
| op.src_extents = [] |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckReplaceOperation, |
| op, data_length, (data_length + block_size - 1) / block_size, 'foo') |
| |
| def testCheckMoveOperation_Pass(self): |
| """Tests _CheckMoveOperation(); pass case.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = common.OpType.MOVE |
| |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 128))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((16, 128), (512, 6))) |
| self.assertIsNone( |
| payload_checker._CheckMoveOperation(op, None, 134, 134, 'foo')) |
| |
| def testCheckMoveOperation_FailContainsData(self): |
| """Tests _CheckMoveOperation(); fails, message contains data.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = common.OpType.MOVE |
| |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 128))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((16, 128), (512, 6))) |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckMoveOperation, |
| op, 1024, 134, 134, 'foo') |
| |
| def testCheckMoveOperation_FailInsufficientSrcBlocks(self): |
| """Tests _CheckMoveOperation(); fails, not enough actual src blocks.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = common.OpType.MOVE |
| |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 127))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((16, 128), (512, 6))) |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckMoveOperation, |
| op, None, 134, 134, 'foo') |
| |
| def testCheckMoveOperation_FailInsufficientDstBlocks(self): |
| """Tests _CheckMoveOperation(); fails, not enough actual dst blocks.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = common.OpType.MOVE |
| |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 128))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((16, 128), (512, 5))) |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckMoveOperation, |
| op, None, 134, 134, 'foo') |
| |
| def testCheckMoveOperation_FailExcessSrcBlocks(self): |
| """Tests _CheckMoveOperation(); fails, too many actual src blocks.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = common.OpType.MOVE |
| |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 128))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((16, 128), (512, 5))) |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckMoveOperation, |
| op, None, 134, 134, 'foo') |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 129))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((16, 128), (512, 6))) |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckMoveOperation, |
| op, None, 134, 134, 'foo') |
| |
| def testCheckMoveOperation_FailExcessDstBlocks(self): |
| """Tests _CheckMoveOperation(); fails, too many actual dst blocks.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = common.OpType.MOVE |
| |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 128))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((16, 128), (512, 7))) |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckMoveOperation, |
| op, None, 134, 134, 'foo') |
| |
| def testCheckMoveOperation_FailStagnantBlocks(self): |
| """Tests _CheckMoveOperation(); fails, there are blocks that do not move.""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = common.OpType.MOVE |
| |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 4), (12, 2), (1024, 128))) |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((8, 128), (512, 6))) |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckMoveOperation, |
| op, None, 134, 134, 'foo') |
| |
| def testCheckBsdiff(self): |
| """Tests _CheckMoveOperation().""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| |
| # Pass. |
| self.assertIsNone( |
| payload_checker._CheckBsdiffOperation(10000, 3, 'foo')) |
| |
| # Fail, missing data blob. |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckBsdiffOperation, |
| None, 3, 'foo') |
| |
| # Fail, too big of a diff blob (unjustified). |
| self.assertRaises( |
| update_payload.PayloadError, |
| payload_checker._CheckBsdiffOperation, |
| 10000, 2, 'foo') |
| |
| def DoCheckOperationTest(self, op_type_name, is_last, allow_signature, |
| allow_unhashed, fail_src_extents, fail_dst_extents, |
| fail_mismatched_data_offset_length, |
| fail_missing_dst_extents, fail_src_length, |
| fail_dst_length, fail_data_hash, |
| fail_prev_data_offset): |
| """Parametric testing of _CheckOperation(). |
| |
| Args: |
| op_type_name: 'REPLACE', 'REPLACE_BZ', 'MOVE' or 'BSDIFF' |
| is_last: whether we're testing the last operation in a sequence |
| allow_signature: whether we're testing a signature-capable operation |
| allow_unhashed: whether we're allowing to not hash the data |
| fail_src_extents: tamper with src extents |
| fail_dst_extents: tamper with dst extents |
| fail_mismatched_data_offset_length: make data_{offset,length} inconsistent |
| fail_missing_dst_extents: do not include dst extents |
| fail_src_length: make src length inconsistent |
| fail_dst_length: make dst length inconsistent |
| fail_data_hash: tamper with the data blob hash |
| fail_prev_data_offset: make data space uses incontiguous |
| |
| """ |
| op_type = _OpTypeByName(op_type_name) |
| |
| # Create the test object. |
| payload = self.MockPayload() |
| payload_checker = checker.PayloadChecker(payload) |
| block_size = payload_checker.block_size |
| |
| # Create auxiliary arguments. |
| old_part_size = _MiB(4) |
| new_part_size = _MiB(8) |
| old_block_counters = array.array( |
| 'B', [0] * ((old_part_size + block_size - 1) / block_size)) |
| new_block_counters = array.array( |
| 'B', [0] * ((new_part_size + block_size - 1) / block_size)) |
| prev_data_offset = 1876 |
| blob_hash_counts = collections.defaultdict(int) |
| |
| # Create the operation object for the test. |
| op = update_metadata_pb2.DeltaArchiveManifest.InstallOperation() |
| op.type = op_type |
| |
| total_src_blocks = 0 |
| if op_type in (common.OpType.MOVE, common.OpType.BSDIFF): |
| if fail_src_extents: |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 0))) |
| else: |
| self.AddToMessage(op.src_extents, |
| self.NewExtentList((0, 16))) |
| total_src_blocks = 16 |
| |
| if op_type != common.OpType.MOVE: |
| if not fail_mismatched_data_offset_length: |
| op.data_length = 16 * block_size - 8 |
| if fail_prev_data_offset: |
| op.data_offset = prev_data_offset + 16 |
| else: |
| op.data_offset = prev_data_offset |
| |
| fake_data = 'fake-data'.ljust(op.data_length) |
| if not (allow_unhashed or (is_last and allow_signature and |
| op_type == common.OpType.REPLACE)): |
| if not fail_data_hash: |
| # Create a valid data blob hash. |
| op.data_sha256_hash = hashlib.sha256(fake_data).digest() |
| payload.ReadDataBlob(op.data_offset, op.data_length).AndReturn( |
| fake_data) |
| elif fail_data_hash: |
| # Create an invalid data blob hash. |
| op.data_sha256_hash = hashlib.sha256( |
| fake_data.replace(' ', '-')).digest() |
| payload.ReadDataBlob(op.data_offset, op.data_length).AndReturn( |
| fake_data) |
| |
| total_dst_blocks = 0 |
| if not fail_missing_dst_extents: |
| total_dst_blocks = 16 |
| if fail_dst_extents: |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((4, 16), (32, 0))) |
| else: |
| self.AddToMessage(op.dst_extents, |
| self.NewExtentList((4, 8), (64, 8))) |
| |
| if total_src_blocks: |
| if fail_src_length: |
| op.src_length = total_src_blocks * block_size + 8 |
| else: |
| op.src_length = total_src_blocks * block_size |
| elif fail_src_length: |
| # Add an orphaned src_length. |
| op.src_length = 16 |
| |
| if total_dst_blocks: |
| if fail_dst_length: |
| op.dst_length = total_dst_blocks * block_size + 8 |
| else: |
| op.dst_length = total_dst_blocks * block_size |
| |
| self.mox.ReplayAll() |
| should_fail = (fail_src_extents or fail_dst_extents or |
| fail_mismatched_data_offset_length or |
| fail_missing_dst_extents or fail_src_length or |
| fail_dst_length or fail_data_hash or fail_prev_data_offset) |
| largs = [op, 'foo', is_last, old_block_counters, new_block_counters, |
| old_part_size, new_part_size, prev_data_offset, allow_signature, |
| allow_unhashed, blob_hash_counts] |
| if should_fail: |
| self.assertRaises(update_payload.PayloadError, |
| payload_checker._CheckOperation, *largs) |
| else: |
| self.assertEqual(payload_checker._CheckOperation(*largs), |
| op.data_length if op.HasField('data_length') else 0) |
| |
| def testAllocBlockCounters(self): |
| """Tests _CheckMoveOperation().""" |
| payload_checker = checker.PayloadChecker(self.MockPayload()) |
| block_size = payload_checker.block_size |
| |
| # Check allocation for block-aligned partition size, ensure it's integers. |
| result = payload_checker._AllocBlockCounters(16 * block_size) |
| self.assertEqual(len(result), 16) |
| self.assertEqual(type(result[0]), int) |
| |
| # Check allocation of unaligned partition sizes. |
| result = payload_checker._AllocBlockCounters(16 * block_size - 1) |
| self.assertEqual(len(result), 16) |
| result = payload_checker._AllocBlockCounters(16 * block_size + 1) |
| self.assertEqual(len(result), 17) |
| |
| def DoCheckOperationsTest(self, fail_bad_type, |
| fail_nonexhaustive_full_update): |
| # Generate a test payload. For this test, we only care about one |
| # (arbitrary) set of operations, so we'll only be generating kernel and |
| # test with them. |
| payload_gen = test_utils.PayloadGenerator() |
| |
| block_size = _KiB(4) |
| payload_gen.SetBlockSize(block_size) |
| |
| rootfs_part_size = _MiB(8) |
| |
| # Fake rootfs operations in a full update, tampered with as required. |
| rootfs_op_type = common.OpType.REPLACE |
| if fail_bad_type: |
| # Choose a type value that's bigger than the highest valid value. |
| for valid_op_type in common.OpType.ALL: |
| rootfs_op_type = max(rootfs_op_type, valid_op_type) |
| rootfs_op_type += 1 |
| |
| rootfs_data_length = rootfs_part_size |
| if fail_nonexhaustive_full_update: |
| rootfs_data_length -= block_size |
| |
| payload_gen.AddOperation(False, rootfs_op_type, |
| dst_extents=[(0, rootfs_data_length / block_size)], |
| data_offset=0, |
| data_length=rootfs_data_length) |
| |
| # Create the test object. |
| payload_checker = _GetPayloadChecker(payload_gen.WriteToFile) |
| payload_checker.payload_type = checker._TYPE_FULL |
| report = checker._PayloadReport() |
| |
| should_fail = (fail_bad_type or fail_nonexhaustive_full_update) |
| largs = (payload_checker.payload.manifest.install_operations, report, |
| 'foo', 0, rootfs_part_size, 0, True, False) |
| if should_fail: |
| self.assertRaises(update_payload.PayloadError, |
| payload_checker._CheckOperations, *largs) |
| else: |
| self.assertEqual(payload_checker._CheckOperations(*largs), |
| rootfs_data_length) |
| |
| def DoCheckSignaturesTest(self, fail_empty_sigs_blob, fail_missing_pseudo_op, |
| fail_mismatched_pseudo_op, fail_sig_missing_fields, |
| fail_unknown_sig_version, fail_incorrect_sig): |
| # Generate a test payload. For this test, we only care about the signature |
| # block and how it relates to the payload hash. Therefore, we're generating |
| # a random (otherwise useless) payload for this purpose. |
| payload_gen = test_utils.EnhancedPayloadGenerator() |
| block_size = _KiB(4) |
| payload_gen.SetBlockSize(block_size) |
| rootfs_part_size = _MiB(2) |
| payload_gen.SetPartInfo(False, True, rootfs_part_size, |
| hashlib.sha256('fake-new-rootfs-content').digest()) |
| payload_gen.SetPartInfo(True, True, _KiB(16), |
| hashlib.sha256('fake-new-kernel-content').digest()) |
| payload_gen.AddOperationWithData( |
| False, common.OpType.REPLACE, |
| dst_extents=[(0, rootfs_part_size / block_size)], |
| data_blob=os.urandom(rootfs_part_size)) |
| |
| do_forge_pseudo_op = (fail_missing_pseudo_op or fail_mismatched_pseudo_op) |
| do_forge_sigs_data = (do_forge_pseudo_op or fail_empty_sigs_blob or |
| fail_sig_missing_fields or fail_unknown_sig_version |
| or fail_incorrect_sig) |
| |
| sigs_data = None |
| if do_forge_sigs_data: |
| sigs_gen = test_utils.SignaturesGenerator() |
| if not fail_empty_sigs_blob: |
| if fail_sig_missing_fields: |
| sig_data = None |
| else: |
| sig_data = test_utils.SignSha256('fake-payload-content', |
| _PRIVKEY_FILE_NAME) |
| sigs_gen.AddSig(5 if fail_unknown_sig_version else 1, sig_data) |
| |
| sigs_data = sigs_gen.ToBinary() |
| payload_gen.SetSignatures(payload_gen.curr_offset, len(sigs_data)) |
| |
| if do_forge_pseudo_op: |
| assert sigs_data is not None, 'should have forged signatures blob by now' |
| sigs_len = len(sigs_data) |
| payload_gen.AddOperation( |
| False, common.OpType.REPLACE, |
| data_offset=payload_gen.curr_offset / 2, |
| data_length=sigs_len / 2, |
| dst_extents=[(0, (sigs_len / 2 + block_size - 1) / block_size)]) |
| |
| # Generate payload (complete w/ signature) and create the test object. |
| payload_checker = _GetPayloadChecker( |
| payload_gen.WriteToFileWithData, sigs_data=sigs_data, |
| privkey_file_name=_PRIVKEY_FILE_NAME, |
| do_add_pseudo_operation=(not do_forge_pseudo_op)) |
| payload_checker.payload_type = checker._TYPE_FULL |
| report = checker._PayloadReport() |
| |
| # We have to check the manifest first in order to set signature attributes. |
| payload_checker._CheckManifest(report) |
| |
| should_fail = (fail_empty_sigs_blob or fail_missing_pseudo_op or |
| fail_mismatched_pseudo_op or fail_sig_missing_fields or |
| fail_unknown_sig_version or fail_incorrect_sig) |
| largs = (report, _PUBKEY_FILE_NAME) |
| if should_fail: |
| self.assertRaises(update_payload.PayloadError, |
| payload_checker._CheckSignatures, *largs) |
| else: |
| self.assertIsNone(payload_checker._CheckSignatures(*largs)) |
| |
| def DoRunTest(self, fail_wrong_payload_type, fail_invalid_block_size, |
| fail_mismatched_block_size, fail_excess_data): |
| # Generate a test payload. For this test, we generate a full update that |
| # has samle kernel and rootfs operations. Since most testing is done with |
| # internal PayloadChecker methods that are tested elsewhere, here we only |
| # tamper with what's actually being manipulated and/or tested in the Run() |
| # method itself. Note that the checker doesn't verify partition hashes, so |
| # they're safe to fake. |
| payload_gen = test_utils.EnhancedPayloadGenerator() |
| block_size = _KiB(4) |
| payload_gen.SetBlockSize(block_size) |
| kernel_part_size = _KiB(16) |
| rootfs_part_size = _MiB(2) |
| payload_gen.SetPartInfo(False, True, rootfs_part_size, |
| hashlib.sha256('fake-new-rootfs-content').digest()) |
| payload_gen.SetPartInfo(True, True, kernel_part_size, |
| hashlib.sha256('fake-new-kernel-content').digest()) |
| payload_gen.AddOperationWithData( |
| False, common.OpType.REPLACE, |
| dst_extents=[(0, rootfs_part_size / block_size)], |
| data_blob=os.urandom(rootfs_part_size)) |
| payload_gen.AddOperationWithData( |
| True, common.OpType.REPLACE, |
| dst_extents=[(0, kernel_part_size / block_size)], |
| data_blob=os.urandom(kernel_part_size)) |
| |
| # Generate payload (complete w/ signature) and create the test object. |
| payload_checker = _GetPayloadChecker( |
| payload_gen.WriteToFileWithData, |
| privkey_file_name=_PRIVKEY_FILE_NAME, |
| do_add_pseudo_operation=True, is_pseudo_in_kernel=True, |
| padding=os.urandom(1024) if fail_excess_data else None) |
| |
| if fail_invalid_block_size: |
| use_block_size = block_size + 5 # not a power of two |
| elif fail_mismatched_block_size: |
| use_block_size = block_size * 2 # different that payload stated |
| else: |
| use_block_size = block_size |
| dargs = { |
| 'pubkey_file_name': _PUBKEY_FILE_NAME, |
| 'assert_type': 'delta' if fail_wrong_payload_type else 'full', |
| 'block_size': use_block_size} |
| |
| should_fail = (fail_wrong_payload_type or fail_invalid_block_size or |
| fail_mismatched_block_size or fail_excess_data) |
| if should_fail: |
| self.assertRaises(update_payload.PayloadError, |
| payload_checker.Run, **dargs) |
| else: |
| self.assertIsNone(payload_checker.Run(**dargs)) |
| |
| |
| # This implements a generic API, hence the occasional unused args. |
| # pylint: disable=W0613 |
| def ValidateCheckOperationTest(op_type_name, is_last, allow_signature, |
| allow_unhashed, fail_src_extents, |
| fail_dst_extents, |
| fail_mismatched_data_offset_length, |
| fail_missing_dst_extents, fail_src_length, |
| fail_dst_length, fail_data_hash, |
| fail_prev_data_offset): |
| """Returns True iff the combination of arguments represents a valid test.""" |
| op_type = _OpTypeByName(op_type_name) |
| |
| # REPLACE/REPLACE_BZ operations don't read data from src partition. |
| if (op_type in (common.OpType.REPLACE, common.OpType.REPLACE_BZ) and ( |
| fail_src_extents or fail_src_length)): |
| return False |
| |
| # MOVE operations don't carry data. |
| if (op_type == common.OpType.MOVE and ( |
| fail_mismatched_data_offset_length or fail_data_hash or |
| fail_prev_data_offset)): |
| return False |
| |
| return True |
| |
| |
| def TestMethodBody(run_method_name, run_dargs): |
| """Returns a function that invokes a named method with named arguments.""" |
| return lambda self: getattr(self, run_method_name)(**run_dargs) |
| |
| |
| def AddParametricTests(tested_method_name, arg_space, validate_func=None): |
| """Enumerates and adds specific parametric tests to PayloadCheckerTest. |
| |
| This function enumerates a space of test parameters (defined by arg_space), |
| then binds a new, unique method name in PayloadCheckerTest to a test function |
| that gets handed the said parameters. This is a preferable approach to doing |
| the enumeration and invocation during the tests because this way each test is |
| treated as a complete run by the unittest framework, and so benefits from the |
| usual setUp/tearDown mechanics. |
| |
| Args: |
| tested_method_name: name of the tested PayloadChecker method |
| arg_space: a dictionary containing variables (keys) and lists of values |
| (values) associated with them |
| validate_func: a function used for validating test argument combinations |
| |
| """ |
| for value_tuple in itertools.product(*arg_space.itervalues()): |
| run_dargs = dict(zip(arg_space.iterkeys(), value_tuple)) |
| if validate_func and not validate_func(**run_dargs): |
| continue |
| run_method_name = 'Do%sTest' % tested_method_name |
| test_method_name = 'test%s' % tested_method_name |
| for arg_key, arg_val in run_dargs.iteritems(): |
| if arg_val or type(arg_val) is int: |
| test_method_name += '__%s=%s' % (arg_key, arg_val) |
| setattr(PayloadCheckerTest, test_method_name, |
| TestMethodBody(run_method_name, run_dargs)) |
| |
| |
| def AddAllParametricTests(): |
| """Enumerates and adds all parametric tests to PayloadCheckerTest.""" |
| # Add all _CheckElem() test cases. |
| AddParametricTests('AddElem', |
| {'linebreak': (True, False), |
| 'indent': (0, 1, 2), |
| 'convert': (str, lambda s: s[::-1]), |
| 'is_present': (True, False), |
| 'is_mandatory': (True, False), |
| 'is_submsg': (True, False)}) |
| |
| # Add all _Add{Mandatory,Optional}Field tests. |
| AddParametricTests('AddField', |
| {'is_mandatory': (True, False), |
| 'linebreak': (True, False), |
| 'indent': (0, 1, 2), |
| 'convert': (str, lambda s: s[::-1]), |
| 'is_present': (True, False)}) |
| |
| # Add all _Add{Mandatory,Optional}SubMsg tests. |
| AddParametricTests('AddSubMsg', |
| {'is_mandatory': (True, False), |
| 'is_present': (True, False)}) |
| |
| # Add all _CheckManifest() test cases. |
| AddParametricTests('CheckManifest', |
| {'fail_mismatched_block_size': (True, False), |
| 'fail_bad_sigs': (True, False), |
| 'fail_mismatched_oki_ori': (True, False), |
| 'fail_bad_oki': (True, False), |
| 'fail_bad_ori': (True, False), |
| 'fail_bad_nki': (True, False), |
| 'fail_bad_nri': (True, False), |
| 'fail_missing_ops': (True, False)}) |
| |
| # Add all _CheckOperation() test cases. |
| AddParametricTests('CheckOperation', |
| {'op_type_name': ('REPLACE', 'REPLACE_BZ', 'MOVE', |
| 'BSDIFF'), |
| 'is_last': (True, False), |
| 'allow_signature': (True, False), |
| 'allow_unhashed': (True, False), |
| 'fail_src_extents': (True, False), |
| 'fail_dst_extents': (True, False), |
| 'fail_mismatched_data_offset_length': (True, False), |
| 'fail_missing_dst_extents': (True, False), |
| 'fail_src_length': (True, False), |
| 'fail_dst_length': (True, False), |
| 'fail_data_hash': (True, False), |
| 'fail_prev_data_offset': (True, False)}, |
| validate_func=ValidateCheckOperationTest) |
| |
| # Add all _CheckOperations() test cases. |
| AddParametricTests('CheckOperations', |
| {'fail_bad_type': (True, False), |
| 'fail_nonexhaustive_full_update': (True, False)}) |
| |
| # Add all _CheckOperations() test cases. |
| AddParametricTests('CheckSignatures', |
| {'fail_empty_sigs_blob': (True, False), |
| 'fail_missing_pseudo_op': (True, False), |
| 'fail_mismatched_pseudo_op': (True, False), |
| 'fail_sig_missing_fields': (True, False), |
| 'fail_unknown_sig_version': (True, False), |
| 'fail_incorrect_sig': (True, False)}) |
| |
| # Add all Run() test cases. |
| AddParametricTests('Run', |
| {'fail_wrong_payload_type': (True, False), |
| 'fail_invalid_block_size': (True, False), |
| 'fail_mismatched_block_size': (True, False), |
| 'fail_excess_data': (True, False)}) |
| |
| |
| if __name__ == '__main__': |
| AddAllParametricTests() |
| unittest.main() |
