return resp
+def _request_put_file(url, file_to_upload, parameters=None):
+ """Execute a request put, return the resp."""
+ resp = {}
+ try:
+ upload_file = open(file_to_upload, 'rb')
+ except FileNotFoundError:
+ raise FileNotFoundError(
+ errno.ENOENT, os.strerror(errno.ENOENT), file_to_upload)
+
+ files = {'file': upload_file}
+ try:
+ if parameters:
+ resp = requests.put(url, data=parameters, files=files)
+ else:
+ resp = requests.put(url, data=upload_file.read())
+ except requests.exceptions.MissingSchema:
+ raise requests.HTTPError("Not valid URL: {}".format(url))
+ except requests.exceptions.ConnectionError:
+ raise requests.HTTPError("Could not connect to URL: {}".format(url))
+ except requests.exceptions.InvalidURL:
+ raise requests.HTTPError("Invalid URL: {}".format(url))
+
+ if resp.status_code == 400:
+ raise requests.HTTPError("Repository is read only")
+ elif resp.status_code == 404:
+ raise requests.HTTPError("Did not find repository.")
+
+ if not str(resp.status_code).startswith('20'):
+ raise requests.HTTPError("Failed to upload to Nexus with status code: {}.\n{}\n{}".format(
+ resp.status_code, resp.text, file_to_upload))
+
+ return resp
+
+
def _get_node_from_xml(xml_data, tag_name):
"""Extract tag data from xml data."""
log.debug('xml={}'.format(xml_data))
tests/fixtures/deploy/zip-test-files
"""
def _deploy_nexus_upload(file):
- """Fix file path, and call _request_post_file."""
+ """Fix file path, and call _request_put_file."""
nexus_url_with_file = '{}/{}'.format(_format_url(nexus_repo_url), file)
log.info('Uploading {}'.format(file))
- _request_post_file(nexus_url_with_file, file)
+ _request_put_file(nexus_url_with_file, file)
log.debug('Uploaded {}'.format(file))
file_list = []
"""Test deploy_nexus with snapshot.
This test will send a directory of files to deploy_nexus, which should
- call requests.post once for every valid (=3) file.
+ call requests.put once for every valid (=3) file.
There are two files that should not be uploaded.
"""
os.chdir(str(datafiles))
'maven-metadata.xml.sha1']
for file in test_files:
success_upload_url = '{}/{}'.format(nexus_url, file)
- responses.add(responses.POST, success_upload_url,
+ responses.add(responses.PUT, success_upload_url,
status=201)
deploy_sys.deploy_nexus(nexus_url, deploy_dir, snapshot)
"""Test deploy_nexus with no snapshot.
This test will send a directory of files to deploy_nexus, which should
- call requests.post once for every valid (=3) file.
+ call requests.put once for every valid (=3) file.
There are six files that should not be uploaded, and three that should.
"""
os.chdir(str(datafiles))
'4.0.3-SNAPSHOT/odlparent-lite-4.0.3-20181120.113136-1.pom.md5']
for file in test_files:
success_upload_url = '{}/{}'.format(nexus_url, file)
- responses.add(responses.POST, success_upload_url,
+ responses.add(responses.PUT, success_upload_url,
status=201)
deploy_sys.deploy_nexus(nexus_url, deploy_dir)
@pytest.mark.datafiles(
os.path.join(FIXTURE_DIR, 'deploy'),
)
-def test_nexus_deploy_stage(datafiles, responses):
- """Test nexus_deploy_stage."""
+def test_deploy_nexus_stage(datafiles, responses):
+ """Test deploy_nexus_stage."""
url='http://valid.deploy.stage'
url_repo = 'service/local/staging/profiles'
staging_profile_id='93fb68073c18'
'4.0.3-SNAPSHOT/odlparent-lite-4.0.3-20181120.113136-1.pom.md5']
for file in test_files:
success_upload_url = '{}/{}'.format(nexus_deploy_url, file)
- responses.add(responses.POST, success_upload_url,
+ responses.add(responses.PUT, success_upload_url,
status=201)
#Setup for nexus_stage_repo_close