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