33import requests
44import subprocess
55import tempfile
6+ import pytest
67
78from urllib .parse import urljoin
89
9- from pulp_smash .pulp3 .bindings import monitor_task , tasks as task_api
10+ from pulp_smash .pulp3 .bindings import monitor_task
1011from pulp_smash .pulp3 .utils import get_added_content_summary , get_content_summary
1112from pulp_python .tests .functional .constants import (
1213 PYTHON_CONTENT_NAME ,
4041PYPI_HOST = urljoin (HOST , PULP_PYPI_BASE_URL )
4142
4243
44+ @pytest .fixture
45+ def python_empty_repo_distro (python_repo_factory , python_distribution_factory ):
46+ """Returns an empty repo with and distribution serving it."""
47+ def _generate_empty_repo_distro (repo_body = None , distro_body = None ):
48+ repo_body = repo_body or {}
49+ distro_body = distro_body or {}
50+ repo = python_repo_factory (** repo_body )
51+ distro = python_distribution_factory (repository = repo , ** distro_body )
52+ return repo , distro
53+
54+ yield _generate_empty_repo_distro
55+
56+
57+ @pytest .fixture (scope = "module" )
58+ def python_package_dist_directory (tmp_path_factory , http_get ):
59+ """Creates a temp dir to hold package distros for uploading."""
60+ dist_dir = tmp_path_factory .mktemp ("dist" )
61+ egg_file = dist_dir / PYTHON_EGG_FILENAME
62+ wheel_file = dist_dir / PYTHON_WHEEL_FILENAME
63+ with open (egg_file , "wb" ) as f :
64+ f .write (http_get (PYTHON_EGG_URL ))
65+ with open (wheel_file , "wb" ) as f :
66+ f .write (http_get (PYTHON_WHEEL_URL ))
67+ yield dist_dir , egg_file , wheel_file
68+
69+
4370class PyPISummaryTestCase (TestCaseUsingBindings , TestHelpersMixin ):
4471 """Tests the summary response of the base url of an index."""
4572
@@ -162,18 +189,51 @@ def test_package_upload_simple(self):
162189 content = get_added_content_summary (repo , f"{ repo .versions_href } 1/" )
163190 self .assertDictEqual ({PYTHON_CONTENT_NAME : 1 }, content )
164191
165- def test_twine_upload (self ):
166- """Tests that packages can be properly uploaded through Twine."""
167- repo , distro = self ._create_empty_repo_and_distribution ()
168- url = urljoin (PYPI_HOST , distro .base_path + "/legacy/" )
169- username , password = "admin" , "password"
192+
193+ @pytest .mark .parallel
194+ def test_twine_upload (
195+ pulpcore_bindings ,
196+ python_content_summary ,
197+ python_empty_repo_distro ,
198+ python_package_dist_directory ,
199+ monitor_task ,
200+ ):
201+ """Tests that packages can be properly uploaded through Twine."""
202+ repo , distro = python_empty_repo_distro ()
203+ url = urljoin (distro .base_url , "legacy/" )
204+ dist_dir , _ , _ = python_package_dist_directory
205+ username , password = "admin" , "password"
206+ subprocess .run (
207+ (
208+ "twine" ,
209+ "upload" ,
210+ "--repository-url" ,
211+ url ,
212+ dist_dir / "*" ,
213+ "-u" ,
214+ username ,
215+ "-p" ,
216+ password ,
217+ ),
218+ capture_output = True ,
219+ check = True ,
220+ )
221+ tasks = pulpcore_bindings .TasksApi .list (reserved_resources = repo .pulp_href ).results
222+ for task in reversed (tasks ):
223+ t = monitor_task (task .pulp_href )
224+ repo_ver_href = t .created_resources [- 1 ]
225+ summary = python_content_summary (repository_version = repo_ver_href )
226+ assert summary .present ["python.python" ]["count" ] == 2
227+
228+ # Test re-uploading same packages gives error
229+ with pytest .raises (subprocess .CalledProcessError ):
170230 subprocess .run (
171231 (
172232 "twine" ,
173233 "upload" ,
174234 "--repository-url" ,
175235 url ,
176- self . dists_dir . name + "/ *" ,
236+ dist_dir / " *" ,
177237 "-u" ,
178238 username ,
179239 "-p" ,
@@ -182,50 +242,26 @@ def test_twine_upload(self):
182242 capture_output = True ,
183243 check = True ,
184244 )
185- tasks = task_api .list (reserved_resources_record = [repo .pulp_href ]).results
186- for task in reversed (tasks ):
187- t = monitor_task (task .pulp_href )
188- repo_ver_href = t .created_resources [- 1 ]
189- content = get_content_summary (repo , f"{ repo_ver_href } " )
190- self .assertDictEqual ({PYTHON_CONTENT_NAME : 2 }, content )
191245
192- # Test re-uploading same packages gives error
193- with self .assertRaises (subprocess .CalledProcessError ):
194- subprocess .run (
195- (
196- "twine" ,
197- "upload" ,
198- "--repository-url" ,
199- url ,
200- self .dists_dir .name + "/*" ,
201- "-u" ,
202- username ,
203- "-p" ,
204- password ,
205- ),
206- capture_output = True ,
207- check = True ,
208- )
209-
210- # Test re-uploading same packages with --skip-existing works
211- output = subprocess .run (
212- (
213- "twine" ,
214- "upload" ,
215- "--repository-url" ,
216- url ,
217- self .dists_dir .name + "/*" ,
218- "-u" ,
219- username ,
220- "-p" ,
221- password ,
222- "--skip-existing" ,
223- ),
224- capture_output = True ,
225- check = True ,
226- text = True
227- )
228- self .assertEqual (output .stdout .count ("Skipping" ), 2 )
246+ # Test re-uploading same packages with --skip-existing works
247+ output = subprocess .run (
248+ (
249+ "twine" ,
250+ "upload" ,
251+ "--repository-url" ,
252+ url ,
253+ dist_dir / "*" ,
254+ "-u" ,
255+ username ,
256+ "-p" ,
257+ password ,
258+ "--skip-existing" ,
259+ ),
260+ capture_output = True ,
261+ check = True ,
262+ text = True
263+ )
264+ assert output .stdout .count ("Skipping" ) == 2
229265
230266
231267class PyPISimpleApi (TestCaseUsingBindings , TestHelpersMixin ):
0 commit comments