From 60eaa295d1027ba011ab98b70a7de45d8e8ab184 Mon Sep 17 00:00:00 2001
From: Robert Habermann <mail@rhab.de>
Date: Mon, 11 Apr 2016 22:38:26 +0200
Subject: [PATCH] wip
---
pyotrs/__init__.py | 2 +-
pyotrs/client.py | 150 ++++------
pyotrs/pyotrs.py | 638 -----------------------------------------
tests/test_doctests.py | 3 +-
tests/test_pyotrs.py | 21 +-
tox.ini | 4 +-
6 files changed, 67 insertions(+), 751 deletions(-)
delete mode 100644 pyotrs/pyotrs.py
diff --git a/pyotrs/__init__.py b/pyotrs/__init__.py
index f8c1702..53a363d 100644
--- a/pyotrs/__init__.py
+++ b/pyotrs/__init__.py
@@ -1,2 +1,2 @@
""" __init__.py """
-from .pyotrs import PyOTRS
+from .client import *
diff --git a/pyotrs/client.py b/pyotrs/client.py
index 1c0e670..0ee2f7b 100644
--- a/pyotrs/client.py
+++ b/pyotrs/client.py
@@ -111,30 +111,31 @@ class Client(object):
self.webservicename = webservicename
if not session_id_file:
- self.session_id_file = ".session_id.tmp"
+ self.session_id_file = "/tmp/.session_id.tmp"
+ else:
+ self.session_id_file = session_id_file
if not session_timeout:
self.session_timeout = 28800 # 8 hours is OTRS default
+ self.https_verify = https_verify
+
+ # dummy initialization
+ self.data = None
+ self.data_type = None
+ self.data_json = None
+
# if session_id is passed in with a value other than None
# ignore username and password and use session_id
# if session_id is None then restore or create a Session!
self.username = username
self.password = password
- self.session_id_file = session_id_file
if not session_id:
self.restore_or_set_up_session()
else:
self.session_id = session_id
- self.https_verify = https_verify
-
- # dummy initialization
- self.data = None
- self.data_type = None
- self.data_json = None
-
"""
GenericInterface::Operation::Session::SessionCreate
Client public methods:
@@ -180,9 +181,9 @@ class Client(object):
"""
if not session_id:
- client = TicketGet(self.baseurl, self.webservicename, session_id=self.session_id)
+ client = Client(self.baseurl, self.webservicename, session_id=self.session_id)
else:
- client = TicketGet(self.baseurl, self.webservicename, session_id=session_id)
+ client = Client(self.baseurl, self.webservicename, session_id=session_id)
return client.ticket_get_by_id(1, dynamic_fields=0, all_articles=0)
@@ -192,12 +193,21 @@ class Client(object):
"""
if os.path.isfile(self.session_id_file):
with open(self.session_id_file, "r") as f:
- data = json.loads(f.read())
- created = data['created']
+ content = f.read()
+ try:
+ data = json.loads(content)
+ created = datetime.datetime.utcfromtimestamp(int(data['created']))
+
expires = created + datetime.timedelta(minutes=self.session_timeout)
if expires > datetime.datetime.utcnow():
self.session_id = data['session_id'] # still valid
return True
+ except ValueError as err:
+ logger.error("JSON Parse Exception: {}".format(err))
+ pass # pass this exception as
+ except Exception as err:
+ logger.error("Some Exception: {}".format(err))
+ raise Exception("Some Exception: {}".format(err))
return False
@@ -210,7 +220,7 @@ class Client(object):
"""
with open(self.session_id_file, "w") as f:
- f.write(json.dumps({'created': datetime.datetime.utcnow(),
+ f.write(json.dumps({'created': str(int(time.time())),
'session_id': self.session_id}))
return True
@@ -224,7 +234,7 @@ class Client(object):
payload = {
"UserLogin": self.username,
- "Password": self.password
+ "Password": self.password
}
try:
@@ -246,20 +256,18 @@ class Client(object):
return True
-
-class TicketCreate(Client):
"""
GenericInterface::Operation::Ticket::TicketCreate
- TicketCreate public methods:
+ public methods:
+ * ticket_create
"""
- pass
+ def ticket_create(self):
+ pass
-
-class TicketGet(Client):
"""
GenericInterface::Operation::Ticket::TicketGet
- TicketGet public methods:
+ public methods:
* ticket_get_by_id
"""
@@ -343,8 +351,8 @@ class TicketGet(Client):
"{0.webservicename}/Ticket/{1}".format(self, ticket_id)
payload = {
- "SessionID": self.session_id,
- "AllArticles": all_articles,
+ "SessionID": self.session_id,
+ "AllArticles": all_articles,
"DynamicFields": dynamic_fields
}
@@ -354,11 +362,9 @@ class TicketGet(Client):
logger.error("Exception: {0}".format(err))
return False
-
-class TicketSearch(Client):
"""
GenericInterface::Operation::Ticket::TicketSearch
- TicketSearch public methods:
+ public methods:
* ticket_search
* ticket_search_by_past_days
"""
@@ -488,7 +494,7 @@ class TicketSearch(Client):
older = newer - 1440 * days
payload = {
- "SessionID": self.session_id,
+ "SessionID": self.session_id,
"TicketLastChangeTimeNewerMinutes": newer,
"TicketLastChangeTimeOlderMinutes": older
}
@@ -501,17 +507,13 @@ class TicketSearch(Client):
logger.error("Exception: {0}".format(err))
return False
-
-class TicketUpdate(Client):
- """
- GenericInterface::Operation::Ticket::TicketUpdate
- TicketUpdate public methods:
- * ticket_update_add_article
- * ticket_update_set_dynamic_field_value
- * ticket_update_set_state
- * ticket_update_set_state_pending
- * ticket_update_set_title
- """
+ # GenericInterface :: Operation::Ticket :: TicketUpdate
+ # public methods
+ # * ticket_update_add_article
+ # * ticket_update_set_dynamic_field_value
+ # * ticket_update_set_state
+ # * ticket_update_set_state_pending
+ # * ticket_update_set_title
def _update_by_ticket_id(self, url, payload):
""" update_by_ticket_id
@@ -589,9 +591,9 @@ class TicketUpdate(Client):
payload = {
"SessionID": self.session_id,
"Article": {
- "Subject": article_subject,
- "Body": article_body,
- "TimeUnit": 0,
+ "Subject": article_subject,
+ "Body": article_body,
+ "TimeUnit": 0,
"ContentType": "text/plain; charset=ISO-8859-15"
}
}
@@ -629,7 +631,7 @@ class TicketUpdate(Client):
"{0.webservicename}/Ticket/{1}".format(self, ticket_id)
payload = {
- "SessionID": self.session_id,
+ "SessionID": self.session_id,
"DynamicField": {"Name": dynamic_field_name, "Value": dynamic_field_value}
}
@@ -658,7 +660,7 @@ class TicketUpdate(Client):
payload = {
"SessionID": self.session_id,
- "Ticket": {"State": new_state}
+ "Ticket": {"State": new_state}
}
logger.debug("Updating {0} with: ".format(ticket_id))
@@ -693,16 +695,16 @@ class TicketUpdate(Client):
pending_time = time.gmtime(
time.time() + (pending_hours * 3600) + (pending_days * 3600 * 24))
pending_time_str = {
- "Year": pending_time.tm_year,
- "Month": pending_time.tm_mon,
- "Day": pending_time.tm_mday,
- "Hour": pending_time.tm_hour,
+ "Year": pending_time.tm_year,
+ "Month": pending_time.tm_mon,
+ "Day": pending_time.tm_mday,
+ "Hour": pending_time.tm_hour,
"Minute": pending_time.tm_min
}
payload = {
"SessionID": self.session_id,
- "Ticket": {"State": new_state, "PendingTime": pending_time_str}
+ "Ticket": {"State": new_state, "PendingTime": pending_time_str}
}
logger.debug("Updating {0} with: ".format(ticket_id))
@@ -742,54 +744,4 @@ class TicketUpdate(Client):
logger.error("Exception: {0}".format(err))
return False
-
-def _get_numeric_logger_level_from_string(level):
- """Parse string and return appropriate numeric logger level
-
- Args:
- level (str): the string representation of a logging level.
- Level is one of these: [NOTSET|TRACE|DEBUG|INFO|WARNING|ERROR|CRITICAL]
-
- Returns:
- int: value of the log level - see doctests below for mapping
- if for some reason the level is not known return DEBUG (10)
-
- Examples:
-
- >>> print(_get_numeric_logger_level_from_string("NOTSET"))
- 0
- >>> print(_get_numeric_logger_level_from_string("TRACE"))
- 5
- >>> print(_get_numeric_logger_level_from_string("DEBUG"))
- 10
- >>> print(_get_numeric_logger_level_from_string("INFO"))
- 20
- >>> print(_get_numeric_logger_level_from_string("WARNING"))
- 30
- >>> print(_get_numeric_logger_level_from_string("ERROR"))
- 40
- >>> print(_get_numeric_logger_level_from_string("CRITICAL"))
- 50
- >>> print(_get_numeric_logger_level_from_string("FOO_BAR"))
- 10
-
- """
-
- if level == "NOTSET":
- return 0
- elif level == "TRACE":
- return 5
- elif level == "DEBUG":
- return 10
- elif level == "INFO":
- return 20
- elif level == "WARNING":
- return 30
- elif level == "ERROR":
- return 40
- elif level == "CRITICAL":
- return 50
- else:
- return 10
-
# EOF
diff --git a/pyotrs/pyotrs.py b/pyotrs/pyotrs.py
deleted file mode 100644
index 62f694a..0000000
--- a/pyotrs/pyotrs.py
+++ /dev/null
@@ -1,638 +0,0 @@
-# -*- coding: utf-8 -*-
-"""Python wrapper for OTRS (using REST API)
-Name: pyotrs.py
-Description: PyOTRS
-
-Author: robert.habermann@dlh.de
-Date: 2015-07-20
-"""
-
-# Versioning
-from .version import __version__
-
-import os
-import sys
-import json
-import time
-import datetime
-import logging
-
-# import pymisc
-
-import requests
-from requests.packages.urllib3 import disable_warnings
-
-### turn of plattform insecurity warnings from urllib3
-disable_warnings()
-
-logger = logging.getLogger(__name__)
-
-### path to certificate bundle and set to environment
-# ca_certs = os.path.join(os.path.dirname(os.path.realpath(__file__)), os.pardir) +
-# "/LufthansaCA.crt"
-#ca_certs = os.path.abspath("LufthansaCA.crt")
-ca_certs = os.path.abspath("/etc/ssl/certs/ca-certificates.crt")
-
-os.environ['REQUESTS_CA_BUNDLE'] = ca_certs
-os.environ['CURL_CA_BUNDLE'] = ca_certs
-
-### disable any proxies
-proxies = {"http": "", "https": "", "no": ""}
-
-
-class PyOTRS(object):
- """PyOTRS class"""
-
- def __init__(self,
- baseurl,
- webservicename,
- username=None,
- password=None,
- session_id=None,
- session_id_file="/tmp/.session_id.tmp",
- https_verify=True):
-
- self.baseurl = baseurl
- self.webservicename = webservicename
- self.data = None
- self.https_verify = https_verify
-
- # dummy initialization
- self.data_type = None
- self.data_json = None
-
- # if session_id is passed in with a value other than None
- # ignore username and password and use session_id
- # if session_id is None then restore or create a Session!
- self.username = username
- self.password = password
- self.session_id_file = session_id_file
-
- if session_id == None:
- self.restore_or_set_up_session()
- else:
- self.session_id = session_id
-
- # Session Handling
-
- def restore_or_set_up_session(self):
- """ Try to restore Session ID from file otherwise create new one
-
- """
-
- # try to read session id from file
- if self._read_session_id_from_file():
- # got one.. check whether it's still valid
- if self._check_is_valid_session_id():
- logger.debug("Using valid Session ID from file (" + self.session_id_file + ")")
- return True
-
- # got no (valid) Session ID from file.. try to create new one
- if self.create_session():
- # safe new Session ID to file
- if self._write_session_id_to_file():
- logger.debug("Saved new Session ID to file: " + self.session_id_file)
- return True
- else:
- logger.error("Failed to save Session ID to file: " + self.session_id_file)
- # pymisc.send_message_via_mail(subject="OTRS PyOTRS - Error - Failed to save
- # Session ID!")
- sys.exit(1)
- else:
- logger.error("Failed to create a Session ID!")
- # pymisc.send_message_via_mail(subject="OTRS PyOTRS - Error - Failed to create a
- # Session ID!")
- sys.exit(1)
-
- def _check_is_valid_session_id(self, session_id=None):
- """ check whether given Session ID is still valid
-
- Args:
- session_id (str): A string containing the OTRS Session ID (default stored ID
- from obj)
-
- """
- if session_id == None:
- session_id = self.session_id
-
- return self.get_json_ticket_data_by_ticket_id(1, DynamicFields=0, AllArticles=0)
-
- def _read_session_id_from_file(self):
- """ Retrieve a stored Session ID from file
-
- """
- if os.path.isfile(self.session_id_file):
- with open(self.session_id_file, 'r') as f:
- data = f.read()
- self.session_id = data
- return True
-
- return False
-
- def _write_session_id_to_file(self):
- """ Write and store a Session ID to file
-
- Todo:
- Error Handling and return True/False
-
- """
-
- with open(self.session_id_file, 'w') as f:
- f.write(self.session_id)
- return True
-
- return False
-
- def create_session(self):
- """ create new OTRS Session and store Session ID
-
- """
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + \
- self.webservicename + "/Session"
-
- payload = {
- 'UserLogin': self.username,
- 'Password' : self.password
- }
-
- try:
- self.data = requests.post(url, params=payload, proxies=proxies,
- verify=self.https_verify)
-
- except requests.exceptions.RequestException as e:
- print("Requests Exception: {}".format(e))
- sys.exit(1)
-
- try:
- self.data_json = json.loads(self.data.content.decode())
- self.session_id = self.data_json['SessionID']
- return True
- except Exception as e:
- print("JSON Parse Exception: {}".format(e))
- sys.exit(1)
-
-
- ### Gets (GenericInterface::Operation::Ticket::TicketGet)
-
- def _get_json_ticket_data(self, url, payload):
- ''' helper function
- _get_json_ticket_data
- url [str]
- payload [dict]
- '''
-
- try:
- self.data = requests.get(url, proxies=proxies, params=payload,
- verify=self.https_verify)
-
- # Just assume that we get valid JSON
- self.data_json = json.loads(self.data.content.decode())
- logger.debug(self.data_json)
-
- if self.data_json.get('Ticket') is not None:
- self.data_type = 'Ticket'
- elif self.data_json.get('Error') is not None:
- self.data_type = 'Error'
- elif self.data_json.get('Session') is not None:
- self.data_type = 'Session'
- else:
- logger.error('Unkown key in response JSON DICT!')
- # critical error: Unknown response from OTRS API - FAIL NOW!
- sys.exit(1)
- logger.debug('Type was set to: ' + self.data_type)
-
- get_request_result_bool = True
-
- # critical error: HTTP request resulted in an error!
- except Exception as e:
- logger.error('Error with http communication: %s', e)
- logger.error(
- "Failed to access OTRS HTTP. Check Hostname, Proxy, SSL Certificate...!")
-
- get_request_result_bool = False
- raise Exception('get http')
-
- # critical error: OTRS API sent an error!
- if self.data_type == "Error":
- logger.error(self.data_json['Error'])
- logger.error(
- "Failed to access OTRS API. Check Username and Password! Session ID expired?! "
- "Does requested Ticket exist?")
- get_request_result_bool = False
- raise Exception('get api')
-
- _df_dict = {}
-
- _df_name_value_list = self.data_json['Ticket'][0]['DynamicField']
- for _df_name_value in _df_name_value_list:
- _df_dict[_df_name_value['Name']] = _df_name_value['Value']
-
- self.ticket_dynamic_fields = _df_dict
-
- return get_request_result_bool
-
-
- def get_json_ticket_data_by_ticket_id(self, ticket_id, DynamicFields=1, AllArticles=0):
- """ get_json_ticket_data_by_ticket_id
- DynamicFields [0|1] will request OTRS to include all Dynamic Fields. Defaults to on
- AllArticles [0|1] will request OTRS to include all Articles. Defaults to off
-
- Args:
- ticket_id (int): Integer value of a Ticket ID
- DynamicFields (int): will request OTRS to include all Dynamic Fields (default: 1)
- AllArticles (int): will request OTRS to include all Articles (default: 0)
-
- Returns:
- bool [true|false]
- failed or successful
-
- """
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket/" + str(
- ticket_id)
-
- payload = {
- 'SessionID' : self.session_id,
- 'AllArticles' : AllArticles,
- 'DynamicFields': DynamicFields
- }
-
- try:
- return self._get_json_ticket_data(url, payload)
- except:
- return False
-
-
- ### Updates (GenericInterface::Operation::Ticket::TicketUpdate)
-
- def _update_by_ticket_id(self, url, payload):
- ''' update_by_ticket_id
- url [str]
- payload [dict]
- '''
-
- try:
- self.data = requests.patch(url, proxies=proxies, data=json.dumps(payload),
- verify=self.https_verify)
- logger.debug("Received HTTP Status Code: " + str(self.data.status_code))
- logger.debug(self.data.content)
-
- # Just assume that we get valid JSON
- self.data_json = json.loads(self.data.content.decode())
- # logger.debug(self.data_json)
-
- if self.data_json.get('Error') is not None:
- self.data_type = 'Error'
- elif self.data_json.get('TicketID') is not None:
- self.data_type = 'TicketID'
- else:
- logger.error('Unkown key in response JSON DICT!')
- patch_request_result_bool = False
- logger.debug('Type was set to: ' + self.data_type)
-
- patch_request_result_bool = True
-
- # critical error: HTTP request resulted in an error!
- except Exception as e:
- logger.error('Error with http communication: %s', e)
- logger.error(
- "Failed to access OTRS HTTP. Check Hostname, Proxy, SSL Certificate and so on!")
-
- patch_request_result_bool = False
- raise Exception('update http')
-
- # critical error: OTRS API sent an error!
- if self.data_type == "Error":
- logger.error(self.data_json['Error'])
- logger.error(
- "Failed to access OTRS API. Check Username and Password! Does requested Ticket "
- "exist?")
-
- patch_request_result_bool = False
- raise Exception('update api')
-
- return patch_request_result_bool
-
-
- def update_by_ticket_id_set_title(self, ticket_id, title):
- ''' update_by_ticket_id_set_scout_id
- ticket_id [int]
- title [str]
- '''
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket/" + str(
- ticket_id)
-
- payload = {
- 'SessionID': self.session_id,
- 'Ticket' : {'Title': title}
- }
-
- logger.debug("Updating " + str(ticket_id) + " with: ")
- logger.debug(json.dumps(payload))
-
- try:
- return self._update_by_ticket_id(url, payload)
- except:
- return False
-
-
- def update_by_ticket_id_set_scout_id(self, ticket_id, scout_id):
- ''' update_by_ticket_id_set_scout_id
- ticket_id [int]
- scout_id [str]
- '''
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket/" + str(
- ticket_id)
-
- payload = {
- 'SessionID' : self.session_id,
- 'DynamicField': {'Name': 'ScoutIM', 'Value': scout_id}
- }
-
- logger.debug("Updating " + str(ticket_id) + " with: ")
- logger.debug(json.dumps(payload))
-
- try:
- return self._update_by_ticket_id(url, payload)
- except:
- return False
-
-
- def update_by_ticket_id_set_state(self, ticket_id, new_state="open"):
- ''' update_by_ticket_id_set_scout_id
- ticket_id [int]
- new_state [str] default: "open"
- '''
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket/" + str(
- ticket_id)
-
- payload = {
- 'SessionID': self.session_id,
- 'Ticket' : {'State': new_state}
- }
-
- logger.debug("Updating " + str(ticket_id) + " with: ")
- logger.debug(json.dumps(payload))
-
- try:
- return self._update_by_ticket_id(url, payload)
- except:
- return False
-
-
- def update_by_ticket_id_set_pending_state(self, ticket_id, new_state="pending reminder",
- pending_days=1, pending_hours=0):
- ''' update_by_ticket_id_set_scout_id
- ticket_id [int]
- state [str]
- pending_days [int] default: 1
- pending_hours [int] default: 0
- '''
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket/" + str(
- ticket_id)
-
- pending_time = time.gmtime(
- time.time() + (pending_hours * 3600) + (pending_days * 3600 * 24))
- pending_time_str = {
- "Year" : pending_time.tm_year,
- "Month" : pending_time.tm_mon,
- "Day" : pending_time.tm_mday,
- "Hour" : pending_time.tm_hour,
- "Minute": pending_time.tm_min
- }
-
- payload = {
- 'SessionID': self.session_id,
- 'Ticket' : {'State': new_state, 'PendingTime': pending_time_str}
- }
-
- logger.debug("Updating " + str(ticket_id) + " with: ")
- logger.debug(json.dumps(payload))
-
- try:
- return self._update_by_ticket_id(url, payload)
- except:
- return False
-
-
- def update_by_ticket_id_add_article(self, ticket_id, article_subject, article_body):
- ''' update_by_ticket_id_set_scout_id
- article_subject [str]
- article_body [str]
- '''
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket/" + str(
- ticket_id)
-
- ### TODO ISO-8859-15 ? Oder UTF-8?!
- payload = {
- 'SessionID': self.session_id,
- 'Article' : {
- 'Subject' : article_subject,
- 'Body' : article_body,
- 'TimeUnit' : 0,
- 'ContentType': 'text/plain; charset=ISO-8859-15'
- }
- }
-
- logger.debug("Updating " + str(ticket_id) + " with: ")
- logger.debug(json.dumps(payload))
-
- try:
- return self._update_by_ticket_id(url, payload)
- except:
- return False
-
-
- ### Searches (GenericInterface::Operation::Ticket::TicketSearch)
-
- def search_ticket(self, payload):
- """Wrapper for search ticket
-
- Args:
- payload (dict)
-
- Raises:
-
-
- Returns:
- True|False
-
- """
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket"
-
- return self._search_json_ticket_data(url, payload)
-
-
- def _search_json_ticket_data(self, url, payload):
- ''' helper function
- _search_json_ticket_data
- url [str]
- payload [dict]
- '''
-
- headers = {'Content-Type': 'application/json'}
-
- json_payload = json.dumps(payload)
-
- try:
- self.data = requests.get(url, proxies=proxies, data=json_payload, headers=headers,
- verify=self.https_verify)
-
- # Just assume that we get valid JSON
- self.data_json = json.loads(self.data.content.decode())
- logger.debug(self.data_json)
-
- if self.data_json.get('TicketID') is not None:
- self.data_type = 'TicketID'
- get_request_result_bool = True
- else:
- logger.debug('search found no results')
- get_request_result_bool = False
-
- # critical error: HTTP request resulted in an error!
- except Exception as e:
- logger.error('Error with http communication: %s', e)
- logger.error(
- "Failed to access OTRS HTTP. Check Hostname, Proxy, SSL Certificate and so on!")
-
- get_request_result_bool = False
- raise Exception('get http')
-
- # critical error: OTRS API sent an error!
- if self.data_type == "Error":
- logger.error(self.data_json['Error'])
- logger.error(
- "Failed to access OTRS API. Check Username and Password! Session ID expired?! "
- "Does requested Ticket exist?")
- get_request_result_bool = False
- raise Exception('get api')
-
- return get_request_result_bool
-
-
- def _minutes_since_midnight(self, now_obj):
- seconds_since_midnight = (
- now_obj - now_obj.replace(hour=0, minute=0, second=0, microsecond=0)).total_seconds()
- minutes_since_midnight = int(seconds_since_midnight / 60)
- return minutes_since_midnight
-
-
- def get_ticket_id_list_by_past_days(self, from_past_day=1, days=1, DynamicFields=1,
- AllArticles=0):
- """Search for tickets that where changed in a certain timeframe in the
- past. Result is stored in object. This method returns True|False.
-
- Args:
- from_past_day (int): day in the past to start from (e.g. 1 for yesterday)
- days (int): day in the past (or today) where to end
- DynamicFields (int): will request OTRS to include all Dynamic Fields (default: 1)
- AllArticles (int): will request OTRS to include all Articles (default: 0)
-
- Notes:
- from_past_day defaults to 1
- days defaults to 1
- This means that the ticket list will contain tickets between
- yesteray 00:00:00 and yesterday 23:59:59.
- e.g. use 0, 1 for today
- e.g. use 1, 1 for yesterday
- e.g. use 2, 1 for day before yesterday
- e.g. use 8, 7 for last 7 days
- e.g. on a Monday use 3, 1 to get tickets from last Friday
-
- Returns:
- bool [true|false]
- failed or successful
-
- """
-
- url = self.baseurl + "/otrs/nph-genericinterface.pl/Webservice/" + self.webservicename +\
- "/Ticket"
-
- minute_offset = self._minutes_since_midnight(datetime.datetime.now())
-
- # from_past_day = 0 means today (from 0:00:00 till now)
- if from_past_day == 0:
- newer = minute_offset
- older = 0
- else:
- newer = from_past_day * 1440 + minute_offset
- older = newer - 1440 * days
-
- payload = {
- 'SessionID' : self.session_id,
- 'TicketLastChangeTimeNewerMinutes': newer,
- 'TicketLastChangeTimeOlderMinutes': older
- }
-
- logger.debug(payload)
-
- try:
- return self._search_json_ticket_data(url, payload)
- except Exception:
- return False
-
-
-def _get_numeric_logger_level_from_string(level):
- """Parse string and return appropriate numeric logger level
-
- Args:
- level (str): the string representation of a logging level.
- Level is one of these: [NOTSET|TRACE|DEBUG|INFO|WARNING|ERROR|CRITICAL]
-
- Returns:
- int: value of the log level - see doctests below for mapping
- if for some reason the level is not known return DEBUG (10)
-
- Examples:
-
- >>> print(_get_numeric_logger_level_from_string("NOTSET"))
- 0
- >>> print(_get_numeric_logger_level_from_string("TRACE"))
- 5
- >>> print(_get_numeric_logger_level_from_string("DEBUG"))
- 10
- >>> print(_get_numeric_logger_level_from_string("INFO"))
- 20
- >>> print(_get_numeric_logger_level_from_string("WARNING"))
- 30
- >>> print(_get_numeric_logger_level_from_string("ERROR"))
- 40
- >>> print(_get_numeric_logger_level_from_string("CRITICAL"))
- 50
- >>> print(_get_numeric_logger_level_from_string("FOO_BAR"))
- 10
-
- """
-
- if level == "NOTSET":
- return 0
- elif level == "TRACE":
- return 5
- elif level == "DEBUG":
- return 10
- elif level == "INFO":
- return 20
- elif level == "WARNING":
- return 30
- elif level == "ERROR":
- return 40
- elif level == "CRITICAL":
- return 50
- else:
- return 10
-
-### EOF
diff --git a/tests/test_doctests.py b/tests/test_doctests.py
index 5b4128d..3d30015 100644
--- a/tests/test_doctests.py
+++ b/tests/test_doctests.py
@@ -1,7 +1,8 @@
from doctest import DocTestSuite
from unittest import TestSuite
+
def load_tests(loader, tests, pattern):
suite = TestSuite()
- suite.addTests(DocTestSuite('pyotrs.pyotrs'))
+ suite.addTests(DocTestSuite('pyotrs.client'))
return suite
diff --git a/tests/test_pyotrs.py b/tests/test_pyotrs.py
index 52d7854..4402395 100644
--- a/tests/test_pyotrs.py
+++ b/tests/test_pyotrs.py
@@ -21,7 +21,8 @@ import responses
from mock import MagicMock, patch
from pyotrs import Client
-from pyotrs import TicketCreate, TicketGet, TicketSearch, TicketUpdate
+from pyotrs import NoBaseURL, NoWebServiceName, NoCredentials
+from pyotrs import SessionCreateError, SessionIDFileError, OTRSAPIError, OTRSHTTPError
class PyOTRSTests(unittest.TestCase):
@@ -89,9 +90,9 @@ class PyOTRSTests(unittest.TestCase):
'DynamicFields': 1
}
- obj = TicketGet(self.base_url, self.webservice_name, username, password)
+ obj = Client(self.base_url, self.webservice_name, username, password)
- self.assertRaisesRegexp(Exception, 'get api', obj._get_json_ticket_data, url=url,
+ self.assertRaisesRegexp(OTRSAPIError, 'get api', obj._get_json_ticket_data, url=url,
payload=payload)
""" Test Responses: when calling: get_json_ticket_data_by_ticket_id - Test 200 OK; Body Error
@@ -117,13 +118,13 @@ class PyOTRSTests(unittest.TestCase):
username = "wrong_username"
password = "wrong_password"
- obj = TicketGet(self.base_url, self.webservice_name, username, password)
+ obj = Client(self.base_url, self.webservice_name, username, password)
obj.ticket_get_by_id(1)
# self.logger.debug(obj.data)
self.assertEqual({"Error": {"ErrorMessage": "TicketGet: Authorization failing!",
- "ErrorCode" : "TicketGet.AuthFail"}}, obj.data.json())
+ "ErrorCode": "TicketGet.AuthFail"}}, obj.data.json())
self.assertEqual(200, obj.data.status_code)
""" Test Responses: update title on valid ticket """
@@ -148,7 +149,7 @@ class PyOTRSTests(unittest.TestCase):
username = "fake_username"
password = "fake_password"
- obj = TicketUpdate(self.base_url, self.webservice_name, username, password)
+ obj = Client(self.base_url, self.webservice_name, username, password)
obj.ticket_update_set_title(10, "Full New Title")
self.assertEqual({"TicketNumber": "1970010100000010", "TicketID": "10"}, obj.data.json())
@@ -176,10 +177,10 @@ class PyOTRSTests(unittest.TestCase):
username = "fake_username"
password = "fake_password"
- obj = TicketUpdate(self.base_url, self.webservice_name, username, password)
+ obj = Client(self.base_url, self.webservice_name, username, password)
obj.ticket_update_set_title(20, "Full New Title")
- self.assertEqual({"Error": {"ErrorCode" : "TicketUpdate.AccessDenied",
+ self.assertEqual({"Error": {"ErrorCode": "TicketUpdate.AccessDenied",
"ErrorMessage": "TicketUpdate: User does not have access to the ticket!"}},
obj.data.json())
self.assertEqual(200, obj.data.status_code)
@@ -209,7 +210,7 @@ class PyOTRSTests(unittest.TestCase):
}
### create object
- obj = TicketGet(self.base_url, self.webservice_name, username, password)
+ obj = Client(self.base_url, self.webservice_name, username, password)
### enable Mock
obj._get_json_ticket_data = MagicMock()
@@ -241,7 +242,7 @@ class PyOTRSTests(unittest.TestCase):
mock_username = "wrong_username"
mock_password = "wrong_password"
- obj = TicketGet(self.base_url, self.webservice_name, mock_username, mock_password)
+ obj = Client(self.base_url, self.webservice_name, mock_username, mock_password)
with patch('requests.get') as patched_get:
obj.ticket_get_by_id(1)
diff --git a/tox.ini b/tox.ini
index 1648f53..4aa37ed 100644
--- a/tox.ini
+++ b/tox.ini
@@ -53,8 +53,8 @@ basepython = python2.7
commands = flake8 \
--max-complexity=15 \
- --exclude=./build,.venv,.tox,dist,doc \
- --ignore=F403 \
+ --exclude=./build,.venv,.tox,dist,doc,test_pyotrs.py \
+ --ignore=F403,Q000 \
--max-line-length=99 \
[]
--
GitLab