| 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=79a4e9c1838c0631d5d8fa49a26efd6e9a364f6b38d9597c0f6df112271a0e28">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() |
