| #!/usr/bin/env vpython3 | |
| # Copyright 2026 The Chromium Authors | |
| # Use of this source code is governed by a BSD-style license that can be | |
| # found in the LICENSE file. | |
| """Unit tests for the ensure_node_modules.py gclient hook.""" | |
| import contextlib | |
| import hashlib | |
| import io | |
| import os | |
| import tarfile | |
| import tempfile | |
| import time | |
| import unittest | |
| import ensure_node_modules | |
| DIGEST = 'a' * 64 | |
| STAMPS = ( | |
| f'.devtools-frontend_{DIGEST}_hash', | |
| f'.devtools-frontend_{DIGEST}_is_first_class_gcs', | |
| f'.devtools-frontend_{DIGEST}_content_names', | |
| ) | |
| def write(root, relpath, content=''): | |
| full = os.path.join(root, relpath) | |
| os.makedirs(os.path.dirname(full), exist_ok=True) | |
| with open(full, 'w', encoding='utf-8') as f: | |
| f.write(content) | |
| return full | |
| class EnsureNodeModulesTest(unittest.TestCase): | |
| def setUp(self): | |
| tmpdir = tempfile.TemporaryDirectory() | |
| self.addCleanup(tmpdir.cleanup) | |
| self.node_modules = os.path.join(tmpdir.name, 'node_modules') | |
| os.makedirs(self.node_modules) | |
| def write_archive(self, entries): | |
| """Writes node_modules.tar.gz with the given relative files.""" | |
| with tempfile.TemporaryDirectory() as src: | |
| for relpath in entries: | |
| write(src, relpath, relpath) | |
| archive = os.path.join(self.node_modules, | |
| ensure_node_modules.ARCHIVE_NAME) | |
| with tarfile.open(archive, 'w:gz') as tar: | |
| for relpath in sorted(entries): | |
| info = tar.gettarinfo(os.path.join(src, relpath), | |
| arcname=relpath) | |
| info.mtime = 0 | |
| with open(os.path.join(src, relpath), 'rb') as f: | |
| tar.addfile(info, f) | |
| with open(archive, 'rb') as f: | |
| return hashlib.sha256(f.read()).hexdigest() | |
| def write_stamps(self, digest): | |
| for name in STAMPS: | |
| content = digest if name.endswith('_hash') else '1' | |
| write(self.node_modules, name, content + '\n') | |
| def run_hook(self): | |
| stdout, stderr = io.StringIO(), io.StringIO() | |
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr( | |
| stderr): | |
| code = ensure_node_modules.main(['hook', self.node_modules]) | |
| return code, stdout.getvalue(), stderr.getvalue() | |
| def exists(self, relpath): | |
| return os.path.lexists(os.path.join(self.node_modules, relpath)) | |
| def test_does_nothing_without_stamps(self): | |
| code, out, err = self.run_hook() | |
| self.assertEqual(code, 0) | |
| self.assertEqual(out + err, '') | |
| self.assertEqual(os.listdir(self.node_modules), []) | |
| def test_does_nothing_for_missing_directory(self): | |
| os.rmdir(self.node_modules) | |
| code, out, err = self.run_hook() | |
| self.assertEqual(code, 0) | |
| self.assertEqual(out + err, '') | |
| self.assertFalse(os.path.exists(self.node_modules)) | |
| def test_does_nothing_when_tree_is_present(self): | |
| self.write_stamps(DIGEST) | |
| write(self.node_modules, ensure_node_modules.SENTINEL, '{}') | |
| code, out, err = self.run_hook() | |
| self.assertEqual(code, 0) | |
| self.assertEqual(out + err, '') | |
| self.assertTrue(all(self.exists(name) for name in STAMPS)) | |
| def test_reextracts_matching_archive(self): | |
| digest = self.write_archive( | |
| [ensure_node_modules.SENTINEL, 'typescript/lib/tsc.js']) | |
| self.write_stamps(digest) | |
| code, out, err = self.run_hook() | |
| self.assertEqual(code, 0, err) | |
| self.assertIn('Re-extracting', out) | |
| self.assertEqual(err, '') | |
| self.assertTrue(self.exists(ensure_node_modules.SENTINEL)) | |
| self.assertTrue(self.exists('typescript/lib/tsc.js')) | |
| # Stamps and archive stay valid for gclient. | |
| self.assertTrue(all(self.exists(name) for name in STAMPS)) | |
| self.assertTrue(self.exists(ensure_node_modules.ARCHIVE_NAME)) | |
| # Like gclient, the archive's zeroed mtimes are not applied. | |
| if hasattr(tarfile, 'data_filter'): | |
| mtime = os.path.getmtime( | |
| os.path.join(self.node_modules, ensure_node_modules.SENTINEL)) | |
| self.assertGreater(mtime, time.time() - 3600) | |
| def test_matches_any_hash_stamp(self): | |
| # gclient leaves the stamps of previously pinned archives behind. | |
| digest = self.write_archive([ensure_node_modules.SENTINEL]) | |
| self.write_stamps(DIGEST) | |
| write(self.node_modules, f'.devtools-frontend_{digest}_hash', | |
| digest + '\n') | |
| code, _, err = self.run_hook() | |
| self.assertEqual(code, 0, err) | |
| self.assertTrue(self.exists(ensure_node_modules.SENTINEL)) | |
| @unittest.skipUnless(hasattr(tarfile, 'data_filter'), | |
| 'needs tarfile extraction filters (PEP 706)') | |
| def test_rejects_archive_escaping_node_modules(self): | |
| archive = os.path.join(self.node_modules, | |
| ensure_node_modules.ARCHIVE_NAME) | |
| with tarfile.open(archive, 'w:gz') as tar: | |
| info = tarfile.TarInfo('../escaped') | |
| info.size = 0 | |
| tar.addfile(info, io.BytesIO(b'')) | |
| with open(archive, 'rb') as f: | |
| self.write_stamps(hashlib.sha256(f.read()).hexdigest()) | |
| code, _, err = self.run_hook() | |
| self.assertEqual(code, 1) | |
| self.assertIn('failed', err) | |
| self.assertIn('run `gclient sync` again', err) | |
| self.assertFalse( | |
| os.path.exists(os.path.join(self.node_modules, '..', 'escaped'))) | |
| # The archive is unusable, so the next sync has to download it again. | |
| self.assertEqual(os.listdir(self.node_modules), []) | |
| def test_removes_stamps_without_archive(self): | |
| self.write_stamps(DIGEST) | |
| code, out, err = self.run_hook() | |
| self.assertEqual(code, 1) | |
| self.assertEqual(out, '') | |
| self.assertIn('run `gclient sync` again', err) | |
| self.assertEqual(os.listdir(self.node_modules), []) | |
| def test_removes_stamps_and_archive_on_hash_mismatch(self): | |
| self.write_archive([ensure_node_modules.SENTINEL]) | |
| self.write_stamps(DIGEST) | |
| code, _, err = self.run_hook() | |
| self.assertEqual(code, 1) | |
| self.assertIn('run `gclient sync` again', err) | |
| self.assertEqual(os.listdir(self.node_modules), []) | |
| self.assertFalse(self.exists(ensure_node_modules.SENTINEL)) | |
| def test_keeps_stamps_when_archive_lacks_sentinel(self): | |
| # A re-download would not help; fail loudly instead of looping. | |
| digest = self.write_archive(['typescript/lib/tsc.js']) | |
| self.write_stamps(digest) | |
| code, _, err = self.run_hook() | |
| self.assertEqual(code, 1) | |
| self.assertIn('SENTINEL', err) | |
| self.assertTrue(all(self.exists(name) for name in STAMPS)) | |
| self.assertTrue(self.exists('typescript/lib/tsc.js')) | |
| def test_default_location_is_the_checkout(self): | |
| expected = os.path.join( | |
| os.path.dirname( | |
| os.path.dirname( | |
| os.path.dirname( | |
| os.path.abspath(ensure_node_modules.__file__)))), | |
| 'node_modules') | |
| self.assertEqual(ensure_node_modules.NODE_MODULES, expected) | |
| if __name__ == '__main__': | |
| unittest.main() |