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 (scope = "module" )
45+ def python_package_dist_directory (tmp_path_factory , http_get ):
46+ """Creates a temp dir to hold package distros for uploading."""
47+ dist_dir = tmp_path_factory .mktemp ("dist" )
48+ egg_file = dist_dir / PYTHON_EGG_FILENAME
49+ wheel_file = dist_dir / PYTHON_WHEEL_FILENAME
50+ with open (egg_file , "wb" ) as f :
51+ f .write (http_get (PYTHON_EGG_URL ))
52+ with open (wheel_file , "wb" ) as f :
53+ f .write (http_get (PYTHON_WHEEL_URL ))
54+ yield dist_dir , egg_file , wheel_file
55+
56+
4357class PyPISummaryTestCase (TestCaseUsingBindings , TestHelpersMixin ):
4458 """Tests the summary response of the base url of an index."""
4559
@@ -162,18 +176,51 @@ def test_package_upload_simple(self):
162176 content = get_added_content_summary (repo , f"{ repo .versions_href } 1/" )
163177 self .assertDictEqual ({PYTHON_CONTENT_NAME : 1 }, content )
164178
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"
179+
180+ @pytest .mark .parallel
181+ def test_twine_upload (
182+ pulpcore_bindings ,
183+ python_content_summary ,
184+ python_empty_repo_distro ,
185+ python_package_dist_directory ,
186+ monitor_task ,
187+ ):
188+ """Tests that packages can be properly uploaded through Twine."""
189+ repo , distro = python_empty_repo_distro ()
190+ url = urljoin (distro .base_url , "legacy/" )
191+ dist_dir , _ , _ = python_package_dist_directory
192+ username , password = "admin" , "password"
193+ subprocess .run (
194+ (
195+ "twine" ,
196+ "upload" ,
197+ "--repository-url" ,
198+ url ,
199+ dist_dir / "*" ,
200+ "-u" ,
201+ username ,
202+ "-p" ,
203+ password ,
204+ ),
205+ capture_output = True ,
206+ check = True ,
207+ )
208+ tasks = pulpcore_bindings .TasksApi .list (reserved_resources = repo .pulp_href ).results
209+ for task in reversed (tasks ):
210+ t = monitor_task (task .pulp_href )
211+ repo_ver_href = t .created_resources [- 1 ]
212+ summary = python_content_summary (repository_version = repo_ver_href )
213+ assert summary .present ["python.python" ]["count" ] == 2
214+
215+ # Test re-uploading same packages gives error
216+ with pytest .raises (subprocess .CalledProcessError ):
170217 subprocess .run (
171218 (
172219 "twine" ,
173220 "upload" ,
174221 "--repository-url" ,
175222 url ,
176- self . dists_dir . name + "/ *" ,
223+ dist_dir / " *" ,
177224 "-u" ,
178225 username ,
179226 "-p" ,
@@ -182,50 +229,26 @@ def test_twine_upload(self):
182229 capture_output = True ,
183230 check = True ,
184231 )
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 )
191-
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- )
209232
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 )
233+ # Test re-uploading same packages with --skip-existing works
234+ output = subprocess .run (
235+ (
236+ "twine" ,
237+ "upload" ,
238+ "--repository-url" ,
239+ url ,
240+ dist_dir / " *" ,
241+ "-u" ,
242+ username ,
243+ "-p" ,
244+ password ,
245+ "--skip-existing" ,
246+ ),
247+ capture_output = True ,
248+ check = True ,
249+ text = True
250+ )
251+ assert output .stdout .count ("Skipping" ) == 2
229252
230253
231254class PyPISimpleApi (TestCaseUsingBindings , TestHelpersMixin ):
0 commit comments