diff --git a/pyotrs/lib.py b/pyotrs/lib.py
index 5b2d5ea2157daa64af2cdef4cc3f0f8431c98df2..2900b0afbcdb4302cd80ce10bd5066a633540521 100644
--- a/pyotrs/lib.py
+++ b/pyotrs/lib.py
@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals # support both Python2 and 3
+
""" lib.py
PyOTRS lib
@@ -21,6 +22,21 @@ from requests.packages.urllib3 import disable_warnings
# turn of platform insecurity warnings from urllib3
disable_warnings() # TODO 2016-04-23 (RH) verify this
+OPERATION_MAPPING_DEFAULT = {
+ 'SessionCreate':
+ {'ResultType': 'SessionID', 'RequestMethod': 'POST', 'Route': '/Session'},
+ 'TicketCreate':
+ {'ResultType': 'TicketID', 'RequestMethod': 'POST', 'Route': '/Ticket'},
+ 'TicketGet':
+ {'ResultType': 'Ticket', 'RequestMethod': 'GET', 'Route': '/Ticket/:TicketID'},
+ 'TicketGetList':
+ {'ResultType': 'Ticket', 'RequestMethod': 'GET', 'Route': '/TicketList'},
+ 'TicketSearch':
+ {'ResultType': 'TicketID', 'RequestMethod': 'GET', 'Route': '/Ticket'},
+ 'TicketUpdate':
+ {'ResultType': 'TicketID', 'RequestMethod': 'PATCH', 'Route': '/Ticket/:TicketID'}
+}
+
class PyOTRSError(Exception):
def __init__(self, message):
@@ -28,15 +44,15 @@ class PyOTRSError(Exception):
self.message = message
-class ArgumentMissingError(Exception):
+class ArgumentMissingError(PyOTRSError):
pass
-class ArgumentInvalidError(Exception):
+class ArgumentInvalidError(PyOTRSError):
pass
-class ResponseJSONParseError(PyOTRSError):
+class ResponseParseError(PyOTRSError):
pass
@@ -44,15 +60,11 @@ class SessionCreateError(PyOTRSError):
pass
-class TicketError(PyOTRSError):
- pass
-
-
-class OTRSAPIError(PyOTRSError):
+class APIError(PyOTRSError):
pass
-class OTRSHTTPError(PyOTRSError):
+class HTTPError(PyOTRSError):
pass
@@ -192,7 +204,7 @@ class Attachment(object):
Attachment
"""
with open(file_path, 'rb') as f:
- content = f.read()
+ content = f.read().encode('utf-8')
content_type = mimetypes.guess_type(file_path)[0]
if not content_type:
@@ -201,7 +213,7 @@ class Attachment(object):
'ContentType': content_type,
'Filename': os.path.basename(file_path)})
- def save_to_disc(self, folder="/tmp"):
+ def save_to_dir(self, folder="/tmp"):
"""save Attachment to a folder on disc
Args:
folder (unicode):
@@ -238,6 +250,7 @@ class DynamicField(object):
DynamicField representation changed between OTRS 4 and OTRS 5.
**PyOTRS only supports OTRS 5 style!**
"""
+
def __init__(self, name, value):
self.name = name
self.value = value
@@ -405,21 +418,21 @@ class Ticket(object):
Title="Bäsic Ticket")
@staticmethod
- def datetime_to_pending_time_text(datetime_obj=None):
+ def datetime_to_pending_time_text(datetime_object=None):
"""datetime_to_pending_time_str
Args:
- datetime_obj (Datetime):
+ datetime_object (Datetime):
Returns:
str: The pending time in the format required for OTRS REST interface
"""
return {
- "Year": datetime_obj.year,
- "Month": datetime_obj.month,
- "Day": datetime_obj.day,
- "Hour": datetime_obj.hour,
- "Minute": datetime_obj.minute
+ "Year": datetime_object.year,
+ "Month": datetime_object.month,
+ "Day": datetime_object.day,
+ "Hour": datetime_object.hour,
+ "Minute": datetime_object.minute
}
@@ -441,7 +454,9 @@ class SessionIDStore(object):
needs to decide whether to use it..?!
"""
- def __init__(self, file_path=None, session_timeout=None):
+
+ def __init__(self, file_path=None, session_timeout=None,
+ value=None, created=None, expires=None):
if not file_path:
raise ArgumentMissingError("Argument file_path is required!")
@@ -450,9 +465,9 @@ class SessionIDStore(object):
self.file_path = file_path
self.timeout = session_timeout
- self.value = None
- self.created = None
- self.expires = None
+ self.value = value
+ self.created = created
+ self.expires = expires
def __repr__(self):
return "<{0}: {1}>".format(self.__class__.__name__, self.file_path)
@@ -575,6 +590,7 @@ class Client(object):
https_verify (bool): Should HTTPS certificates be verified (defaults to True)
ca_cert_bundle (unicode): file path - if specified overrides python/system default for
Root CA bundle that will be used. Settings are (temporarily) exported to environment.
+ operation_map (dict)
"""
def __init__(self,
@@ -587,7 +603,8 @@ class Client(object):
session_validation_ticket_id=1,
proxies=None,
https_verify=True,
- ca_cert_bundle=None):
+ ca_cert_bundle=None,
+ operation_map=OPERATION_MAPPING_DEFAULT):
if not baseurl:
raise ArgumentMissingError("baseurl")
@@ -618,14 +635,16 @@ class Client(object):
raise ValueError("Proxy settings need to be provided as dict!")
self.proxies = proxies
- self.https_verify = https_verify
+ if https_verify:
+ if not ca_cert_bundle:
+ self.https_verify = https_verify
+ else:
+ ca_certs = os.path.abspath(ca_cert_bundle)
+ if not os.path.isfile(ca_certs):
+ raise ValueError("Certificate file does not exist: {0}".format(ca_certs))
+ self.https_verify = ca_certs
- if ca_cert_bundle:
- ca_certs = os.path.abspath(ca_cert_bundle)
- if not os.path.isfile(ca_certs):
- raise ValueError("Certificate file does not exist: {0}".format(ca_certs))
- os.environ["REQUESTS_CA_BUNDLE"] = ca_certs
- os.environ["CURL_CA_BUNDLE"] = ca_certs
+ self.operation_map = operation_map
# credentials
self.username = username
@@ -635,6 +654,7 @@ class Client(object):
self.operation = None
self.result_json = None
self.result = []
+ self._result_type = None
"""
GenericInterface::Operation::Session::SessionCreate
@@ -657,18 +677,18 @@ class Client(object):
bool: **True** if valid, otherwise **False**
.. note::
- **Calls _ticket_get_json (GET)**
+ Uses HTTP Method: GET
"""
+ self.operation = "TicketGet"
+
if not session_id:
raise ArgumentMissingError("session_id")
# TODO 2016-04-13 (RH): Is there a nicer way to check whether session is valid?!
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket/{1}".format(self, self.session_validation_ticket_id))
-
payload = {"SessionID": session_id}
- return self._ticket_get_json(url, payload)
+ response = self._send_request(payload, ticket_id=self.session_validation_ticket_id)
+ return self._parse_and_validate_response(response)
def session_create(self):
"""create new (temporary) session (and Session ID)
@@ -683,19 +703,14 @@ class Client(object):
Uses HTTP Method: POST
"""
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Session".format(self))
+ self.operation = "SessionCreate"
payload = {
"UserLogin": self.username,
"Password": self.password
}
- self.operation = "SessionCreate"
- http_method = "POST"
- response = self._send_request(http_method, url, payload,
- self.proxies, self.https_verify)
- if not self._parse_and_validate_response(response):
+ if not self._parse_and_validate_response(self._send_request(payload)):
return False
self.session_id_store.value = self.result_json['SessionID']
@@ -727,7 +742,7 @@ class Client(object):
print("Using valid Session ID "
"from ({0})".format(self.session_id_store.file_path))
return True
- except OTRSAPIError:
+ except APIError:
"""most likely invalid session_id so pass. Remove clear session_id_store.."""
# got no (valid) session_id; clean store
@@ -749,6 +764,7 @@ class Client(object):
Methods (public):
* ticket_create
"""
+
def ticket_create(self,
ticket=None,
article=None,
@@ -767,8 +783,7 @@ class Client(object):
Returns:
dict or False: dict if successful, otherwise **False**
"""
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket".format(self))
+ self.operation = "TicketCreate"
payload = {"SessionID": self.session_id_store.value}
@@ -792,36 +807,11 @@ class Client(object):
# noinspection PyTypeChecker
payload.update({"DynamicField": [df.to_dct() for df in dynamic_field_list]})
- # return self._ticket_create_json(url, payload)
- _result = self._ticket_create_json(url, payload)
- if not _result:
+ if not self._parse_and_validate_response(self._send_request(payload)):
return False
else:
return self.result_json
- def _ticket_create_json(self, url, payload):
- """_ticket_create_json
-
- Args:
- url (unicode):
- payload (dict):
-
- Raises:
- OTRSAPIError
- ResponseJSONParseError
-
- Returns:
- bool: **True** if successful, otherwise **False**
- """
- self.operation = "TicketCreate"
- http_method = "POST"
- response = self._send_request(http_method, url, payload,
- self.proxies, self.https_verify)
- if not self._parse_and_validate_response(response):
- return False
-
- return True
-
"""
GenericInterface::Operation::Ticket::TicketGet
Methods (public):
@@ -846,22 +836,20 @@ class Client(object):
Dynamic Fields (*default: True*)
Returns:
- Ticket or None: Ticket object if successful, otherwise **False**
+ Ticket or False: Ticket object if successful, otherwise **False**
"""
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket/{1}".format(self, ticket_id))
+ self.operation = "TicketGetList"
payload = {
"SessionID": self.session_id_store.value,
+ "TicketID": "{0}".format(ticket_id),
"AllArticles": int(articles),
"Attachments": int(attachments),
"DynamicFields": int(dynamic_fields)
}
- # return self._ticket_get_json(url, payload)
- _result = self._ticket_get_json(url, payload)
- if not _result:
+ if not self._parse_and_validate_response(self._send_request(payload)):
return False
else:
return self.result[0]
@@ -885,12 +873,11 @@ class Client(object):
list: Ticket objects (as list) if successful, otherwise **False**
"""
+ self.operation = "TicketGetList"
+
if isinstance(ticket_id_list, int):
raise ArgumentInvalidError("Please provide list of IDs!")
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/TicketList".format(self))
-
payload = {
"SessionID": self.session_id_store.value,
"TicketID": ','.join([str(item) for item in ticket_id_list]),
@@ -899,9 +886,7 @@ class Client(object):
"DynamicFields": int(dynamic_fields)
}
- # return self._ticket_get_json(url, payload)
- _result = self._ticket_get_json(url, payload)
- if not _result:
+ if not self._parse_and_validate_response(self._send_request(payload)):
return False
else:
return self.result
@@ -915,7 +900,7 @@ class Client(object):
Args:
attachments:
- ticket_number (unicode): Integer value of a Ticket ID
+ ticket_number (unicode): Ticket Number as str
attachments (bool): will request OTRS to include attachments (*default: False*)
articles (bool): will request OTRS to include all
Articles (*default: False*)
@@ -932,52 +917,25 @@ class Client(object):
if isinstance(ticket_number, int):
raise ArgumentInvalidError("Provide ticket_number as str/unicode. "
"Got ticket_number as int.")
-
result_list = self.ticket_search(TicketNumber=ticket_number)
if not result_list:
return False
if len(result_list) == 1:
- _result = self.ticket_get_by_id(result_list[0],
- articles=articles,
- attachments=attachments,
- dynamic_fields=dynamic_fields)
- if not _result:
+ result = self.ticket_get_by_id(result_list[0],
+ articles=articles,
+ attachments=attachments,
+ dynamic_fields=dynamic_fields)
+ if not result:
return False
else:
- return self.result
+ return result
else:
- # TODO
+ # TODO more than one ticket found for a specific ticket number
raise ValueError("Found more that one result for "
"Ticket Number: {0}".format(ticket_number))
- def _ticket_get_json(self, url, payload):
- """_ticket_get_json
-
- Args:
- url (unicode):
- payload (dict):
-
- Raises:
- OTRSAPIError
- ResponseJSONParseError
-
- Returns:
- bool: **True** if successful, otherwise **False**
-
- .. note::
- Uses HTTP Method: GET
- """
- self.operation = "TicketGet"
- http_method = "GET"
- response = self._send_request(http_method, url, payload,
- self.proxies, self.https_verify)
- if not self._parse_and_validate_response(response):
- return False
-
- return True
-
"""
GenericInterface::Operation::Ticket::TicketSearch
Methods (public):
@@ -1001,9 +959,7 @@ class Client(object):
.. note::
tickets that were found as list of **str** stored self.ticket_search_result_list
"""
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket".format(self))
-
+ self.operation = "TicketSearch"
payload = {
"SessionID": self.session_id_store.value,
}
@@ -1014,9 +970,7 @@ class Client(object):
value = value.strftime("%Y-%m-%d %H:%M:%S")
payload.update({key: value})
- # return self._ticket_search_json(url, payload)
- _result = self._ticket_search_json(url, payload)
- if not _result:
+ if not self._parse_and_validate_response(self._send_request(payload)):
return False
else:
return self.result
@@ -1036,6 +990,7 @@ class Client(object):
.. note::
Waiting for progress on OTRS Bug: http://bugs.otrs.org/show_bug.cgi?id=11981
"""
+ self.operation = "TicketSearch"
pattern_wildcard = "%{0}%".format(pattern)
return self.ticket_search(FullTextIndex="1",
@@ -1043,39 +998,13 @@ class Client(object):
Subject=pattern_wildcard,
Body=pattern_wildcard)
- def _ticket_search_json(self, url, payload):
- """_ticket_search_json
- _search_json_ticket_data
-
- Args:
- url (unicode):
- payload (dict):
-
- Raises:
- OTRSAPIError
- ResponseJSONParseError
-
- Returns:
- bool: **True** if successful, otherwise **False**
-
- .. note::
- Uses HTTP Method: GET
- """
- self.operation = "TicketSearch"
- http_method = "GET"
- response = self._send_request(http_method, url, payload,
- self.proxies, self.https_verify)
- if not self._parse_and_validate_response(response):
- return False
-
- return True
-
"""
GenericInterface::Operation::Ticket::TicketUpdate
Methods (public):
* ticket_update
* ticket_update_set_pending
"""
+
def ticket_update(self,
ticket_id,
article=None,
@@ -1091,13 +1020,12 @@ class Client(object):
attachment_list (list): list of one or more *Attachment* objects that will
be added to ticket. Also requires an *Article*!
dynamic_field_list (list): *DynamicField* objects
- **kwargs: any regular OTRS Fields (not for Dynamic Fields!)
+ **kwargs: any regular Ticket Fields (not for Dynamic Fields!)
Returns:
dict or False: dict if successful, otherwise **False**
"""
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket/{1}".format(self, ticket_id))
+ self.operation = "TicketUpdate"
payload = {"SessionID": self.session_id_store.value}
@@ -1115,12 +1043,16 @@ class Client(object):
# noinspection PyTypeChecker
payload.update({"DynamicField": [df.to_dct() for df in dynamic_field_list]})
- # return self._ticket_update_json(url, payload)
- _result = self._ticket_update_json(url, payload)
- if not _result:
+ if kwargs is not None and not kwargs == {}:
+ ticket_dct = {}
+ for key, value in kwargs.items():
+ ticket_dct.update({key: value})
+ payload.update({"Ticket": ticket_dct})
+
+ if not self._parse_and_validate_response(self._send_request(payload, ticket_id)):
return False
- else:
- return self.result_json
+
+ return self.result_json
def ticket_update_set_pending(self,
ticket_id,
@@ -1137,71 +1069,61 @@ class Client(object):
Returns:
dict or False: dict if successful, otherwise **False**
- """
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket/{1}".format(self, ticket_id))
+ .. note::
+ Operates in UTC
+ """
datetime_now = datetime.datetime.utcnow()
pending_till = datetime_now + datetime.timedelta(days=pending_days, hours=pending_hours)
- pt = Ticket.datetime_to_pending_time_text(datetime_obj=pending_till)
-
- payload = {
- "SessionID": self.session_id_store.value,
- "Ticket": {"State": new_state, "PendingTime": pt}
- }
+ pt = Ticket.datetime_to_pending_time_text(datetime_object=pending_till)
- # return self._ticket_update_json(url, payload)
- _result = self._ticket_update_json(url, payload)
- if not _result:
- return False
- else:
- return self.result_json
+ return self.ticket_update(ticket_id, State=new_state, PendingTime=pt)
- def _ticket_update_json(self, url, payload):
- """_ticket_update_json
+ def _build_url(self, ticket_id=None):
+ """build url for request
Args:
- url (unicode):
- payload (dict):
-
- Raises:
- ValueError
+ ticket_id (optional[int])
Returns:
- bool: **True** if successful, otherwise **False**
+ url as str
- .. note::
- Uses HTTP Method: PATCH
"""
- self.operation = "TicketUpdate"
- http_method = "PATCH"
- response = self._send_request(http_method, url, payload,
- self.proxies, self.https_verify)
- if not self._parse_and_validate_response(response):
- return False
-
- # TODO 2016-04-17 (RH): is this "extra net" needed?!
- if "Article" in payload.keys():
- try:
- article_id = self.result_json.get('ArticleID', None)
- if not article_id:
- raise ValueError("No new Article was created?!")
+ route = self.operation_map[self.operation]["Route"]
+
+ if not (route.startswith("/Ticket") or route.startswith("/Session")):
+ raise ValueError("Route misconfigured: {}".format(route))
+
+ if ":" in route:
+ route_split = route.split(":")
+ route = route_split[0]
+ route_arg = route_split[1]
+
+ if route_arg == "TicketID":
+ if not ticket_id:
+ raise ValueError("TicketID is None but Route requires "
+ "TicketID: {}".format(route))
+
+ self._url = ("{0}/otrs/nph-genericinterface.pl/Webservice/"
+ "{1}{2}{3}".format(self.baseurl,
+ self.webservicename,
+ route,
+ ticket_id))
+ else:
+ self._url = ("{0}/otrs/nph-genericinterface.pl/Webservice/"
+ "{1}{2}".format(self.baseurl,
+ self.webservicename,
+ route))
- except Exception as err:
- raise ValueError("Unknown Exception: {0}".format(err))
- return True
+ return self._url
- @staticmethod
- def _send_request(http_method=None, url=None, payload=None, proxies=None, https_verify=True):
+ def _send_request(self, payload=None, ticket_id=None):
"""send the API request using the *requests.request* method
Args:
- http_method (unicode): HTTP Method to be used
- url (unicode): URL
- payload (dict): Payload to be sent
- proxies (dict): Proxy settings - default to no proxies
- https_verify (bool): (optional) whether the SSL cert will be verified.
+ payload (dict)
+ ticket_id (optional[dict])
Raises:
OTRSHTTPError:
@@ -1215,34 +1137,43 @@ class Client(object):
.. note::
Supported HTTP Methods: GET, HEAD, PATCH, POST, PUT
"""
- if not (http_method and url and payload):
- raise ValueError("http_method, url and payload are required.")
+ if not payload:
+ raise ArgumentMissingError("payload")
+
+ self._result_type = self.operation_map[self.operation]["ResultType"]
+
+ url = self._build_url(ticket_id)
+
+ http_method = self.operation_map[self.operation]["RequestMethod"]
- if not http_method.upper() in ["GET", "HEAD", "PATCH", "POST", "PUT"]:
- raise NotImplementedError("Accepted HTTP Methods: GET, HEAD, PATCH, POST, PUT")
+ if http_method not in ["GET", "HEAD", "PATCH", "POST", "PUT"]:
+ raise ValueError("invalid http_method")
headers = {"Content-Type": "application/json"}
json_payload = json.dumps(payload)
- # print("sending {0} to {1} as {2}".format(payload, url, http_method.upper()))
+ # ("sending {0} to {1} as {2}".format(payload, url, http_method.upper()))
try:
response = requests.request(http_method.upper(),
url,
- proxies=proxies,
- data=json_payload,
headers=headers,
- verify=https_verify)
+ data=json_payload,
+ proxies=self.proxies,
+ verify=self.https_verify)
+
+ # store a copy of the request
+ self._request = response.request
# critical error: HTTP request resulted in an error!
except Exception as err:
# raise OTRSHTTPError("get http")
- raise OTRSHTTPError("Failed to access OTRS. Check Hostname, Proxy, SSL Certificate!\n"
- "Error with http communication: {0}".format(err))
+ raise HTTPError("Failed to access OTRS. Check Hostname, Proxy, SSL Certificate!\n"
+ "Error with http communication: {0}".format(err))
if not response.status_code == 200:
- raise OTRSHTTPError("Received HTTP Error. Check Hostname and WebServiceName.\n"
- "HTTP Status Code: {0.status_code}\n"
- "HTTP Message: {0.content}".format(response))
+ raise HTTPError("Received HTTP Error. Check Hostname and WebServiceName.\n"
+ "HTTP Status Code: {0.status_code}\n"
+ "HTTP Message: {0.content}".format(response))
return response
def _parse_and_validate_response(self, response):
@@ -1264,6 +1195,9 @@ class Client(object):
if not isinstance(response, requests.models.Response):
raise ValueError("requests.Response object expected!")
+ if self.operation not in self.operation_map.keys():
+ raise ValueError("invalid operation")
+
# clear data from Client
self.result = None
self._result_error = False
@@ -1276,23 +1210,6 @@ class Client(object):
self._result_status_code = response.status_code
self._result_content = response.content
- # store a copy of the request
- if hasattr(response, 'request'): # TODO (this is just a cheat so that the tests pass)
- self._request = response.request
-
- if self.operation == "SessionCreate":
- self._result_type = "SessionID"
- elif self.operation == "TicketGet":
- self._result_type = "Ticket"
- elif self.operation == "TicketCreate":
- self._result_type = "TicketID"
- elif self.operation == "TicketSearch":
- self._result_type = "TicketID"
- elif self.operation == "TicketUpdate":
- self._result_type = "TicketID"
- else:
- raise NotImplementedError("Unknown Operation!")
-
# handle TicketSearch operation first. special: empty search result has no "TicketID"
if self.operation == "TicketSearch":
if not self.result_json:
@@ -1310,20 +1227,34 @@ class Client(object):
else:
self._result_error = True
# critical error: Unknown response from OTRS API - FAIL NOW!
- raise ResponseJSONParseError("Unknown key in response JSON DICT!")
+ raise ResponseParseError("Unknown key in response JSON DICT!")
# report error
if self._result_error:
- raise OTRSAPIError("Failed to access OTRS API. Check Username and Password! "
- "Session ID expired?! Does Ticket exist?\n"
- "OTRS Error Code: {0}\nOTRS Error Message: {1}"
- "".format(self.result_json["Error"]["ErrorCode"],
- self.result_json["Error"]["ErrorMessage"]))
+ raise APIError("Failed to access OTRS API. Check Username and Password! "
+ "Session ID expired?! Does Ticket exist?\n"
+ "OTRS Error Code: {0}\nOTRS Error Message: {1}"
+ "".format(self.result_json["Error"]["ErrorCode"],
+ self.result_json["Error"]["ErrorMessage"]))
# for operation TicketGet: parse result list into Ticket object list
- if self.operation == "TicketGet":
+ if self.operation == "TicketGet" or self.operation == "TicketGetList":
self.result = [Ticket(item) for item in self.result_json['Ticket']]
+ # TODO 2016-04-17 (RH): is this "extra net" needed?!
+ # for operation TicketUpdate: if Article was updated, check that response contains
+ # new ArticleID.
+ # if self.operation == "TicketUpdate":
+ # _request_body = json.loads(response.request.body)
+ # if "Article" in _request_body.keys():
+ # try:
+ # article_id = self.result_json.get('ArticleID', None)
+ # if not article_id:
+ # raise ValueError("No new Article was created?!")
+ #
+ # except Exception as err:
+ # raise ValueError("Unknown Exception: {0}".format(err))
+
return True
# EOF
diff --git a/tests/test_attachment.py b/tests/test_attachment.py
index 587a7f3d146d3a585467f132de65119fca4ab1af..186f6cf4cac7db8f931b112fa633a27277513513 100644
--- a/tests/test_attachment.py
+++ b/tests/test_attachment.py
@@ -9,6 +9,7 @@ Test for PyOTRS Attachment class
import os.path
import sys
import unittest2 as unittest
+import mock
# from mock import MagicMock, patch
current_path = os.path.dirname(os.path.realpath(__file__))
@@ -58,9 +59,69 @@ class AttachmentTests(unittest.TestCase):
'ContentType': 'text/plain',
'Filename': 'd\xfcmmy5.txt'})
- # TODO 2016-04-24 (RH) missing tests:
- # * def create_from_file(cls, file_path):
- # * def save_to_disc(self, folder="/tmp"):
+ def test_attachment_create_from_file(self):
+ """create_from_file test ending in .txt"""
+ r_data = "this is the content of a file!"
+ with mock.patch('pyotrs.lib.open', mock.mock_open(read_data=r_data)) as m:
+ att = Attachment.create_from_file("/tmp/a_file.txt")
+
+ self.assertIsInstance(att, Attachment)
+ self.assertEqual(att.ContentType, "text/plain")
+ self.assertEqual(m.call_count, 1)
+ self.assertEqual(m.call_args_list, [mock.call("/tmp/a_file.txt", 'rb')])
+
+ def test_attachment_create_from_file_no_file_ext(self):
+ """create_from_file test - no file ending -> mime needs to default"""
+ r_data = "this is the content of a file!"
+ with mock.patch('pyotrs.lib.open', mock.mock_open(read_data=r_data)) as m:
+ att = Attachment.create_from_file("/tmp/b_file")
+
+ self.assertIsInstance(att, Attachment)
+ self.assertEqual(att.ContentType, "application/octet-stream")
+ self.assertEqual(m.call_count, 1)
+ self.assertEqual(m.call_args_list, [mock.call("/tmp/b_file", 'rb')])
+
+ def test_attachment_save_to_dir_invalid_no_filename(self):
+ """save_to_dir test - invalid Attachment"""
+
+ att = Attachment.create_basic("YmFyCg==", "text/plain", "dümmy5.txt")
+ att.__delattr__("Filename")
+
+ self.assertRaisesRegex(ValueError,
+ "invalid Attachment",
+ att.save_to_dir,
+ "/tmp")
+
+ def test_attachment_save_to_dir_invalid_no_content(self):
+ """save_to_dir test - invalid Attachment"""
+
+ att = Attachment.create_basic("mFyCg==", "text/plain", "foo.txt")
+ att.__delattr__("Content")
+
+ self.assertRaisesRegex(ValueError,
+ "invalid Attachment",
+ att.save_to_dir,
+ "/tmp")
+
+ def test_attachment_save_to_dir(self):
+ """save_to_dir test - ok"""
+ att = Attachment.create_basic("YmFyCg==", "text/plain", "foo.txt")
+
+ with mock.patch('pyotrs.lib.open', mock.mock_open()) as m:
+ att.save_to_dir("/tmp")
+
+ self.assertEqual(m.call_count, 1)
+ self.assertEqual(m.call_args_list, [mock.call("/tmp/foo.txt", 'wb')])
+
+ def test_attachment_save_to_dir_two(self):
+ """save_to_dir test - ok"""
+ att = Attachment.create_basic("YmFyCg==", "text/plain", "dümmy5.txt")
+
+ with mock.patch('pyotrs.lib.open', mock.mock_open()) as m:
+ att.save_to_dir()
+
+ self.assertEqual(m.call_count, 1)
+ self.assertEqual(m.call_args_list, [mock.call("/tmp/dümmy5.txt", 'wb')])
def main():
diff --git a/tests/test_client.py b/tests/test_client.py
index 596d050d6a43ece0795169f67037de8fd0b0d753..04993d5ed4e70bc8262d0c633e8ae887eaf0f5ea 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -9,25 +9,19 @@ Test for PyOTRS Client class
import unittest2 as unittest
import mock
-# TODO
-# try:
-# from test.test_support import EnvironmentVarGuard
-# except ImportError:
-# from test.support import EnvironmentVarGuard # noqa
-
import datetime # noqa
import requests # noqa
from pyotrs import Article, Attachment, Client, DynamicField, Ticket # noqa
-from pyotrs.lib import ResponseJSONParseError # noqa
+from pyotrs.lib import ResponseParseError # noqa
from pyotrs.lib import ArgumentMissingError, ArgumentInvalidError # noqa
from pyotrs.lib import SessionCreateError # noqa
-from pyotrs.lib import OTRSAPIError, OTRSHTTPError # noqa
+from pyotrs.lib import APIError, HTTPError # noqa
class ClientTests(unittest.TestCase):
def test_init(self):
- client = Client(baseurl="http://localhost/", webservicename="dummy")
+ client = Client(baseurl="http://fqdn/", webservicename="dummy")
self.assertIsInstance(client, Client)
def test_init_no_base(self):
@@ -45,18 +39,18 @@ class ClientTests(unittest.TestCase):
self.assertRaisesRegex(ArgumentMissingError, 'webservicename', Client, 'http://localhost')
def test_init_session_id_store(self):
- client = Client(baseurl="http://localhost/", webservicename="foo", session_id_file=".sid")
+ client = Client(baseurl="http://fqdn/", webservicename="foo", session_id_file=".sid")
self.assertEqual(client.session_id_store.file_path, '.sid')
def test_init_session_id_store_timeout_default(self):
- client = Client(baseurl="http://localhost/",
+ client = Client(baseurl="http://fqdn/",
webservicename="foo",
session_id_file=".sid")
self.assertEqual(client.session_id_store.file_path, '.sid')
self.assertEqual(client.session_id_store.timeout, 28800)
def test_init_session_id_store_timeout(self):
- client = Client(baseurl="http://localhost/",
+ client = Client(baseurl="http://fqdn/",
webservicename="foo",
session_id_file=".sid",
session_timeout=815)
@@ -64,22 +58,18 @@ class ClientTests(unittest.TestCase):
self.assertEqual(client.session_id_store.timeout, 815)
def test_init_proxies_default(self):
- client = Client(baseurl="http://localhost/",
+ client = Client(baseurl="http://fqdn/",
webservicename="foo")
self.assertDictEqual(client.proxies, {'http': '', 'https': '', 'no': ''})
- # TODO
- # @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
- # def test_init_ca_cert_bundle(self, mock_isfile):
- # mock_isfile.return_value = True
- # with EnvironmentVarGuard() as self.env:
- # Client(baseurl="http://localhost/",
- # webservicename="foo",
- # ca_cert_bundle="/tmp/certs.pem")
- # # os.environ["REQUESTS_CA_BUNDLE"] = ca_certs
- # # os.environ["CURL_CA_BUNDLE"] = ca_certs
- # self.assertEqual(self.env.get("REQUESTS_CA_BUNDLE", None), "/tmp/certs.pem")
- # self.assertEqual(self.env.get("CURL_CA_BUNDLE", None), "/tmp/certs.pem")
+ @mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
+ def test_init_ca_cert_bundle(self, mock_isfile):
+ obj = Client(baseurl="http://fqdn/",
+ webservicename="foo",
+ ca_cert_bundle="/tmp/certs.pem")
+ mock_isfile.return_value = True
+
+ self.assertEqual(obj.https_verify, "/tmp/certs.pem")
@mock.patch('pyotrs.lib.os.path.isfile', autospec=True)
def test_init_ca_cert_bundle_non_existent(self, mock_isfile):
@@ -87,7 +77,7 @@ class ClientTests(unittest.TestCase):
self.assertRaisesRegex(ValueError,
'Certificate file does not exist.*',
Client,
- baseurl="http://localhost/",
+ baseurl="http://fqdn/",
webservicename="foo",
ca_cert_bundle="/tmp/certs.pem")
@@ -95,12 +85,12 @@ class ClientTests(unittest.TestCase):
self.assertRaisesRegex(ValueError,
'Proxy settings need to be provided as dict!',
Client,
- baseurl="http://localhost/",
+ baseurl="http://fqdn/",
webservicename="foo",
proxies='http://proxy:3128')
def test_init_proxies_override_valid(self):
- client = Client(baseurl="http://localhost/",
+ client = Client(baseurl="http://fqdn/",
webservicename="foo",
proxies={'http': 'http://proxy:3128',
'https': 'http://proxy:3128',
@@ -111,61 +101,109 @@ class ClientTests(unittest.TestCase):
def test_session_check_is_valid_no_session_id_error(self):
"""Test"""
- client = Client(baseurl="http://localhost/", webservicename="foo")
+ client = Client(baseurl="http://fqdn/", webservicename="foo")
self.assertRaisesRegex(ArgumentMissingError,
'session_id',
client.session_check_is_valid)
- @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- def test_session_check_is_valid_session_id(self, mock_ticket_get_json):
- """Test session_check_is_valid with a given session id"""
- obj = Client(baseurl="http://localhost/", webservicename="foo")
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_session_check_is_valid_session_id(self, mock_parse_validate, mock_send_req):
+ """Test session_check_is_valid with a given valid session id"""
+ obj = Client(baseurl="http://fqdn/", webservicename="foo")
obj.session_id_store.value = "some_other_value"
- mock_ticket_get_json.return_value = True
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
result = obj.session_check_is_valid(session_id="some_value")
self.assertTrue(result)
- @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- def test_session_check_is_valid_self_session_id(self, mock_ticket_get_json):
- """Test session_check_is_valid with a given session id"""
- obj = Client(baseurl="http://localhost/", webservicename="foo")
- mock_ticket_get_json.return_value = True
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_session_check_is_valid_invalid_session_id(self, mock_parse_validate, mock_send_req):
+ """Test session_check_is_valid with a given invalid session id"""
+ obj = Client(baseurl="http://fqdn/", webservicename="foo")
+ obj.session_id_store.value = "some_other_value"
- result = obj.session_check_is_valid("some_other_value2")
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
- self.assertTrue(result)
+ result = obj.session_check_is_valid(session_id="some_value")
+ self.assertFalse(result)
def test_init_session_default_session_timeout(self):
- client = Client(baseurl="http://localhost/", webservicename="foo")
+ client = Client(baseurl="http://fqdn/", webservicename="foo")
self.assertEqual(client.session_timeout, 28800)
def test_init_session_manual_session_timeout(self):
- client = Client(baseurl="http://localhost/", webservicename="foo", session_timeout=4711)
+ client = Client(baseurl="http://fqdn/", webservicename="foo", session_timeout=4711)
self.assertEqual(client.session_timeout, 4711)
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_session_create_ok(self, mock_parse_validate, mock_send_req):
+ """Test session create ok"""
+ obj = Client(baseurl="http://fqdn/", webservicename="foo")
+
+ obj.session_id_store.value = None
+ obj.result_json = {'SessionID': 'fake'}
+ obj.session_id_store.value = obj.result_json['SessionID']
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
+
+ result = obj.session_create()
+ self.assertTrue(result)
+ self.assertEqual(obj.session_id_store.value, 'fake')
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
+
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_session_create_fail(self, mock_parse_validate, mock_send_req):
+ """Test session create ok"""
+ obj = Client(baseurl="http://fqdn/", webservicename="foo")
+
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
+
+ result = obj.session_create()
+ self.assertFalse(result)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
+
+ @mock.patch('pyotrs.SessionIDStore.write', autospec=True)
@mock.patch('pyotrs.Client.session_check_is_valid', autospec=True)
@mock.patch('pyotrs.SessionIDStore.read', autospec=True)
- def test_session_restore_or_set_up_new_with_file_nok(self, mock_read_s_id, mock_is_valid):
+ def test_session_restore_or_set_up_new_with_file_nok(self,
+ mock_read_s_id,
+ mock_is_valid,
+ mock_wr):
"""Tests session_restore_or_set_up_new when read from file successful but session nok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id_store.value = "foobar"
- otrs_api_error = OTRSHTTPError("Failed to access OTRS. Check Hostname...")
+ otrs_api_error = HTTPError("Failed to access OTRS. Check Hostname...")
mock_read_s_id.return_value = "SomeSessionID1"
mock_is_valid.side_effect = otrs_api_error
- self.assertRaisesRegex(OTRSHTTPError,
+ self.assertRaisesRegex(HTTPError,
'Failed to access OTRS. Check Hostname...',
obj.session_restore_or_set_up_new)
self.assertEqual(mock_read_s_id.call_count, 1)
self.assertEqual(mock_is_valid.call_count, 1)
+ self.assertEqual(mock_wr.call_count, 0)
+ @mock.patch('pyotrs.SessionIDStore.write', autospec=True)
@mock.patch('pyotrs.Client.session_check_is_valid', autospec=True)
@mock.patch('pyotrs.SessionIDStore.read', autospec=True)
- def test_session_restore_or_set_up_new_with_file_ok(self, mock_read_s_id, mock_is_valid):
+ def test_session_restore_or_set_up_new_with_file_ok(self,
+ mock_read_s_id,
+ mock_is_valid,
+ mock_wr):
"""Tests session_restore_or_set_up_new when read from file successful and session ok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id_store.value = "foobar2"
@@ -177,34 +215,39 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_read_s_id.call_count, 1)
self.assertEqual(mock_is_valid.call_count, 1)
+ self.assertEqual(mock_wr.call_count, 0)
self.assertTrue(result)
+ @mock.patch('pyotrs.SessionIDStore.write', autospec=True)
@mock.patch('pyotrs.Client.session_create', autospec=True)
@mock.patch('pyotrs.Client.session_check_is_valid', autospec=True)
@mock.patch('pyotrs.SessionIDStore.read', autospec=True)
def test_session_restore_or_set_up_new_with_file_pass(self,
mock_read_s_id,
mock_is_valid,
- mock_s_create):
+ mock_s_create,
+ mock_wr):
"""Tests session_restore_or_set_up_new when read from file successful but session nok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id_store.value = "foobar"
mock_read_s_id.return_value = "SomeSessionID"
- mock_is_valid.side_effect = OTRSHTTPError("Failed to access OTRS. Check Hostname...")
+ mock_is_valid.side_effect = HTTPError("Failed to access OTRS. Check Hostname...")
mock_s_create.return_value = True
try:
obj.session_restore_or_set_up_new()
- except OTRSHTTPError:
+ except HTTPError:
pass
self.assertEqual(mock_read_s_id.call_count, 1)
self.assertEqual(mock_is_valid.call_count, 1)
+ self.assertEqual(mock_wr.call_count, 0)
+ @mock.patch('pyotrs.SessionIDStore.write', autospec=True)
@mock.patch('pyotrs.Client.session_create', autospec=True)
@mock.patch('pyotrs.SessionIDStore.read', autospec=True)
- def test_session_restore_or_set_up_new_nok(self, mock_read_s_id, mock_s_create):
+ def test_session_restore_or_set_up_new_nok(self, mock_read_s_id, mock_s_create, mock_wr):
"""Tests session_restore_or_set_up_new no file; create unsuccessful"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.session_id_store.value = "foobar"
@@ -218,6 +261,7 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_read_s_id.call_count, 1)
self.assertEqual(mock_s_create.call_count, 1)
+ self.assertEqual(mock_wr.call_count, 1)
@mock.patch('pyotrs.SessionIDStore.write', autospec=True)
@mock.patch('pyotrs.Client.session_create', autospec=True)
@@ -259,69 +303,93 @@ class ClientTests(unittest.TestCase):
self.assertEqual(mock_s_create.call_count, 1)
self.assertEqual(mock_wr.call_count, 2)
- @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- def test_ticket_get_by_id_fail(self, mock_ticket_get_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_get_by_id_fail(self, mock_parse_validate, mock_send_req):
"""Tests ticket_get_by_id fail"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- mock_ticket_get_json.return_value = False
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
+
result = obj.ticket_get_by_id(1)
- self.assertEqual(mock_ticket_get_json.call_count, 1)
self.assertFalse(result)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- # TODO
- # @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- # def test_ticket_get_by_id_ok(self, mock_ticket_get_json):
- # """Tests ticket_get_by_id ok"""
- # obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- # obj.result_json = {'Ticket': [{'Some': 'Ticket'}]}
- # mock_ticket_get_json.return_value = True
- # result = obj.ticket_get_by_id(1)
- #
- # self.assertEqual(mock_ticket_get_json.call_count, 1)
- # self.assertIsInstance(result, Ticket)
- # self.assertDictEqual(result, {'Some': 'Ticket'})
-
- @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- def test_ticket_get_by_ids_fail(self, mock_ticket_get_json):
- """Tests ticket_get_by_id fail"""
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_get_by_id_ok(self, mock_parse_validate, mock_send_req):
+ """Tests ticket_get_by_id ok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- mock_ticket_get_json.return_value = False
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
+ obj.result = [Ticket._dummy()]
+
+ result = obj.ticket_get_by_id(1)
+
+ self.assertIsInstance(result, Ticket)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
+
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_get_by_list_fail(self, mock_parse_validate, mock_send_req):
+ """Tests ticket_get_by_list fail"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
+
result = obj.ticket_get_by_list([1])
- self.assertEqual(mock_ticket_get_json.call_count, 1)
self.assertFalse(result)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- # How
- # @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- # def test_ticket_get_by_ids_ok(self, mock_ticket_get_json):
- # """Tests ticket_get_by_id fail"""
- # obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- #
- # def mock_result(obj):
- # obj.result = [Ticket._dummy()]
- # return obj
- #
- # mock_ticket_get_json.return_value = True
- # result = obj.ticket_get_by_list([1])
- #
- # self.assertEqual(mock_ticket_get_json.call_count, 1)
- # self.assertIsInstance(result[0], Ticket)
- #
- # @mock.patch('pyotrs.Client._ticket_get_json', autospec=True)
- # def test_ticket_get_by_ids_ok_two(self, mock_ticket_get_json):
- # """Tests ticket_get_by_id fail"""
- # obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- # tic1 = Ticket._dummy()
- # tic2 = Ticket._dummy()
- #
- # mock_ticket_get_json.return_value = True
- # result = obj.ticket_get_by_list([1, 2])
- #
- # self.assertEqual(mock_ticket_get_json.call_count, 1)
- # self.assertIsInstance(result[0], Ticket)
+ def test_ticket_get_by_list_fail_int_provided(self):
+ """Tests ticket_get_by_list fail int was provided"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+
+ self.assertRaisesRegex(ArgumentInvalidError,
+ "Please provide list of IDs!",
+ obj.ticket_get_by_list,
+ 1)
+
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_get_by_list_ok(self, mock_parse_validate, mock_send_req):
+ """Tests ticket_get_by_list ok"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
+ obj.result = [Ticket._dummy()]
+
+ result = obj.ticket_get_by_list([1])
+
+ self.assertIsInstance(result[0], Ticket)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
+
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_get_by_list_ok_two(self, mock_parse_validate, mock_send_req):
+ """Tests ticket_get_by_list ok"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
+ obj.result = [Ticket._dummy(), Ticket._dummy()]
+
+ result = obj.ticket_get_by_list([1, 2])
+
+ self.assertIsInstance(result[0], Ticket)
+ self.assertEqual(len(result), 2)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
def test_ticket_get_by_number_with_int(self):
"""Tests ticket_get_by_number provided int not str -> fail"""
@@ -341,24 +409,36 @@ class ClientTests(unittest.TestCase):
self.assertFalse(result)
self.assertEqual(mock_ticket_search.call_count, 1)
- # TODO
- # @mock.patch('pyotrs.Client.ticket_get_by_id', autospec=True)
- # @mock.patch('pyotrs.Client.ticket_search', autospec=True)
- # def test_ticket_get_by_number_with_string_one_result(self, mock_t_search, mock_t_get_by_id):
- # """Tests ticket_get_by_number provided as int -> ok; 1 result"""
- # obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- # tic = Ticket._dummy()
- #
- # mock_t_search.return_value = [u'4711']
- # obj.result_json = {'Ticket': [{'SomeOne': 'TicketOne'}]}
- # mock_t_get_by_id.return_value = tic
- #
- # result = obj.ticket_get_by_number("SomeOtherNumber1")
- #
- # self.assertIsInstance(result, Ticket)
- # self.assertDictEqual(result.to_dct(), {'SomeOne': 'TicketOne'})
- # self.assertEqual(mock_t_search.call_count, 1)
- # self.assertEqual(mock_t_search.call_args, [4711, "dynamic_fields=1"])
+ @mock.patch('pyotrs.Client.ticket_get_by_id', autospec=True)
+ @mock.patch('pyotrs.Client.ticket_search', autospec=True)
+ def test_ticket_get_by_number_one_result_fail(self, mock_ticket_search, mock_t_get_id):
+ """Tests ticket_get_by_number - one result but fail to get"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+
+ mock_ticket_search.return_value = [u'11']
+ mock_t_get_id.return_value = False
+
+ result = obj.ticket_get_by_number("4711")
+
+ self.assertFalse(result)
+ self.assertEqual(mock_ticket_search.call_count, 1)
+ self.assertEqual(mock_t_get_id.call_count, 1)
+
+ @mock.patch('pyotrs.Client.ticket_get_by_id', autospec=True)
+ @mock.patch('pyotrs.Client.ticket_search', autospec=True)
+ def test_ticket_get_by_number_one_result_ok(self, mock_ticket_search, mock_t_get_id):
+ """Tests ticket_get_by_number - one result - get ok"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+
+ mock_ticket_search.return_value = [u'12']
+ mock_t_get_id.return_value = Ticket._dummy()
+ obj.result = [Ticket._dummy()]
+
+ result = obj.ticket_get_by_number("4712")
+
+ self.assertIsInstance(result, Ticket)
+ self.assertEqual(mock_ticket_search.call_count, 1)
+ self.assertEqual(mock_t_get_id.call_count, 1)
@mock.patch('pyotrs.Client.ticket_search', autospec=True)
def test_ticket_get_by_number_with_string_three_results(self, mock_ticket_search):
@@ -393,8 +473,9 @@ class ClientTests(unittest.TestCase):
'Article',
obj.ticket_create, ticket=tic)
- @mock.patch('pyotrs.lib.Client._ticket_create_json', autospec=True)
- def test_ticket_create_mocked_none(self, mock_t_create_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_create_mocked_none(self, mock_parse_validate, mock_send_req):
"""Test ticket_create - mocked result None"""
obj = Client(baseurl="http://fqdn",
webservicename="GenericTicketConnectorREST")
@@ -405,12 +486,18 @@ class ClientTests(unittest.TestCase):
'TimeUnit': 0,
'MimeType': 'text/plain',
'Charset': 'UTF8'})
- mock_t_create_json.return_value = False
+
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
+
result = obj.ticket_create(ticket=tic, article=art)
self.assertFalse(result)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- @mock.patch('pyotrs.lib.Client._ticket_create_json', autospec=True)
- def test_ticket_create_mocked_true(self, mock_t_create_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_create_mocked_true(self, mock_parse_validate, mock_send_req):
"""Test ticket_create - mocked result """
obj = Client(baseurl="http://fqdn",
webservicename="GenericTicketConnectorREST")
@@ -421,12 +508,17 @@ class ClientTests(unittest.TestCase):
'TimeUnit': 0,
'MimeType': 'text/plain',
'Charset': 'UTF8'})
- mock_t_create_json.return_value = True
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
+
obj.ticket_create(ticket=tic, article=art)
- self.assertEqual(mock_t_create_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- @mock.patch('pyotrs.lib.Client._ticket_create_json', autospec=True)
- def test_ticket_create_mocked_attachment_true(self, mock_t_create_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_create_mocked_attachment_true(self, mock_parse_validate, mock_send_req):
"""Test ticket_create - mocked attachment result """
obj = Client(baseurl="http://fqdn",
webservicename="GenericTicketConnectorREST")
@@ -439,12 +531,17 @@ class ClientTests(unittest.TestCase):
'TimeUnit': 0,
'MimeType': 'text/plain',
'Charset': 'UTF8'})
- mock_t_create_json.return_value = True
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
+
obj.ticket_create(ticket=tic, article=art, attachment_list=[att1, att2])
- self.assertEqual(mock_t_create_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- @mock.patch('pyotrs.lib.Client._ticket_create_json', autospec=True)
- def test_ticket_create_mocked_dynamic_field_true(self, mock_t_create_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_create_mocked_dynamic_field_true(self, mock_parse_validate, mock_send_req):
"""Test ticket_create - mocked dynamic field result """
obj = Client(baseurl="http://fqdn",
webservicename="GenericTicketConnectorREST")
@@ -457,31 +554,56 @@ class ClientTests(unittest.TestCase):
'TimeUnit': 0,
'MimeType': 'text/plain',
'Charset': 'UTF8'})
- mock_t_create_json.return_value = True
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
+
obj.ticket_create(ticket=tic, article=art, dynamic_field_list=[dyn1, dyn2])
- self.assertEqual(mock_t_create_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
+
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_search_ticket_id(self, mock_parse_validate, mock_send_req):
+ """Tests ticket_search ticket id"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.result = [1]
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
- @mock.patch('pyotrs.Client._ticket_search_json', autospec=True)
- def test_ticket_search_ticket_id(self, mock_ticket_search_json):
+ obj.ticket_search(TicketID="1")
+
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
+
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_search_ticket_id_fail(self, mock_parse_validate, mock_send_req):
"""Tests ticket_search ticket id"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.result = [1]
- mock_ticket_search_json.return_value = True
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
obj.ticket_search(TicketID="1")
- self.assertEqual(mock_ticket_search_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- @mock.patch('pyotrs.Client._ticket_search_json', autospec=True)
- def test_ticket_search_datetime(self, mock_ticket_search_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_search_datetime(self, mock_parse_validate, mock_send_req):
"""Tests ticket_search datetime"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
obj.result = [1]
- mock_ticket_search_json.return_value = True
+
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
obj.ticket_search(TicketCreateTimeOlderDate=datetime.datetime.utcnow())
- self.assertEqual(mock_ticket_search_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
@mock.patch('pyotrs.Client.ticket_search')
def test_ticket_search_full_text(self, mock_ticket_search):
@@ -506,29 +628,36 @@ class ClientTests(unittest.TestCase):
FullTextIndex=u'1',
Subject=u'%Something%')
- @mock.patch('pyotrs.Client._ticket_update_json', autospec=True)
- def test_ticket_update_queue_id_ok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_queue_id_ok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update queue_id ok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- mock_ticket_update_json.return_value = True
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
obj.ticket_update(1, QueueID="1")
- self.assertEqual(mock_ticket_update_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- @mock.patch('pyotrs.Client._ticket_update_json', autospec=True)
- def test_ticket_update_queue_id_nok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_queue_id_nok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update queue_id nok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- mock_ticket_update_json.return_value = False
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
result = obj.ticket_update(1, QueueID="1")
- self.assertEqual(mock_ticket_update_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
self.assertFalse(result)
- @mock.patch('pyotrs.Client._ticket_update_json', autospec=True)
- def test_ticket_update_article_nok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_article_nok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update article nok"""
# create object
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
@@ -537,19 +666,24 @@ class ClientTests(unittest.TestCase):
'TimeUnit': 0,
'MimeType': 'text/plain',
'Charset': 'UTF8'})
- mock_ticket_update_json.return_value = False
+
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
result = obj.ticket_update(1, article=art)
- self.assertEqual(mock_ticket_update_json.call_count, 1)
self.assertFalse(result)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- @mock.patch('pyotrs.Client._ticket_update_json', autospec=True)
- def test_ticket_update_attachment_list_no_article_nok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_attach_list_no_article_nok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update attachment list no article nok"""
# create object
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- mock_ticket_update_json.return_value = False
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
att1 = Attachment.create_basic("mFyCg==", "text/plain", "foo.txt")
att2 = Attachment.create_basic("YmFyCg==", "text/plain", "dümmy.txt")
@@ -559,12 +693,11 @@ class ClientTests(unittest.TestCase):
obj.ticket_update,
1, attachment_list=[att1, att2])
- @mock.patch('pyotrs.Client._ticket_update_json')
- def test_ticket_update_attachment_list_article_nok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_attachment_list_article_nok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update attachment list article nok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket/{1}".format(obj, 8))
payload = {u'Attachment': [{u'Content': u'mFyCg==',
u'ContentType': u'text/plain',
@@ -587,22 +720,23 @@ class ClientTests(unittest.TestCase):
att1 = Attachment.create_basic("mFyCg==", "text/plain", "foo.txt")
att2 = Attachment.create_basic("YmFyCg==", "text/plain", "dümmy.txt")
- mock_ticket_update_json.return_value = False
+
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
result = obj.ticket_update(8, article=art, attachment_list=[att1, att2])
- self.assertEqual(mock_ticket_update_json.call_count, 1)
self.assertFalse(result)
- mock_ticket_update_json.assert_called_once_with(url, payload)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
+ mock_send_req.assert_called_once_with(payload, 8)
- @mock.patch('pyotrs.Client._ticket_update_json')
- def test_ticket_update_dynamic_field_list_nok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_dynamic_field_list_nok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update dynamic field list nok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket/{1}".format(obj, 7))
-
payload = {u'DynamicField': [{u'Name': u'firstname', u'Value': u'Jane'},
{u'Name': u'lastname', u'Value': u'Doe'}],
u'SessionID': None,
@@ -621,63 +755,188 @@ class ClientTests(unittest.TestCase):
dyn1 = DynamicField(name="firstname", value="Jane")
dyn2 = DynamicField.from_dct({'Name': 'lastname', 'Value': 'Doe'})
- mock_ticket_update_json.return_value = False
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
result = obj.ticket_update(7, article=art, dynamic_field_list=[dyn1, dyn2])
- self.assertEqual(mock_ticket_update_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
self.assertFalse(result)
- mock_ticket_update_json.assert_called_once_with(url, payload)
+ mock_send_req.assert_called_once_with(payload, 7)
- @mock.patch('pyotrs.Client._ticket_update_json', autospec=True)
- def test_ticket_update_set_pending_ok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_set_pending_ok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update_set_pending ok"""
# create object
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- mock_ticket_update_json.return_value = True
+ mock_parse_validate.return_value = True
+ mock_send_req.return_value = "mock"
obj.ticket_update_set_pending(1)
- self.assertEqual(mock_ticket_update_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
- @mock.patch('pyotrs.Client._ticket_update_json', autospec=True)
- def test_ticket_update_set_pending_nok(self, mock_ticket_update_json):
+ @mock.patch('pyotrs.Client._send_request')
+ @mock.patch('pyotrs.Client._parse_and_validate_response', autospec=True)
+ def test_ticket_update_set_pending_nok(self, mock_parse_validate, mock_send_req):
"""Tests ticket_update_set_pending nok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
- mock_ticket_update_json.return_value = False
+ mock_parse_validate.return_value = False
+ mock_send_req.return_value = "mock"
result = obj.ticket_update_set_pending(1)
- self.assertEqual(mock_ticket_update_json.call_count, 1)
+ self.assertEqual(mock_parse_validate.call_count, 1)
+ self.assertEqual(mock_send_req.call_count, 1)
self.assertFalse(result)
- def test__send_request_invalid_signature(self):
- """Test _send_request with invalid signature """
+ def test__build_url_session_create(self):
+ """Test _build_url for session create"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "SessionCreate"
+
+ self.assertEqual("http://fqdn/otrs/nph-genericinterface.pl/Webservice/"
+ "GenericTicketConnectorREST/Session",
+ obj._build_url())
+
+ def test__build_url_ticket_create(self):
+ """Test _build_url for ticket create"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketCreate"
+
+ self.assertEqual("http://fqdn/otrs/nph-genericinterface.pl/Webservice/"
+ "GenericTicketConnectorREST/Ticket",
+ obj._build_url())
+
+ def test__build_url_ticket_get(self):
+ """Test _build_url for ticket get"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketGet"
+
+ self.assertEqual("http://fqdn/otrs/nph-genericinterface.pl/Webservice/"
+ "GenericTicketConnectorREST/Ticket/508",
+ obj._build_url(508))
+
+ def test__build_url_ticket_get_list(self):
+ """Test _build_url for ticket get list"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketGetList"
+
+ self.assertEqual("http://fqdn/otrs/nph-genericinterface.pl/Webservice/"
+ "GenericTicketConnectorREST/TicketList",
+ obj._build_url())
+
+ def test__build_url_ticket_search(self):
+ """Test _build_url for ticket search"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketSearch"
+
+ self.assertEqual("http://fqdn/otrs/nph-genericinterface.pl/Webservice/"
+ "GenericTicketConnectorREST/Ticket",
+ obj._build_url())
+
+ def test__build_url_ticket_update(self):
+ """Test _build_url for ticket update"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketUpdate"
+
+ self.assertEqual("http://fqdn/otrs/nph-genericinterface.pl/Webservice/"
+ "GenericTicketConnectorREST/Ticket/509",
+ obj._build_url(509))
+
+ def test__build_url_ticket_update_invalid(self):
+ """Test _build_url for ticket update when required TicketID is not given"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketUpdate"
+
+ self.assertRaisesRegex(ValueError,
+ "TicketID is None but Route requires TicketID.*",
+ obj._build_url)
+
+ def test__build_url_route_invalid(self):
+ """Test _build_url with invalid route"""
+
+ operation_mapping_invalid = {
+ 'SessionCreate':
+ {'ResultType': 'SessionID', 'RequestMethod': 'POST', 'Route': '/Session'},
+ 'TicketCreate':
+ {'ResultType': 'TicketID', 'RequestMethod': 'POST', 'Route': '/barfoo'},
+ 'TicketGet':
+ {'ResultType': 'Ticket', 'RequestMethod': 'GET', 'Route': '/Ticket/:TicketID'},
+ 'TicketGetList':
+ {'ResultType': 'Ticket', 'RequestMethod': 'GET', 'Route': '/TicketList'},
+ 'TicketSearch':
+ {'ResultType': 'TicketID', 'RequestMethod': 'GET', 'Route': '/Ticket'},
+ 'TicketUpdate':
+ {'ResultType': 'TicketID', 'RequestMethod': 'PATCH', 'Route': '/foobar'}
+ }
+
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST",
+ operation_map=operation_mapping_invalid)
+
+ obj.operation = "TicketUpdate"
+ self.assertRaisesRegex(ValueError,
+ "Route misconfigured.*",
+ obj._build_url)
+
+ obj.operation = "TicketCreate"
self.assertRaisesRegex(ValueError,
- 'http_method, url and payload are required.',
- Client._send_request)
+ "Route misconfigured.*",
+ obj._build_url)
- def test_send_request_invalid_method(self):
+ def test__send_request_no_payload(self):
+ """Test _send_request no payload"""
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
+
+ self.assertRaisesRegex(ArgumentMissingError,
+ 'payload',
+ obj._send_request)
+
+ def test__send_request_invalid_method(self):
"""Test _send_request with invalid http method """
- self.assertRaisesRegex(NotImplementedError,
- '^Accepted HTTP Methods.*',
- Client._send_request,
- http_method="FOOBAR",
- url="http://localhost/",
+ operation_mapping_invalid = {
+ 'SessionCreate':
+ {'ResultType': 'SessionID', 'RequestMethod': 'FOOBAR', 'Route': '/Session'},
+ 'TicketCreate':
+ {'ResultType': 'TicketID', 'RequestMethod': 'FOOBAR', 'Route': '/Ticket'},
+ 'TicketGet':
+ {'ResultType': 'Ticket', 'RequestMethod': 'FOOBAR', 'Route': '/Ticket/:TicketID'},
+ 'TicketGetList':
+ {'ResultType': 'Ticket', 'RequestMethod': 'FOOBAR', 'Route': '/TicketList'},
+ 'TicketSearch':
+ {'ResultType': 'TicketID', 'RequestMethod': 'FOOBAR', 'Route': '/Ticket'},
+ 'TicketUpdate':
+ {'ResultType': 'TicketID', 'RequestMethod': 'FOOBAR', 'Route': '/Ticket/:TicketID'}
+ }
+
+ obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST",
+ operation_map=operation_mapping_invalid)
+
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
+
+ self.assertRaisesRegex(ValueError,
+ 'invalid http_method',
+ obj._send_request,
payload={"foo": "bar"})
@mock.patch('pyotrs.lib.requests.request')
def test__send_request_ok(self, mock_requests_req):
"""Tests _send_request ok"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mock_requests_req.return_value.result = True
mock_requests_req.return_value.status_code = 200
mock_requests_req.return_value.content = "all good"
- mocked_result = obj._send_request(http_method='GET',
- url='http://fqdn/bla/Generic/',
- payload={'foo': 'bar'})
+ mocked_result = obj._send_request(payload={'foo': 'bar'})
self.assertEqual(mock_requests_req.call_count, 1)
self.assertTrue(mocked_result.result)
@@ -686,16 +945,16 @@ class ClientTests(unittest.TestCase):
def test__send_request_http_status_code_nok(self, mock_requests_req):
"""Tests _send_request fail http status code not 200"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mock_requests_req.return_value.result = False
mock_requests_req.return_value.status_code = 500
mock_requests_req.return_value.content = "no me gusta"
- self.assertRaisesRegex(OTRSHTTPError,
+ self.assertRaisesRegex(HTTPError,
'Received HTTP Error. Check Hostname and We.*',
obj._send_request,
- http_method='GET',
- url='http://fqdn/bla/Generic/',
payload={'fooEs': 'barSp'})
self.assertEqual(mock_requests_req.call_count, 1)
@@ -704,57 +963,62 @@ class ClientTests(unittest.TestCase):
def test__send_request_fail(self, mock_requests_req):
"""Tests _send_request fail"""
obj = Client(baseurl="http://fqdn", webservicename="GenericTicketConnectorREST")
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
+
mock_requests_req.side_effect = Exception("Some Exception")
- self.assertRaisesRegex(OTRSHTTPError,
+ self.assertRaisesRegex(HTTPError,
'Failed to access OTRS. Check Hostname, Proxy, SSL.*',
obj._send_request,
- http_method='GET',
- url='http://fqdn/bla/Generic/',
payload={'foo': 'bar'})
self.assertEqual(mock_requests_req.call_count, 1)
def test__validate_response_init_invalid(self):
"""Test _validate_response_init_invalid - missing response """
- client = Client(baseurl="http://localhost", webservicename="foo")
+ obj = Client(baseurl="http://localhost", webservicename="foo")
self.assertRaisesRegex(ValueError,
'requests.Response object expected!',
- client._parse_and_validate_response,
+ obj._parse_and_validate_response,
'just_some_string')
def test__validate_response_invalid_operation(self):
"""Test _validate_response with an invalid operation"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "DoTheMagicRainDance"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "DoTheMagicRainDance"
+ # obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mocked_response = mock.Mock(spec=requests.Response)
mocked_response.status_code = 200
mocked_response.json.return_value = {}
- self.assertRaisesRegex(NotImplementedError,
- 'Unknown Operation!',
- client._parse_and_validate_response,
+ self.assertRaisesRegex(ValueError,
+ 'invalid operation',
+ obj._parse_and_validate_response,
mocked_response)
def test__validate_response_operation_session_create(self):
"""Test _validate_response with SessionCreate"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "SessionCreate"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "SessionCreate"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mocked_response = mock.Mock(spec=requests.Response)
mocked_response.status_code = 200
mocked_response.json.return_value = {u'SessionID': u'tMtTFDg1PxCXfoobarue4W5oQtNsFd0k'}
- client._parse_and_validate_response(mocked_response)
+ obj._parse_and_validate_response(mocked_response)
- self.assertEqual(client._result_type, 'SessionID')
+ self.assertEqual(obj._result_type, 'SessionID')
def test__validate_response_operation_ticket_get(self):
"""Test _validate_response with TicketGet"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "TicketGet"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "TicketGetList"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
+
tkt = {u'Ticket': [{u'Age': 24040576,
u'CreateBy': u'1',
u'CustomerID': None,
@@ -769,15 +1033,16 @@ class ClientTests(unittest.TestCase):
mocked_response.status_code = 200
mocked_response.json.return_value = tkt
- client._parse_and_validate_response(mocked_response)
+ obj._parse_and_validate_response(mocked_response)
- self.assertEqual(client._result_type, 'Ticket')
- self.assertDictEqual(client.result_json, tkt)
+ self.assertEqual(obj._result_type, 'Ticket')
+ self.assertDictEqual(obj.result_json, tkt)
def test__validate_response_operation_ticket_create(self):
"""Test _validate_response with TicketCreate"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "TicketCreate"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "TicketCreate"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mocked_response = mock.Mock(spec=requests.Response)
mocked_response.status_code = 200
@@ -785,73 +1050,78 @@ class ClientTests(unittest.TestCase):
u'TicketID': u'9',
u'TicketNumber': u'000008'}
- client._parse_and_validate_response(mocked_response)
+ obj._parse_and_validate_response(mocked_response)
- self.assertEqual(client._result_type, 'TicketID')
- self.assertDictEqual(client.result_json, {u'ArticleID': u'26',
- u'TicketID': u'9',
- u'TicketNumber': u'000008'})
+ self.assertEqual(obj._result_type, 'TicketID')
+ self.assertDictEqual(obj.result_json, {u'ArticleID': u'26',
+ u'TicketID': u'9',
+ u'TicketNumber': u'000008'})
def test__validate_response_operation_ticket_update(self):
"""Test _validate_response with TicketUpdate"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "TicketCreate"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "TicketCreate"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mocked_response = mock.Mock(spec=requests.Response)
mocked_response.status_code = 200
mocked_response.json.return_value = {u'TicketID': u'9', u'TicketNumber': u'000008'}
- client._parse_and_validate_response(mocked_response)
- self.assertEqual(client._result_type, 'TicketID')
- self.assertDictEqual(client.result_json, {u'TicketID': u'9',
- u'TicketNumber': u'000008'})
+ obj._parse_and_validate_response(mocked_response)
+ self.assertEqual(obj._result_type, 'TicketID')
+ self.assertDictEqual(obj.result_json, {u'TicketID': u'9',
+ u'TicketNumber': u'000008'})
def test__validate_response_operation_ticket_search(self):
"""Test _validate_response with TicketSearch"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "TicketSearch"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mocked_response = mock.Mock(spec=requests.Response)
mocked_response.status_code = 200
mocked_response.json.return_value = {u'TicketID': [u'9']}
- client._parse_and_validate_response(mocked_response)
- self.assertEqual(client._result_type, 'TicketID')
- self.assertDictEqual(client.result_json, {u'TicketID': [u'9']})
+ obj._parse_and_validate_response(mocked_response)
+ self.assertEqual(obj._result_type, 'TicketID')
+ self.assertDictEqual(obj.result_json, {u'TicketID': [u'9']})
def test__validate_response_operation_ticket_search_empty(self):
"""Test _validate_response with TicketSearch with empty result"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "TicketSearch"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mocked_response = mock.Mock(spec=requests.Response)
mocked_response.status_code = 200
mocked_response.json.return_value = {}
- client._parse_and_validate_response(mocked_response)
- self.assertEqual(client._result_type, 'TicketID')
- self.assertDictEqual(client.result_json, {})
+ obj._parse_and_validate_response(mocked_response)
+ self.assertEqual(obj._result_type, 'TicketID')
+ self.assertDictEqual(obj.result_json, {})
def test__validate_response_operation_ticket_search_nonsense(self):
"""Test _validate_response with TicketSearch with a nonsence response"""
- client = Client(baseurl="http://localhost", webservicename="foo")
- client.operation = "TicketSearch"
+ obj = Client(baseurl="http://localhost", webservicename="foo")
+ obj.operation = "TicketSearch"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
mocked_response = mock.Mock(spec=requests.Response)
mocked_response.status_code = 200
mocked_response.json.return_value = {u'FooBar': [u'1', u'3']}
- self.assertRaises(ResponseJSONParseError,
- client._parse_and_validate_response,
+ self.assertRaises(ResponseParseError,
+ obj._parse_and_validate_response,
mocked_response)
- self.assertEqual(client._result_type, 'TicketID')
- self.assertTrue(client._result_error)
- self.assertDictEqual(client.result_json, {u'FooBar': [u'1', u'3']})
+ self.assertEqual(obj._result_type, 'TicketID')
+ self.assertTrue(obj._result_error)
+ self.assertDictEqual(obj.result_json, {u'FooBar': [u'1', u'3']})
def test__validate_response_operation_ticket_get_error(self):
"""Test _validate_response with TicketGet when an error is received"""
obj = Client(baseurl="http://localhost", webservicename="foo")
obj.operation = "TicketGet"
+ obj._result_type = obj.operation_map[obj.operation]["ResultType"]
tkt = {"Error": {"ErrorMessage": "TicketGet: Authorization failing!",
"ErrorCode": "TicketGet.AuthFail"}}
@@ -860,15 +1130,11 @@ class ClientTests(unittest.TestCase):
mocked_response.status_code = 200
mocked_response.json.return_value = tkt
- self.assertRaisesRegex(OTRSAPIError,
+ self.assertRaisesRegex(APIError,
'Failed to access OTRS API. Check Username and Password.*',
obj._parse_and_validate_response,
mocked_response)
- # TODO 2016-04-24 (RH) missing tests:
- # * ticket_get single id and multiple ids
- # * session check is valid too
-
def main():
unittest.main()
diff --git a/tests/test_pyotrs_responses.py b/tests/test_pyotrs_responses.py
index dd891d0e76484a0711c44af1525c247d2145188d..8321abf793b29a6f9fd8a0a6ad858d6a4af45278 100644
--- a/tests/test_pyotrs_responses.py
+++ b/tests/test_pyotrs_responses.py
@@ -7,426 +7,175 @@ Test for PyOTRS using **responses**
# make sure (early) that parent dir (main app) is in path
import unittest2 as unittest
-import responses
-import mock
+import responses # noqa
+import mock # noqa
+
+import requests # noqa
from pyotrs import Client # noqa
-from pyotrs.lib import OTRSAPIError # noqa
+from pyotrs.lib import APIError # noqa
-class PyOTRSResponsesTests(unittest.TestCase):
+class FullResponsesTests(unittest.TestCase):
"""Tests using the responses module"""
- def test_session_create_req_mocked_valid(self):
- """Test session_create and _parse_and_validate_response; _send_request mocked; valid"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
-
- self.assertIsNone(obj.result_json)
- self.assertIsNone(obj.session_id_store.value)
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.POST,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Session',
- json={u'SessionID': u'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k'},
- status=200,
- content_type='application/json')
-
- result = obj.session_create()
-
- self.assertEqual(result, True)
- self.assertFalse(obj._result_error)
- self.assertEqual(obj.operation, 'SessionCreate')
- self.assertEqual(obj._result_status_code, 200)
- self.assertDictEqual(obj.result_json,
- {u'SessionID': u'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k'})
- self.assertEqual(obj.session_id_store.value, 'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k')
-
- def test_session_create_req_mocked_invalid(self):
- """Test session_create and _parse_and_validate_response; _send_request mocked; invalid"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
-
- self.assertIsNone(obj.result_json)
- self.assertIsNone(obj.session_id_store.value)
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.POST,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Session',
- json={"Error": {"ErrorCode": "SessionCreate.AuthFail",
- "ErrorMessage": "SessionCreate: Authorization failing!"}},
- status=200,
- content_type='application/json')
-
- self.assertRaisesRegex(OTRSAPIError,
- 'Failed to access OTRS API. Check Username and Password.*',
- obj.session_create)
-
- self.assertTrue(obj._result_error)
- self.assertEqual(obj.operation, 'SessionCreate')
- self.assertEqual(obj._result_status_code, 200)
- self.assertDictEqual(obj.result_json,
- {"Error": {"ErrorCode": "SessionCreate.AuthFail",
- "ErrorMessage": "SessionCreate: Authorization failing!"}})
- self.assertIsNone(obj.session_id_store.value)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test_session_create_req_mocked_failed_validation(self, mock_validate_resp):
- """Test session_create; _parse_and_validate_response and _send_request mocked; fail vali"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
-
- self.assertIsNone(obj.result_json)
- self.assertIsNone(obj.session_id_store.value)
-
- mock_validate_resp.return_value = False
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.POST,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Session',
- json={u'SessionID': u'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k'},
- status=200,
- content_type='application/json')
-
- result = obj.session_create()
-
- self.assertFalse(result)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_create_json_mocked(self, mock_validate_resp):
- """Test _ticket_create_json - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket".format(obj)
-
- mock_validate_resp.return_value = True
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.POST,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket',
- json={u'ArticleID': u'2', u'TicketID': u'2', u'TicketNumber': u'000001'},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_create_json(url=url, payload={"bar": "ticket-create"})
-
- self.assertEqual(obj.operation, 'TicketCreate')
- self.assertTrue(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_create_json_mocked_fail_validation(self, mock_validate_resp):
- """Test _ticket_create_json - mocked _fail_validation"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket".format(obj)
-
- mock_validate_resp.return_value = False
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.POST,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket',
- json={u'ArticleID': u'2', u'TicketID': u'2', u'TicketNumber': u'000001'},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_create_json(url=url, payload={"bar": "ticket-create"})
-
- self.assertEqual(obj.operation, 'TicketCreate')
- self.assertFalse(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_get_json_mocked(self, mock_validate_resp):
- """Test _ticket_get_json - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket".format(obj)
-
- mock_validate_resp.return_value = True
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.GET,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket',
- json={u'Ticket': [{u'Age': 24040576,
- u'ArchiveFlag': u'n',
- u'ChangeBy': u'1',
- u'Changed': u'2016-04-13 20:41:19',
- u'CreateBy': u'1',
- u'StateType': u'open',
- u'TicketID': u'1',
- u'TicketNumber': u'2015071510123456',
- u'Title': u'Welcome to OTRS!',
- u'Type': u'Unclassified',
- u'TypeID': 1,
- u'UnlockTimeout': u'0',
- u'UntilTime': 0}]},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_get_json(url=url, payload={"bla": "ticket-get"})
-
- self.assertEqual(obj.operation, 'TicketGet')
- self.assertTrue(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_get_json_mocked_fail_validation(self, mock_validate_resp):
- """Test _ticket_create_json - mocked _fail_validation"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket".format(obj)
-
- mock_validate_resp.return_value = False
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.GET,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket',
- json={u'Ticket': [{u'Age': 24040576,
- u'ArchiveFlag': u'n',
- u'ChangeBy': u'1',
- u'TicketID': u'1',
- u'TicketNumber': u'2015071510123456',
- u'Title': u'Welcome to OTRS!',
- u'Type': u'Unclassified',
- u'TypeID': 1,
- u'UnlockTimeout': u'0',
- u'UntilTime': 0}]},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_get_json(url=url, payload={"bar": "ticket-create"})
-
- self.assertEqual(obj.operation, 'TicketGet')
- self.assertFalse(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_search_json_mocked(self, mock_validate_resp):
- """Test _ticket_search_json - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket".format(obj)
-
- mock_validate_resp.return_value = True
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.GET,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket',
- json={u'TicketID': [u'1']},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_search_json(url=url, payload={"bla": "ticket-search"})
-
- self.assertEqual(obj.operation, 'TicketSearch')
- self.assertTrue(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_search_json_mocked_fail_validation(self, mock_validate_resp):
- """Test _ticket_search_json - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket".format(obj)
-
- mock_validate_resp.return_value = False
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.GET,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket',
- json={u'TicketID': [u'1']},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_search_json(url=url, payload={"bla": "ticket-search"})
-
- self.assertEqual(obj.operation, 'TicketSearch')
- self.assertFalse(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_update_json_mocked(self, mock_validate_resp):
- """Test _ticket_update_json - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket/{1}".format(obj, 1)
-
- mock_validate_resp.return_value = True
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.PATCH,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket/1',
- json={u'TicketID': u'9', u'TicketNumber': u'000008'},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_update_json(url=url, payload={"alb": "ticket-update"})
-
- self.assertEqual(obj.operation, 'TicketUpdate')
- self.assertTrue(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_update_json_mocked_fail_validation(self, mock_validate_resp):
- """Test _ticket_update_json - mocked _fail_validation"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket/{1}".format(obj, 1)
-
- mock_validate_resp.return_value = False
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.PATCH,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket/1',
- json={u'TicketID': u'9', u'TicketNumber': u'000008'},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_update_json(url=url, payload={"alb": "ticket-update"})
-
- self.assertEqual(obj.operation, 'TicketUpdate')
- self.assertFalse(result)
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- def test__ticket_update_json_w_article_ok_mocked(self):
- """Test _ticket_update_json with article and parse+validate - ok - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket/{1}".format(obj, 2)
-
- art = {'Article': {'Subject': 'Dümmy Subject',
- 'Body': 'Hallo Bjørn,\n[kt]\n\n -- The End',
- 'TimeUnit': 0,
- 'MimeType': 'text/plain',
- 'Charset': 'UTF8'}}
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.PATCH,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket/2',
- json={u'ArticleID': u'2', u'TicketID': u'2', u'TicketNumber': u'000002'},
- status=200,
- content_type='application/json')
-
- result = obj._ticket_update_json(url=url, payload=art)
-
- self.assertEqual(obj.operation, 'TicketUpdate')
- self.assertTrue(result)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_update_json_w_article_nok_unknown_exception_mocked(self, mock_validate_resp):
- """Test _ticket_update_json with article - nok - exception (unknown) - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket/{1}".format(obj, 3)
-
- art = {'Article': {'Subject': 'Dümmy Subject',
- 'Body': 'Hallo Bjørn,\n[kt]\n\n -- The End',
- 'TimeUnit': 0,
- 'MimeType': 'text/plain',
- 'Charset': 'UTF8'}}
-
- mock_validate_resp.return_value = True
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.PATCH,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket/3',
- json={u'TicketID': u'3', u'TicketNumber': u'000003'},
- status=200,
- content_type='application/json')
-
- self.assertRaisesRegex(ValueError,
- 'Unknown Exception',
- obj._ticket_update_json,
- url=url, payload=art)
-
- self.assertEqual(obj.operation, 'TicketUpdate')
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
- def test__ticket_update_json_w_article_nok_exception_mocked(self, mock_validate_resp):
- """Test _ticket_update_json with article - nok - reraised exception - mocked"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = "{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/" \
- "{0.webservicename}/Ticket/{1}".format(obj, 4)
-
- art = {'Article': {'Subject': 'Dümmy Subject',
- 'Body': 'Hallo Bjørn,\n[kt]\n\n -- The End',
- 'TimeUnit': 0,
- 'MimeType': 'text/plain',
- 'Charset': 'UTF8'}}
-
- mock_validate_resp.return_value = True
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.PATCH,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket/4',
- json={u'ArticleID': u'', u'TicketID': u'4', u'TicketNumber': u'000004'},
- status=200,
- content_type='application/json')
-
- self.assertRaisesRegex(ValueError,
- 'Unknown Ex.*',
- obj._ticket_update_json,
- url=url, payload=art)
-
- self.assertEqual(obj.operation, 'TicketUpdate')
- self.assertEqual(mock_validate_resp.call_count, 1)
-
- def test__ticket_update_json_w_article_nok_exception_mocked_no_art(self):
- """Test _ticket_update_json with article - nok - reraised exception - mocked - no art"""
- obj = Client(baseurl="http://fqdn",
- webservicename="GenericTicketConnectorREST")
- url = ("{0.baseurl}/otrs/nph-genericinterface.pl/Webservice/"
- "{0.webservicename}/Ticket/{1}".format(obj, 4))
-
- art = {'Article': {'Subject': 'Dümmy Subject',
- 'Body': 'Hallo Bjørn,\n[kt]\n\n -- The End',
- 'TimeUnit': 0,
- 'MimeType': 'text/plain',
- 'Charset': 'UTF8'}}
-
- with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
- rsps.add(responses.PATCH,
- 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
- 'GenericTicketConnectorREST/Ticket/4',
- json={u'TicketID': u'4', u'TicketNumber': u'000004'},
- status=200,
- content_type='application/json')
-
- self.assertRaisesRegex(ValueError,
- 'Unknown Ex.*',
- obj._ticket_update_json,
- url=url, payload=art)
-
- self.assertEqual(obj.operation, 'TicketUpdate')
-
- # TODO 2016-04-24 (RH) missing tests:
- # * ticket_get single id and multiple ids
+ # def test_session_create_req_mocked_valid(self):
+ # """Test session_create and _parse_and_validate_response; _send_request mocked; valid"""
+ # obj = Client(baseurl="http://fqdn",
+ # webservicename="GenericTicketConnectorREST")
+ #
+ # self.assertIsNone(obj.result_json)
+ # self.assertIsNone(obj.session_id_store.value)
+ #
+ # with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
+ # rsps.add(responses.POST,
+ # 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
+ # 'GenericTicketConnectorREST/Session',
+ # json={u'SessionID': u'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k'},
+ # status=200,
+ # content_type='application/json')
+ #
+ # result = obj.session_create()
+ #
+ # self.assertEqual(result, True)
+ # self.assertFalse(obj._result_error)
+ # self.assertEqual(obj.operation, 'SessionCreate')
+ # self.assertEqual(obj._result_status_code, 200)
+ # self.assertDictEqual(obj.result_json,
+ # {u'SessionID': u'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k'})
+ # self.assertEqual(obj.session_id_store.value, 'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k')
+ #
+ # def test_session_create_req_mocked_invalid(self):
+ # """Test session_create and _parse_and_validate_response; _send_request mocked; invalid"""
+ # obj = Client(baseurl="http://fqdn",
+ # webservicename="GenericTicketConnectorREST")
+ #
+ # self.assertIsNone(obj.result_json)
+ # self.assertIsNone(obj.session_id_store.value)
+ #
+ # with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
+ # rsps.add(responses.POST,
+ # 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
+ # 'GenericTicketConnectorREST/Session',
+ # json={"Error": {"ErrorCode": "SessionCreate.AuthFail",
+ # "ErrorMessage": "SessionCreate: Authorization failing!"}},
+ # status=200,
+ # content_type='application/json')
+ #
+ # self.assertRaisesRegex(APIError,
+ # 'Failed to access OTRS API. Check Username and Password.*',
+ # obj.session_create)
+ #
+ # self.assertTrue(obj._result_error)
+ # self.assertEqual(obj.operation, 'SessionCreate')
+ # self.assertEqual(obj._result_status_code, 200)
+ # expected_dct = {"Error": {"ErrorCode": "SessionCreate.AuthFail",
+ # "ErrorMessage": "SessionCreate: Authorization failing!"}}
+ # self.assertDictEqual(obj.result_json, expected_dct)
+ # self.assertIsNone(obj.session_id_store.value)
+ #
+ # @mock.patch('pyotrs.lib.Client._parse_and_validate_response', autospec=True)
+ # def test_session_create_req_mocked_failed_validation(self, mock_validate_resp):
+ # """Test session_create; _parse_and_val_response and _send_request mocked; fail vali"""
+ # obj = Client(baseurl="http://fqdn",
+ # webservicename="GenericTicketConnectorREST")
+ #
+ # self.assertIsNone(obj.result_json)
+ # self.assertIsNone(obj.session_id_store.value)
+ #
+ # mock_validate_resp.return_value = False
+ #
+ # with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
+ # rsps.add(responses.POST,
+ # 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
+ # 'GenericTicketConnectorREST/Session',
+ # json={u'SessionID': u'tMtTFDg1PxCX51dWnjue4W5oQtNsFd0k'},
+ # status=200,
+ # content_type='application/json')
+ #
+ # result = obj.session_create()
+ #
+ # self.assertFalse(result)
+
+ # def test_ticket_update_w_art_nok_unknown_exception(self):
+ # """Test ticket_update with article - nok - exception (unknown) - mocked"""
+ # obj = Client(baseurl="http://fqdn",
+ # webservicename="GenericTicketConnectorREST")
+ # obj.operation = "TicketUpdate"
+ # obj._result_type = obj.operation_map[obj.operation]["ResultType"]
+ #
+ # art = {'Article': {'Subject': 'Dümmy Subject',
+ # 'Body': 'Hallo Bjørn,\n[kt]\n\n -- The End',
+ # 'TimeUnit': 0,
+ # 'MimeType': 'text/plain',
+ # 'Charset': 'UTF8'}}
+ #
+ # with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
+ # rsps.add(responses.PATCH,
+ # 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
+ # 'GenericTicketConnectorREST/Ticket/3',
+ # json={u'TicketID': u'3', u'TicketNumber': u'000003'},
+ # status=200,
+ # content_type='application/json')
+ #
+ # self.assertRaisesRegex(ValueError,
+ # 'Unknown Exception',
+ # obj.ticket_update,
+ # payload=art,
+ # ticket_id=3)
+ #
+ # def test_ticket_update_w_article_nok_exception(self):
+ # """Test ticket_update with article - nok - reraised exception - mocked"""
+ # obj = Client(baseurl="http://fqdn",
+ # webservicename="GenericTicketConnectorREST")
+ # obj.operation = "TicketUpdate"
+ # obj._result_type = obj.operation_map[obj.operation]["ResultType"]
+ #
+ # art = {'Article': {'Subject': 'Dümmy Subject',
+ # 'Body': 'Hallo Bjørn,\n[kt]\n\n -- The End',
+ # 'TimeUnit': 0,
+ # 'MimeType': 'text/plain',
+ # 'Charset': 'UTF8'}}
+ #
+ # with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
+ # rsps.add(responses.PATCH,
+ # 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
+ # 'GenericTicketConnectorREST/Ticket/4',
+ # json={u'ArticleID': u'', u'TicketID': u'4', u'TicketNumber': u'000004'},
+ # status=200,
+ # content_type='application/json')
+ #
+ # self.assertRaisesRegex(ValueError,
+ # 'Unknown Ex.*',
+ # obj.ticket_update,
+ # payload=art,
+ # ticket_id=4)
+ #
+ # def test_ticket_update_w_article_nok_exception_mocked_no_art(self):
+ # """Test ticket_update with article - nok - reraised excep mocked - no art"""
+ # obj = Client(baseurl="http://fqdn",
+ # webservicename="GenericTicketConnectorREST")
+ # obj.operation = "TicketUpdate"
+ # obj._result_type = obj.operation_map[obj.operation]["ResultType"]
+ #
+ # art = {'Article': {'Subject': 'Dümmy Subject',
+ # 'Body': 'Hallo Bjørn,\n[kt]\n\n -- The End',
+ # 'TimeUnit': 0,
+ # 'MimeType': 'text/plain',
+ # 'Charset': 'UTF8'}}
+ #
+ # with responses.RequestsMock(assert_all_requests_are_fired=True) as rsps:
+ # rsps.add(responses.PATCH,
+ # 'http://fqdn/otrs/nph-genericinterface.pl/Webservice/'
+ # 'GenericTicketConnectorREST/Ticket/5',
+ # json={u'TicketID': u'4', u'TicketNumber': u'000004'},
+ # status=200,
+ # content_type='application/json')
+ #
+ # self.assertRaisesRegex(ValueError,
+ # 'Unknown Ex.*',
+ # obj.ticket_update,
+ # payload=art,
+ # ticket_id=5)
# Main
@@ -437,4 +186,4 @@ def main():
if __name__ == '__main__':
main()
- # EOF
+# EOF