| #!/usr/bin/env vpython3 |
| # Copyright 2020 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| """ |
| Tests for the Traffic Annotation Auditor. |
| """ |
| |
| import argparse |
| import itertools |
| import os |
| import platform |
| import sys |
| import tempfile |
| import unittest |
| |
| from typing import cast, Tuple |
| |
| from auditor import * |
| from error import * |
| from util import * |
| |
| # Path to the test_data/ dir. |
| TEST_DATA_DIR = SCRIPT_DIR.parent / "test_data" |
| |
| |
| class AuditorTest(unittest.TestCase): |
| def setUp(self): |
| build_path = TEST_DATA_DIR / "out" / "Debug" |
| |
| unittest.TestCase.setUp(self) |
| |
| path_filters = [ |
| (TEST_DATA_DIR / |
| "test_sample_annotations.cc").relative_to(SRC_DIR).as_posix() |
| ] |
| self.auditor_ui = AuditorUI(build_path, |
| path_filters, |
| no_filtering=False, |
| test_only=True) |
| self.auditor = self.auditor_ui.auditor |
| self.auditor.file_filter.git_file_for_testing = (TEST_DATA_DIR / |
| "git_list.txt") |
| |
| all_annotations = self.auditor.run_extractor(self.auditor_ui.build_path, |
| self.auditor_ui.path_filters, |
| skip_compdb=True) |
| |
| self.sample_annotations = {} |
| for annotation in all_annotations: |
| self.sample_annotations[annotation.unique_id] = annotation |
| |
| # Expose the traffic_annotation_pb2.py import from auditor.py. |
| global traffic_annotation |
| traffic_annotation = self.auditor_ui.traffic_annotation |
| |
| def deserialize(self, |
| file_name: str) -> Tuple[Annotation, List[AuditorError]]: |
| file_path = TEST_DATA_DIR / "extractor_outputs" / file_name |
| lines = file_path.read_text(encoding="utf-8").splitlines() |
| |
| annotation = Annotation() |
| language = extractor.LANGUAGE_MAPPING[Path(lines[0]).suffix] |
| type_name = extractor.AnnotationType(lines[2]) |
| extracted_annotation = extractor.Annotation(language=language, |
| file_path=lines[0], |
| line_number=int(lines[1]), |
| type_name=type_name, |
| unique_id=lines[3], |
| extra_id=lines[4], |
| text="\n".join(lines[5:])) |
| errors = annotation.deserialize(extracted_annotation) |
| return annotation, errors |
| |
| def create_annotation_sample(self, |
| type=Annotation.Type.COMPLETE, |
| unique_id=0, |
| second_id=0) -> Annotation: |
| if not unique_id: |
| instance, errors = self.deserialize("good_complete_annotation.txt") |
| assert not errors |
| return instance |
| instance = Annotation() |
| instance.type = type |
| instance.unique_id = UniqueId("S{}".format(unique_id)) |
| if second_id: |
| instance.second_id = UniqueId("S{}".format(second_id)) |
| else: |
| instance.second_id = UniqueId("") |
| return instance |
| |
| def run_id_checker(self, annotation: Annotation) -> List[AuditorError]: |
| id_checker = IdChecker(RESERVED_IDS) |
| return id_checker.check_ids([annotation]) |
| |
| def test_iterative_hash(self): |
| """Tests that the hash computation is identical to the C++ version.""" |
| # Reference values obtained from recursive_hash() in C++. |
| self.assertEqual(3556498, iterative_hash("test")) |
| self.assertEqual(10236504, iterative_hash("unique_id")) |
| self.assertEqual(70581310, iterative_hash("123_id")) |
| self.assertEqual(69491511, iterative_hash("ID123")) |
| self.assertEqual( |
| 98652091, |
| iterative_hash( |
| "a_unique_looooooooooooooooooooooooooooooooooooooooooooooooooooooong" |
| "_id")) |
| self.assertEqual(124751853, iterative_hash("bébé")) |
| |
| def test_get_files_from_git(self): |
| """Tests that FileFilter.get_files_from_git() returns correct files given |
| a mock git_list.txt file. It also inherently checks |
| FileFilter._is_supported_source_file().""" |
| filter = FileFilter([".cc", ".mm"]) |
| filter.git_file_for_testing = TEST_DATA_DIR / "git_list.txt" |
| filter.get_files_from_git() |
| |
| relevant_files = [ |
| "tools/traffic_annotation/scripts/test_data/" |
| "objective_cpp.mm", |
| "tools/traffic_annotation/scripts/test_data/" |
| "test_sample_annotations.cc", |
| ] |
| self.assertCountEqual([Path(f) for f in relevant_files], filter.git_files) |
| |
| def test_get_source_files(self): |
| """Tests that FileFilter.get_source_files() gives the correct list of |
| files, given a mock git_list.txt file.""" |
| filter = FileFilter([".cc", ".mm"]) |
| filter.git_file_for_testing = TEST_DATA_DIR / "git_list.txt" |
| filter.get_files_from_git() |
| |
| # Check if all files are returned with no ignore list and directory. |
| ignore_list = {} |
| self.assertCountEqual(filter.git_files, |
| filter.get_source_files(ignore_list, "")) |
| |
| # Check if a file is ignored when added to the ignore list. |
| ignore_list = { |
| ExceptionType.ALL: [re.compile(filter.git_files[0].as_posix())] |
| } |
| self.assertCountEqual( |
| set(filter.git_files) - set(filter.git_files[:1]), |
| filter.get_source_files(ignore_list, "")) |
| |
| # Check if files are filtered based on given directory. |
| ignore_list = {} |
| self.assertCountEqual( |
| filter.git_files, |
| filter.get_source_files(ignore_list, "tools/traffic_annotation")) |
| self.assertEqual([], filter.get_source_files(ignore_list, "content")) |
| |
| def test_is_safelisted(self): |
| """Tests if Auditor._is_safe_listed() works as expected. Inherently checks |
| Auditor.load_safe_list() as well.""" |
| auditor = Auditor(get_current_platform()) # use the real safe_list.txt |
| for t in ExceptionType: |
| # Anything in /tools directory is safelisted for all types. |
| self.assertTrue(auditor._is_safe_listed(Path("tools/something.cc"), t)) |
| self.assertTrue( |
| auditor._is_safe_listed(Path("tools/somewhere/something.mm"), t)) |
| |
| # Anything in a general folder is not safelisted for any type. |
| self.assertFalse(auditor._is_safe_listed(Path("something.cc"), t)) |
| self.assertFalse(auditor._is_safe_listed(Path("content/something.mm"), t)) |
| |
| # Files defining missing annotation functions in net/ are exception of |
| # 'missing' type. |
| self.assertTrue( |
| auditor._is_safe_listed(Path("net/url_request/url_fetcher.cc"), |
| ExceptionType.MISSING)) |
| self.assertTrue( |
| auditor._is_safe_listed(Path("net/url_request/url_request_context.cc"), |
| ExceptionType.MISSING)) |
| |
| # Files with the word "test" in their path can have the "test" annotation. |
| self.assertFalse( |
| auditor._is_safe_listed(Path("net/url_request/url_fetcher.cc"), |
| ExceptionType.TEST_ANNOTATION)) |
| self.assertTrue( |
| auditor._is_safe_listed(Path("chrome/browser/test_something.cc"), |
| ExceptionType.TEST_ANNOTATION)) |
| self.assertTrue( |
| auditor._is_safe_listed(Path("test/send_something.cc"), |
| ExceptionType.TEST_ANNOTATION)) |
| |
| def test_annotation_deserialization(self) -> None: |
| test_cases = [ |
| ("good_complete_annotation.txt", None, Annotation.Type.COMPLETE), |
| ("good_branched_completing_annotation.txt", None, |
| Annotation.Type.BRANCHED_COMPLETING), |
| ("good_completing_annotation.txt", None, Annotation.Type.COMPLETING), |
| ("good_partial_annotation.txt", None, Annotation.Type.PARTIAL), |
| ("good_test_annotation.txt", ErrorType.TEST_ANNOTATION, None), |
| ("missing_annotation.txt", ErrorType.MISSING_TAG_USED, None), |
| ("good_no_annotation.txt", ErrorType.NO_ANNOTATION, None), |
| ("bad_syntax_annotation1.txt", ErrorType.SYNTAX, None), |
| ("bad_syntax_annotation2.txt", ErrorType.SYNTAX, None), |
| ("bad_syntax_annotation3.txt", ErrorType.SYNTAX, None), |
| ("bad_syntax_annotation4.txt", ErrorType.SYNTAX, None), |
| # "fatal" means a Python exception gets raised. |
| ("fatal_annotation1.txt", "fatal", None), |
| ("fatal_annotation2.txt", "fatal", None), |
| ("fatal_annotation3.txt", "fatal", None), |
| ] |
| |
| for test_case in test_cases: |
| logger.debug("Testing: {}".format(test_case)) |
| |
| (file_name, error_type, annotation_type) = test_case |
| |
| if error_type == "fatal": |
| # Should raise a Python exception. |
| with self.assertRaises(Exception): |
| annotation, errors = self.deserialize(file_name) |
| elif error_type: |
| # Should raise an AuditorError of the specified type. |
| annotation, errors = self.deserialize(file_name) |
| self.assertEqual(1, len(errors)) |
| self.assertEqual(error_type, errors[0].type) |
| else: |
| # Should not raise an error. |
| annotation, errors = self.deserialize(file_name) |
| self.assertEqual([], errors) |
| self.assertEqual(annotation_type, annotation.type, file_name) |
| |
| # Check contents for one complete sample. |
| if file_name != "good_complete_annotation.txt": |
| continue |
| |
| self.assertEqual(annotation.unique_id, |
| "supervised_user_refresh_token_fetcher") |
| self.assertEqual( |
| annotation.file, |
| Path("chrome/browser/supervised_user/legacy/" |
| "supervised_user_refresh_token_fetcher.cc")) |
| self.assertEqual(annotation.line, 166) |
| self.assertEqual(annotation.proto.semantics.sender, "Supervised Users") |
| self.assertEqual(annotation.proto.policy.cookies_allowed, 1) |
| |
| def test_get_reserved_ids_coverage(self): |
| """Tests if RESERVED_IDS has all known ids.""" |
| self.assertEqual(["test", "test_partial", "missing", "undefined"], |
| RESERVED_IDS) |
| |
| def test_reserved_ids_usage_detection(self): |
| """Tests if use of reserved ids are detected.""" |
| for reserved_id in RESERVED_IDS: |
| annotation = Annotation() |
| annotation.type = Annotation.Type.COMPLETE |
| annotation.unique_id = reserved_id |
| errors = self.run_id_checker(annotation) |
| self.assertEqual(1, len(errors)) |
| self.assertEqual(ErrorType.RESERVED_ID_HASH_CODE, errors[0].type) |
| |
| annotation.type = Annotation.Type.PARTIAL |
| annotation.unique_id = "nonempty" |
| annotation.second_id = reserved_id |
| errors = self.run_id_checker(annotation) |
| self.assertEqual(1, len(errors)) |
| self.assertEqual(ErrorType.RESERVED_ID_HASH_CODE, errors[0].type) |
| |
| def test_repeated_ids_detection(self): |
| """Tests if use of repeated ids are detected.""" |
| id_checker = IdChecker([]) |
| |
| # Check if several different hash codes result in no error. |
| annotations = [ |
| self.create_annotation_sample(unique_id=i) for i in range(10) |
| ] |
| errors = id_checker.check_ids(annotations) |
| self.assertEqual([], errors) |
| |
| # Check if repeating the same hash codes results in errors. |
| annotations = [ |
| self.create_annotation_sample(unique_id=i // 2) for i in range(20) |
| ] |
| errors = id_checker.check_ids(annotations) |
| self.assertCountEqual([ErrorType.REPEATED_ID] * 10, |
| [e.type for e in errors]) |
| |
| def test_similar_unique_and_second_ids_detection(self): |
| """Tests if having the same unique id and second id is detected.""" |
| for t in Annotation.Type: |
| annotation = Annotation() |
| annotation.type = t |
| annotation.unique_id = "the_same" |
| annotation.second_id = "the_same" |
| errors = self.run_id_checker(annotation) |
| if annotation.needs_two_ids(): |
| self.assertEqual(1, len(errors), t) |
| else: |
| self.assertEqual(0, len(errors), t) |
| |
| def test_duplicate_ids_detection(self): |
| """Tests unique id and second id collision cases.""" |
| T = Annotation.Type |
| annotation_types = list(T) |
| for type1, type2 in itertools.product(*[list(T)] * 2): |
| if annotation_types.index(type2) < annotation_types.index(type1): |
| continue |
| |
| for id1, id2, id3, id4 in \ |
| itertools.product(*[range(1, 5) for i in range(4)]): |
| logger.debug("Testing ({}, {}, {}, {}, {}, {})".format( |
| type1, type2, id1, id2, id3, id4)) |
| |
| annotation1 = Annotation() |
| annotation1.type = type1 |
| annotation1.unique_id = str(id1) |
| annotation1.second_id = str(id2) |
| |
| annotation2 = Annotation() |
| annotation2.type = type2 |
| annotation2.unique_id = str(id3) |
| annotation2.second_id = str(id4) |
| |
| id_checker = IdChecker([]) |
| errors = id_checker.check_ids([annotation1, annotation2]) |
| |
| first_needs_two = annotation1.needs_two_ids() |
| second_needs_two = annotation2.needs_two_ids() |
| |
| unique_ids = set(id for a in [annotation1, annotation2] |
| for id in a.get_ids()) |
| |
| if first_needs_two and second_needs_two: |
| # If both need 2 ids, either the 4 ids should be different, or the |
| # second ids should be equal and both annotations should be of types |
| # partial/branched-completing. |
| if len(unique_ids) == 4: |
| self.assertFalse(errors) |
| elif len(unique_ids) == 3: |
| acceptable = (id2 == id4 |
| and type1 in [T.PARTIAL, T.BRANCHED_COMPLETING] |
| and type2 in [T.PARTIAL, T.BRANCHED_COMPLETING]) |
| self.assertEqual(not acceptable, bool(errors)) |
| else: |
| self.assertTrue(errors) |
| elif first_needs_two and not second_needs_two: |
| # If just the first one needs two ids, then either the 3 ids should be |
| # different or the first annotation would be partial and the second |
| # completing, with one common id. |
| if len(unique_ids) == 3: |
| self.assertFalse(errors) |
| elif len(unique_ids) == 2: |
| acceptable = (id2 == id3 and type1 == T.PARTIAL |
| and type2 == T.COMPLETING) |
| self.assertEqual(not acceptable, bool(errors)) |
| else: |
| self.assertTrue(errors) |
| elif not first_needs_two and second_needs_two: |
| # Can only be valid if all 3 are different. |
| self.assertEqual(len(unique_ids) != 3, bool(errors)) |
| else: |
| # If none requires 2 ids, it can only be valid if ids are different. |
| self.assertEqual(len(unique_ids) != 2, bool(errors)) |
| |
| def test_check_ids_format(self): |
| """Tests if IDs' format is correctly checked.""" |
| test_cases = [ |
| ("ID1", True), |
| ("id2", True), |
| ("Id_3", True), |
| ("ID?4", False), |
| ("ID:5", False), |
| ("ID>>6", False), |
| ] |
| |
| annotation = self.create_annotation_sample() |
| for test_case in test_cases: |
| # Set type to complete to require just unique ID. |
| annotation.type = Annotation.Type.COMPLETE |
| annotation.unique_id = test_case[0] |
| errors = self.run_id_checker(annotation) |
| self.assertEqual(not test_case[1], bool(errors), test_case[0]) |
| |
| # Set type to partial to require both ids. |
| annotation.type = Annotation.Type.PARTIAL |
| annotation.unique_id = "Something_Good" |
| annotation.second_id = test_case[0] |
| self.run_id_checker(annotation) |
| self.assertEqual(not test_case[1], bool(errors), test_case[0]) |
| |
| # Test all cases together. |
| annotations = [] |
| |
| false_sample_count = 0 |
| for test_case in test_cases: |
| annotation = self.create_annotation_sample() |
| annotation.type = Annotation.Type.COMPLETE |
| annotation.unique_id = test_case[0] |
| annotations.append(annotation) |
| if not test_case[1]: |
| false_sample_count += 1 |
| |
| id_checker = IdChecker([]) |
| errors = id_checker.check_ids(annotations) |
| self.assertEqual(false_sample_count, len(errors)) |
| |
| def test_check_complete_annotations(self): |
| """Tests if Auditor.check_annotation_contents() works as expected for |
| COMPLETE annotations. It also inherently checks |
| Auditor.is_annotation_complete(), Auditor.is_annotation_consistent(), and |
| Auditor.is_in_grouping_xml().""" |
| annotations = [] |
| expected_errors_count = 0 |
| |
| Destination = traffic_annotation.TrafficSemantics.Destination |
| CookiesAllowed = traffic_annotation.TrafficPolicy.CookiesAllowed |
| |
| test_no = 0 |
| while True: |
| annotation = self.create_annotation_sample() |
| annotation.unique_id = "foobar_policy_fetcher{}".format(test_no) |
| test_description = "" |
| expect_error = True |
| if test_no == 0: |
| test_description = "All fields OK." |
| expect_error = False |
| elif test_no == 1: |
| test_description = "Missing semantics::sender" |
| annotation.proto.semantics.sender = "" |
| elif test_no == 2: |
| test_description = "Missing semantics::description" |
| annotation.proto.semantics.description = "" |
| elif test_no == 3: |
| test_description = "Missing semantics::trigger" |
| annotation.proto.semantics.trigger = "" |
| elif test_no == 4: |
| test_description = "Missing semantics::data" |
| annotation.proto.semantics.data = "" |
| elif test_no == 5: |
| test_description = "Missing semantics::destination" |
| annotation.proto.semantics.destination = Destination.UNSPECIFIED |
| elif test_no == 6: |
| test_description = "Missing policy::cookies_allowed" |
| annotation.proto.policy.cookies_allowed = CookiesAllowed.UNSPECIFIED |
| elif test_no == 7: |
| test_description = \ |
| "policy::cookies_allowed = NO with existing policy::cookies_store." |
| annotation.proto.policy.cookies_allowed = CookiesAllowed.NO |
| annotation.proto.policy.cookies_store = "somewhere" |
| elif test_no == 8: |
| test_description = \ |
| "policy::cookies_allowed = NO and no policy::cookies_store." |
| annotation.proto.policy.cookies_allowed = CookiesAllowed.NO |
| annotation.proto.policy.cookies_store = "" |
| expect_error = False |
| elif test_no == 9: |
| test_description = \ |
| "policy::cookies_allowed = YES and policy::cookies_store exists." |
| annotation.proto.policy.cookies_allowed = CookiesAllowed.YES |
| annotation.proto.policy.cookies_store = "somewhere" |
| expect_error = False |
| elif test_no == 10: |
| test_description = \ |
| "policy::cookies_allowed = YES and no policy::cookies_store." |
| annotation.proto.policy.cookies_allowed = CookiesAllowed.YES |
| annotation.proto.policy.cookies_store = "" |
| elif test_no == 11: |
| test_description = "Missing policy::setting." |
| annotation.proto.policy.setting = "" |
| break |
| elif test_no == 12: |
| test_description = \ |
| "Missing policy::chrome_policy and " \ |
| "policy::policy_exception_justification." |
| annotation.proto.policy.chrome_policy = "" |
| annotation.proto.policy.policy_exception_justification = "" |
| elif test_no == 13: |
| test_description = \ |
| "Missing policy::chrome_policy and existing " \ |
| "policy::policy_exception_justification." |
| annotation.proto.policy.chrome_policy = "" |
| annotation.proto.policy.policy_exception_justification = "Because!" |
| expect_error = False |
| elif test_no == 14: |
| test_description = \ |
| "Existing policy::chrome_policy and no " \ |
| "policy::policy_exception_justification." |
| self.assertTrue(annotation.proto.policy.chrome_policy) |
| annotation.proto.policy.policy_exception_justification = "" |
| expect_error = False |
| elif test_no == 15: |
| test_description = \ |
| "Existing policy::chrome_policy and existing" \ |
| "policy::policy_exception_justification." |
| self.assertTrue(annotation.proto.policy.chrome_policy) |
| annotation.proto.policy.policy_exception_justification = "Because!" |
| else: |
| # Done checking individual test cases. |
| break |
| |
| logger.debug("Testing: {}".format(test_description)) |
| |
| self.auditor.extracted_annotations = [annotation] |
| errors = self.auditor.check_annotation_contents() |
| |
| if expect_error: |
| self.assertEqual(1, len(errors), |
| "test_no={}, errors={}".format(test_no, errors)) |
| else: |
| self.assertEqual([], errors, |
| "test_no={}, errors={}".format(test_no, errors)) |
| |
| annotations.append(annotation) |
| |
| if expect_error: |
| expected_errors_count += 1 |
| |
| test_no += 1 |
| |
| # Check all. |
| self.auditor.extracted_annotations = annotations |
| errors = self.auditor.check_annotation_contents() |
| self.assertEqual(expected_errors_count, len(errors)) |
| |
| def test_is_completable_with(self): |
| """Tests is Annotation.is_completable_with() works as expected.""" |
| T = Annotation.Type |
| for type1, type2 in itertools.product(*[list(T)] * 2): |
| for ids in range(256): |
| annotation1 = self.create_annotation_sample(type1, ids % 4, |
| (ids >> 2) % 4) |
| annotation2 = self.create_annotation_sample(type2, (ids >> 4) % 4, |
| (ids >> 6)) |
| expectation = False |
| if annotation1.type == T.PARTIAL and annotation1.second_id: |
| expectation = expectation or \ |
| (annotation2.type == T.COMPLETING and |
| annotation1.second_id_hash_code == annotation2.unique_id_hash_code) |
| expectation = expectation or \ |
| (annotation2.type == T.BRANCHED_COMPLETING and |
| annotation1.second_id_hash_code == annotation2.second_id_hash_code) |
| self.assertEqual(annotation1.is_completable_with(annotation2), |
| expectation, "{} <=> {}".format(type1, type2)) |
| |
| def test_create_complete_annotation(self): |
| """Tests is Annotation.create_complete_annotation() works as |
| expected.""" |
| instance = self.create_annotation_sample() |
| other = self.create_annotation_sample() |
| |
| instance.proto.semantics.Clear() |
| instance.proto.policy.Clear() |
| |
| instance.type = Annotation.Type.PARTIAL |
| other.type = Annotation.Type.COMPLETING |
| |
| # Partial and Completing. |
| instance.second_id = "SomeID" |
| other.unique_id = "SomeID" |
| combination, errors = instance.create_complete_annotation(other) |
| self.assertEqual([], errors) |
| self.assertEqual(combination.unique_id_hash_code, |
| instance.unique_id_hash_code) |
| |
| # Partial and Branched-completing. |
| other.type = Annotation.Type.BRANCHED_COMPLETING |
| other.second_id = "SomeID" |
| instance.second_id = "SomeID" |
| combination, errors = instance.create_complete_annotation(other) |
| self.assertEqual([], errors) |
| self.assertEqual(combination.unique_id_hash_code, other.unique_id_hash_code) |
| |
| # Inconsistent field. |
| Destination = traffic_annotation.TrafficSemantics.Destination |
| other.proto.MergeFrom(instance.proto) |
| other.type = Annotation.Type.BRANCHED_COMPLETING |
| other.second_id = "SomeID" |
| instance.second_id = "SomeID" |
| instance.proto.semantics.destination = Destination.WEBSITE |
| other.proto.semantics.destination = Destination.LOCAL |
| annotation, errors = instance.create_complete_annotation(other) |
| self.assertEqual(1, len(errors)) |
| |
| def test_load_from_archive(self): |
| """Tests that Annotation.load_from_archive() works as expected.""" |
| archived = ArchivedAnnotation(type=Annotation.Type.PARTIAL, |
| id="foobar", |
| second_id="baz", |
| content_hash_code=32, |
| os_list=["linux", "windows"], |
| added_in_milestone=62, |
| semantics_fields=[2, 3], |
| policy_fields=[-1, 3, 4], |
| file_path=Path("foobar.cc")) |
| annotation = Annotation.load_from_archive(archived) |
| self.assertTrue(annotation.is_loaded_from_archive) |
| self.assertEqual(annotation.type, archived.type) |
| self.assertEqual(annotation.unique_id, archived.id) |
| self.assertEqual(annotation.unique_id_hash_code, archived.hash_code) |
| self.assertEqual(annotation.second_id, archived.second_id) |
| self.assertEqual(annotation.archived_content_hash_code, 32) |
| self.assertEqual(annotation.archived_added_in_milestone, 62) |
| self.assertEqual(annotation.get_semantics_field_numbers(), |
| archived.semantics_fields) |
| self.assertEqual(annotation.get_policy_field_numbers(), |
| archived.policy_fields) |
| self.assertEqual(annotation.file, archived.file_path) |
| |
| def test_annotations_xml(self): |
| """Tests is annotations.xml has proper content.""" |
| # annotations.xml should parse without errors. |
| exporter = Exporter(get_current_platform()) |
| exporter.load_annotations_xml() |
| errors = exporter.check_archived_annotations() |
| self.assertEqual([], errors) |
| |
| # The content of annotations.xml shouldn't change when writing it. |
| old_xml = Exporter.ANNOTATIONS_XML_PATH.read_text(encoding="utf-8") |
| new_xml = exporter._generate_serialized_xml() |
| self.assertEqual(old_xml, new_xml) |
| |
| def test_grouping_xml(self): |
| """Tests is grouping.xml has proper content.""" |
| # grouping.xml should parse without errors. |
| exporter = Exporter(get_current_platform()) |
| exporter.load_grouping_xml(Exporter.GROUPING_XML_PATH) |
| |
| # The content of grouping.xml shouldn't change when writing it. |
| old_xml = Exporter.GROUPING_XML_PATH.read_text(encoding="utf-8") |
| new_xml = exporter._generate_serialized_grouping_xml() |
| self.assertEqual(old_xml, new_xml) |
| |
| def test_grouping_required_fields_errors(self) -> None: |
| """Tests is grouping.xml has no content.""" |
| # grouping.xml should parse with errors. |
| grouping_erro_xml_path = \ |
| TEST_DATA_DIR / "test_required_field_error_grouping.xml" |
| exporter = Exporter(get_current_platform()) |
| self.assertRaises(ValueError, |
| lambda: exporter.load_grouping_xml(grouping_erro_xml_path)) |
| |
| def test_annotations_xml_differences(self): |
| """Tests if annotations.xml changes are correctly reported.""" |
| exporter = Exporter(get_current_platform()) |
| |
| xml1 = (TEST_DATA_DIR / |
| "annotations_sample1.xml").read_text(encoding="utf-8") |
| xml2 = (TEST_DATA_DIR / |
| "annotations_sample2.xml").read_text(encoding="utf-8") |
| xml3 = (TEST_DATA_DIR / |
| "annotations_sample3.xml").read_text(encoding="utf-8") |
| |
| diff12 = exporter._get_xml_differences(xml1, xml2) |
| diff13 = exporter._get_xml_differences(xml1, xml3) |
| diff23 = exporter._get_xml_differences(xml2, xml3) |
| |
| expected_diff12 = (TEST_DATA_DIR / |
| "annotations_diff12.txt").read_text(encoding="utf-8") |
| expected_diff13 = (TEST_DATA_DIR / |
| "annotations_diff13.txt").read_text(encoding="utf-8") |
| expected_diff23 = (TEST_DATA_DIR / |
| "annotations_diff23.txt").read_text(encoding="utf-8") |
| |
| self.assertEqual(expected_diff12, diff12) |
| self.assertEqual(expected_diff13, diff13) |
| self.assertEqual(expected_diff23, diff23) |
| |
| def test_annotation_grouping(self): |
| """Tests if an annotation is in test_grouping.xml or not.""" |
| grouping_xml_path = TEST_DATA_DIR / "test_grouping.xml" |
| errors = self.auditor.run_all_checks([], True, grouping_xml_path) |
| self.assertTrue(errors) |
| grouping_xml_ids = self.auditor._get_grouping_xml_ids(grouping_xml_path) |
| self.assertCountEqual([ |
| "foobar_policy_fetcher", "foobar_info_fetcher", |
| "fizzbuzz_handle_front_end_messages", "fizzbuzz_hard_coded_data_source", |
| "fizzbuzz_http_handler", "widget_grabber" |
| ], grouping_xml_ids) |
| |
| def test_setup(self) -> None: |
| """|self.sample_annotations| should include all those inside |
| test_data/test_sample_annotations.cc""" |
| expected = [ |
| "ok_annotation", "syntax_error_annotation", |
| "incomplete_error_annotation" |
| ] |
| self.assertCountEqual(expected, self.sample_annotations.keys()) |
| |
| def test_ensure_errors(self) -> None: |
| """In the |test_sample_annotations.cc| there are some broken annotations. |
| |
| This test ensures that AuditorUI catches these errors by |
| running from start to finish via |.main()| |
| """ |
| # Suppress |self.auditor_ui.main()| prints to stdout. |
| with open(os.devnull, "w") as devnull: |
| sys.stdout = devnull |
| self.assertEqual(1, self.auditor_ui.main()) # 1 indicates errors caught. |
| sys.stdout = sys.__stdout__ |
| |
| def test_result_ok(self) -> None: |
| errors = self.auditor.parse_extractor_output( |
| [self.sample_annotations["ok_annotation"]]) |
| |
| # Assert that correct annotation has been extracted and is OK (no errors). |
| self.assertTrue(self.auditor.extracted_annotations) |
| self.assertFalse(errors) |
| |
| def test_syntax_error(self) -> None: |
| errors = self.auditor.parse_extractor_output( |
| [self.sample_annotations["syntax_error_annotation"]]) |
| |
| self.assertTrue(errors) |
| result = errors[0] |
| self.assertEqual(ErrorType.SYNTAX, result.type) |
| self.assertTrue("sender: \"Cloud Policy\"': Expected \"{\"" in str(result)) |
| |
| def test_incomplete_error(self) -> None: |
| self.auditor.parse_extractor_output( |
| [self.sample_annotations["incomplete_error_annotation"]]) |
| |
| self.assertTrue(self.auditor.extracted_annotations) |
| errors = self.auditor.run_all_checks([], True, Exporter.GROUPING_XML_PATH) |
| self.assertTrue(errors) |
| result = errors[0] |
| self.assertEqual(ErrorType.INCOMPLETE_ANNOTATION, result.type) |
| |
| expected_missing_fields = [ |
| "sender", "chrome_policy", "cookies_store", |
| "policy_exception_justification" |
| ] |
| missing_fields = str(result).split("missing fields:", |
| 1)[1].lstrip().split(", ") |
| self.assertCountEqual(expected_missing_fields, missing_fields) |
| |
| def test_get_current_platform(self) -> None: |
| host_platform = platform.system().lower() |
| |
| if host_platform == "windows": |
| self.assertEqual("windows", |
| get_current_platform(TEST_DATA_DIR / "out" / "Debug")) |
| elif host_platform == "linux": |
| self.assertEqual("linux", |
| get_current_platform(TEST_DATA_DIR / "out" / "Debug")) |
| self.assertEqual("android", |
| get_current_platform(TEST_DATA_DIR / "out" / "Android")) |
| else: |
| raise ValueError("Unrecognized host platform {}".format(host_platform)) |
| |
| def test_write_annotations_tsv_file(self) -> None: |
| annotation, errors = self.deserialize("good_complete_annotation.txt") |
| self.assertEqual([], errors) |
| annotations = [annotation] |
| |
| self.maxDiff = None |
| tsv_path = Path(tempfile.mktemp()) |
| write_annotations_tsv_file(tsv_path, annotations, []) |
| self.assertTrue(tsv_path.exists()) |
| tsv_contents = tsv_path.read_text(encoding="utf-8") |
| expected_contents = """Unique ID\tLast Update\tSender\tDescription\tTrigger\tData\tDestination\tCookies Allowed\tCookies Store\tSetting\tChrome Policy\tComments\tSource File |
| supervised_user_refresh_token_fetcher\t\tSupervised Users\tFetches an OAuth2 refresh token scoped down to the Supervised User Sync scope and tied to the given Supervised User ID, identifying the Supervised User Profile to be created.\tCalled when creating a new Supervised User profile in Chromium to fetch OAuth credentials for using Sync with the new profile.\t"The request is authenticated with an OAuth2 access token identifying the Google account and contains the following information: |
| * The Supervised User ID, a randomly generated 64-bit identifier for the profile. |
| * The device name, to identify the refresh token in account management."\tGoogle\tNo\t\tUsers can disable this feature by toggling 'Let anyone add a person to Chrome' in Chromium settings, under People.\tSupervisedUserCreationEnabled: false\t\thttps://cs.chromium.org/chromium/src/?l=0 |
| """ |
| self.assertEqual(expected_contents, tsv_contents) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |
