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