Fix AttributeError due to cgi.escape removal in Python 3.8. (#66040)
* Fix cgi.escape AttributeError due to cgi.escape removal in Python 3.8. * Fix linter errors.
This commit is contained in:
Karthikeyan Singaravelan
ansibot
parent
d3427bb860
commit
891c759832
2 changed files with 103 additions and 3 deletions
|
|
@ -72,7 +72,14 @@ EXAMPLES = '''
|
||||||
msg: Task completed ... with feeling.
|
msg: Task completed ... with feeling.
|
||||||
'''
|
'''
|
||||||
|
|
||||||
import cgi
|
try:
|
||||||
|
from html import escape as html_escape
|
||||||
|
except ImportError:
|
||||||
|
# Python-3.2 or later
|
||||||
|
import cgi
|
||||||
|
|
||||||
|
def html_escape(text, quote=True):
|
||||||
|
return cgi.escape(text, quote)
|
||||||
|
|
||||||
from ansible.module_utils.basic import AnsibleModule
|
from ansible.module_utils.basic import AnsibleModule
|
||||||
from ansible.module_utils.urls import fetch_url
|
from ansible.module_utils.urls import fetch_url
|
||||||
|
|
@ -126,14 +133,14 @@ def main():
|
||||||
|
|
||||||
# Send some audible notification if requested
|
# Send some audible notification if requested
|
||||||
if notify:
|
if notify:
|
||||||
response, info = fetch_url(module, target_url, data=NSTR % cgi.escape(notify), headers=headers)
|
response, info = fetch_url(module, target_url, data=NSTR % html_escape(notify), headers=headers)
|
||||||
if info['status'] not in [200, 201]:
|
if info['status'] not in [200, 201]:
|
||||||
module.fail_json(msg="unable to send msg: '%s', campfire api"
|
module.fail_json(msg="unable to send msg: '%s', campfire api"
|
||||||
" returned error code: '%s'" %
|
" returned error code: '%s'" %
|
||||||
(notify, info['status']))
|
(notify, info['status']))
|
||||||
|
|
||||||
# Send the message
|
# Send the message
|
||||||
response, info = fetch_url(module, target_url, data=MSTR % cgi.escape(msg), headers=headers)
|
response, info = fetch_url(module, target_url, data=MSTR % html_escape(msg), headers=headers)
|
||||||
if info['status'] not in [200, 201]:
|
if info['status'] not in [200, 201]:
|
||||||
module.fail_json(msg="unable to send msg: '%s', campfire api"
|
module.fail_json(msg="unable to send msg: '%s', campfire api"
|
||||||
" returned error code: '%s'" %
|
" returned error code: '%s'" %
|
||||||
|
|
|
||||||
93
test/units/modules/notification/test_campfire.py
Normal file
93
test/units/modules/notification/test_campfire.py
Normal file
|
|
@ -0,0 +1,93 @@
|
||||||
|
|
||||||
|
from __future__ import (absolute_import, division, print_function)
|
||||||
|
__metaclass__ = type
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from units.compat.mock import patch
|
||||||
|
from ansible.modules.notification import campfire
|
||||||
|
from units.modules.utils import AnsibleExitJson, AnsibleFailJson, ModuleTestCase, set_module_args
|
||||||
|
|
||||||
|
|
||||||
|
class TestCampfireModule(ModuleTestCase):
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
super(TestCampfireModule, self).setUp()
|
||||||
|
self.module = campfire
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
super(TestCampfireModule, self).tearDown()
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def fetch_url_mock(self, mocker):
|
||||||
|
return mocker.patch('ansible.module_utils.notification.campfire.fetch_url')
|
||||||
|
|
||||||
|
def test_without_required_parameters(self):
|
||||||
|
"""Failure must occurs when all parameters are missing"""
|
||||||
|
with self.assertRaises(AnsibleFailJson):
|
||||||
|
set_module_args({})
|
||||||
|
self.module.main()
|
||||||
|
|
||||||
|
def test_successful_message(self):
|
||||||
|
"""Test failure message"""
|
||||||
|
set_module_args({
|
||||||
|
'subscription': 'test',
|
||||||
|
'token': 'abc',
|
||||||
|
'room': 'test',
|
||||||
|
'msg': 'test'
|
||||||
|
})
|
||||||
|
|
||||||
|
with patch.object(campfire, "fetch_url") as fetch_url_mock:
|
||||||
|
fetch_url_mock.return_value = (None, {"status": 200})
|
||||||
|
with self.assertRaises(AnsibleExitJson):
|
||||||
|
self.module.main()
|
||||||
|
|
||||||
|
assert fetch_url_mock.call_count == 1
|
||||||
|
url = fetch_url_mock.call_args[0][1]
|
||||||
|
data = fetch_url_mock.call_args[1]['data']
|
||||||
|
|
||||||
|
assert url == 'https://test.campfirenow.com/room/test/speak.xml'
|
||||||
|
assert data == '<message><body>test</body></message>'
|
||||||
|
|
||||||
|
def test_successful_message_with_notify(self):
|
||||||
|
"""Test failure message"""
|
||||||
|
set_module_args({
|
||||||
|
'subscription': 'test',
|
||||||
|
'token': 'abc',
|
||||||
|
'room': 'test',
|
||||||
|
'msg': 'test',
|
||||||
|
'notify': 'bell'
|
||||||
|
})
|
||||||
|
|
||||||
|
with patch.object(campfire, "fetch_url") as fetch_url_mock:
|
||||||
|
fetch_url_mock.return_value = (None, {"status": 200})
|
||||||
|
with self.assertRaises(AnsibleExitJson):
|
||||||
|
self.module.main()
|
||||||
|
|
||||||
|
assert fetch_url_mock.call_count == 2
|
||||||
|
notify_call = fetch_url_mock.mock_calls[0]
|
||||||
|
url = notify_call[1][1]
|
||||||
|
data = notify_call[2]['data']
|
||||||
|
|
||||||
|
assert url == 'https://test.campfirenow.com/room/test/speak.xml'
|
||||||
|
assert data == '<message><type>SoundMessage</type><body>bell</body></message>'
|
||||||
|
|
||||||
|
message_call = fetch_url_mock.mock_calls[1]
|
||||||
|
url = message_call[1][1]
|
||||||
|
data = message_call[2]['data']
|
||||||
|
|
||||||
|
assert url == 'https://test.campfirenow.com/room/test/speak.xml'
|
||||||
|
assert data == '<message><body>test</body></message>'
|
||||||
|
|
||||||
|
def test_failure_message(self):
|
||||||
|
"""Test failure message"""
|
||||||
|
set_module_args({
|
||||||
|
'subscription': 'test',
|
||||||
|
'token': 'abc',
|
||||||
|
'room': 'test',
|
||||||
|
'msg': 'test'
|
||||||
|
})
|
||||||
|
|
||||||
|
with patch.object(campfire, "fetch_url") as fetch_url_mock:
|
||||||
|
fetch_url_mock.return_value = (None, {"status": 403})
|
||||||
|
with self.assertRaises(AnsibleFailJson):
|
||||||
|
self.module.main()
|
||||||
Loading…
Reference in a new issue