+# -*- coding: utf-8 -*-
# SPDX-License-Identifier: EPL-1.0
##############################################################################
# Copyright (c) 2017 The Linux Foundation and others.
import tempfile
import zipfile
+from defusedxml.minidom import parseString
import glob2 # Switch to glob when Python < 3.5 support is dropped
import requests
try:
resp = requests.post(url, data=data, headers=headers)
except requests.exceptions.MissingSchema:
+ log.debug("in _request_post. MissingSchema")
_log_error_and_exit("Not valid URL: {}".format(url))
except requests.exceptions.ConnectionError:
+ log.debug("in _request_post. ConnectionError")
_log_error_and_exit("Could not connect to URL: {}".format(url))
except requests.exceptions.InvalidURL:
+ log.debug("in _request_post. InvalidURL")
_log_error_and_exit("Invalid URL: {}".format(url))
return resp
+def _get_node_from_xml(xml_data, tag_name):
+ """Extract tag data from xml data."""
+ log.debug('xml={}'.format(xml_data))
+
+ try:
+ dom1 = parseString(xml_data)
+ childnode = dom1.getElementsByTagName(tag_name)[0]
+ except:
+ _log_error_and_exit("Received bad XML, can not find tag {}".format(tag_name), xml_data)
+ return childnode.firstChild.data
+
+
def copy_archives(workspace, pattern=None):
"""Copy files matching PATTERN in a WORKSPACE to the current directory.
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, zipfile.ZipFile(zip_file).infolist()))
+
+
+def nexus_stage_repo_create(nexus_url, staging_profile_id):
+ """Create a Nexus staging repo.
+
+ Parameters:
+ nexus_url: URL to Nexus server. (Ex: https://nexus.example.org)
+ staging_profile_id: The staging profile id as defined in Nexus for the
+ staging repo.
+
+ Returns: staging_repo_id
+
+ Sample:
+ lftools deploy nexus-stage-repo-create 192.168.1.26:8081/nexsus/ 93fb68073c18
+ """
+ nexus_url = '{0}/service/local/staging/profiles/{1}/start'.format(
+ _format_url(nexus_url),
+ staging_profile_id)
+
+ log.debug("Nexus URL = {}".format(nexus_url))
+
+ xml = """
+ <promoteRequest>
+ <data>
+ <description>Create staging repository.</description>
+ </data>
+ </promoteRequest>
+ """
+
+ headers = {'Content-Type': 'application/xml'}
+ resp = _request_post(nexus_url, xml, headers)
+
+ log.debug("resp.status_code = {}".format(resp.status_code))
+ log.debug("resp.text = {}".format(resp.text))
+
+ if re.search('nexus-error', resp.text):
+ error_msg = _get_node_from_xml(resp.text, 'msg')
+ if re.search('.*profile with id:.*does not exist.', error_msg):
+ _log_error_and_exit("Staging profile id {} not found.".format(staging_profile_id))
+ _log_error_and_exit(error_msg)
+
+ if resp.status_code == 405:
+ _log_error_and_exit("HTTP method POST is not supported by this URL", nexus_url)
+ if resp.status_code == 404:
+ _log_error_and_exit("Did not find nexus site: {}".format(nexus_url))
+ if not resp.status_code == 201:
+ _log_error_and_exit("Failed with status code {}".format(resp.status_code), resp.text)
+
+ staging_repo_id = _get_node_from_xml(resp.text, 'stagedRepositoryId')
+ log.debug("staging_repo_id = {}".format(staging_repo_id))
+
+ return staging_repo_id
+
+
+def nexus_stage_repo_close(nexus_url, staging_profile_id, staging_repo_id):
+ """Close a Nexus staging repo.
+
+ Parameters:
+ nexus_url: URL to Nexus server. (Ex: https://nexus.example.org)
+ staging_profile_id: The staging profile id as defined in Nexus for the
+ staging repo.
+ staging_repo_id: The ID of the repo to close.
+
+ Sample:
+ lftools deploy nexus-stage-repo-close 192.168.1.26:8081/nexsus/ 93fb68073c18 test1-1031
+ """
+ nexus_url = '{0}/service/local/staging/profiles/{1}/finish'.format(
+ _format_url(nexus_url),
+ staging_profile_id)
+
+ log.debug("Nexus URL = {}".format(nexus_url))
+ log.debug("staging_repo_id = {}".format(staging_repo_id))
+
+ xml = """
+ <promoteRequest>
+ <data>
+ <stagedRepositoryId>{0}</stagedRepositoryId>
+ <description>Close staging repository.</description>
+ </data>
+ </promoteRequest>
+ """.format(staging_repo_id)
+
+ headers = {'Content-Type': 'application/xml'}
+ resp = _request_post(nexus_url, xml, headers)
+
+ log.debug("resp.status_code = {}".format(resp.status_code))
+ log.debug("resp.text = {}".format(resp.text))
+
+ if re.search('nexus-error', resp.text):
+ error_msg = _get_node_from_xml(resp.text, 'msg')
+ else:
+ error_msg = resp.text
+
+ if resp.status_code == 404:
+ _log_error_and_exit("Did not find nexus site: {}".format(nexus_url))
+
+ if re.search('invalid state: closed', error_msg):
+ _log_error_and_exit("Staging repository is already closed.")
+ if re.search('Missing staging repository:', error_msg):
+ _log_error_and_exit("Staging repository do not exist.")
+
+ if not resp.status_code == 201:
+ _log_error_and_exit("Failed with status code {}".format(resp.status_code), resp.text)
from lftools import cli
import lftools.deploy as deploy_sys
+
FIXTURE_DIR = os.path.join(
os.path.dirname(os.path.realpath(__file__)),
'fixtures',
)
+def test_log_and_exit():
+ """Test exit."""
+ with pytest.raises(SystemExit) as excinfo:
+ deploy_sys._log_error_and_exit("testmsg")
+ assert excinfo.type == SystemExit
+
+
def test_format_url():
"""Test url format."""
test_url=[["192.168.1.1", "http://192.168.1.1"],
assert os.path.exists(os.path.join(stage_dir, 'test.log'))
+
@pytest.mark.datafiles(
os.path.join(FIXTURE_DIR, 'deploy'),
)
stage_dir, 'aaa', 'aaa-cert', 'target', 'surefire-reports',
'org.opendaylight.aaa.cert.test.AaaCertMdsalProviderTest-output.txt'))
+
@pytest.mark.datafiles(
os.path.join(FIXTURE_DIR, 'deploy'),
)
obj={})
assert result.exit_code == 0
+
@pytest.mark.datafiles(
os.path.join(FIXTURE_DIR, 'deploy'),
)
obj={})
assert result.exit_code == 1
-def mocked_log_error(msg1=None, msg2=None):
- """Mock local_log_error_and_exit function.
+
+def test_get_node_from_xml():
+ """Test extracting from xml."""
+ document = """\
+ <slideshow>
+ <title>Demo slideshow</title>
+ <slide><title>Slide title</title>
+ <point>This is a demo</point>
+ <point>Of a program for processing slides</point>
+ </slide>
+
+ <slide><title>Another demo slide</title><stagedRepositoryId>432</stagedRepositoryId><point>It is important</point>
+ <point>To have more than</point>
+ <point>one slide</point>
+ </slide>
+ </slideshow>
+ """
+ assert deploy_sys._get_node_from_xml(document, 'stagedRepositoryId') == '432'
+ with pytest.raises(SystemExit) as excinfo:
+ deploy_sys._get_node_from_xml(document, 'NotFoundTag')
+ assert excinfo.type == SystemExit
+
+
+def mocked_log_error(*msg_list):
+ """Mock _log_error_and_exit function.
This function is modified to simply raise an Exception.
The original will print msg1 & msg2, then call sys.exit(1)."""
+ msg1=msg_list[0]
if 'Could not connect to URL:' in msg1:
raise ValueError('connection_error')
if 'Invalid URL:' in msg1:
raise ValueError('invalid_url')
if 'Not valid URL:' in msg1:
raise ValueError('missing_schema')
+ if "profile with id 'INVALID' does not exist" in msg1:
+ raise ValueError('profile.id.not.exist')
+ if "OTHER create error" in msg1:
+ raise ValueError('other.create.error')
+ if "HTTP method POST is not supported by this URL" in msg1:
+ raise ValueError('post.not.supported')
+ if "Did not find nexus site" in msg1:
+ raise ValueError('site.not.found')
+ if "Failed with status code " in msg1:
+ raise ValueError('other.error.occured')
+ if "Staging repository do not exist." in msg1:
+ raise ValueError('missing.staging.repository')
+ if "Staging repository is already closed." in msg1:
+ raise ValueError('staging.already.closed')
raise ValueError('fail')
-def mocked_requests_post(*args, **kwargs):
- """Mock requests.post function."""
- class MockResponse:
- def __init__(self, json_data, status_code):
- self.json_data = json_data
- self.status_code = status_code
- self.text = json_data
+def test__request_post(responses, mocker):
+ """Test _request_post."""
+ mocker.patch('lftools.deploy._log_error_and_exit', side_effect=mocked_log_error)
+ xml_doc="""
+ <promoteRequest><data>
+ <stagedRepositoryId>test1-1027</stagedRepositoryId>
+ <description>Close staging repository.</description>
+ </data></promoteRequest>
+ """
+ headers = {'Content-Type': 'application/xml'}
+
+ test_url='http://connection.error.test'
+ exception = requests.exceptions.ConnectionError(test_url)
+ responses.add(responses.POST, test_url, body=exception)
+ with pytest.raises(ValueError) as excinfo:
+ deploy_sys._request_post(test_url, xml_doc, headers)
+ assert 'connection_error' in str(excinfo.value)
- def json(self):
- return self.json_data
+ test_url='http://invalid.url.test:8081'
+ exception = requests.exceptions.InvalidURL(test_url)
+ responses.add(responses.POST, test_url, body=exception)
+ with pytest.raises(ValueError) as excinfo:
+ deploy_sys._request_post(test_url, xml_doc, headers)
+ assert 'invalid_url' in str(excinfo.value)
- if 'connection.error.test' in args[0]:
- raise requests.exceptions.ConnectionError
- if 'invalid.url.test' in args[0]:
- raise requests.exceptions.InvalidURL
- if 'missing.schema.test' in args[0]:
- raise requests.exceptions.MissingSchema
- return MockResponse(None, 404)
+ test_url='http://missing.schema.test:8081'
+ exception = requests.exceptions.MissingSchema(test_url)
+ responses.add(responses.POST, test_url, body=exception)
+ with pytest.raises(ValueError) as excinfo:
+ deploy_sys._request_post(test_url, xml_doc, headers)
+ assert 'missing_schema' in str(excinfo.value)
-def test_local_request_post(mocker):
- """Test local_request_post."""
- mocker.patch('requests.post', side_effect=mocked_requests_post)
+def test_nexus_stage_repo_close(responses, mocker):
+ """Test nexus_stage_repo_close."""
mocker.patch('lftools.deploy._log_error_and_exit', side_effect=mocked_log_error)
+ url='service/local/staging/profiles'
+
+ responses.add(responses.POST, 'http://valid.create.post/{}/{}/finish'.format(url, '93fb68073c18' ),
+ body=None, status=201)
+ deploy_sys.nexus_stage_repo_close('valid.create.post', '93fb68073c18', 'test1-1027')
+
+ xml_site_not_found = """
+ <html><head><title>404 - Site Not Found</title></head>
+ <body><h1>404 - Site not found</h1></body>
+ </html>
+ """
+ responses.add(responses.POST, 'http://site.not.found/{}/{}/finish'.format(url, 'INVALID'),
+ body=xml_site_not_found, status=404)
+ with pytest.raises(ValueError) as excinfo:
+ deploy_sys.nexus_stage_repo_close('site.not.found', 'INVALID', 'test1-1027')
+ assert 'site.not.found' in str(excinfo.value)
+
+ xml_missing_staging_repository = """
+ <nexus-error><errors><error>
+ <id>*</id>
+ <msg>Unhandled: Missing staging repository: test1-1</msg>
+ </error></errors></nexus-error>
+ """
+ responses.add(responses.POST, 'http://missing.staging.repository/{}/{}/finish'.format(url, 'INVALID'),
+ body=xml_missing_staging_repository, status=500)
+ with pytest.raises(ValueError) as excinfo:
+ deploy_sys.nexus_stage_repo_close('missing.staging.repository', 'INVALID', 'test1-1027')
+ assert 'missing.staging.repository' in str(excinfo.value)
+
+ xml_staging_already_closed = """
+ <nexus-error><errors><error>
+ <id>*</id>
+ <msg>Unhandled: Repository: test1-1000 has invalid state: closed</msg>
+ </error></errors></nexus-error>
+ """
+ responses.add(responses.POST, 'http://staging.already.closed/{}/{}/finish'.format(url, 'INVALID'),
+ body=xml_staging_already_closed, status=500)
+ with pytest.raises(ValueError) as excinfo:
+ deploy_sys.nexus_stage_repo_close('staging.already.closed', 'INVALID', 'test1-1027')
+ assert 'staging.already.closed' in str(excinfo.value)
+
+ xml_other_error_occured = """
+ <html><head><title>303 - See Other</title></head>
+ <body><h1>303 - See Other</h1></body>
+ </html>
+ """
+ responses.add(responses.POST, 'http://other.error.occured/{}/{}/finish'.format(url, 'INVALID'),
+ body=xml_other_error_occured, status=303)
+ with pytest.raises(ValueError) as excinfo:
+ deploy_sys.nexus_stage_repo_close('other.error.occured', 'INVALID', 'test1-1027')
+ assert 'other.error.occured' in str(excinfo.value)
- xml_doc='''
- <promoteRequest>
- <data>
- <stagedRepositoryId>test1-1027</stagedRepositoryId>
- <description>Close staging repository.</description>
- </data>
- </promoteRequest>
- '''
+
+def test_nexus_stage_repo_create(responses, mocker):
+ """Test nexus_stage_repo_create."""
+ mocker.patch('lftools.deploy._log_error_and_exit', side_effect=mocked_log_error)
+ url = 'service/local/staging/profiles'
+
+ xml_created = "<stagedRepositoryId>test1-1030</stagedRepositoryId>"
+ responses.add(responses.POST, 'http://valid.create.post/{}/{}/start'.format(url, '93fb68073c18' ),
+ body=xml_created, status=201)
+ res = deploy_sys.nexus_stage_repo_create('valid.create.post', '93fb68073c18')
+ assert res == 'test1-1030'
+
+ xml_profile_id_dont_exist = """
+ <nexus-error><errors><error>
+ <id>*</id>
+ <msg>Cannot create Staging Repository, profile with id 'INVALID' does not exist.</msg>
+ </error></errors></nexus-error>
+ """
+ responses.add(responses.POST, 'http://profile.id_not.exist/{}/{}/start'.format(url, 'INVALID' ),
+ body=xml_profile_id_dont_exist, status=404)
with pytest.raises(ValueError) as excinfo:
- deploy_sys._request_post('connection.error.test', xml_doc, "{'Content-Type': 'application/xml'}")
- assert 'connection_error' in str(excinfo.value)
+ res = deploy_sys.nexus_stage_repo_create('profile.id_not.exist', 'INVALID')
+ assert 'profile.id.not.exist' in str(excinfo.value)
+
+ xml_other_create_error = "<nexus-error><errors><error><id>*</id><msg>OTHER create error.</msg></error></errors></nexus-error>"
+ responses.add(responses.POST, 'http://other.create.error/{}/{}/start'.format(url, 'INVALID' ),
+ body=xml_other_create_error, status=404)
with pytest.raises(ValueError) as excinfo:
- deploy_sys._request_post('invalid.url.test:8081nexus', xml_doc, "{'Content-Type': 'application/xml'}")
- assert 'invalid_url' in str(excinfo.value)
+ res = deploy_sys.nexus_stage_repo_create('other.create.error', 'INVALID')
+ assert 'other.create.error' in str(excinfo.value)
+
+ xml_other_error_occured = """
+ <html>
+ <head><title>303 - See Other</title></head>
+ <body><h1>303 - See Other</h1></body>
+ </html>
+ """
+ responses.add(responses.POST, 'http://other.error.occured/{}/{}/start'.format(url, 'INVALID' ),
+ body=xml_other_error_occured, status=303)
with pytest.raises(ValueError) as excinfo:
- deploy_sys._request_post('http:/missing.schema.test:8081nexus', xml_doc, "{'Content-Type': 'application/xml'}")
- assert 'missing_schema' in str(excinfo.value)
+ res = deploy_sys.nexus_stage_repo_create('other.error.occured', 'INVALID')
+ assert 'other.error.occured' in str(excinfo.value)
+
+ xml_post_not_supported = """
+ <html>
+ <head>
+ <title>405 - HTTP method POST is not supported by this URL</title>
+ <meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>
+ <link rel="icon" type="image/png" href="http://192.168.1.26:8081/nexus/favicon.png">
+ <!--[if IE]>
+ <link rel="SHORTCUT ICON" href="http://192.168.1.26:8081/nexus/favicon.ico"/>
+ <![endif]-->
+ <link rel="stylesheet" href="http://192.168.1.26:8081/nexus/static/css/Sonatype-content.css?2.14.10-01" type="text/css" media="screen" title="no title" charset="utf-8">
+ </head>
+ <body>
+ <h1>405 - HTTP method POST is not supported by this URL</h1>
+ <p>HTTP method POST is not supported by this URL</p>
+ </body>
+ </html>
+ """
+ responses.add(responses.POST, 'http://post.not.supported/{}/{}/start'.format(url, 'INVALID' ),
+ body=xml_post_not_supported, status=405)
+ with pytest.raises(ValueError) as excinfo:
+ res = deploy_sys.nexus_stage_repo_create('post.not.supported', 'INVALID')
+ assert 'post.not.supported' in str(excinfo.value)
+
+ xml_site_not_found = """
+ <html><head><title>404 - Site Not Found</title></head>
+ <body><h1>404 - Site not found</h1></body>
+ </html>
+ """
+ responses.add(responses.POST, 'http://site.not.found/{}/{}/start'.format(url, 'INVALID' ),
+ body=xml_site_not_found, status=404)
+ with pytest.raises(ValueError) as excinfo:
+ res = deploy_sys.nexus_stage_repo_create('site.not.found', 'INVALID')
+ assert 'site.not.found' in str(excinfo.value)
Code Review / releng / lftools.git / commitdiff