| 
 | 1 | +import subprocess  | 
 | 2 | +import sys  | 
 | 3 | +import tempfile  | 
 | 4 | +import unittest  | 
 | 5 | +from pathlib import Path  | 
 | 6 | + | 
 | 7 | +from python import runfiles  | 
 | 8 | + | 
 | 9 | +rfiles = runfiles.Create()  | 
 | 10 | + | 
 | 11 | + | 
 | 12 | +def _relative_rpath(path: str) -> Path:  | 
 | 13 | +    p = (Path("_main") / "tests" / "uv" / "lock" / path).as_posix()  | 
 | 14 | +    rpath = rfiles.Rlocation(p)  | 
 | 15 | +    if not rpath:  | 
 | 16 | +        raise ValueError(f"Could not find file: {p}")  | 
 | 17 | + | 
 | 18 | +    return Path(rpath)  | 
 | 19 | + | 
 | 20 | + | 
 | 21 | +class LockTests(unittest.TestCase):  | 
 | 22 | +    def test_requirements_updating_for_the_first_time(self):  | 
 | 23 | +        # Given  | 
 | 24 | +        copier_path = _relative_rpath("requirements_new_file.update")  | 
 | 25 | + | 
 | 26 | +        # When  | 
 | 27 | +        with tempfile.TemporaryDirectory() as dir:  | 
 | 28 | +            workspace_dir = Path(dir)  | 
 | 29 | +            want_path = workspace_dir / "tests" / "uv" / "lock" / "does_not_exist.txt"  | 
 | 30 | + | 
 | 31 | +            self.assertFalse(  | 
 | 32 | +                want_path.exists(), "The path should not exist after the test"  | 
 | 33 | +            )  | 
 | 34 | +            output = subprocess.run(  | 
 | 35 | +                copier_path,  | 
 | 36 | +                capture_output=True,  | 
 | 37 | +                env={  | 
 | 38 | +                    "BUILD_WORKSPACE_DIRECTORY": f"{workspace_dir}",  | 
 | 39 | +                },  | 
 | 40 | +            )  | 
 | 41 | + | 
 | 42 | +            # Then  | 
 | 43 | +            self.assertEqual(0, output.returncode, output.stderr)  | 
 | 44 | +            self.assertIn(  | 
 | 45 | +                "cp <bazel-sandbox>/tests/uv/lock/requirements_new_file.out",  | 
 | 46 | +                output.stdout.decode("utf-8"),  | 
 | 47 | +            )  | 
 | 48 | +            self.assertTrue(want_path.exists(), "The path should exist after the test")  | 
 | 49 | +            self.assertNotEqual(want_path.read_text(), "")  | 
 | 50 | + | 
 | 51 | +    def test_requirements_updating(self):  | 
 | 52 | +        # Given  | 
 | 53 | +        copier_path = _relative_rpath("requirements.update")  | 
 | 54 | +        existing_file = _relative_rpath("testdata/requirements.txt")  | 
 | 55 | +        want_text = existing_file.read_text()  | 
 | 56 | + | 
 | 57 | +        # When  | 
 | 58 | +        with tempfile.TemporaryDirectory() as dir:  | 
 | 59 | +            workspace_dir = Path(dir)  | 
 | 60 | +            want_path = (  | 
 | 61 | +                workspace_dir  | 
 | 62 | +                / "tests"  | 
 | 63 | +                / "uv"  | 
 | 64 | +                / "lock"  | 
 | 65 | +                / "testdata"  | 
 | 66 | +                / "requirements.txt"  | 
 | 67 | +            )  | 
 | 68 | +            want_path.parent.mkdir(parents=True)  | 
 | 69 | +            want_path.write_text(  | 
 | 70 | +                want_text + "\n\n"  | 
 | 71 | +            )  # Write something else to see that it is restored  | 
 | 72 | + | 
 | 73 | +            output = subprocess.run(  | 
 | 74 | +                copier_path,  | 
 | 75 | +                capture_output=True,  | 
 | 76 | +                env={  | 
 | 77 | +                    "BUILD_WORKSPACE_DIRECTORY": f"{workspace_dir}",  | 
 | 78 | +                },  | 
 | 79 | +            )  | 
 | 80 | + | 
 | 81 | +            # Then  | 
 | 82 | +            self.assertEqual(0, output.returncode)  | 
 | 83 | +            self.assertIn(  | 
 | 84 | +                "cp <bazel-sandbox>/tests/uv/lock/requirements.out",  | 
 | 85 | +                output.stdout.decode("utf-8"),  | 
 | 86 | +            )  | 
 | 87 | +            self.assertEqual(want_path.read_text(), want_text)  | 
 | 88 | + | 
 | 89 | +    def test_requirements_run_on_the_first_time(self):  | 
 | 90 | +        # Given  | 
 | 91 | +        copier_path = _relative_rpath("requirements_new_file.run")  | 
 | 92 | + | 
 | 93 | +        # When  | 
 | 94 | +        with tempfile.TemporaryDirectory() as dir:  | 
 | 95 | +            workspace_dir = Path(dir)  | 
 | 96 | +            want_path = workspace_dir / "tests" / "uv" / "lock" / "does_not_exist.txt"  | 
 | 97 | +            # NOTE @aignas 2025-03-18: right now we require users to have the folder  | 
 | 98 | +            # there already  | 
 | 99 | +            want_path.parent.mkdir(parents=True)  | 
 | 100 | + | 
 | 101 | +            self.assertFalse(  | 
 | 102 | +                want_path.exists(), "The path should not exist after the test"  | 
 | 103 | +            )  | 
 | 104 | +            output = subprocess.run(  | 
 | 105 | +                copier_path,  | 
 | 106 | +                capture_output=True,  | 
 | 107 | +                env={  | 
 | 108 | +                    "BUILD_WORKSPACE_DIRECTORY": f"{workspace_dir}",  | 
 | 109 | +                },  | 
 | 110 | +            )  | 
 | 111 | + | 
 | 112 | +            # Then  | 
 | 113 | +            self.assertEqual(0, output.returncode, output.stderr)  | 
 | 114 | +            self.assertTrue(want_path.exists(), "The path should exist after the test")  | 
 | 115 | +            got_contents = want_path.read_text()  | 
 | 116 | +            self.assertNotEqual(got_contents, "")  | 
 | 117 | +            self.assertIn(  | 
 | 118 | +                got_contents,  | 
 | 119 | +                output.stdout.decode("utf-8"),  | 
 | 120 | +            )  | 
 | 121 | + | 
 | 122 | +    def test_requirements_run(self):  | 
 | 123 | +        # Given  | 
 | 124 | +        copier_path = _relative_rpath("requirements.run")  | 
 | 125 | +        existing_file = _relative_rpath("testdata/requirements.txt")  | 
 | 126 | +        want_text = existing_file.read_text()  | 
 | 127 | + | 
 | 128 | +        # When  | 
 | 129 | +        with tempfile.TemporaryDirectory() as dir:  | 
 | 130 | +            workspace_dir = Path(dir)  | 
 | 131 | +            want_path = (  | 
 | 132 | +                workspace_dir  | 
 | 133 | +                / "tests"  | 
 | 134 | +                / "uv"  | 
 | 135 | +                / "lock"  | 
 | 136 | +                / "testdata"  | 
 | 137 | +                / "requirements.txt"  | 
 | 138 | +            )  | 
 | 139 | + | 
 | 140 | +            want_path.parent.mkdir(parents=True)  | 
 | 141 | +            want_path.write_text(  | 
 | 142 | +                want_text + "\n\n"  | 
 | 143 | +            )  # Write something else to see that it is restored  | 
 | 144 | + | 
 | 145 | +            output = subprocess.run(  | 
 | 146 | +                copier_path,  | 
 | 147 | +                capture_output=True,  | 
 | 148 | +                env={  | 
 | 149 | +                    "BUILD_WORKSPACE_DIRECTORY": f"{workspace_dir}",  | 
 | 150 | +                },  | 
 | 151 | +            )  | 
 | 152 | + | 
 | 153 | +            # Then  | 
 | 154 | +            self.assertEqual(0, output.returncode, output.stderr)  | 
 | 155 | +            self.assertTrue(want_path.exists(), "The path should exist after the test")  | 
 | 156 | +            got_contents = want_path.read_text()  | 
 | 157 | +            self.assertNotEqual(got_contents, "")  | 
 | 158 | +            self.assertIn(  | 
 | 159 | +                got_contents,  | 
 | 160 | +                output.stdout.decode("utf-8"),  | 
 | 161 | +            )  | 
 | 162 | + | 
 | 163 | + | 
 | 164 | +if __name__ == "__main__":  | 
 | 165 | +    unittest.main()  | 
0 commit comments