Internal func _request_post_file function

Added one internal function
_request_post_file : requests.post call to upload a file.

Change-Id: Ie5a3ed3fc7c7c2b553d52b66eb8e7dcd9700eb43
Signed-off-by: Bengt Thuree <bthuree@linuxfoundation.org>

diff --git a/lftools/deploy.py b/lftools/deploy.py
index eb6058e..d64d759 100644 (file)
--- a/lftools/deploy.py
+++ b/lftools/deploy.py
@@ -10,6 +10,7 @@
 ##############################################################################
 """Library of functions for deploying artifacts to Nexus."""
 
+import errno
 import gzip
 import logging
 import os
@@ -85,6 +86,41 @@ def _request_post(url, data, headers):
     return resp
 
 
+def _request_post_file(url, file_to_upload):
+    """Execute a request post, 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:
+        resp = requests.post(url, files=files)
+    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'):
+        if zipfile.is_zipfile(file_to_upload):
+            raise requests.HTTPError("Failed to upload to Nexus with status code: {}.\n{}\n{}".format(
+                resp.status_code, resp.text, zipfile.ZipFile(file_to_upload).infolist()))
+        else:
+            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))
@@ -284,6 +320,13 @@ def deploy_nexus_zip(nexus_url, nexus_repo, nexus_path, zip_file):
                       Maven Ex: org/opendaylight/odlparent
                       Site Ex: org.opendaylight.odlparent
         zip_file:     The zip to deploy. (Ex: /tmp/artifacts.zip)
+
+    Sample:
+    lftools deploy nexus-zip \
+        192.168.1.26:8081/nexus \
+        snapshots \
+        tst_path \
+        tests/fixtures/deploy/zip-test-files/test.zip
     """
     url = '{}/service/local/repositories/{}/content-compressed/{}'.format(
         _format_url(nexus_url),

diff --git a/tests/test_deploy.py b/tests/test_deploy.py
index af7b4c7..d51e6ec 100644 (file)
--- a/tests/test_deploy.py
+++ b/tests/test_deploy.py
@@ -288,6 +288,50 @@ def test__request_post(responses, mocker):
         deploy_sys._request_post(test_url, xml_doc, headers)
     assert 'missing_schema' in str(excinfo.value)
 
+def test__request_post_file(responses, mocker):
+    """Test _request_post_file."""
+
+    zip_file='zip-test-files/test.zip'
+    resp = {}
+    test_url='http://connection.error.test'
+    exception = requests.exceptions.ConnectionError(test_url)
+    responses.add(responses.POST, test_url, body=exception)
+    with pytest.raises(requests.HTTPError) as excinfo:
+        resp = deploy_sys._request_post_file(test_url, zip_file)
+    assert 'Could not connect to URL' in str(excinfo.value)
+
+    test_url='http://invalid.url.test:8081'
+    exception = requests.exceptions.InvalidURL(test_url)
+    responses.add(responses.POST, test_url, body=exception)
+    with pytest.raises(requests.HTTPError) as excinfo:
+        resp = deploy_sys._request_post_file(test_url, zip_file)
+    assert 'Invalid URL' in str(excinfo.value)
+
+    test_url='http://missing.schema.test:8081'
+    exception = requests.exceptions.MissingSchema(test_url)
+    responses.add(responses.POST, test_url, body=exception)
+    with pytest.raises(requests.HTTPError) as excinfo:
+        resp = deploy_sys._request_post_file(test_url, zip_file)
+    assert 'Not valid URL' in str(excinfo.value)
+
+    test_url='http://repository.read.only:8081'
+    responses.add(responses.POST, test_url, body=None, status=400)
+    with pytest.raises(requests.HTTPError) as excinfo:
+        resp = deploy_sys._request_post_file(test_url, zip_file)
+    assert 'Repository is read only' in str(excinfo.value)
+
+    test_url='http://repository.not.found:8081'
+    responses.add(responses.POST, test_url, body=None, status=404)
+    with pytest.raises(requests.HTTPError) as excinfo:
+        resp = deploy_sys._request_post_file(test_url, zip_file)
+    assert 'Did not find repository' in str(excinfo.value)
+
+    test_url='http://other.upload.error:8081'
+    responses.add(responses.POST, test_url, body=None, status=500)
+    with pytest.raises(requests.HTTPError) as excinfo:
+        resp = deploy_sys._request_post_file(test_url, zip_file)
+    assert 'Failed to upload to Nexus with status code' in str(excinfo.value)
+
 
 def test_nexus_stage_repo_close(responses, mocker):
     """Test nexus_stage_repo_close."""