From e85142123aa2ff6dc3f3d21aa3b738acd77dc952 Mon Sep 17 00:00:00 2001 From: Richard Levasseur Date: Wed, 22 Jul 2026 19:26:43 +0000 Subject: [PATCH 1/3] tests: convert private tool tests to pytest_test and pytest style Convert private tool tests (tests/tools/zipapp/ and tools/private/update_deps/) to pytest_test and pytest style to be more idiomatic Python tests. Also converts to use pytest fixtures. --- tests/tools/zipapp/BUILD.bazel | 8 +- tests/tools/zipapp/exe_zip_maker_test.py | 87 +-- tests/tools/zipapp/zip_main_maker_test.py | 169 +++-- tests/tools/zipapp/zipper_test.py | 588 ++++++++---------- tools/private/update_deps/BUILD.bazel | 4 +- tools/private/update_deps/update_file_test.py | 112 ++-- 6 files changed, 441 insertions(+), 527 deletions(-) diff --git a/tests/tools/zipapp/BUILD.bazel b/tests/tools/zipapp/BUILD.bazel index b71e9b2589..31bbc8b3bb 100644 --- a/tests/tools/zipapp/BUILD.bazel +++ b/tests/tools/zipapp/BUILD.bazel @@ -1,18 +1,18 @@ -load("//python:py_test.bzl", "py_test") +load("//tests/support/pytest_test:pytest_test.bzl", "pytest_test") -py_test( +pytest_test( name = "zipper_test", srcs = ["zipper_test.py"], deps = ["//tools/private/zipapp:zipper_lib"], ) -py_test( +pytest_test( name = "exe_zip_maker_test", srcs = ["exe_zip_maker_test.py"], deps = ["//tools/private/zipapp:exe_zip_maker_lib"], ) -py_test( +pytest_test( name = "zip_main_maker_test", srcs = ["zip_main_maker_test.py"], deps = ["//tools/private/zipapp:zip_main_maker_lib"], diff --git a/tests/tools/zipapp/exe_zip_maker_test.py b/tests/tools/zipapp/exe_zip_maker_test.py index 73c509bdbe..97df258e92 100644 --- a/tests/tools/zipapp/exe_zip_maker_test.py +++ b/tests/tools/zipapp/exe_zip_maker_test.py @@ -1,71 +1,46 @@ import hashlib -import pathlib -import shutil import stat -import tempfile -import unittest from tools.private.zipapp import exe_zip_maker -class ExeZipMakerTest(unittest.TestCase): - def setUp(self): - self.test_dir = pathlib.Path(tempfile.mkdtemp()) - self.preamble_path = self.test_dir / "preamble.txt" - self.zip_path = self.test_dir / "data.zip" - self.output_path = self.test_dir / "output.exe" +def test_create_exe_zip(tmp_path): + preamble_path = tmp_path / "preamble.txt" + zip_path = tmp_path / "data.zip" + output_path = tmp_path / "output.exe" - def tearDown(self): - shutil.rmtree(self.test_dir) + # Create dummy zip file + zip_content = b"PK\x03\x04dummyzipcontent" + zip_path.write_bytes(zip_content) - def assertStartsWith(self, actual, expected): - if not actual.startswith(expected): - self.fail(f"{actual!r} does not start with {expected!r}") + # Calculate expected hash + expected_hash = hashlib.sha256(zip_content).hexdigest().encode("utf-8") - def test_create_exe_zip(self): - # Create dummy zip file - zip_content = b"PK\x03\x04dummyzipcontent" - self.zip_path.write_bytes(zip_content) + # Create preamble with placeholder + preamble_text = b"#!/bin/bash\nEXPECTED_HASH='%ZIP_HASH%'\n# ... logic ...\n" + preamble_path.write_bytes(preamble_text) - # Calculate expected hash - expected_hash = hashlib.sha256(zip_content).hexdigest().encode("utf-8") + # Call create_exe_zip directly + exe_zip_maker.create_exe_zip(str(preamble_path), str(zip_path), str(output_path)) - # Create preamble with placeholder - preamble_text = b"#!/bin/bash\nEXPECTED_HASH='%ZIP_HASH%'\n# ... logic ...\n" - self.preamble_path.write_bytes(preamble_text) + # Verify output exists + assert output_path.exists(), f"Output path '{output_path}' should exist" - # Call create_exe_zip directly - exe_zip_maker.create_exe_zip( - str(self.preamble_path), str(self.zip_path), str(self.output_path) - ) + # Verify executable bit + st = output_path.stat() + assert st.st_mode & stat.S_IEXEC, ( + f"Output path '{output_path}' should be executable" + ) - # Verify output exists - self.assertTrue( - self.output_path.exists(), - msg=f"Output path '{self.output_path}' should exist", - ) + # Verify content + content = output_path.read_bytes() - # Verify executable bit - st = self.output_path.stat() - self.assertTrue( - st.st_mode & stat.S_IEXEC, - msg=f"Output path '{self.output_path}' should be executable", - ) + # Split content back into preamble and zip + # We know the preamble text length after substitution. + expected_preamble = preamble_text.replace(b"%ZIP_HASH%", expected_hash) - # Verify content - content = self.output_path.read_bytes() - - # Split content back into preamble and zip - # We know the preamble text length after substitution. - expected_preamble = preamble_text.replace(b"%ZIP_HASH%", expected_hash) - - self.assertStartsWith(content, expected_preamble) - self.assertTrue( - content.endswith(zip_content), - msg="Output content should end with the zip content", - ) - self.assertEqual(len(content), len(expected_preamble) + len(zip_content)) - - -if __name__ == "__main__": - unittest.main() + assert content.startswith(expected_preamble) + assert content.endswith(zip_content), ( + "Output content should end with the zip content" + ) + assert len(content) == len(expected_preamble) + len(zip_content) diff --git a/tests/tools/zipapp/zip_main_maker_test.py b/tests/tools/zipapp/zip_main_maker_test.py index dd8e8e8029..5c7f57c590 100644 --- a/tests/tools/zipapp/zip_main_maker_test.py +++ b/tests/tools/zipapp/zip_main_maker_test.py @@ -1,101 +1,90 @@ import hashlib import os -import tempfile -import unittest -from unittest import mock from tools.private.zipapp import zip_main_maker -class ZipMainMakerTest(unittest.TestCase): - def setUp(self): - self.temp_dir = tempfile.TemporaryDirectory() - self.addCleanup(self.temp_dir.cleanup) - - def test_creates_zip_main(self): - template_path = os.path.join(self.temp_dir.name, "template.py") - with open(template_path, "w", encoding="utf-8") as f: - f.write("hash=%APP_HASH%\nfoo=%FOO%\n") - - output_path = os.path.join(self.temp_dir.name, "output.py") - - file1_path = os.path.join(self.temp_dir.name, "file1.txt") - with open(file1_path, "wb") as f: - f.write(b"content1") - - file2_path = os.path.join(self.temp_dir.name, "file2.txt") - with open(file2_path, "wb") as f: - f.write(b"content2") - - # Add a symlink to test symlink hashing - symlink_path = os.path.join(self.temp_dir.name, "symlink.txt") - os.symlink(file1_path, symlink_path) - - manifest_path = os.path.join(self.temp_dir.name, "manifest.txt") - with open(manifest_path, "w", encoding="utf-8") as f: - f.write(f"rf-file|0|file1.txt|{file1_path}\n") - f.write(f"rf-file|0|file2.txt|{file2_path}\n") - f.write(f"rf-symlink|1|symlink.txt|{symlink_path}\n") - f.write("rf-empty|empty_file.txt\n") - - argv = [ - "zip_main_maker.py", - "--template", - template_path, - "--output", - output_path, - "--substitution", - "%FOO%=bar", - "--hash_files_manifest", - manifest_path, - ] - - with mock.patch("sys.argv", argv): - zip_main_maker.main() - - # Calculate expected hash - h = hashlib.sha256() - line1 = f"rf-file|0|file1.txt|{file1_path}" - line2 = f"rf-file|0|file2.txt|{file2_path}" - line3 = f"rf-symlink|1|symlink.txt|{symlink_path}" - line4 = "rf-empty|empty_file.txt" - - # Sort lines like the program does - lines = sorted([line1, line2, line3, line4]) - for line in lines: - parts = line.split("|") - if len(parts) > 1: - _, rest = line.split("|", 1) - h.update(rest.encode("utf-8")) - else: - h.update(line.encode("utf-8")) - - type_ = parts[0] - if type_ == "rf-empty": +def test_creates_zip_main(tmp_path, monkeypatch): + temp_dir = str(tmp_path) + template_path = os.path.join(temp_dir, "template.py") + with open(template_path, "w", encoding="utf-8") as f: + f.write("hash=%APP_HASH%\nfoo=%FOO%\n") + + output_path = os.path.join(temp_dir, "output.py") + + file1_path = os.path.join(temp_dir, "file1.txt") + with open(file1_path, "wb") as f: + f.write(b"content1") + + file2_path = os.path.join(temp_dir, "file2.txt") + with open(file2_path, "wb") as f: + f.write(b"content2") + + # Add a symlink to test symlink hashing + symlink_path = os.path.join(temp_dir, "symlink.txt") + os.symlink(file1_path, symlink_path) + + manifest_path = os.path.join(temp_dir, "manifest.txt") + with open(manifest_path, "w", encoding="utf-8") as f: + f.write(f"rf-file|0|file1.txt|{file1_path}\n") + f.write(f"rf-file|0|file2.txt|{file2_path}\n") + f.write(f"rf-symlink|1|symlink.txt|{symlink_path}\n") + f.write("rf-empty|empty_file.txt\n") + + argv = [ + "zip_main_maker.py", + "--template", + template_path, + "--output", + output_path, + "--substitution", + "%FOO%=bar", + "--hash_files_manifest", + manifest_path, + ] + + monkeypatch.setattr("sys.argv", argv) + zip_main_maker.main() + + # Calculate expected hash + h = hashlib.sha256() + line1 = f"rf-file|0|file1.txt|{file1_path}" + line2 = f"rf-file|0|file2.txt|{file2_path}" + line3 = f"rf-symlink|1|symlink.txt|{symlink_path}" + line4 = "rf-empty|empty_file.txt" + + # Sort lines like the program does + lines = sorted([line1, line2, line3, line4]) + for line in lines: + parts = line.split("|") + if len(parts) > 1: + _, rest = line.split("|", 1) + h.update(rest.encode("utf-8")) + else: + h.update(line.encode("utf-8")) + + type_ = parts[0] + if type_ == "rf-empty": + continue + if len(parts) >= 4: + is_symlink_str = parts[1] + path = parts[-1] + if not path: continue - if len(parts) >= 4: - is_symlink_str = parts[1] - path = parts[-1] - if not path: - continue - if is_symlink_str == "-1": - is_symlink = not os.path.exists(path) - else: - is_symlink = is_symlink_str == "1" - - if is_symlink: - h.update(os.readlink(path).encode("utf-8")) - else: - with open(path, "rb") as f: - h.update(f.read()) - - expected_hash = h.hexdigest() + if is_symlink_str == "-1": + is_symlink = not os.path.exists(path) + else: + is_symlink = is_symlink_str == "1" - with open(output_path, "r", encoding="utf-8") as f: - content = f.read() + if is_symlink: + h.update(os.readlink(path).encode("utf-8")) + else: + with open(path, "rb") as f: + h.update(f.read()) - self.assertEqual(content, f"hash={expected_hash}\nfoo=bar\n") + expected_hash = h.hexdigest() + with open(output_path, "r", encoding="utf-8") as f: + content = f.read() -if __name__ == "__main__": - unittest.main() + assert content == f"hash={expected_hash}\nfoo=bar\n" diff --git a/tests/tools/zipapp/zipper_test.py b/tests/tools/zipapp/zipper_test.py index ac70917c30..bbd85b6536 100644 --- a/tests/tools/zipapp/zipper_test.py +++ b/tests/tools/zipapp/zipper_test.py @@ -1,8 +1,5 @@ import os -import pathlib import shutil -import tempfile -import unittest import zipfile from tools.private.zipapp import zipper @@ -12,333 +9,294 @@ def symlink_target_path(p): return p.replace("/", os.sep) -class ZipperTest(unittest.TestCase): - def setUp(self): - self.test_dir = pathlib.Path(tempfile.mkdtemp()) - self.manifest_path = self.test_dir / "manifest.txt" - self.output_zip = self.test_dir / "output.zip" - - def tearDown(self): - shutil.rmtree(self.test_dir) - - def _create_zip(self, **kwargs): - defaults = { - "manifest_path": self.manifest_path, - "output_zip": self.output_zip, - "compress_level": 0, - "workspace_name": "my_ws", - "legacy_external_runfiles": False, - "runfiles_dir": "runfiles", - # We need to generate paths for the platform we're running on. - "platform_pathsep": os.sep, +def is_symlink(zip_info): + # Check upper 4 bits of external_attr for S_IFLNK + # S_IFLNK is 0o120000 = 0xA000 + attr = zip_info.external_attr >> 16 + return (attr & 0xF000) == 0xA000 + + +def assert_zip_file_content(zf, path, content=None, is_symlink_file=False, target=None): + info = zf.getinfo(path) + if is_symlink_file: + assert is_symlink(info), f"{path} should be a symlink but is not" + assert zf.read(path).decode() == target + else: + assert not is_symlink(info), f"{path} should NOT be a symlink but is" + assert zf.read(path).decode() == content + + +def create_zip(manifest_path, output_zip, **kwargs): + defaults = { + "manifest_path": manifest_path, + "output_zip": output_zip, + "compress_level": 0, + "workspace_name": "my_ws", + "legacy_external_runfiles": False, + "runfiles_dir": "runfiles", + "platform_pathsep": os.sep, + } + defaults.update(kwargs) + zipper.create_zip(**defaults) + + +def extract_zip(zip_path, extract_dir): + # Manually extract to preserve symlinks + with zipfile.ZipFile(zip_path, "r") as zf: + for info in zf.infolist(): + extract_path = extract_dir / info.filename + extract_path.parent.mkdir(parents=True, exist_ok=True) + if is_symlink(info): + target = zf.read(info).decode() + os.symlink(target, extract_path) + else: + with zf.open(info) as src, open(extract_path, "wb") as dst: + shutil.copyfileobj(src, dst) + + +def test_create_zip_with_files_and_symlinks(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" + + file1_path = tmp_path / "file1.txt" + file1_path.write_text("content1") + + link_target_path = "target.txt" + symlink_path = tmp_path / "symlink_source" + symlink_path.symlink_to(link_target_path) + + manifest_content = [ + f"regular|0|file1.txt|{file1_path}", + f"rf-file|0|foo/bar.txt|{file1_path}", + f"rf-symlink|1|link1|{symlink_path}", + f"rf-root-symlink|0|root_file|{file1_path}", + "rf-empty|empty_file", + ] + manifest_path.write_text("\n".join(manifest_content)) + + create_zip(manifest_path, output_zip) + + assert output_zip.exists() + + with zipfile.ZipFile(output_zip, "r") as zf: + assert set(zf.namelist()) == { + "file1.txt", + "runfiles/my_ws/foo/bar.txt", + "runfiles/my_ws/link1", + "runfiles/root_file", + "runfiles/my_ws/empty_file", } - defaults.update(kwargs) - zipper.create_zip(**defaults) - - def assertZipFileContent( - self, zf, path, content=None, is_symlink=False, target=None - ): - info = zf.getinfo(path) - if is_symlink: - self.assertTrue( - self.is_symlink(info), - f"{path} should be a symlink but is not", - ) - self.assertEqual(zf.read(path).decode(), target) - else: - self.assertFalse( - self.is_symlink(info), - f"{path} should NOT be a symlink but is", - ) - self.assertEqual(zf.read(path).decode(), content) - - def test_create_zip_with_files_and_symlinks(self): - file1_path = self.test_dir / "file1.txt" - file1_path.write_text("content1") - - link_target_path = "target.txt" # Relative target - symlink_path = self.test_dir / "symlink_source" - symlink_path.symlink_to(link_target_path) - - manifest_content = [ - f"regular|0|file1.txt|{file1_path}", - f"rf-file|0|foo/bar.txt|{file1_path}", - f"rf-symlink|1|link1|{symlink_path}", # Should read target 'target.txt' - f"rf-root-symlink|0|root_file|{file1_path}", - "rf-empty|empty_file", - ] - self.manifest_path.write_text("\n".join(manifest_content)) - - self._create_zip() - - self.assertTrue(self.output_zip.exists()) - - with zipfile.ZipFile(self.output_zip, "r") as zf: - self.assertEqual( - set(zf.namelist()), - { - "file1.txt", - "runfiles/my_ws/foo/bar.txt", - "runfiles/my_ws/link1", - "runfiles/root_file", - "runfiles/my_ws/empty_file", - }, - ) - - self.assertZipFileContent(zf, "file1.txt", content="content1") - self.assertZipFileContent( - zf, "runfiles/my_ws/foo/bar.txt", content="content1" - ) - self.assertZipFileContent( - zf, "runfiles/my_ws/link1", is_symlink=True, target="target.txt" - ) - self.assertZipFileContent(zf, "runfiles/root_file", content="content1") - self.assertZipFileContent(zf, "runfiles/my_ws/empty_file", content="") - - def test_create_zip_with_direct_symlink(self): - # Test the 'symlink' manifest entry type - manifest_content = [ - "symlink|path/to/link|target/path", - ] - self.manifest_path.write_text("\n".join(manifest_content)) - - self._create_zip() - - with zipfile.ZipFile(self.output_zip, "r") as zf: - self.assertEqual(zf.namelist(), ["runfiles/path/to/link"]) - self.assertZipFileContent( - zf, - "runfiles/path/to/link", - is_symlink=True, - target=symlink_target_path("../../target/path"), - ) - - def test_pathsep_normalization(self): - # Test that pathsep="\\" normalizes paths - file1_path = self.test_dir / "file1.txt" - file1_path.write_text("content1") - - manifest_content = [ - f"regular|0|dir/file.txt|{file1_path}", - "symlink|link/path|target/path", - ] - self.manifest_path.write_text("\n".join(manifest_content)) - - # Use backslash as platform_pathsep - self._create_zip(platform_pathsep="\\") - - with zipfile.ZipFile(self.output_zip, "r") as zf: - # zipfile.namelist() always returns with forward slashes - # But the content of the symlink should be normalized if it was passed through path_norm - self.assertEqual( - set(zf.namelist()), - {"dir/file.txt", "runfiles/link/path"}, - ) - # The target of the symlink should have backslashes - self.assertZipFileContent( - zf, - "runfiles/link/path", - is_symlink=True, - target="..\\target\\path", - ) - - def test_symlink_precedence(self): - # Test that 'symlink' entries take precedence over others for the same path - file1_path = self.test_dir / "file1.txt" - file1_path.write_text("content1") - - manifest_content = [ - # Same zip path: runfiles/my_ws/path/to/file - f"rf-file|0|path/to/file|{file1_path}", - "symlink|my_ws/path/to/file|symlink/target", - ] - self.manifest_path.write_text("\n".join(manifest_content)) - - self._create_zip() - - with zipfile.ZipFile(self.output_zip, "r") as zf: - self.assertEqual(zf.namelist(), ["runfiles/my_ws/path/to/file"]) - # It should be the symlink, not the file - self.assertZipFileContent( - zf, - "runfiles/my_ws/path/to/file", - is_symlink=True, - target=symlink_target_path("../../../symlink/target"), - ) - - def test_timestamps_are_deterministic(self): - # Create a content file with a specific recent timestamp - file1_path = self.test_dir / "file1.txt" - file1_path.write_text("content1") - - # Set mtime to something recent (e.g. now) - os.utime(file1_path, None) - - manifest_content = [ - f"regular|0|file1.txt|{file1_path}", - ] - self.manifest_path.write_text("\n".join(manifest_content)) + assert_zip_file_content(zf, "file1.txt", content="content1") + assert_zip_file_content(zf, "runfiles/my_ws/foo/bar.txt", content="content1") + assert_zip_file_content( + zf, "runfiles/my_ws/link1", is_symlink_file=True, target="target.txt" + ) + assert_zip_file_content(zf, "runfiles/root_file", content="content1") + assert_zip_file_content(zf, "runfiles/my_ws/empty_file", content="") - self._create_zip() - with zipfile.ZipFile(self.output_zip, "r") as zf: - info = zf.getinfo("file1.txt") - # DOS epoch is 1980-01-01 00:00:00 - expected_date_time = (1980, 1, 1, 0, 0, 0) - self.assertEqual(info.date_time, expected_date_time) +def test_create_zip_with_direct_symlink(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" - def test_runfiles_mapping_with_cross_repo_paths(self): - # Create content file - file1_path = self.test_dir / "file1.txt" - file1_path.write_text("content1") + manifest_content = ["symlink|path/to/link|target/path"] + manifest_path.write_text("\n".join(manifest_content)) - manifest_content = [ - f"rf-file|0|../other_repo/foo.txt|{file1_path}", - "rf-empty|../other_repo/empty_file", - ] + create_zip(manifest_path, output_zip) - self.manifest_path.write_text("\n".join(manifest_content)) - - self._create_zip(workspace_name="my_ws") - - with zipfile.ZipFile(self.output_zip, "r") as zf: - self.assertEqual( - set(zf.namelist()), - { - "runfiles/other_repo/foo.txt", - "runfiles/other_repo/empty_file", - }, - ) - self.assertZipFileContent( - zf, "runfiles/other_repo/foo.txt", content="content1" - ) - self.assertZipFileContent(zf, "runfiles/other_repo/empty_file", content="") - - def test_runfiles_mapping_with_legacy_external_paths(self): - file1_path = self.test_dir / "file1.txt" - file1_path.write_text("content1") - - manifest_content = [ - f"rf-file|0|external/other_repo/foo.txt|{file1_path}", - "rf-empty|external/other_repo/empty_file", - ] + with zipfile.ZipFile(output_zip, "r") as zf: + assert zf.namelist() == ["runfiles/path/to/link"] + assert_zip_file_content( + zf, + "runfiles/path/to/link", + is_symlink_file=True, + target=symlink_target_path("../../target/path"), + ) - self.manifest_path.write_text("\n".join(manifest_content)) - - self._create_zip(workspace_name="my_ws", legacy_external_runfiles=True) - - with zipfile.ZipFile(self.output_zip, "r") as zf: - self.assertEqual( - set(zf.namelist()), - { - "runfiles/other_repo/foo.txt", - "runfiles/other_repo/empty_file", - }, - ) - self.assertZipFileContent( - zf, "runfiles/other_repo/foo.txt", content="content1" - ) - self.assertZipFileContent(zf, "runfiles/other_repo/empty_file", content="") - - def test_output_deterministic(self): - # Create files - file1 = self.test_dir / "file1" - file1.write_text("1") - file2 = self.test_dir / "file2" - file2.write_text("2") - file3 = self.test_dir / "file3" - file3.write_text("3") - - # Manifest entries mixed up - # We want the final order to be: - # 1. a/regular (regular) - # 2. runfiles/a_root_link (rf-root-symlink) - # 3. runfiles/my_ws/b_rf_file (rf-file) - # 4. runfiles/my_ws/c_rf_link (rf-symlink) - # 5. runfiles/my_ws/d_rf_empty (rf-empty) - # 6. z/regular (regular) - - manifest_content = [ - f"regular|0|z/regular|{file1}", - f"rf-file|0|b_rf_file|{file2}", # -> runfiles/my_ws/b_rf_file - f"rf-root-symlink|0|a_root_link|{file3}", # -> runfiles/a_root_link - f"regular|0|a/regular|{file3}", - "rf-empty|d_rf_empty", # -> runfiles/my_ws/d_rf_empty - f"rf-symlink|0|c_rf_link|{file3}", # -> runfiles/my_ws/c_rf_link - ] - self.manifest_path.write_text("\n".join(manifest_content)) - - self._create_zip(workspace_name="my_ws") - - with zipfile.ZipFile(self.output_zip, "r") as zf: - self.assertEqual( - zf.namelist(), - [ - "a/regular", - "runfiles/a_root_link", - "runfiles/my_ws/b_rf_file", - "runfiles/my_ws/c_rf_link", - "runfiles/my_ws/d_rf_empty", - "z/regular", - ], - ) - - def _extract_zip(self, zip_path, extract_dir): - # Manually extract to preserve symlinks - with zipfile.ZipFile(zip_path, "r") as zf: - for info in zf.infolist(): - extract_path = extract_dir / info.filename - extract_path.parent.mkdir(parents=True, exist_ok=True) - if self.is_symlink(info): - target = zf.read(info).decode() - # On Windows, relative symlinks must use backslashes to be readable - os.symlink(target, extract_path) - else: - with zf.open(info) as src, open(extract_path, "wb") as dst: - shutil.copyfileobj(src, dst) - - def test_symlink_extraction(self): - # Test that 'symlink' entries extract correctly as relative symlinks - # Create a file that the symlink will point to - target_file = self.test_dir / "target_file.txt" - target_file.write_text("target content") - - manifest_content = [ - f"rf-file|0|target/path|{target_file}", - "symlink|my_ws/path/to/link|my_ws/target/path", - f"rf-file|0|same_dir_target|{target_file}", - "symlink|my_ws/same_dir_link|my_ws/same_dir_target", - ] - self.manifest_path.write_text("\n".join(manifest_content)) +def test_pathsep_normalization(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" - self._create_zip(workspace_name="my_ws") + file1_path = tmp_path / "file1.txt" + file1_path.write_text("content1") - extract_dir = self.test_dir / "extract" - extract_dir.mkdir() + manifest_content = [ + f"regular|0|dir/file.txt|{file1_path}", + "symlink|link/path|target/path", + ] + manifest_path.write_text("\n".join(manifest_content)) - self._extract_zip(self.output_zip, extract_dir) + create_zip(manifest_path, output_zip, platform_pathsep="\\") - link_path = extract_dir / "runfiles/my_ws/path/to/link" - self.assertTrue(link_path.is_symlink(), f"{link_path} should be a symlink") - self.assertEqual( - os.readlink(link_path), "../../target/path".replace("/", os.path.sep) + with zipfile.ZipFile(output_zip, "r") as zf: + assert set(zf.namelist()) == {"dir/file.txt", "runfiles/link/path"} + assert_zip_file_content( + zf, + "runfiles/link/path", + is_symlink_file=True, + target="..\\target\\path", ) - self.assertEqual(link_path.read_text(), "target content") - link2_path = extract_dir / "runfiles/my_ws/same_dir_link" - self.assertTrue(link2_path.is_symlink(), f"{link2_path} should be a symlink") - # Relative path from runfiles/my_ws/ to runfiles/my_ws/same_dir_target is just same_dir_target - self.assertEqual(os.readlink(link2_path), "same_dir_target") - self.assertEqual(link2_path.read_text(), "target content") - def is_symlink(self, zip_info): - # Check upper 4 bits of external_attr for S_IFLNK - # S_IFLNK is 0o120000 = 0xA000 - attr = zip_info.external_attr >> 16 - return (attr & 0xF000) == 0xA000 +def test_symlink_precedence(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" + + file1_path = tmp_path / "file1.txt" + file1_path.write_text("content1") + + manifest_content = [ + f"rf-file|0|path/to/file|{file1_path}", + "symlink|my_ws/path/to/file|symlink/target", + ] + manifest_path.write_text("\n".join(manifest_content)) + + create_zip(manifest_path, output_zip) + + with zipfile.ZipFile(output_zip, "r") as zf: + assert zf.namelist() == ["runfiles/my_ws/path/to/file"] + assert_zip_file_content( + zf, + "runfiles/my_ws/path/to/file", + is_symlink_file=True, + target=symlink_target_path("../../../symlink/target"), + ) + + +def test_timestamps_are_deterministic(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" + + file1_path = tmp_path / "file1.txt" + file1_path.write_text("content1") + os.utime(file1_path, None) + + manifest_content = [f"regular|0|file1.txt|{file1_path}"] + manifest_path.write_text("\n".join(manifest_content)) + + create_zip(manifest_path, output_zip) + + with zipfile.ZipFile(output_zip, "r") as zf: + info = zf.getinfo("file1.txt") + expected_date_time = (1980, 1, 1, 0, 0, 0) + assert info.date_time == expected_date_time + + +def test_runfiles_mapping_with_cross_repo_paths(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" + + file1_path = tmp_path / "file1.txt" + file1_path.write_text("content1") + + manifest_content = [ + f"rf-file|0|../other_repo/foo.txt|{file1_path}", + "rf-empty|../other_repo/empty_file", + ] + manifest_path.write_text("\n".join(manifest_content)) + + create_zip(manifest_path, output_zip, workspace_name="my_ws") + + with zipfile.ZipFile(output_zip, "r") as zf: + assert set(zf.namelist()) == { + "runfiles/other_repo/foo.txt", + "runfiles/other_repo/empty_file", + } + assert_zip_file_content(zf, "runfiles/other_repo/foo.txt", content="content1") + assert_zip_file_content(zf, "runfiles/other_repo/empty_file", content="") + + +def test_runfiles_mapping_with_legacy_external_paths(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" + + file1_path = tmp_path / "file1.txt" + file1_path.write_text("content1") + + manifest_content = [ + f"rf-file|0|external/other_repo/foo.txt|{file1_path}", + "rf-empty|external/other_repo/empty_file", + ] + manifest_path.write_text("\n".join(manifest_content)) + + create_zip( + manifest_path, output_zip, workspace_name="my_ws", legacy_external_runfiles=True + ) + + with zipfile.ZipFile(output_zip, "r") as zf: + assert set(zf.namelist()) == { + "runfiles/other_repo/foo.txt", + "runfiles/other_repo/empty_file", + } + assert_zip_file_content(zf, "runfiles/other_repo/foo.txt", content="content1") + assert_zip_file_content(zf, "runfiles/other_repo/empty_file", content="") + + +def test_output_deterministic(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" + + file1 = tmp_path / "file1" + file1.write_text("1") + file2 = tmp_path / "file2" + file2.write_text("2") + file3 = tmp_path / "file3" + file3.write_text("3") + + manifest_content = [ + f"regular|0|z/regular|{file1}", + f"rf-file|0|b_rf_file|{file2}", + f"rf-root-symlink|0|a_root_link|{file3}", + f"regular|0|a/regular|{file3}", + "rf-empty|d_rf_empty", + f"rf-symlink|0|c_rf_link|{file3}", + ] + + manifest_path.write_text("\n".join(manifest_content)) + + create_zip(manifest_path, output_zip, workspace_name="my_ws") + + with zipfile.ZipFile(output_zip, "r") as zf: + assert zf.namelist() == [ + "a/regular", + "runfiles/a_root_link", + "runfiles/my_ws/b_rf_file", + "runfiles/my_ws/c_rf_link", + "runfiles/my_ws/d_rf_empty", + "z/regular", + ] + + +def test_symlink_extraction(tmp_path): + manifest_path = tmp_path / "manifest.txt" + output_zip = tmp_path / "output.zip" + + target_file = tmp_path / "target_file.txt" + target_file.write_text("target content") + + manifest_content = [ + f"rf-file|0|target/path|{target_file}", + "symlink|my_ws/path/to/link|my_ws/target/path", + f"rf-file|0|same_dir_target|{target_file}", + "symlink|my_ws/same_dir_link|my_ws/same_dir_target", + ] + manifest_path.write_text("\n".join(manifest_content)) + + create_zip(manifest_path, output_zip, workspace_name="my_ws") + + extract_dir = tmp_path / "extract" + extract_dir.mkdir() + + extract_zip(output_zip, extract_dir) + link_path = extract_dir / "runfiles/my_ws/path/to/link" + assert link_path.is_symlink(), f"{link_path} should be a symlink" + assert os.readlink(link_path) == "../../target/path".replace("/", os.path.sep) + assert link_path.read_text() == "target content" -if __name__ == "__main__": - unittest.main() + link2_path = extract_dir / "runfiles/my_ws/same_dir_link" + assert link2_path.is_symlink(), f"{link2_path} should be a symlink" + assert os.readlink(link2_path) == "same_dir_target" + assert link2_path.read_text() == "target content" diff --git a/tools/private/update_deps/BUILD.bazel b/tools/private/update_deps/BUILD.bazel index c3f94c9814..da3dd2a4b7 100644 --- a/tools/private/update_deps/BUILD.bazel +++ b/tools/private/update_deps/BUILD.bazel @@ -13,7 +13,7 @@ # limitations under the License. load("//python:py_binary.bzl", "py_binary") load("//python:py_library.bzl", "py_library") -load("//python:py_test.bzl", "py_test") +load("//tests/support/pytest_test:pytest_test.bzl", "pytest_test") licenses(["notice"]) @@ -65,7 +65,7 @@ py_binary( ], ) -py_test( +pytest_test( name = "update_file_test", srcs = ["update_file_test.py"], imports = ["../../.."], diff --git a/tools/private/update_deps/update_file_test.py b/tools/private/update_deps/update_file_test.py index a3cb1c0a6d..b1d6545ead 100644 --- a/tools/private/update_deps/update_file_test.py +++ b/tools/private/update_deps/update_file_test.py @@ -12,14 +12,13 @@ # See the License for the specific language governing permissions and # limitations under the License. -import unittest +import pytest from tools.private.update_deps.update_file import replace_snippet, unified_diff -class TestReplaceSnippet(unittest.TestCase): - def test_replace_simple(self): - current = """\ +def test_replace_simple(): + current = """\ Before the snippet # Start marker @@ -30,15 +29,14 @@ def test_replace_simple(self): After the snippet """ - snippet = "Replaced" # noqa: F841 - got = replace_snippet( - current=current, - snippet="Replaced", - start_marker="# Start marker", - end_marker="# End marker", - ) - - want = """\ + got = replace_snippet( + current=current, + snippet="Replaced", + start_marker="# Start marker", + end_marker="# End marker", + ) + + want = """\ Before the snippet # Start marker @@ -47,10 +45,11 @@ def test_replace_simple(self): After the snippet """ - self.assertEqual(want, got) + assert got == want - def test_replace_indented(self): - current = """\ + +def test_replace_indented(): + current = """\ Before the snippet # Start marker @@ -59,14 +58,14 @@ def test_replace_indented(self): After the snippet """ - got = replace_snippet( - current=current, - snippet=" Replaced", - start_marker="# Start marker", - end_marker="# End marker", - ) - - want = """\ + got = replace_snippet( + current=current, + snippet=" Replaced", + start_marker="# Start marker", + end_marker="# End marker", + ) + + want = """\ Before the snippet # Start marker @@ -75,45 +74,42 @@ def test_replace_indented(self): After the snippet """ - self.assertEqual(want, got) - - def test_raises_if_start_is_not_found(self): - with self.assertRaises(RuntimeError) as exc: - replace_snippet( - current="foo", - snippet="", - start_marker="start", - end_marker="end", - ) - - self.assertEqual(exc.exception.args[0], "Start marker 'start' was not found") - - def test_raises_if_end_is_not_found(self): - with self.assertRaises(RuntimeError) as exc: - replace_snippet( - current="start", - snippet="", - start_marker="start", - end_marker="end", - ) - - self.assertEqual(exc.exception.args[0], "End marker 'end' was not found") - - -class TestUnifiedDiff(unittest.TestCase): - def test_diff(self): - give_a = """\ + assert got == want + + +def test_raises_if_start_is_not_found(): + with pytest.raises(RuntimeError, match="Start marker 'start' was not found"): + replace_snippet( + current="foo", + snippet="", + start_marker="start", + end_marker="end", + ) + + +def test_raises_if_end_is_not_found(): + with pytest.raises(RuntimeError, match="End marker 'end' was not found"): + replace_snippet( + current="start", + snippet="", + start_marker="start", + end_marker="end", + ) + + +def test_diff(): + give_a = """\ First line second line Third line """ - give_b = """\ + give_b = """\ First line Second line Third line """ - got = unified_diff("filename", give_a, give_b) - want = """\ + got = unified_diff("filename", give_a, give_b) + want = """\ --- a/filename +++ b/filename @@ -1,3 +1,3 @@ @@ -121,8 +117,4 @@ def test_diff(self): -second line +Second line Third line""" - self.assertEqual(want, got) - - -if __name__ == "__main__": - unittest.main() + assert got == want From 9d0b03d43a1ce3366e443cd8e44255e227e0ee8d Mon Sep 17 00:00:00 2001 From: Richard Levasseur Date: Wed, 22 Jul 2026 19:39:31 +0000 Subject: [PATCH 2/3] fix(tests): add SUPPORTS_BZLMOD to private tool pytest_test targets Set target_compatible_with = SUPPORTS_BZLMOD on private tool pytest_test targets to prevent ModuleNotFoundError: No module named 'pytest_bazel' in non-bzlmod CI test matrix runs. --- tests/tools/zipapp/BUILD.bazel | 4 ++++ tools/private/update_deps/BUILD.bazel | 2 ++ 2 files changed, 6 insertions(+) diff --git a/tests/tools/zipapp/BUILD.bazel b/tests/tools/zipapp/BUILD.bazel index 31bbc8b3bb..97c8096ce9 100644 --- a/tests/tools/zipapp/BUILD.bazel +++ b/tests/tools/zipapp/BUILD.bazel @@ -1,19 +1,23 @@ +load("//tests/support:support.bzl", "SUPPORTS_BZLMOD") load("//tests/support/pytest_test:pytest_test.bzl", "pytest_test") pytest_test( name = "zipper_test", srcs = ["zipper_test.py"], + target_compatible_with = SUPPORTS_BZLMOD, deps = ["//tools/private/zipapp:zipper_lib"], ) pytest_test( name = "exe_zip_maker_test", srcs = ["exe_zip_maker_test.py"], + target_compatible_with = SUPPORTS_BZLMOD, deps = ["//tools/private/zipapp:exe_zip_maker_lib"], ) pytest_test( name = "zip_main_maker_test", srcs = ["zip_main_maker_test.py"], + target_compatible_with = SUPPORTS_BZLMOD, deps = ["//tools/private/zipapp:zip_main_maker_lib"], ) diff --git a/tools/private/update_deps/BUILD.bazel b/tools/private/update_deps/BUILD.bazel index da3dd2a4b7..8746d82c17 100644 --- a/tools/private/update_deps/BUILD.bazel +++ b/tools/private/update_deps/BUILD.bazel @@ -13,6 +13,7 @@ # limitations under the License. load("//python:py_binary.bzl", "py_binary") load("//python:py_library.bzl", "py_library") +load("//tests/support:support.bzl", "SUPPORTS_BZLMOD") load("//tests/support/pytest_test:pytest_test.bzl", "pytest_test") licenses(["notice"]) @@ -69,6 +70,7 @@ pytest_test( name = "update_file_test", srcs = ["update_file_test.py"], imports = ["../../.."], + target_compatible_with = SUPPORTS_BZLMOD, deps = [ ":update_file", ], From a1c69ef32479128e3904b02faed82c927cadd928 Mon Sep 17 00:00:00 2001 From: Richard Levasseur Date: Thu, 23 Jul 2026 02:10:54 +0000 Subject: [PATCH 3/3] chore(agents): update monitor_remote_ci.py script Update monitor_remote_ci.py to send completion notifications when all PR checks pass. --- .../scripts/monitor_remote_ci.py | 59 ++++++++++++++++++- 1 file changed, 58 insertions(+), 1 deletion(-) diff --git a/.agents/skills/monitor-ci-results/scripts/monitor_remote_ci.py b/.agents/skills/monitor-ci-results/scripts/monitor_remote_ci.py index fc1f8955d3..e5f4e81d10 100755 --- a/.agents/skills/monitor-ci-results/scripts/monitor_remote_ci.py +++ b/.agents/skills/monitor-ci-results/scripts/monitor_remote_ci.py @@ -199,9 +199,66 @@ def main(): with open(state_file, "w") as f: json.dump(monitored, f) + # Check if all CI checks and jobs are finished and passed + if checks: + all_completed = True + any_failed = False + total_passed = 0 + + for check in checks: + state = check.get("state", "UNKNOWN") + name = check.get("name", "") + link = check.get("link", "") + if "buildkite" in name.lower() and link: + jobs = get_buildkite_jobs(link) + if not jobs: + all_completed = False + for job in jobs: + jstate = job.get("state", "unknown") + exit_status = job.get("exit_status") + is_soft_failed = job.get("soft_failed") is True + is_failed = ( + jstate in ["failed", "failing"] + or (exit_status != 0 and exit_status is not None) + ) and not is_soft_failed + is_passed = ( + jstate in ["passed", "success"] + or (jstate == "finished" and exit_status == 0) + or is_soft_failed + ) + if is_failed: + any_failed = True + elif is_passed: + total_passed += 1 + else: + all_completed = False + else: + if state in ["SUCCESS", "success"]: + total_passed += 1 + elif state in ["FAILURE", "failed"]: + any_failed = True + elif state in ["PENDING", "IN_PROGRESS", "queued", "running"]: + all_completed = False + + if all_completed and not any_failed and total_passed > 0: + print( + f"✅ All {total_passed} remote CI checks for PR #{args.pr} PASSED!" + ) + msg = f"✅ Remote CI checks for PR #{args.pr} completed successfully! ({total_passed} checks passed)" + subprocess.run( + [ + "agentapi", + "send-message", + "--title=CI Checks Passed", + args.conv_id, + msg, + ] + ) + break + time.sleep(args.interval) - print("🏁 CI monitoring service completed its scheduled iterations.") + print("🏁 CI monitoring service completed.") if __name__ == "__main__":