Ensure blinkpy mock text files encode to and decode from bytes

This change ensures all file contents in MockFileSystem are stored
internally as bytes. Previously, open_text_file_for_{reading,writing}
assumed file contents were Unicode strings, while {read,write}_text_file
assumed contents were bytes. Calling a write method from one pair with
the read method from the other caused type errors.

This change also removes direct accesses in the unit tests to
MockFileSystem.files. The unit tests sometimes incorrectly wrote Unicode
strings as well.

Bug: 1307533
Test: vpython run_blinkpy_tests.py
Test: vpython3 run_blinkpy_tests.py
Change-Id: Ic15bc378817a7c928e473361108eb62525ea31c5
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/3538824
Reviewed-by: Weizhong Xia <weizhong@google.com>
Commit-Queue: Jonathan Lee <jonathanjlee@google.com>
Cr-Commit-Position: refs/heads/main@{#983944}

third_party/blink/tools/blinkpy/common/system/filesystem_mock.py

@@ -34,7 +34,17 @@
 
 from blinkpy.common.system.filesystem import _remove_contents, _sanitize_filename
 
-from six import StringIO
+from six import ensure_binary, StringIO
+
+_TEXT_ENCODING = 'utf-8'
+
+
+def _ensure_binary_contents(file_contents):
+    # Iterate over a copy while the underlying mapping is mutated.
+    for path, contents in list(file_contents.items()):
+        if contents is not None:
+            contents = ensure_binary(contents, _TEXT_ENCODING)
+        file_contents[path] = contents
 
 
 class MockFileSystem(object):
@@ -59,6 +69,7 @@ def __init__(self, files=None, dirs=None, cwd='/'):
         self.cwd = cwd
         self.dirs = set(dirs or [])
         self.dirs.add(cwd)
+        _ensure_binary_contents(self.files)
         for file_path in self.files:
             directory = self.dirname(file_path)
             while directory not in self.dirs:
@@ -352,7 +363,8 @@ def open_text_tempfile(self, suffix=''):
     def open_text_file_for_reading(self, path):
         if self.files[path] is None:
             self._raise_not_found(path)
-        return ReadableTextFileObject(self, path, self.files[path])
+        contents = self.files[path].decode(_TEXT_ENCODING)
+        return ReadableTextFileObject(self, path, contents)
 
     def open_text_file_for_writing(self, path):
         return WritableTextFileObject(self, path)
@@ -361,10 +373,10 @@ def open_text_file_for_appending(self, path):
         return WritableTextFileObject(self, path, append=True)
 
     def read_text_file(self, path):
-        return self.read_binary_file(path).decode('utf-8')
+        return self.read_binary_file(path).decode(_TEXT_ENCODING)
 
     def write_text_file(self, path, contents):
-        return self.write_binary_file(path, contents.encode('utf-8'))
+        return self.write_binary_file(path, contents.encode(_TEXT_ENCODING))
 
     def sha1(self, path):
         contents = self.read_binary_file(path)
@@ -490,14 +502,16 @@ def __init__(self, fs, path, append=False):
         super(WritableTextFileObject, self).__init__(fs, path, append)
 
     def write(self, string):
-        WritableBinaryFileObject.write(self, string)
+        contents = string.encode(_TEXT_ENCODING)
+        super(WritableTextFileObject, self).write(contents)
 
     def writelines(self, lines):
-        self.fs.files[self.path] = "".join(lines)
-        self.fs.written_files[self.path] = self.fs.files[self.path]
+        contents = b''.join(line.encode(_TEXT_ENCODING) for line in lines)
+        self.fs.files[self.path] = contents
+        self.fs.written_files[self.path] = contents
 
     def setup_path(self, path):
-        self.fs.files[path] = ''
+        super(WritableTextFileObject, self).setup_path(path)
 
 
 class ReadableBinaryFileObject(object):
@@ -510,7 +524,7 @@ def __init__(self, fs, path, data):
         try:
             # Maintain a text version if possible to
             # support readline.
-            data_str = data.decode('utf8', 'replace')
+            data_str = data.decode(_TEXT_ENCODING, 'replace')
             self.text = StringIO(data_str)
         except:
             pass
@@ -532,10 +546,10 @@ def read(self, num_bytes=None):
         return self.data[start:self.offset]
 
     def readline(self, length=None):
-        return self.text.readline(length).encode('utf8', 'replace')
+        return self.text.readline(length).encode(_TEXT_ENCODING, 'replace')
 
     def readlines(self):
-        return self.text.readlines().encode('utf8', 'replace')
+        return self.text.readlines().encode(_TEXT_ENCODING, 'replace')
 
     def seek(self, offset, whence=os.SEEK_SET):
         if whence == os.SEEK_SET:
@@ -586,6 +600,7 @@ def __init__(self, test_case, mock_filesystem, expected_files):
             self.test_case = test_case
             self.mock_filesystem = mock_filesystem
             self.expected_files = expected_files
+            _ensure_binary_contents(self.expected_files)
 
         def __enter__(self):
             # Make sure that the expected_files aren't already in the mock

third_party/blink/tools/blinkpy/w3c/directory_owners_extractor_unittest.py

@@ -161,17 +161,16 @@ def test_find_owners_file_out_of_external(self):
         self.assertIsNone(self.extractor.find_owners_file('third_party'))
 
     def test_extract_owners(self):
-        self.host.filesystem.files = {
-            ABS_WPT_BASE + '/foo/OWNERS':
-            b'#This is a comment\n'
-            b'*\n'
-            b'foo@chromium.org\n'
-            b'bar@chromium.org\n'
-            b'foobar\n'
-            b'#foobar@chromium.org\n'
-            b'# TEAM: some-team@chromium.org\n'
-            b'# COMPONENT: Blink>Layout\n'
-        }
+        fs = self.host.filesystem
+        fs.write_text_file(fs.join(ABS_WPT_BASE, 'foo', 'OWNERS'),
+                           ('#This is a comment\n'
+                            '*\n'
+                            'foo@chromium.org\n'
+                            'bar@chromium.org\n'
+                            'foobar\n'
+                            '#foobar@chromium.org\n'
+                            '# TEAM: some-team@chromium.org\n'
+                            '# COMPONENT: Blink>Layout\n'))
         self.assertEqual(
             self.extractor.extract_owners(ABS_WPT_BASE + '/foo/OWNERS'),
             ['foo@chromium.org', 'bar@chromium.org'])

third_party/blink/tools/blinkpy/w3c/wpt_expectations_updater_unittest.py

@@ -1399,24 +1399,27 @@ def test_same_platform_one_without_results(self):
     def test_cleanup_all_deleted_tests_in_expectations_files(self):
         host = MockHost()
         port = host.port_factory.get()
-        host.filesystem.files[MOCK_WEB_TESTS + 'TestExpectations'] = (
-            b'# results: [ Failure ]\n'
-            b'external/wpt/some/test/a.html?hello%20world [ Failure ]\n'
-            b'some/test/b.html [ Failure ]\n'
-            b'# This line should be deleted\n'
-            b'some/test/c.html [ Failure ]\n'
-            b'# line below should exist in new file\n'
-            b'some/test/d.html [ Failure ]\n')
-        host.filesystem.files[MOCK_WEB_TESTS + 'VirtualTestSuites'] = b'[]'
-        host.filesystem.files[MOCK_WEB_TESTS + 'new/a.html'] = b''
-        host.filesystem.files[MOCK_WEB_TESTS + 'new/b.html'] = b''
-        host.filesystem.files[
-            host.filesystem.join(
-                port.web_tests_dir(), 'some', 'test', 'd.html')] = b''
+        fs = host.filesystem
+        expectations_path = fs.join(MOCK_WEB_TESTS, 'TestExpectations')
+
+        fs.write_text_file(
+            expectations_path,
+            ('# results: [ Failure ]\n'
+             'external/wpt/some/test/a.html?hello%20world [ Failure ]\n'
+             'some/test/b.html [ Failure ]\n'
+             '# This line should be deleted\n'
+             'some/test/c.html [ Failure ]\n'
+             '# line below should exist in new file\n'
+             'some/test/d.html [ Failure ]\n'))
+        fs.write_text_file(fs.join(MOCK_WEB_TESTS, 'VirtualTestSuites'), '[]')
+        fs.write_text_file(fs.join(MOCK_WEB_TESTS, 'new', 'a.html'), '')
+        fs.write_text_file(fs.join(MOCK_WEB_TESTS, 'new', 'b.html'), '')
+        fs.write_text_file(
+            fs.join(port.web_tests_dir(), 'some', 'test', 'd.html'), '')
         # TODO(rmhasan): Remove creation of Android files within
         # tests.
         for path in PRODUCTS_TO_EXPECTATION_FILE_PATHS.values():
-            host.filesystem.write_text_file(path, '')
+            fs.write_text_file(path, '')
 
         updater = WPTExpectationsUpdater(host)
 
@@ -1428,63 +1431,70 @@ def _git_command_return_val(cmd):
         updater.git.run = _git_command_return_val
         updater._relative_to_web_test_dir = lambda test_path: test_path
         updater.cleanup_test_expectations_files()
-        self.assertMultiLineEqual(
-            host.filesystem.read_text_file(MOCK_WEB_TESTS +
-                                           'TestExpectations'),
-            ('# results: [ Failure ]\n'
-             '# line below should exist in new file\n'
-             'some/test/d.html [ Failure ]\n'))
+        self.assertMultiLineEqual(fs.read_text_file(expectations_path),
+                                  ('# results: [ Failure ]\n'
+                                   '# line below should exist in new file\n'
+                                   'some/test/d.html [ Failure ]\n'))
 
     def test_skip_slow_timeout_tests(self):
         host = MockHost()
-        host.filesystem.write_text_file(
-            MOCK_WEB_TESTS + 'SlowTests',
-            ('# results: [ Slow ]\n'
-             'foo/slow_timeout.html [ Slow ]\n'
-             'bar/slow.html [ Slow ]\n'))
+        fs = host.filesystem
+        expectations_path = fs.join(MOCK_WEB_TESTS, 'TestExpectations')
         data = ('# results: [ Pass Failure Crash Timeout Skip ]\n'
                 'foo/failure.html [ Failure ]\n'
                 'foo/slow_timeout.html [ Timeout ]\n'
                 'bar/text.html [ Pass ]\n')
-        host.filesystem.write_text_file(
-            MOCK_WEB_TESTS + 'TestExpectations', data)
+
+        fs.write_text_file(fs.join(MOCK_WEB_TESTS, 'SlowTests'),
+                           ('# results: [ Slow ]\n'
+                            'foo/slow_timeout.html [ Slow ]\n'
+                            'bar/slow.html [ Slow ]\n'))
+        fs.write_text_file(expectations_path, data)
         for path in PRODUCTS_TO_EXPECTATION_FILE_PATHS.values():
-            host.filesystem.write_text_file(path, '')
+            fs.write_text_file(path, '')
+
         newdata = data.replace('foo/slow_timeout.html [ Timeout ]',
                                'foo/slow_timeout.html [ Skip Timeout ]')
         updater = WPTExpectationsUpdater(host)
         rv = updater.skip_slow_timeout_tests(host.port_factory.get())
         self.assertTrue(rv)
-        self.assertEqual(
-            newdata,
-            host.filesystem.read_text_file(MOCK_WEB_TESTS + 'TestExpectations'))
+        self.assertEqual(newdata, fs.read_text_file(expectations_path))
 
     def test_cleanup_all_test_expectations_files(self):
         host = MockHost()
-        host.filesystem.files[MOCK_WEB_TESTS + 'TestExpectations'] = (
-            b'# results: [ Failure ]\n'
-            b'some/test/a.html [ Failure ]\n'
-            b'some/test/b.html [ Failure ]\n'
-            b'ignore/globs/* [ Failure ]\n'
-            b'some/test/c\*.html [ Failure ]\n'
-            # default test case, line below should exist in new file
-            b'some/test/d.html [ Failure ]\n')
-        host.filesystem.files[MOCK_WEB_TESTS + 'WebDriverExpectations'] = (
-            b'# results: [ Failure ]\n'
-            b'external/wpt/webdriver/some/test/a\*.html>>foo\* [ Failure ]\n'
-            b'external/wpt/webdriver/some/test/a\*.html>>bar [ Failure ]\n'
-            b'external/wpt/webdriver/some/test/b.html>>foo [ Failure ]\n'
-            b'external/wpt/webdriver/some/test/c.html>>a [ Failure ]\n'
-            # default test case, line below should exist in new file
-            b'external/wpt/webdriver/some/test/d.html>>foo [ Failure ]\n')
-        host.filesystem.files[MOCK_WEB_TESTS + 'VirtualTestSuites'] = b'[]'
-        host.filesystem.files[MOCK_WEB_TESTS + 'new/a.html'] = b''
-        host.filesystem.files[MOCK_WEB_TESTS + 'new/b.html'] = b''
+        fs = host.filesystem
+        test_expect_path = fs.join(MOCK_WEB_TESTS, 'TestExpectations')
+        webdriver_expect_path = fs.join(MOCK_WEB_TESTS,
+                                        'WebDriverExpectations')
+
+        fs.write_text_file(
+            test_expect_path,
+            (
+                '# results: [ Failure ]\n'
+                'some/test/a.html [ Failure ]\n'
+                'some/test/b.html [ Failure ]\n'
+                'ignore/globs/* [ Failure ]\n'
+                'some/test/c\*.html [ Failure ]\n'
+                # default test case, line below should exist in new file
+                'some/test/d.html [ Failure ]\n'))
+        fs.write_text_file(
+            webdriver_expect_path,
+            (
+                '# results: [ Failure ]\n'
+                'external/wpt/webdriver/some/test/a\*.html>>foo\* [ Failure ]\n'
+                'external/wpt/webdriver/some/test/a\*.html>>bar [ Failure ]\n'
+                'external/wpt/webdriver/some/test/b.html>>foo [ Failure ]\n'
+                'external/wpt/webdriver/some/test/c.html>>a [ Failure ]\n'
+                # default test case, line below should exist in new file
+                'external/wpt/webdriver/some/test/d.html>>foo [ Failure ]\n'))
+        fs.write_text_file(fs.join(MOCK_WEB_TESTS, 'VirtualTestSuites'), '[]')
+        fs.write_text_file(fs.join(MOCK_WEB_TESTS, 'new', 'a.html'), '')
+        fs.write_text_file(fs.join(MOCK_WEB_TESTS, 'new', 'b.html'), '')
 
         # TODO(rmhasan): Remove creation of Android files within
         # tests.
         for path in PRODUCTS_TO_EXPECTATION_FILE_PATHS.values():
-            host.filesystem.write_text_file(path, '')
+            fs.write_text_file(path, '')
 
         updater = WPTExpectationsUpdater(
             host, ['--clean-up-test-expectations-only',
@@ -1501,17 +1511,14 @@ def test_cleanup_all_test_expectations_files(self):
         updater._list_deleted_files = lambda: deleted_files
         updater._list_renamed_files = lambda: renamed_file_pairs
         updater.cleanup_test_expectations_files()
+        self.assertMultiLineEqual(fs.read_text_file(test_expect_path),
+                                  ('# results: [ Failure ]\n'
+                                   'new/a.html [ Failure ]\n'
+                                   'ignore/globs/* [ Failure ]\n'
+                                   'new/c\*.html [ Failure ]\n'
+                                   'some/test/d.html [ Failure ]\n'))
         self.assertMultiLineEqual(
-            host.filesystem.read_text_file(MOCK_WEB_TESTS +
-                                           'TestExpectations'),
-            ('# results: [ Failure ]\n'
-             'new/a.html [ Failure ]\n'
-             'ignore/globs/* [ Failure ]\n'
-             'new/c\*.html [ Failure ]\n'
-             'some/test/d.html [ Failure ]\n'))
-        self.assertMultiLineEqual(
-            host.filesystem.read_text_file(MOCK_WEB_TESTS +
-                                           'WebDriverExpectations'),
+            fs.read_text_file(webdriver_expect_path),
             ('# results: [ Failure ]\n'
              'old/a\*.html>>foo\* [ Failure ]\n'
              'old/a\*.html>>bar [ Failure ]\n'

third_party/blink/tools/blinkpy/web_tests/lint_test_expectations_unittest.py

@@ -432,16 +432,17 @@ def test_check_virtual_test_suites_readme(self):
         self.port.virtual_test_suites = lambda: [
             VirtualTestSuite(prefix='foo', bases=['test'], args=['--foo']),
             VirtualTestSuite(prefix='bar', bases=['test'], args=['--bar']), ]
-        self.host.filesystem.maybe_make_directory(WEB_TEST_DIR + '/test')
+        fs = self.host.filesystem
+        fs.maybe_make_directory(fs.join(WEB_TEST_DIR, 'test'))
 
         res = lint_test_expectations.check_virtual_test_suites(
             self.host, self.options)
         self.assertEqual(len(res), 2)
 
-        self.host.filesystem.files[WEB_TEST_DIR +
-                                   '/virtual/foo/README.md'] = ''
-        self.host.filesystem.files[WEB_TEST_DIR +
-                                   '/virtual/bar/test/README.txt'] = ''
+        fs.write_text_file(
+            fs.join(WEB_TEST_DIR, 'virtual', 'foo', 'README.md'), '')
+        fs.write_text_file(
+            fs.join(WEB_TEST_DIR, 'virtual', 'bar', 'test', 'README.txt'), '')
         res = lint_test_expectations.check_virtual_test_suites(
             self.host, self.options)
         self.assertFalse(res)
@@ -473,10 +474,11 @@ def test_check_virtual_test_suites_non_existent_base(self):
             VirtualTestSuite(prefix='foo', bases=['base1', 'base2', 'base3.html'], args=['-foo']),
         ]
 
-        self.host.filesystem.maybe_make_directory(WEB_TEST_DIR + '/base1')
-        self.host.filesystem.files[WEB_TEST_DIR + '/base3.html'] = ''
-        self.host.filesystem.files[WEB_TEST_DIR +
-                                   '/virtual/foo/README.md'] = ''
+        fs = self.host.filesystem
+        fs.maybe_make_directory(fs.join(WEB_TEST_DIR, 'base1'))
+        fs.write_text_file(fs.join(WEB_TEST_DIR, 'base3.html'), '')
+        fs.write_text_file(
+            fs.join(WEB_TEST_DIR, 'virtual', 'foo', 'README.md'), '')
         res = lint_test_expectations.check_virtual_test_suites(
             self.host, self.options)
         self.assertEqual(len(res), 1)

third_party/blink/tools/blinkpy/web_tests/merge_results_unittest.py

@@ -1460,26 +1460,27 @@ def test(self):
 
         for fname, expected_contents in self.web_test_output_filesystem.items(
         ):
-            self.assertIn(fname, fs.files)
+            self.assertTrue(fs.isfile(fname))
             if fname.endswith(".json"):
-                actual_json_str = fs.files[fname]
+                actual_json_str = fs.read_text_file(fname)
                 expected_json_str = expected_contents
                 if "failing_results" in fname:
                     self.assertTrue(
                         MergeFilesJSONPTests.check_before_after(
-                            fs.files[fname], b'ADD_RESULTS(', b");"))
+                            actual_json_str, 'ADD_RESULTS(', ');'))
                     self.assertTrue(
                         MergeFilesJSONPTests.check_before_after(
-                            expected_contents, 'ADD_RESULTS(', ");"))
+                            expected_contents, 'ADD_RESULTS(', ');'))
                     actual_json_str = MergeFilesJSONPTests.remove_before_after(
-                        fs.files[fname], b'ADD_RESULTS(', b");")
+                        actual_json_str, 'ADD_RESULTS(', ');')
                     expected_json_str = MergeFilesJSONPTests.remove_before_after(
-                        expected_contents, b'ADD_RESULTS(', b");")
+                        expected_contents, 'ADD_RESULTS(', ');')
 
                 self.assertEqual(json.loads(actual_json_str),
                                  json.loads(expected_json_str))
             else:
-                self.assertMultiLineEqual(expected_contents, fs.files[fname])
+                self.assertMultiLineEqual(expected_contents,
+                                          fs.read_text_file(fname))
 
 
 class MarkMissingShardsTest(unittest.TestCase):
@@ -1560,9 +1561,9 @@ class MarkMissingShardsTest(unittest.TestCase):
     web_test_filesystem = {
         '/out/output.json': output_output_json,
         '/swarm/summary.json': summary_json,
-        '/0/output.json': {
+        '/0/output.json': json.dumps({
             'successes': ['fizz', 'baz'],
-        },
+        }),
     }
 
     final_output_json = """\
@@ -1637,6 +1638,6 @@ def test_mark_missing_shards(self):
             ['/0'],  #only dir paths
             '/out/output.json',
             fs)
-        final_merged_output_json = fs.files['/out/output.json']
+        final_merged_output_json = fs.read_text_file('/out/output.json')
         self.assertEqual(json.loads(final_merged_output_json),
                          json.loads(self.final_output_json))