diff --git a/pyipptool/__init__.py b/pyipptool/__init__.py index 0d3873e..b86f100 100644 --- a/pyipptool/__init__.py +++ b/pyipptool/__init__.py @@ -1,496 +1,60 @@ import ConfigParser -import functools import os -import plistlib -import subprocess -import tempfile -import time -import threading -import urlparse -import colander +from .core import IPPToolWrapper -from .forms import (cancel_job_form, - release_job_form, - create_printer_subscription_form, - cups_add_modify_class_form, - cups_add_modify_printer_form, - cups_delete_printer_form, - cups_delete_class_form, - cups_get_classes_form, - cups_get_devices_form, - cups_get_ppds_form, - cups_get_printers_form, - cups_move_job_form, - cups_reject_jobs_form, - get_job_attributes_form, - get_jobs_form, - get_printer_attributes_form, - get_subscriptions_form, - pause_printer_form, - resume_printer_form, - hold_new_jobs_form, - release_held_new_jobs_form, - cancel_subscription_form, - ) - -config = ConfigParser.SafeConfigParser() -config.read(['/etc/pyipptool/pyipptool.cfg', - os.path.join(os.path.expanduser('~'), '.pyipptool.cfg')]) -IPPTOOL_PATH = config.get('main', 'ipptool_path') -try: - LOGIN = config.get('main', 'login') -except ConfigParser.NoOptionError: - LOGIN = '' -try: - PASSWORD = config.get('main', 'password') -except ConfigParser.NoOptionError: - PASSWORD = '' -try: - GRACEFUL_SHUTDOWN_TIME = config.getint('main', 'graceful_shutdown_time') -except ConfigParser.NoOptionError: - GRACEFUL_SHUTDOWN_TIME = 1 -try: - TIMEOUT = config.getint('main', 'timeout') -except ConfigParser.NoOptionError: - TIMEOUT = 5 - - -class TimeoutError(Exception): - pass - - -def authenticate_uri(uri): - if LOGIN and PASSWORD: - parsed_url = urlparse.urlparse(uri) - authenticated_netloc = '{}:{}@{}'.format(LOGIN, PASSWORD, - parsed_url.netloc) - authenticated_uri = urlparse.ParseResult(parsed_url[0], - authenticated_netloc, - *parsed_url[2:]) - return authenticated_uri.geturl() - return uri - - -def timeout_handler(process, future): - future.append(True) - beginning = time.time() - process.terminate() - while process.poll() is None: - if time.time() - beginning > GRACEFUL_SHUTDOWN_TIME: - try: - process.kill() - except OSError: - pass - break - time.sleep(.1) - - -def _call_ipptool(uri, request): - with tempfile.NamedTemporaryFile(delete=False) as temp_file: - temp_file.write(request) - process = subprocess.Popen([IPPTOOL_PATH, - authenticate_uri(uri), - '-X', - temp_file.name], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - future = [] - timer = threading.Timer(TIMEOUT, timeout_handler, (process, future)) - timer.start() +def read_config(paths=('/etc/opt/pyipptool/pyipptool.cfg', + os.path.join(os.path.expanduser('~'), + '.pyipptool.cfg'))): + config = {} + fs_config = ConfigParser.ConfigParser() + fs_config.read(paths) + config['ipptool_path'] = fs_config.get('main', 'ipptool_path') try: - stdout, stderr = process.communicate() - finally: - os.unlink(temp_file.name) - timer.cancel() - if future: - raise TimeoutError - return plistlib.readPlistFromString(stdout) - - -def release_job(uri, - printer_uri=colander.null, - job_id=colander.null, - job_uri=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'job_id': job_id, - 'job_uri': job_uri}}} - request = release_job_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - - -def cancel_job(uri, - printer_uri=colander.null, - job_id=colander.null, - job_uri=colander.null, - purge_job=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'job_id': job_id, - 'job_uri': job_uri, - 'purge_job': purge_job}}} - request = cancel_job_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - - -def create_printer_subscription(uri, - printer_uri=None, - requesting_user_name=None, - notify_recipient_uri=None, - notify_events=None, - notify_lease_duration=colander.null, - notify_lease_expiration_time=colander.null): - """ - Create a new subscription and return its id - """ - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name}}, - 'notify_recipient_uri': notify_recipient_uri, - 'notify_events': notify_events, - 'notify_lease_duration': notify_lease_duration, - 'notify_lease_expiration_time': notify_lease_expiration_time} - request = create_printer_subscription_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'][1]['notify-subscription-id'] - - -def cups_add_modify_printer(uri, - printer_uri=None, - auth_info_required=colander.null, - job_sheets_default=colander.null, - device_uri=colander.null, - port_monitor=colander.null, - ppd_name=colander.null, - printer_is_accepting_jobs=colander.null, - printer_info=colander.null, - printer_location=colander.null, - printer_more_info=colander.null, - printer_op_policy=colander.null, - printer_state=colander.null, - printer_state_message=colander.null, - requesting_user_name_allowed=colander.null, - requesting_user_name_denied=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri}}, - 'auth_info_required': auth_info_required, - 'job_sheets_default': job_sheets_default, - 'device_uri': device_uri, - 'port_monitor': port_monitor, - 'ppd_name': ppd_name, - 'printer_is_accepting_jobs': printer_is_accepting_jobs, - 'printer_info': printer_info, - 'printer_location': printer_location, - 'printer_more_info': printer_more_info, - 'printer_op_policy': printer_op_policy, - 'printer_state': printer_state, - 'printer_state_message': printer_state_message, - 'requesting_user_name_allowed ': requesting_user_name_allowed, - 'requesting_user_name_denied': requesting_user_name_denied, - } - - request = cups_add_modify_printer_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - return True - - -def cups_add_modify_class(uri, - printer_uri=None, - auth_info_required=colander.null, - member_uris=colander.null, - printer_is_accepting_jobs=colander.null, - printer_info=colander.null, - printer_location=colander.null, - printer_more_info=colander.null, - printer_op_policy=colander.null, - printer_state=colander.null, - printer_state_message=colander.null, - requesting_user_name_allowed=colander.null, - requesting_user_name_denied=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri}}, - 'auth_info_required': auth_info_required, - 'member_uris': member_uris, - 'printer_is_accepting_jobs': printer_is_accepting_jobs, - 'printer_info': printer_info, - 'printer_location': printer_location, - 'printer_more_info': printer_more_info, - 'printer_op_policy': printer_op_policy, - 'printer_state': printer_state, - 'printer_state_message': printer_state_message, - 'requesting_user_name_allowed ': requesting_user_name_allowed, - 'requesting_user_name_denied': requesting_user_name_denied, - } - - request = cups_add_modify_class_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - return True - - -def cups_delete_printer(uri, printer_uri=None): - kw = {'header': {'operation_attributes': {'printer_uri': printer_uri}}} - request = cups_delete_printer_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - return True - - -def cups_delete_class(uri, printer_uri=None): - kw = {'header': {'operation_attributes': {'printer_uri': printer_uri}}} - request = cups_delete_class_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - return True - - -def cups_get_classes(uri, - first_printer_name=colander.null, - limit=colander.null, - printer_location=colander.null, - printer_type=colander.null, - printer_type_mask=colander.null, - requested_attributes=colander.null, - requested_user_name=colander.null): - kw = {'header': {'operation_attributes': - {'first_printer_name': first_printer_name, - 'limit': limit, - 'printer_location': printer_location, - 'printer_type': printer_type, - 'printer_type_mask': printer_type_mask, - 'requested_attributes': requested_attributes, - 'requested_user_name': requested_user_name}}} - request = cups_get_classes_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def cups_get_devices(uri, - device_class=colander.null, - exclude_schemes=colander.null, - include_schemes=colander.null, - limit=colander.null, - requested_attributes=colander.null, - timeout=colander.null): - kw = {'header': {'operation_attributes': - {'device_class': device_class, - 'exclude_schemes': exclude_schemes, - 'include-schemes': include_schemes, - 'limit': limit, - 'requested_attributes': requested_attributes, - 'timeout': timeout}}} - request = cups_get_devices_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def cups_get_ppds(uri, - exclude_schemes=colander.null, - include_schemes=colander.null, - limit=colander.null, - ppd_make=colander.null, - ppd_make_and_model=colander.null, - ppd_model_number=colander.null, - ppd_natural_language=colander.null, - ppd_product=colander.null, - ppd_psversion=colander.null, - ppd_type=colander.null, - requested_attributes=colander.null): - kw = {'header': {'operation_attributes': - {'exclude_schemes': exclude_schemes, - 'include_schemes': include_schemes, - 'limit': limit, - 'ppd_make': ppd_make, - 'ppd_make_and_model': ppd_make_and_model, - 'ppd_model_number': ppd_model_number, - 'ppd_natural_language': ppd_natural_language, - 'ppd_product': ppd_product, - 'ppd_psversion': ppd_psversion, - 'ppd_type': ppd_type, - 'requested_attributes': requested_attributes - }}} - request = cups_get_ppds_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def cups_get_printers(uri, - first_printer_name=colander.null, - limit=colander.null, - printer_location=colander.null, - printer_type=colander.null, - printer_type_mask=colander.null, - requested_attributes=colander.null, - requested_user_name=colander.null): - kw = {'header': {'operation_attributes': - {'first_printer_name': first_printer_name, - 'limit': limit, - 'printer_location': printer_location, - 'printer_type': printer_type, - 'printer_type_mask': printer_type_mask, - 'requested_attributes': requested_attributes, - 'requested_user_name': requested_user_name}}} - request = cups_get_printers_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def cups_move_job(uri, - printer_uri=colander.null, - job_id=colander.null, - job_uri=colander.null, - job_printer_uri=None, - printer_state_message=None): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'job_id': job_id, - 'job_uri': job_uri}}, - 'job_printer_uri': job_printer_uri, - 'printer_state_message': printer_state_message} - request = cups_move_job_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - return True - - -def cups_reject_jobs(uri, - printer_uri=None, - requesting_user_name=None, - printer_state_message=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name}}, - 'printer_state_message': printer_state_message} - request = cups_reject_jobs_form.render(kw) - response = _call_ipptool(uri, request) - assert response['Tests'][0]['StatusCode'] == 'successful-ok', response - return True - - -def get_job_attributes(uri, - printer_uri=colander.null, - job_id=colander.null, - job_uri=colander.null, - requesting_user_name=colander.null, - requested_attributes=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'job_id': job_id, - 'job_uri': job_uri, - 'requesting_user_name': requesting_user_name, - 'requested_attributes': requested_attributes}}} - request = get_job_attributes_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def get_jobs(uri, - printer_uri=None, - requesting_user_name=colander.null, - limit=colander.null, - requested_attributes=colander.null, - which_jobs=colander.null, - my_jobs=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name, - 'limit': limit, - 'requested_attributes': requested_attributes, - 'which_jobs': which_jobs, - 'my_jobs': my_jobs}}} - request = get_jobs_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def get_printer_attributes(uri, - printer_uri=None, - requesting_user_name=colander.null, - requested_attributes=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name, - 'requested_attributes': requested_attributes}}} - request = get_printer_attributes_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def get_subscriptions(uri, - printer_uri=None, - requesting_user_name=colander.null, - notify_job_id=colander.null, - limit=colander.null, - requested_attributes=colander.null, - my_subscriptions=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name, - 'notify_job_id': notify_job_id, - 'limit': limit, - 'requested_attributes': requested_attributes, - 'my_subscriptions': my_subscriptions}}} - request = get_subscriptions_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def cancel_subscription(uri, - printer_uri=None, - requesting_user_name=colander.null, - notify_subscription_id=None): - kw = {'header': - {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name, - 'notify_subscription_id': notify_subscription_id}}} - request = cancel_subscription_form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -def _pause_or_resume_printer(form, uri, printer_uri=None, - requesting_user_name=colander.null): - kw = {'header': {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name}}} - request = form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -pause_printer = functools.partial(_pause_or_resume_printer, - pause_printer_form) - -resume_printer = functools.partial(_pause_or_resume_printer, - resume_printer_form) - - -def _hold_or_release_new_jobs(form, uri, printer_uri=None, - requesting_user_name=colander.null, - printer_message_from_operator=colander.null): - kw = { - 'header': { - 'operation_attributes': { - 'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name, - 'printer_message_from_operator': printer_message_from_operator - } - } - } - request = form.render(kw) - response = _call_ipptool(uri, request) - return response['Tests'][0]['ResponseAttributes'] - - -hold_new_jobs = functools.partial(_hold_or_release_new_jobs, - hold_new_jobs_form) - -release_held_new_jobs = functools.partial(_hold_or_release_new_jobs, - release_held_new_jobs_form) + config['login'] = fs_config.get('main', 'login') + except ConfigParser.NoOptionError: + pass + try: + config['password'] = fs_config.get('main', 'password') + except ConfigParser.NoOptionError: + pass + try: + config['graceful_shutdown_time'] = fs_config.getint( + 'main', + 'graceful_shutdown_time') + except ConfigParser.NoOptionError: + config['graceful_shutdown_time'] = 2 + try: + config['timeout'] = fs_config.getint('main', 'timeout') + except ConfigParser.NoOptionError: + config['timeout'] = 10 + return config + + +config = read_config() + +wrapper = IPPToolWrapper(config) +create_printer_subscription = wrapper. create_printer_subscription +cancel_job = wrapper.cancel_job +release_job = wrapper.release_job +create_printer_subscription = wrapper.create_printer_subscription +cups_add_modify_class = wrapper.cups_add_modify_class +cups_add_modify_printer = wrapper.cups_add_modify_printer +cups_delete_printer = wrapper.cups_delete_printer +cups_delete_class = wrapper.cups_delete_class +cups_get_classes = wrapper.cups_get_classes +cups_get_devices = wrapper.cups_get_devices +cups_get_ppds = wrapper.cups_get_ppds +cups_get_printers = wrapper.cups_get_printers +cups_move_job = wrapper.cups_move_job +cups_reject_jobs = wrapper.cups_reject_jobs +get_job_attributes = wrapper.get_job_attributes +get_jobs = wrapper.get_jobs +get_printer_attributes = wrapper.get_printer_attributes +get_subscriptions = wrapper.get_subscriptions +pause_printer = wrapper.pause_printer +resume_printer = wrapper.resume_printer +hold_new_jobs = wrapper.hold_new_jobs +release_held_new_jobs = wrapper.release_held_new_jobs +cancel_subscription = wrapper.cancel_subscription diff --git a/pyipptool/core.py b/pyipptool/core.py new file mode 100644 index 0000000..15dc156 --- /dev/null +++ b/pyipptool/core.py @@ -0,0 +1,457 @@ +import os +import plistlib +import subprocess +import tempfile +import time +import threading +import urlparse + +import colander + +from .forms import (cancel_job_form, + release_job_form, + create_printer_subscription_form, + cups_add_modify_class_form, + cups_add_modify_printer_form, + cups_delete_printer_form, + cups_delete_class_form, + cups_get_classes_form, + cups_get_devices_form, + cups_get_ppds_form, + cups_get_printers_form, + cups_move_job_form, + cups_reject_jobs_form, + get_job_attributes_form, + get_jobs_form, + get_printer_attributes_form, + get_subscriptions_form, + pause_printer_form, + resume_printer_form, + hold_new_jobs_form, + release_held_new_jobs_form, + cancel_subscription_form, + ) + + +class TimeoutError(Exception): + pass + + +class IPPToolWrapper(object): + + def __init__(self, config): + self.config = config + + def authenticate_uri(self, uri): + if 'login' in self.config and 'password' in self.config: + parsed_url = urlparse.urlparse(uri) + authenticated_netloc = '{}:{}@{}'.format(self.config['login'], + self.config['password'], + parsed_url.netloc) + authenticated_uri = urlparse.ParseResult(parsed_url[0], + authenticated_netloc, + *parsed_url[2:]) + return authenticated_uri.geturl() + return uri + + def timeout_handler(self, process, future): + future.append(True) + beginning = time.time() + process.terminate() + while process.poll() is None: + if time.time() - beginning > self.config['graceful_shutdown_time']: + try: + process.kill() + except OSError: + pass + break + time.sleep(.1) + + def _call_ipptool(self, uri, request): + with tempfile.NamedTemporaryFile(delete=False) as temp_file: + temp_file.write(request) + process = subprocess.Popen([self.config['ipptool_path'], + self.authenticate_uri(uri), + '-X', + temp_file.name], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + future = [] + timer = threading.Timer(self.config['timeout'], + self.timeout_handler, (process, future)) + timer.start() + try: + stdout, stderr = process.communicate() + finally: + os.unlink(temp_file.name) + timer.cancel() + if future: + raise TimeoutError + return plistlib.readPlistFromString(stdout) + + def release_job(self, uri, + printer_uri=colander.null, + job_id=colander.null, + job_uri=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'job_id': job_id, + 'job_uri': job_uri}}} + request = release_job_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + + def cancel_job(self, uri, + printer_uri=colander.null, + job_id=colander.null, + job_uri=colander.null, + purge_job=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'job_id': job_id, + 'job_uri': job_uri, + 'purge_job': purge_job}}} + request = cancel_job_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + + def create_printer_subscription( + self, + uri, + printer_uri=None, + requesting_user_name=None, + notify_recipient_uri=None, + notify_events=None, + notify_lease_duration=colander.null, + notify_lease_expiration_time=colander.null): + """ + Create a new subscription and return its id + """ + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name}}, + 'notify_recipient_uri': notify_recipient_uri, + 'notify_events': notify_events, + 'notify_lease_duration': notify_lease_duration, + 'notify_lease_expiration_time': notify_lease_expiration_time} + request = create_printer_subscription_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'][1][ + 'notify-subscription-id'] + + def cups_add_modify_printer(self, uri, + printer_uri=None, + auth_info_required=colander.null, + job_sheets_default=colander.null, + device_uri=colander.null, + port_monitor=colander.null, + ppd_name=colander.null, + printer_is_accepting_jobs=colander.null, + printer_info=colander.null, + printer_location=colander.null, + printer_more_info=colander.null, + printer_op_policy=colander.null, + printer_state=colander.null, + printer_state_message=colander.null, + requesting_user_name_allowed=colander.null, + requesting_user_name_denied=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri}}, + 'auth_info_required': auth_info_required, + 'job_sheets_default': job_sheets_default, + 'device_uri': device_uri, + 'port_monitor': port_monitor, + 'ppd_name': ppd_name, + 'printer_is_accepting_jobs': printer_is_accepting_jobs, + 'printer_info': printer_info, + 'printer_location': printer_location, + 'printer_more_info': printer_more_info, + 'printer_op_policy': printer_op_policy, + 'printer_state': printer_state, + 'printer_state_message': printer_state_message, + 'requesting_user_name_allowed ': requesting_user_name_allowed, + 'requesting_user_name_denied': requesting_user_name_denied, + } + + request = cups_add_modify_printer_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + return True + + def cups_add_modify_class(self, uri, + printer_uri=None, + auth_info_required=colander.null, + member_uris=colander.null, + printer_is_accepting_jobs=colander.null, + printer_info=colander.null, + printer_location=colander.null, + printer_more_info=colander.null, + printer_op_policy=colander.null, + printer_state=colander.null, + printer_state_message=colander.null, + requesting_user_name_allowed=colander.null, + requesting_user_name_denied=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri}}, + 'auth_info_required': auth_info_required, + 'member_uris': member_uris, + 'printer_is_accepting_jobs': printer_is_accepting_jobs, + 'printer_info': printer_info, + 'printer_location': printer_location, + 'printer_more_info': printer_more_info, + 'printer_op_policy': printer_op_policy, + 'printer_state': printer_state, + 'printer_state_message': printer_state_message, + 'requesting_user_name_allowed ': requesting_user_name_allowed, + 'requesting_user_name_denied': requesting_user_name_denied, + } + + request = cups_add_modify_class_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + return True + + def cups_delete_printer(self, uri, printer_uri=None): + kw = {'header': {'operation_attributes': {'printer_uri': printer_uri}}} + request = cups_delete_printer_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + return True + + def cups_delete_class(self, uri, printer_uri=None): + kw = {'header': {'operation_attributes': {'printer_uri': printer_uri}}} + request = cups_delete_class_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + return True + + def cups_get_classes(self, uri, + first_printer_name=colander.null, + limit=colander.null, + printer_location=colander.null, + printer_type=colander.null, + printer_type_mask=colander.null, + requested_attributes=colander.null, + requested_user_name=colander.null): + kw = {'header': {'operation_attributes': + {'first_printer_name': first_printer_name, + 'limit': limit, + 'printer_location': printer_location, + 'printer_type': printer_type, + 'printer_type_mask': printer_type_mask, + 'requested_attributes': requested_attributes, + 'requested_user_name': requested_user_name}}} + request = cups_get_classes_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def cups_get_devices(self, uri, + device_class=colander.null, + exclude_schemes=colander.null, + include_schemes=colander.null, + limit=colander.null, + requested_attributes=colander.null, + timeout=colander.null): + kw = {'header': {'operation_attributes': + {'device_class': device_class, + 'exclude_schemes': exclude_schemes, + 'include-schemes': include_schemes, + 'limit': limit, + 'requested_attributes': requested_attributes, + 'timeout': timeout}}} + request = cups_get_devices_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def cups_get_ppds(self, uri, + exclude_schemes=colander.null, + include_schemes=colander.null, + limit=colander.null, + ppd_make=colander.null, + ppd_make_and_model=colander.null, + ppd_model_number=colander.null, + ppd_natural_language=colander.null, + ppd_product=colander.null, + ppd_psversion=colander.null, + ppd_type=colander.null, + requested_attributes=colander.null): + kw = {'header': {'operation_attributes': + {'exclude_schemes': exclude_schemes, + 'include_schemes': include_schemes, + 'limit': limit, + 'ppd_make': ppd_make, + 'ppd_make_and_model': ppd_make_and_model, + 'ppd_model_number': ppd_model_number, + 'ppd_natural_language': ppd_natural_language, + 'ppd_product': ppd_product, + 'ppd_psversion': ppd_psversion, + 'ppd_type': ppd_type, + 'requested_attributes': requested_attributes + }}} + request = cups_get_ppds_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def cups_get_printers(self, uri, + first_printer_name=colander.null, + limit=colander.null, + printer_location=colander.null, + printer_type=colander.null, + printer_type_mask=colander.null, + requested_attributes=colander.null, + requested_user_name=colander.null): + kw = {'header': {'operation_attributes': + {'first_printer_name': first_printer_name, + 'limit': limit, + 'printer_location': printer_location, + 'printer_type': printer_type, + 'printer_type_mask': printer_type_mask, + 'requested_attributes': requested_attributes, + 'requested_user_name': requested_user_name}}} + request = cups_get_printers_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def cups_move_job(self, uri, + printer_uri=colander.null, + job_id=colander.null, + job_uri=colander.null, + job_printer_uri=None, + printer_state_message=None): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'job_id': job_id, + 'job_uri': job_uri}}, + 'job_printer_uri': job_printer_uri, + 'printer_state_message': printer_state_message} + request = cups_move_job_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + return True + + def cups_reject_jobs(self, uri, + printer_uri=None, + requesting_user_name=None, + printer_state_message=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name}}, + 'printer_state_message': printer_state_message} + request = cups_reject_jobs_form.render(kw) + response = self._call_ipptool(uri, request) + assert response['Tests'][0]['StatusCode'] == 'successful-ok', response + return True + + def get_job_attributes(self, uri, + printer_uri=colander.null, + job_id=colander.null, + job_uri=colander.null, + requesting_user_name=colander.null, + requested_attributes=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'job_id': job_id, + 'job_uri': job_uri, + 'requesting_user_name': requesting_user_name, + 'requested_attributes': requested_attributes}}} + request = get_job_attributes_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def get_jobs(self, uri, + printer_uri=None, + requesting_user_name=colander.null, + limit=colander.null, + requested_attributes=colander.null, + which_jobs=colander.null, + my_jobs=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'limit': limit, + 'requested_attributes': requested_attributes, + 'which_jobs': which_jobs, + 'my_jobs': my_jobs}}} + request = get_jobs_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def get_printer_attributes(self, uri, + printer_uri=None, + requesting_user_name=colander.null, + requested_attributes=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'requested_attributes': requested_attributes}}} + request = get_printer_attributes_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def get_subscriptions(self, uri, + printer_uri=None, + requesting_user_name=colander.null, + notify_job_id=colander.null, + limit=colander.null, + requested_attributes=colander.null, + my_subscriptions=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'notify_job_id': notify_job_id, + 'limit': limit, + 'requested_attributes': requested_attributes, + 'my_subscriptions': my_subscriptions}}} + request = get_subscriptions_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def cancel_subscription(self, uri, + printer_uri=None, + requesting_user_name=colander.null, + notify_subscription_id=None): + kw = {'header': + {'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'notify_subscription_id': notify_subscription_id}}} + request = cancel_subscription_form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def _pause_or_resume_printer(self, form, uri, printer_uri=None, + requesting_user_name=colander.null): + kw = {'header': {'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name}}} + request = form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def pause_printer(self, *args, **kw): + return self._pause_or_resume_printer(pause_printer_form, *args, **kw) + + def resume_printer(self, *args, **kw): + return self._pause_or_resume_printer(resume_printer_form, *args, **kw) + + def _hold_or_release_new_jobs(self, form, uri, printer_uri=None, + requesting_user_name=colander.null, + printer_message_from_operator=colander.null): + kw = { + 'header': { + 'operation_attributes': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'printer_message_from_operator': printer_message_from_operator + }}} + request = form.render(kw) + response = self._call_ipptool(uri, request) + return response['Tests'][0]['ResponseAttributes'] + + def hold_new_jobs(self, *args, **kw): + return self._hold_or_release_new_jobs(hold_new_jobs_form, *args, **kw) + + def release_held_new_jobs(self, *args, **kw): + return self._hold_or_release_new_jobs(release_held_new_jobs_form, + *args, **kw) diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py index ad63804..c43dc7a 100644 --- a/tests/test_highlevel.py +++ b/tests/test_highlevel.py @@ -1,6 +1,5 @@ import BaseHTTPServer import SocketServer -import functools import threading import time @@ -10,25 +9,7 @@ import pyipptool -def patch_config(**kwconfig): - def wrapper(fn): - @functools.wraps(fn) - def inner(*args, **kw): - import pyipptool - old_values = {} - for k, v in kwconfig.items(): - old_values[k] = getattr(pyipptool, k) - setattr(pyipptool, k, v) - try: - return fn(*args, **kw) - finally: - for k, v in old_values.items(): - setattr(pyipptool, k, v) - return inner - return wrapper - - -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_ipptool_create_printer_subscription(_call_ipptool): from pyipptool import create_printer_subscription create_printer_subscription( @@ -49,7 +30,7 @@ def test_ipptool_create_printer_subscription(_call_ipptool): assert 'notify-lease-expiration-time 0' in request -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_add_modify_printer(_call_ipptool): from pyipptool import cups_add_modify_printer _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -63,7 +44,7 @@ def test_cups_add_modify_printer(_call_ipptool): assert 'device-uri cups-pdf:/' in request -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_job_attributes_with_job_id(_call_ipptool): from pyipptool import get_job_attributes get_job_attributes( @@ -77,7 +58,7 @@ def test_get_job_attributes_with_job_id(_call_ipptool): assert 'job-uri' not in request -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_job_attributes_with_job_uri(_call_ipptool): from pyipptool import get_job_attributes get_job_attributes( @@ -89,9 +70,9 @@ def test_get_job_attributes_with_job_uri(_call_ipptool): assert 'printer-uri' not in request -@patch_config(TIMEOUT=1) def test_timeout(): - from pyipptool import TimeoutError, _call_ipptool + from pyipptool import wrapper + from pyipptool.core import TimeoutError from pyipptool.forms import get_subscriptions_form PORT = 6789 @@ -117,19 +98,27 @@ def do_POST(self): {'operation_attributes': {'printer_uri': 'http://localhost:%s/printers/fake' % PORT}}}) - with pytest.raises(TimeoutError): - _call_ipptool('http://localhost:%s/' % PORT, request) + + old_timeout = wrapper.config['timeout'] + wrapper.config['timeout'] = 1 + try: + with pytest.raises(TimeoutError): + wrapper._call_ipptool('http://localhost:%s/' % PORT, request) + finally: + wrapper.config['timeout'] = old_timeout -@patch_config(LOGIN='ezeep', PASSWORD='secret') def test_authentication(): - from pyipptool import authenticate_uri + from pyipptool import IPPToolWrapper + wrapper = IPPToolWrapper({'login': 'ezeep', + 'password': 'secret'}) - assert (authenticate_uri('http://localhost:631/printers/?arg=value') == - 'http://ezeep:secret@localhost:631/printers/?arg=value') + assert (wrapper.authenticate_uri( + 'http://localhost:631/printers/?arg=value') == + 'http://ezeep:secret@localhost:631/printers/?arg=value') -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_release_job(_call_ipptool): from pyipptool import release_job _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -137,7 +126,7 @@ def test_release_job(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cancel_job(_call_ipptool): from pyipptool import cancel_job _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -145,7 +134,7 @@ def test_cancel_job(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_add_modify_class(_call_ipptool): from pyipptool import cups_add_modify_class _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -154,7 +143,7 @@ def test_cups_add_modify_class(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_delete_printer(_call_ipptool): from pyipptool import cups_delete_printer _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -162,7 +151,7 @@ def test_cups_delete_printer(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_delete_class(_call_ipptool): from pyipptool import cups_delete_class _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -170,7 +159,7 @@ def test_cups_delete_class(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_classes(_call_ipptool): from pyipptool import cups_get_classes _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -178,7 +167,7 @@ def test_cups_get_classes(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_printers(_call_ipptool): from pyipptool import cups_get_printers _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -186,7 +175,7 @@ def test_cups_get_printers(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_devices(_call_ipptool): from pyipptool import cups_get_devices _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -194,7 +183,7 @@ def test_cups_get_devices(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_get_ppds(_call_ipptool): from pyipptool import cups_get_ppds _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -202,7 +191,7 @@ def test_cups_get_ppds(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_move_job(_call_ipptool): from pyipptool import cups_move_job _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -210,7 +199,7 @@ def test_cups_move_job(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cups_reject_jobs(_call_ipptool): from pyipptool import cups_reject_jobs _call_ipptool.return_value = {'Tests': [{'StatusCode': 'successful-ok'}]} @@ -218,7 +207,7 @@ def test_cups_reject_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_jobs(_call_ipptool): from pyipptool import get_jobs _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -226,7 +215,7 @@ def test_get_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_printer_attributes(_call_ipptool): from pyipptool import get_printer_attributes _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -234,7 +223,7 @@ def test_get_printer_attributes(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_subscriptions(_call_ipptool): from pyipptool import get_subscriptions _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -242,7 +231,7 @@ def test_get_subscriptions(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_pause_printer(_call_ipptool): from pyipptool import pause_printer _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -250,7 +239,7 @@ def test_pause_printer(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_hold_new_jobs(_call_ipptool): from pyipptool import hold_new_jobs _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -258,7 +247,7 @@ def test_hold_new_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_release_held_new_jobs(_call_ipptool): from pyipptool import release_held_new_jobs _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -266,7 +255,7 @@ def test_release_held_new_jobs(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_resume_printer(_call_ipptool): from pyipptool import resume_printer _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} @@ -274,7 +263,7 @@ def test_resume_printer(_call_ipptool): assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' -@mock.patch.object(pyipptool, '_call_ipptool') +@mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cancel_subscription(_call_ipptool): from pyipptool import cancel_subscription _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]}