xref: /aosp_15_r20/external/bazelbuild-rules_python/examples/wheel/test_publish.py (revision 60517a1edbc8ecf509223e9af94a7adec7d736b8) 
1*60517a1eSAndroid Build Coastguard Workerimport os
2*60517a1eSAndroid Build Coastguard Workerimport socket
3*60517a1eSAndroid Build Coastguard Workerimport subprocess
4*60517a1eSAndroid Build Coastguard Workerimport textwrap
5*60517a1eSAndroid Build Coastguard Workerimport time
6*60517a1eSAndroid Build Coastguard Workerimport unittest
7*60517a1eSAndroid Build Coastguard Workerfrom contextlib import closing
8*60517a1eSAndroid Build Coastguard Workerfrom pathlib import Path
9*60517a1eSAndroid Build Coastguard Workerfrom urllib.request import urlopen
10*60517a1eSAndroid Build Coastguard Worker
11*60517a1eSAndroid Build Coastguard Worker
12*60517a1eSAndroid Build Coastguard Workerdef find_free_port():
13*60517a1eSAndroid Build Coastguard Worker    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
14*60517a1eSAndroid Build Coastguard Worker        s.bind(("", 0))
15*60517a1eSAndroid Build Coastguard Worker        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
16*60517a1eSAndroid Build Coastguard Worker        return s.getsockname()[1]
17*60517a1eSAndroid Build Coastguard Worker
18*60517a1eSAndroid Build Coastguard Worker
19*60517a1eSAndroid Build Coastguard Workerclass TestTwineUpload(unittest.TestCase):
20*60517a1eSAndroid Build Coastguard Worker    def setUp(self):
21*60517a1eSAndroid Build Coastguard Worker        self.maxDiff = 1000
22*60517a1eSAndroid Build Coastguard Worker        self.port = find_free_port()
23*60517a1eSAndroid Build Coastguard Worker        self.url = f"http://localhost:{self.port}"
24*60517a1eSAndroid Build Coastguard Worker        self.dir = Path(os.environ["TEST_TMPDIR"])
25*60517a1eSAndroid Build Coastguard Worker
26*60517a1eSAndroid Build Coastguard Worker        self.log_file = self.dir / "pypiserver-log.txt"
27*60517a1eSAndroid Build Coastguard Worker        self.log_file.touch()
28*60517a1eSAndroid Build Coastguard Worker        _storage_dir = self.dir / "data"
29*60517a1eSAndroid Build Coastguard Worker        for d in [_storage_dir]:
30*60517a1eSAndroid Build Coastguard Worker            d.mkdir(exist_ok=True)
31*60517a1eSAndroid Build Coastguard Worker
32*60517a1eSAndroid Build Coastguard Worker        print("Starting PyPI server...")
33*60517a1eSAndroid Build Coastguard Worker        self._server = subprocess.Popen(
34*60517a1eSAndroid Build Coastguard Worker            [
35*60517a1eSAndroid Build Coastguard Worker                str(Path(os.environ["SERVER_PATH"])),
36*60517a1eSAndroid Build Coastguard Worker                "run",
37*60517a1eSAndroid Build Coastguard Worker                "--verbose",
38*60517a1eSAndroid Build Coastguard Worker                "--log-file",
39*60517a1eSAndroid Build Coastguard Worker                str(self.log_file),
40*60517a1eSAndroid Build Coastguard Worker                "--host",
41*60517a1eSAndroid Build Coastguard Worker                "localhost",
42*60517a1eSAndroid Build Coastguard Worker                "--port",
43*60517a1eSAndroid Build Coastguard Worker                str(self.port),
44*60517a1eSAndroid Build Coastguard Worker                # Allow unauthenticated access
45*60517a1eSAndroid Build Coastguard Worker                "--authenticate",
46*60517a1eSAndroid Build Coastguard Worker                ".",
47*60517a1eSAndroid Build Coastguard Worker                "--passwords",
48*60517a1eSAndroid Build Coastguard Worker                ".",
49*60517a1eSAndroid Build Coastguard Worker                str(_storage_dir),
50*60517a1eSAndroid Build Coastguard Worker            ],
51*60517a1eSAndroid Build Coastguard Worker        )
52*60517a1eSAndroid Build Coastguard Worker
53*60517a1eSAndroid Build Coastguard Worker        line = "Hit Ctrl-C to quit"
54*60517a1eSAndroid Build Coastguard Worker        interval = 0.1
55*60517a1eSAndroid Build Coastguard Worker        wait_seconds = 40
56*60517a1eSAndroid Build Coastguard Worker        for _ in range(int(wait_seconds / interval)):  # 40 second timeout
57*60517a1eSAndroid Build Coastguard Worker            current_logs = self.log_file.read_text()
58*60517a1eSAndroid Build Coastguard Worker            if line in current_logs:
59*60517a1eSAndroid Build Coastguard Worker                print(current_logs.strip())
60*60517a1eSAndroid Build Coastguard Worker                print("...")
61*60517a1eSAndroid Build Coastguard Worker                break
62*60517a1eSAndroid Build Coastguard Worker
63*60517a1eSAndroid Build Coastguard Worker            time.sleep(0.1)
64*60517a1eSAndroid Build Coastguard Worker        else:
65*60517a1eSAndroid Build Coastguard Worker            raise RuntimeError(
66*60517a1eSAndroid Build Coastguard Worker                f"Could not get the server running fast enough, waited for {wait_seconds}s"
67*60517a1eSAndroid Build Coastguard Worker            )
68*60517a1eSAndroid Build Coastguard Worker
69*60517a1eSAndroid Build Coastguard Worker    def tearDown(self):
70*60517a1eSAndroid Build Coastguard Worker        self._server.terminate()
71*60517a1eSAndroid Build Coastguard Worker        print(f"Stopped PyPI server, all logs:\n{self.log_file.read_text()}")
72*60517a1eSAndroid Build Coastguard Worker
73*60517a1eSAndroid Build Coastguard Worker    def test_upload_and_query_simple_api(self):
74*60517a1eSAndroid Build Coastguard Worker        # Given
75*60517a1eSAndroid Build Coastguard Worker        script_path = Path(os.environ["PUBLISH_PATH"])
76*60517a1eSAndroid Build Coastguard Worker        whl = Path(os.environ["WHEEL_PATH"])
77*60517a1eSAndroid Build Coastguard Worker
78*60517a1eSAndroid Build Coastguard Worker        # When I publish a whl to a package registry
79*60517a1eSAndroid Build Coastguard Worker        subprocess.check_output(
80*60517a1eSAndroid Build Coastguard Worker            [
81*60517a1eSAndroid Build Coastguard Worker                str(script_path),
82*60517a1eSAndroid Build Coastguard Worker                "--no-color",
83*60517a1eSAndroid Build Coastguard Worker                "upload",
84*60517a1eSAndroid Build Coastguard Worker                str(whl),
85*60517a1eSAndroid Build Coastguard Worker                "--verbose",
86*60517a1eSAndroid Build Coastguard Worker                "--non-interactive",
87*60517a1eSAndroid Build Coastguard Worker                "--disable-progress-bar",
88*60517a1eSAndroid Build Coastguard Worker            ],
89*60517a1eSAndroid Build Coastguard Worker            env={
90*60517a1eSAndroid Build Coastguard Worker                "TWINE_REPOSITORY_URL": self.url,
91*60517a1eSAndroid Build Coastguard Worker                "TWINE_USERNAME": "dummy",
92*60517a1eSAndroid Build Coastguard Worker                "TWINE_PASSWORD": "dummy",
93*60517a1eSAndroid Build Coastguard Worker            },
94*60517a1eSAndroid Build Coastguard Worker        )
95*60517a1eSAndroid Build Coastguard Worker
96*60517a1eSAndroid Build Coastguard Worker        # Then I should be able to get its contents
97*60517a1eSAndroid Build Coastguard Worker        with urlopen(self.url + "/example-minimal-library/") as response:
98*60517a1eSAndroid Build Coastguard Worker            got_content = response.read().decode("utf-8")
99*60517a1eSAndroid Build Coastguard Worker            want_content = """
100*60517a1eSAndroid Build Coastguard Worker<!DOCTYPE html>
101*60517a1eSAndroid Build Coastguard Worker<html>
102*60517a1eSAndroid Build Coastguard Worker    <head>
103*60517a1eSAndroid Build Coastguard Worker        <title>Links for example-minimal-library</title>
104*60517a1eSAndroid Build Coastguard Worker    </head>
105*60517a1eSAndroid Build Coastguard Worker    <body>
106*60517a1eSAndroid Build Coastguard Worker        <h1>Links for example-minimal-library</h1>
107*60517a1eSAndroid Build Coastguard Worker             <a href="/packages/example_minimal_library-0.0.1-py3-none-any.whl#sha256=79a4e9c1838c0631d5d8fa49a26efd6e9a364f6b38d9597c0f6df112271a0e28">example_minimal_library-0.0.1-py3-none-any.whl</a><br>
108*60517a1eSAndroid Build Coastguard Worker    </body>
109*60517a1eSAndroid Build Coastguard Worker</html>"""
110*60517a1eSAndroid Build Coastguard Worker            self.assertEqual(
111*60517a1eSAndroid Build Coastguard Worker                textwrap.dedent(want_content).strip(),
112*60517a1eSAndroid Build Coastguard Worker                textwrap.dedent(got_content).strip(),
113*60517a1eSAndroid Build Coastguard Worker            )
114*60517a1eSAndroid Build Coastguard Worker
115*60517a1eSAndroid Build Coastguard Worker
116*60517a1eSAndroid Build Coastguard Workerif __name__ == "__main__":
117*60517a1eSAndroid Build Coastguard Worker    unittest.main()
118