diff --git a/pyipptool/core.py b/pyipptool/core.py index 704a436..727a1fd 100644 --- a/pyipptool/core.py +++ b/pyipptool/core.py @@ -129,6 +129,14 @@ def _get_filename_for_content(content): return name, delete +def pretty_printer(form): + """ + Remove blank lines + """ + return '\n'.join((line.strip() for line in form.splitlines() + if line and not line.isspace())) + + class MetaAsyncShifter(type): """ Based on async flage defined on IPPToolWrapper @@ -216,11 +224,11 @@ def release_job(self, printer_uri=colander.null, job_id=colander.null, job_uri=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'printer_uri': printer_uri, 'job_id': job_id, 'job_uri': job_uri}} - request = release_job_form.render(kw) + request = pretty_printer(release_job_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -230,12 +238,12 @@ def cancel_job(self, job_id=colander.null, job_uri=colander.null, purge_job=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'printer_uri': printer_uri, 'job_id': job_id, 'job_uri': job_uri, 'purge_job': purge_job}} - request = cancel_job_form.render(kw) + request = pretty_printer(cancel_job_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -262,14 +270,14 @@ def create_job(self, job_billing=colander.null, job_sheets=colander.null, media=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'printer_uri': printer_uri, 'job_name': job_name, 'ipp_attribute_fidelity': ipp_attribute_fidelity, 'job_k_octets': job_k_octets, 'job_impressions': job_impressions, 'job_media_sheets': job_media_sheets}, - 'sub_operation_attributes': + 'job_attributes_tag': {'job_priority': job_priority, 'job_hold_until': job_hold_until, 'job_sheets': job_sheets, @@ -282,12 +290,12 @@ def create_job(self, 'orientation_requested': orientation_requested, 'media': media, 'printer_resolution': printer_resolution, - 'print_quality': print_quality}, - 'auth_info': auth_info, - 'job_billing': job_billing, - 'job_sheets': job_sheets, - 'media': media} - request = create_job_form.render(kw) + 'print_quality': print_quality, + 'auth_info': auth_info, + 'job_billing': job_billing, + 'job_sheets': job_sheets, + 'media': media}} + request = pretty_printer(create_job_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -314,13 +322,16 @@ def print_job(self, orientation_requested=colander.null, printer_resolution=colander.null, print_quality=colander.null, + notify_recipient_uri=colander.null, + notify_events=colander.null, + notify_time_interval=colander.null, auth_info=colander.null, job_billing=colander.null, job_sheets=colander.null, media=colander.null, document_content=None): filename, delete = _get_filename_for_content(document_content) - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'printer_uri': printer_uri, 'job_name': job_name, 'ipp_attribute_fidelity': ipp_attribute_fidelity, @@ -331,10 +342,12 @@ def print_job(self, 'job_k_octets': job_k_octets, 'job_impressions': job_impressions, 'job_media_sheets': job_media_sheets}, - 'sub_operation_attributes': + 'job_attributes_tag': {'job_priority': job_priority, 'job_hold_until': job_hold_until, 'job_sheets': job_sheets, + 'auth_info': auth_info, + 'job_billing': job_billing, 'multiple_document_handling': multiple_document_handling, 'copies': copies, 'finishings': finishings, @@ -345,12 +358,13 @@ def print_job(self, 'media': media, 'printer_resolution': printer_resolution, 'print_quality': print_quality}, - 'auth_info': auth_info, - 'job_billing': job_billing, - 'job_sheets': job_sheets, - 'media': media, - 'file': filename} - request = print_job_form.render(kw) + 'subscription_attributes_tag': + {'notify_recipient_uri': notify_recipient_uri, + 'notify_events': notify_events, + 'notify_time_interval': notify_time_interval}, + 'document_attributes_tag': + {'file': filename}} + request = pretty_printer(print_job_form.render(kw)) try: response = yield self._call_ipptool(request) raise Return(response) @@ -381,20 +395,21 @@ def create_job_subscription(self, https://www.cups.org/str.php?L4389 """ - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'printer_uri': printer_uri, 'requesting_user_name': requesting_user_name, 'job_id': job_id, 'job_uri': job_uri}, - 'notify_job_id': notify_job_id, - 'notify_recipient_uri': notify_recipient_uri, - 'notify_pull_method': notify_pull_method, - 'notify_events': notify_events, - 'notify_attributes': notify_attributes, - 'notify_charset': notify_charset, - 'notify_natural_language': notify_natural_language, - 'notify_time_interval': notify_time_interval} - request = create_job_subscription_form.render(kw) + 'subscription_attributes_tag': + {'notify_job_id': notify_job_id, + 'notify_recipient_uri': notify_recipient_uri, + 'notify_pull_method': notify_pull_method, + 'notify_events': notify_events, + 'notify_attributes': notify_attributes, + 'notify_charset': notify_charset, + 'notify_natural_language': notify_natural_language, + 'notify_time_interval': notify_time_interval}} + request = pretty_printer(create_job_subscription_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -414,18 +429,19 @@ def create_printer_subscription( """ Create a new subscription and return its id """ - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'printer_uri': printer_uri, 'requesting_user_name': requesting_user_name}, - 'notify_recipient_uri': notify_recipient_uri, - 'notify_pull_method': notify_pull_method, - 'notify_events': notify_events, - 'notify_attributes': notify_attributes, - 'notify_charset': notify_charset, - 'notify_natural_language': notify_natural_language, - 'notify_lease_duration': notify_lease_duration, - 'notify_time_interval': notify_time_interval} - request = create_printer_subscription_form.render(kw) + 'subscription_attributes_tag': + {'notify_recipient_uri': notify_recipient_uri, + 'notify_pull_method': notify_pull_method, + 'notify_events': notify_events, + 'notify_attributes': notify_attributes, + 'notify_charset': notify_charset, + 'notify_natural_language': notify_natural_language, + 'notify_lease_duration': notify_lease_duration, + 'notify_time_interval': notify_time_interval}} + request = pretty_printer(create_printer_subscription_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -447,26 +463,26 @@ def cups_add_modify_printer(self, requesting_user_name_allowed=colander.null, requesting_user_name_denied=colander.null, printer_is_shared=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'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, - 'printer_is_shared': printer_is_shared, - } - - request = cups_add_modify_printer_form.render(kw) + 'printer_attributes_tag': + {'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, + 'printer_is_shared': printer_is_shared}} + + request = pretty_printer(cups_add_modify_printer_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -485,37 +501,37 @@ def cups_add_modify_class(self, requesting_user_name_allowed=colander.null, requesting_user_name_denied=colander.null, printer_is_shared=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'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, - 'printer_is_shared': printer_is_shared, - } - - request = cups_add_modify_class_form.render(kw) + 'printer_attributes_tag': + {'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, + 'printer_is_shared': printer_is_shared}} + + request = pretty_printer(cups_add_modify_class_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @pyipptool_coroutine def cups_delete_printer(self, printer_uri=None): - kw = {'operation_attributes': {'printer_uri': printer_uri}} - request = cups_delete_printer_form.render(kw) + kw = {'operation_attributes_tag': {'printer_uri': printer_uri}} + request = pretty_printer(cups_delete_printer_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @pyipptool_coroutine def cups_delete_class(self, printer_uri=None): - kw = {'operation_attributes': {'printer_uri': printer_uri}} - request = cups_delete_class_form.render(kw) + kw = {'operation_attributes_tag': {'printer_uri': printer_uri}} + request = pretty_printer(cups_delete_class_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -528,7 +544,7 @@ def cups_get_classes(self, printer_type_mask=colander.null, requested_attributes=colander.null, requested_user_name=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'first_printer_name': first_printer_name, 'limit': limit, 'printer_location': printer_location, @@ -536,7 +552,7 @@ def cups_get_classes(self, 'printer_type_mask': printer_type_mask, 'requested_attributes': requested_attributes, 'requested_user_name': requested_user_name}} - request = cups_get_classes_form.render(kw) + request = pretty_printer(cups_get_classes_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -548,14 +564,14 @@ def cups_get_devices(self, limit=colander.null, requested_attributes=colander.null, timeout=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'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) + request = pretty_printer(cups_get_devices_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -572,7 +588,7 @@ def cups_get_ppds(self, ppd_psversion=colander.null, ppd_type=colander.null, requested_attributes=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'exclude_schemes': exclude_schemes, 'include_schemes': include_schemes, 'limit': limit, @@ -585,7 +601,7 @@ def cups_get_ppds(self, 'ppd_type': ppd_type, 'requested_attributes': requested_attributes }} - request = cups_get_ppds_form.render(kw) + request = pretty_printer(cups_get_ppds_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -598,7 +614,7 @@ def cups_get_printers(self, printer_type_mask=colander.null, requested_attributes=colander.null, requested_user_name=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'first_printer_name': first_printer_name, 'limit': limit, 'printer_location': printer_location, @@ -606,7 +622,7 @@ def cups_get_printers(self, 'printer_type_mask': printer_type_mask, 'requested_attributes': requested_attributes, 'requested_user_name': requested_user_name}} - request = cups_get_printers_form.render(kw) + request = pretty_printer(cups_get_printers_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -617,13 +633,14 @@ def cups_move_job(self, job_uri=colander.null, job_printer_uri=None, printer_state_message=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'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) + 'job_attributes_tag': + {'job_printer_uri': job_printer_uri, + 'printer_state_message': printer_state_message}} + request = pretty_printer(cups_move_job_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -632,11 +649,12 @@ def cups_reject_jobs(self, printer_uri=None, requesting_user_name=None, printer_state_message=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'printer_uri': printer_uri, 'requesting_user_name': requesting_user_name}, - 'printer_state_message': printer_state_message} - request = cups_reject_jobs_form.render(kw) + 'printer_attributes_tag': + {'printer_state_message': printer_state_message}} + request = pretty_printer(cups_reject_jobs_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -647,13 +665,13 @@ def get_job_attributes(self, job_uri=colander.null, requesting_user_name=colander.null, requested_attributes=colander.null): - kw = {'operation_attributes': + kw = {'operation_attributes_tag': {'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) + request = pretty_printer(get_job_attributes_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -665,14 +683,14 @@ def get_jobs(self, 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) + kw = {'operation_attributes_tag': + {'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 = pretty_printer(get_jobs_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -681,11 +699,11 @@ def get_printer_attributes(self, 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) + kw = {'operation_attributes_tag': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'requested_attributes': requested_attributes}} + request = pretty_printer(get_printer_attributes_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -697,14 +715,14 @@ def get_subscriptions(self, 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) + kw = {'operation_attributes_tag': + {'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 = pretty_printer(get_subscriptions_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -715,14 +733,13 @@ def get_notifications(self, requesting_user_name=colander.null, notify_sequence_numbers=colander.null, notify_wait=colander.null): - kw = {'header': - {'operation_attributes': - {'printer_uri': printer_uri, - 'requesting_user_name': requesting_user_name, - 'notify_subscription_ids': notify_subscription_ids, - 'notify_sequence_numbers': notify_sequence_numbers, - 'notify_wait': notify_wait}}} - request = get_notifications_form.render(kw) + kw = {'operation_attributes_tag': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'notify_subscription_ids': notify_subscription_ids, + 'notify_sequence_numbers': notify_sequence_numbers, + 'notify_wait': notify_wait}} + request = pretty_printer(get_notifications_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -731,22 +748,21 @@ def cancel_subscription(self, 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) + kw = {'operation_attributes_tag': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'notify_subscription_id': notify_subscription_id}} + request = pretty_printer(cancel_subscription_form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @pyipptool_coroutine def _pause_or_resume_printer(self, form, 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) + kw = {'operation_attributes_tag': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name}} + request = pretty_printer(form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -760,14 +776,12 @@ def resume_printer(self, *args, **kw): def _hold_or_release_new_jobs(self, form, 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) + kw = {'operation_attributes_tag': + {'printer_uri': printer_uri, + 'requesting_user_name': requesting_user_name, + 'printer_message_from_operator': printer_message_from_operator + }} + request = pretty_printer(form.render(kw)) response = yield self._call_ipptool(request) raise Return(response) @@ -796,19 +810,18 @@ def send_document(self, """ delete = False filename, delete = _get_filename_for_content(document_content) - kw = {'header': - {'operation_attributes': - {'job_uri': job_uri, - 'printer_uri': printer_uri, - 'job_id': job_id, - 'requesting_user_name': requesting_user_name, - 'document_name': document_name, - 'compression': compression, - 'document_format': document_format, - 'document_natural_language': document_natural_language, - 'last_document': last_document}}, - 'file': filename} - request = send_document_form.render(kw) + kw = {'operation_attributes_tag': + {'job_uri': job_uri, + 'printer_uri': printer_uri, + 'job_id': job_id, + 'requesting_user_name': requesting_user_name, + 'document_name': document_name, + 'compression': compression, + 'document_format': document_format, + 'document_natural_language': document_natural_language, + 'last_document': last_document}, + 'document_attributes_tag': {'file': filename}} + request = pretty_printer(send_document_form.render(kw)) try: response = yield self._call_ipptool(request) raise Return(response) diff --git a/pyipptool/schemas.py b/pyipptool/schemas.py index 7461859..903dbed 100644 --- a/pyipptool/schemas.py +++ b/pyipptool/schemas.py @@ -2,8 +2,7 @@ import colander from .widgets import (IPPAttributeWidget, IPPBodyWidget, IPPFileWidget, - IPPGroupWidget, IPPNameWidget, IPPTupleWidget, - IPPConstantTupleWidget) + IPPGroupWidget, IPPNameWidget) class IntegerOrTuple(colander.List): @@ -91,7 +90,7 @@ class Uri(StringOrTuple): pass -class OperationAttributes(colander.Schema): +class OperationAttributesGroup(colander.Schema): attributes_charset = colander.SchemaNode(Charset(), default='utf-8', widget=IPPAttributeWidget()) @@ -99,24 +98,12 @@ class OperationAttributes(colander.Schema): Language(), default='en', widget=IPPAttributeWidget()) - - -class OperationAttributesWithPrinterUri(OperationAttributes): + compression = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) printer_uri = colander.SchemaNode(Uri(), widget=IPPAttributeWidget()) - - -class OperationAttributesWithJobUri(OperationAttributes): job_uri = colander.SchemaNode(Uri(), widget=IPPAttributeWidget()) - - -class JobOperationAttributes(OperationAttributesWithPrinterUri, - OperationAttributesWithJobUri): job_id = colander.SchemaNode(colander.Integer(), widget=IPPAttributeWidget()) - - -class GetDevicesOperationAttributes(OperationAttributes): exclude_schemes = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) include_schemes = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) limit = colander.SchemaNode( @@ -125,24 +112,10 @@ class GetDevicesOperationAttributes(OperationAttributes): requested_attributes = colander.SchemaNode( Keyword(), widget=IPPAttributeWidget()) - - -class CancelJobOperationAttributes(JobOperationAttributes): purge_job = colander.SchemaNode(colander.Boolean(true_val=1, false_val=0), widget=IPPAttributeWidget()) - - -class SubscriptionOperationAttributes(OperationAttributesWithPrinterUri): - requesting_user_name = colander.SchemaNode(Name(), - widget=IPPAttributeWidget()) - - -class JobSubscriptionOperationAttributes(JobOperationAttributes): requesting_user_name = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) - - -class CreateJobOperationAttributes(SubscriptionOperationAttributes): job_name = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) ipp_attribute_fidelity = colander.SchemaNode(colander.Boolean(true_val=1, false_val=0), @@ -152,30 +125,21 @@ class CreateJobOperationAttributes(SubscriptionOperationAttributes): widget=IPPAttributeWidget()) job_media_sheets = colander.SchemaNode(Integer(), widget=IPPAttributeWidget()) - - -class PrintJobOperationAttributes(CreateJobOperationAttributes): document_name = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) - compression = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) + device_class = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) document_format = colander.SchemaNode(MimeMediaType(), widget=IPPAttributeWidget()) document_natural_language = colander.SchemaNode( NaturalLanguage(), widget=IPPAttributeWidget()) - -class CancelSubscriptionOperationAttributes(SubscriptionOperationAttributes): notify_subscription_id = colander.SchemaNode( colander.Integer(), widget=IPPAttributeWidget()) - -class HoldNewJobsOperationAttributes(SubscriptionOperationAttributes): printer_message_from_operator = colander.SchemaNode( Text(), widget=IPPAttributeWidget()) - -class GetJobsOperationAttributes(SubscriptionOperationAttributes): limit = colander.SchemaNode( colander.Integer(), widget=IPPAttributeWidget()) @@ -185,18 +149,6 @@ class GetJobsOperationAttributes(SubscriptionOperationAttributes): which_jobs = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) my_jobs = colander.SchemaNode(colander.Boolean(true_val=1, false_val=0), widget=IPPAttributeWidget()) - - -class MoveJobOperationAttributes(OperationAttributesWithPrinterUri): - job_id = colander.SchemaNode(colander.Integer(), - widget=IPPAttributeWidget()) - job_uri = colander.SchemaNode(Uri(), widget=IPPAttributeWidget()) - - -class GetSubscriptionsAttributes(OperationAttributesWithPrinterUri): - requesting_user_name = colander.SchemaNode( - Name(), - widget=IPPAttributeWidget()) notify_job_id = colander.SchemaNode( colander.Integer(), widget=IPPAttributeWidget()) @@ -210,11 +162,6 @@ class GetSubscriptionsAttributes(OperationAttributesWithPrinterUri): colander.Boolean(false_val=0, true_val=1), widget=IPPAttributeWidget()) - -class GetNotificationsAttributes(OperationAttributesWithPrinterUri): - requesting_user_name = colander.SchemaNode( - Name(), - widget=IPPAttributeWidget()) notify_subscription_ids = colander.SchemaNode( Integer(), widget=IPPAttributeWidget()) @@ -225,8 +172,6 @@ class GetNotificationsAttributes(OperationAttributesWithPrinterUri): colander.Boolean(false_val=0, true_val=1), widget=IPPAttributeWidget()) - -class CupsGetPPDsSchemaOperationAttributes(GetDevicesOperationAttributes): ppd_make = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) ppd_make_and_model = colander.SchemaNode( Text(), @@ -239,21 +184,13 @@ class CupsGetPPDsSchemaOperationAttributes(GetDevicesOperationAttributes): ppd_psversion = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) ppd_type = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) - -class CupsGetDevicesSchemaOperationAttributes(GetDevicesOperationAttributes): - device_class = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) timeout = colander.SchemaNode( colander.Integer(), widget=IPPAttributeWidget()) - -class CupsGetPrintersSchemaOperationAttributes(OperationAttributes): first_printer_name = colander.SchemaNode( Name(), widget=IPPAttributeWidget()) - limit = colander.SchemaNode( - colander.Integer(), - widget=IPPAttributeWidget()) printer_location = colander.SchemaNode( Text(), widget=IPPAttributeWidget()) @@ -267,27 +204,6 @@ class CupsGetPrintersSchemaOperationAttributes(OperationAttributes): requested_user_name = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) - -class GetJobAttributesOperationAttributes(MoveJobOperationAttributes): - requesting_user_name = colander.SchemaNode(Name(), - widget=IPPAttributeWidget()) - requested_attributes = colander.SchemaNode( - Keyword(), - widget=IPPAttributeWidget()) - - -class GetPrinterAttributesOperationAttributes( - OperationAttributesWithPrinterUri): - requesting_user_name = colander.SchemaNode(Name(), - widget=IPPAttributeWidget()) - requested_attributes = colander.SchemaNode( - Keyword(), - widget=IPPAttributeWidget()) - - -class SendDocumentOperationAttribute(JobOperationAttributes): - requesting_user_name = colander.SchemaNode(Name(), - widget=IPPAttributeWidget()) document_name = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) compression = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) document_format = colander.SchemaNode(MimeMediaType(), @@ -300,12 +216,16 @@ class SendDocumentOperationAttribute(JobOperationAttributes): widget=IPPAttributeWidget()) -class JobAttributes(colander.Schema): +class JobAttributesGroup(colander.Schema): job_priority = colander.SchemaNode(Integer(), widget=IPPAttributeWidget()) + job_printer_uri = colander.SchemaNode(Uri(), + widget=IPPAttributeWidget()) job_hold_until = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) job_sheets = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) + auth_info = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) + job_billing = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) multiple_document_handling = colander.SchemaNode( Keyword(), widget=IPPAttributeWidget()) @@ -324,53 +244,32 @@ class JobAttributes(colander.Schema): print_quality = colander.SchemaNode(Enum(), widget=IPPAttributeWidget()) -class HeaderIPPSchema(colander.Schema): - name = colander.SchemaNode(colander.String(), widget=IPPNameWidget()) - operation = colander.SchemaNode(colander.String(), widget=IPPNameWidget()) - operation_attributes_tag = colander.SchemaNode(colander.String(), - widget=IPPGroupWidget()) - - -class SubOperationIPPSchema(colander.Schema): - sub_operation_attributes_tag = colander.SchemaNode(colander.String(), - widget=IPPGroupWidget()) - - # override it to enable sub operations - sub_operation_attributes = colander.null - - -class BaseIPPSchema(colander.Schema): - operation_attributes_tag = 'operation-attributes-tag' - sub_operation_attributes_tag = colander.null - header = HeaderIPPSchema(widget=IPPConstantTupleWidget()) - sub_group = SubOperationIPPSchema(widget=IPPConstantTupleWidget()) - operation_attributes = OperationAttributes(widget=IPPTupleWidget()) - object_attributes_tag = colander.SchemaNode( - colander.String(), - widget=IPPGroupWidget()) - - -class CancelJobSchema(BaseIPPSchema): - name = 'Cancel Job' - operation = 'Cancel-Job' - object_attributes_tag = colander.null - operation_attributes = CancelJobOperationAttributes( - widget=IPPTupleWidget()) - +class SubscriptionGroup(colander.Schema): + notify_recipient_uri = colander.SchemaNode(Uri(), + widget=IPPAttributeWidget()) + notify_events = colander.SchemaNode(Keyword(), + widget=IPPAttributeWidget()) + notify_job_id = colander.SchemaNode(colander.Integer(), + widget=IPPAttributeWidget()) + notify_pull_method = colander.SchemaNode(Keyword(), + widget=IPPAttributeWidget()) + notify_attributes = colander.SchemaNode(Keyword(), + widget=IPPAttributeWidget()) + notify_charset = colander.SchemaNode(Charset(), + widget=IPPAttributeWidget()) + notify_natural_language = colander.SchemaNode(Language(), + widget=IPPAttributeWidget()) + notify_time_interval = colander.SchemaNode(colander.Integer(), + widget=IPPAttributeWidget()) + notify_lease_duration = colander.SchemaNode(colander.Integer(), + widget=IPPAttributeWidget()) -class ReleaseJobSchema(BaseIPPSchema): - name = 'Release Job' - operation = 'Release-Job' - operation_attributes = JobOperationAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null +class DocumentAttributesGroup(colander.Schema): + file = colander.SchemaNode(colander.String(), widget=IPPFileWidget()) -class BaseCupsAddModifyIPPSchema(BaseIPPSchema): - operation_attributes = OperationAttributesWithPrinterUri( - widget=IPPTupleWidget()) - object_attributes_tag = 'printer-attributes-tag' +class PrinterAttributesGroup(colander.Schema): auth_info_required = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) @@ -395,32 +294,47 @@ class BaseCupsAddModifyIPPSchema(BaseIPPSchema): printer_is_shared = colander.SchemaNode(colander.Boolean(true_val=1, false_val=0), widget=IPPAttributeWidget()) - - -class CupsAddModifyPrinterSchema(BaseCupsAddModifyIPPSchema): - name = 'CUPS Add Modify Printer' - operation = 'CUPS-Add-Modify-Printer' - job_sheets_default = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) device_uri = colander.SchemaNode(Uri(), widget=IPPAttributeWidget()) port_monitor = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) ppd_name = colander.SchemaNode(Name(), widget=IPPAttributeWidget()) + member_uris = colander.SchemaNode(Uri(), widget=IPPAttributeWidget()) + + +class BaseIPPSchema(colander.Schema): + name = colander.SchemaNode(colander.String(), widget=IPPNameWidget()) + operation = colander.SchemaNode(colander.String(), widget=IPPNameWidget()) + operation_attributes_tag = OperationAttributesGroup( + widget=IPPGroupWidget()) + + +class CancelJobSchema(BaseIPPSchema): + name = 'Cancel Job' + operation = 'Cancel-Job' + + +class ReleaseJobSchema(BaseIPPSchema): + name = 'Release Job' + operation = 'Release-Job' + + +class CupsAddModifyPrinterSchema(BaseIPPSchema): + name = 'CUPS Add Modify Printer' + operation = 'CUPS-Add-Modify-Printer' + printer_attributes_tag = PrinterAttributesGroup(widget=IPPGroupWidget()) class CupsDeletePrinterSchema(BaseIPPSchema): name = 'CUPS Delete Printer' operation = 'CUPS-Delete-Printer' - operation_attributes = OperationAttributesWithPrinterUri( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null -class CupsAddModifyClassSchema(BaseCupsAddModifyIPPSchema): +class CupsAddModifyClassSchema(BaseIPPSchema): name = 'CUPS Add Modify Class' operation = 'CUPS-Add-Modify-Class' - member_uris = colander.SchemaNode(Uri(), widget=IPPAttributeWidget()) + printer_attributes_tag = PrinterAttributesGroup(widget=IPPGroupWidget()) class CupsDeleteClassSchema(CupsDeletePrinterSchema): @@ -431,25 +345,16 @@ class CupsDeleteClassSchema(CupsDeletePrinterSchema): class CupsGetClassesSchema(BaseIPPSchema): name = 'CUPS Get Classes' operation = 'CUPS-Get-Classes' - object_attributes_tag = colander.null - operation_attributes = CupsGetPrintersSchemaOperationAttributes( - widget=IPPTupleWidget()) class CupsGetDevicesSchema(BaseIPPSchema): name = 'CUPS Get Devices' operation = 'CUPS-Get-Devices' - operation_attributes = CupsGetDevicesSchemaOperationAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null class CupsGetPPDsSchema(BaseIPPSchema): name = 'CUPS Get PPDs' operation = 'CUPS-Get-PPDs' - object_attributes_tag = colander.null - operation_attributes = CupsGetPPDsSchemaOperationAttributes( - widget=IPPTupleWidget()) class CupsGetPrintersSchema(CupsGetClassesSchema): @@ -460,157 +365,89 @@ class CupsGetPrintersSchema(CupsGetClassesSchema): class CupsMoveJobSchema(BaseIPPSchema): name = 'CUPS Move Job' operation = 'CUPS-Move-Job' - object_attributes_tag = 'job-attributes-tag' - operation_attributes = MoveJobOperationAttributes( - widget=IPPTupleWidget()) - job_printer_uri = colander.SchemaNode(Uri(), widget=IPPAttributeWidget()) + job_attributes_tag = JobAttributesGroup(widget=IPPGroupWidget()) class CupsRejectJobsSchema(BaseIPPSchema): name = 'CUPS Reject Jobs' operation = 'CUPS-Reject-Jobs' - object_attributes_tag = 'printer-attributes-tag' - printer_state_message = colander.SchemaNode( - Text(), - widget=IPPAttributeWidget()) + printer_attributes_tag = PrinterAttributesGroup(widget=IPPGroupWidget()) class CreateJobSchema(BaseIPPSchema): + """ + http://www.cups.org/documentation.php/spec-ipp.html#CREATE_JOB + """ name = 'Create Job' operation = 'Create-Job' - object_attributes_tag = colander.null - operation_attributes = PrintJobOperationAttributes( - widget=IPPTupleWidget()) - sub_operation_attributes_tag = 'job-attributes-tag' - sub_operation_attributes = JobAttributes(widget=IPPTupleWidget()) - auth_info = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) - job_billing = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) - job_sheets = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) - media = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) + job_attributes_tag = JobAttributesGroup(widget=IPPGroupWidget()) -class CreateSubscriptionSchema(BaseIPPSchema): - operation_attributes = SubscriptionOperationAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = 'subscription-attributes-tag' - notify_job_id = colander.SchemaNode(colander.Integer(), - widget=IPPAttributeWidget()) - notify_recipient_uri = colander.SchemaNode(Uri(), - widget=IPPAttributeWidget()) - notify_pull_method = colander.SchemaNode(Keyword(), - widget=IPPAttributeWidget()) - notify_events = colander.SchemaNode(Keyword(), - widget=IPPAttributeWidget()) - notify_attributes = colander.SchemaNode(Keyword(), - widget=IPPAttributeWidget()) - notify_charset = colander.SchemaNode(Charset(), - widget=IPPAttributeWidget()) - notify_natural_language = colander.SchemaNode(Language(), - widget=IPPAttributeWidget()) - notify_time_interval = colander.SchemaNode(colander.Integer(), - widget=IPPAttributeWidget()) - - -class CreateJobSubscriptionSchema(CreateSubscriptionSchema): +class CreateJobSubscriptionSchema(BaseIPPSchema): name = 'Create Job Subscription' operation = 'Create-Job-Subscription' - operation_attributes = JobSubscriptionOperationAttributes( - widget=IPPTupleWidget()) + subscription_attributes_tag = SubscriptionGroup( + widget=IPPGroupWidget()) -class CreatePrinterSubscriptionSchema(CreateSubscriptionSchema): +class CreatePrinterSubscriptionSchema(CreateJobSubscriptionSchema): name = 'Create Printer Subscription' operation = 'Create-Printer-Subscription' - notify_lease_duration = colander.SchemaNode(colander.Integer(), - widget=IPPAttributeWidget()) class GetJobAttributesSchema(BaseIPPSchema): name = 'Get Job Attributes' operation = 'Get-Job-Attributes' - object_attributes_tag = colander.null - operation_attributes = GetJobAttributesOperationAttributes( - widget=IPPTupleWidget()) class GetJobsSchema(BaseIPPSchema): name = 'Get Jobs' operation = 'Get-Jobs' - object_attributes_tag = colander.null - operation_attributes = GetJobsOperationAttributes( - widget=IPPTupleWidget()) class GetPrinterAttributesSchema(BaseIPPSchema): name = 'Get Printer Attributes' operation = 'Get-Printer-Attributes' - operation_attributes = GetPrinterAttributesOperationAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null class GetSubscriptionsSchema(BaseIPPSchema): name = 'Get Subscriptions' operation = 'Get-Subscriptions' - operation_attributes = GetSubscriptionsAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null class GetNotificationsSchema(BaseIPPSchema): name = 'Get Notifications' operation = 'Get-Notifications' - operation_attributes = GetNotificationsAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null class PausePrinterSchema(BaseIPPSchema): name = 'Pause Printer' operation = 'Pause-Printer' - operation_attributes = SubscriptionOperationAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null class PrintJobSchema(CreateJobSchema): name = 'Print Job' operation = 'Print-Job' - operation_attributes = PrintJobOperationAttributes( - widget=IPPTupleWidget()) - sub_operation_attributes_tag = 'job-attributes-tag' - sub_operation_attributes = JobAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = 'document-attributes-tag' - auth_info = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) - job_billing = colander.SchemaNode(Text(), widget=IPPAttributeWidget()) - job_sheets = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) - media = colander.SchemaNode(Keyword(), widget=IPPAttributeWidget()) - file = colander.SchemaNode(colander.String(), widget=IPPFileWidget()) + job_attributes_tag = JobAttributesGroup(widget=IPPGroupWidget()) + subscription_attributes_tag = SubscriptionGroup( + widget=IPPGroupWidget()) + document_attributes_tag = DocumentAttributesGroup(widget=IPPGroupWidget()) class ResumePrinterSchema(PausePrinterSchema): name = 'Resume Printer' operation = 'Resume-Printer' - operation_attributes = SubscriptionOperationAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null class SendDocumentSchema(BaseIPPSchema): name = 'Send Document' operation = 'Send-Document' - operation_attributes = SendDocumentOperationAttribute( - widget=IPPTupleWidget()) - object_attributes_tag = 'document-attributes-tag' - file = colander.SchemaNode(colander.String(), widget=IPPFileWidget()) + document_attributes_tag = DocumentAttributesGroup(widget=IPPGroupWidget()) class HoldNewJobsSchema(PausePrinterSchema): name = 'Hold New Jobs' operation = 'Hold-New-Jobs' - operation_attributes = HoldNewJobsOperationAttributes( - widget=IPPTupleWidget()) class ReleaseHeldNewJobsSchema(HoldNewJobsSchema): @@ -621,9 +458,6 @@ class ReleaseHeldNewJobsSchema(HoldNewJobsSchema): class CancelSubscriptionSchema(BaseIPPSchema): name = 'Cancel Subscription' operation = 'Cancel-Subscription' - operation_attributes = CancelSubscriptionOperationAttributes( - widget=IPPTupleWidget()) - object_attributes_tag = colander.null cancel_job_schema = CancelJobSchema(widget=IPPBodyWidget()) diff --git a/pyipptool/templates/ipp/group_tuple.pt b/pyipptool/templates/ipp/group_tuple.pt new file mode 100644 index 0000000..b3adf4f --- /dev/null +++ b/pyipptool/templates/ipp/group_tuple.pt @@ -0,0 +1,5 @@ + + + + diff --git a/pyipptool/templates/ipp/tuple.pt b/pyipptool/templates/ipp/tuple.pt deleted file mode 100644 index a258e60..0000000 --- a/pyipptool/templates/ipp/tuple.pt +++ /dev/null @@ -1,4 +0,0 @@ - - - diff --git a/pyipptool/widgets.py b/pyipptool/widgets.py index 7b485da..5db6d90 100644 --- a/pyipptool/widgets.py +++ b/pyipptool/widgets.py @@ -24,17 +24,6 @@ def serialize(self, field, cstruct=None, readonly=False): return 'FILE {}'.format(cstruct) -class IPPGroupWidget(Widget): - def serialize(self, field, cstruct=None, readonly=False): - name = field.name - while field.parent is not None: - field = field.parent - value = getattr(field.schema, name) - if value is colander.null: - return '' - return 'GROUP {}'.format(value) - - class IPPAttributeWidget(Widget): def serialize(self, field, cstruct=None, readonly=False): if cstruct is colander.null: @@ -55,8 +44,8 @@ class IPPBodyWidget(MappingWidget): item_template = 'ipp/item' -class IPPTupleWidget(SequenceWidget): - readonly_template = 'ipp/tuple' +class IPPGroupWidget(SequenceWidget): + readonly_template = 'ipp/group_tuple' template = readonly_template item_template = 'ipp/item' diff --git a/tests/test_async_subprocess.py b/tests/test_async_subprocess.py index dd51ce2..a0febe1 100644 --- a/tests/test_async_subprocess.py +++ b/tests/test_async_subprocess.py @@ -79,7 +79,7 @@ def do_POST(self): wrapper.config['cups_uri'] = 'http://localhost:%s/' % PORT request = get_subscriptions_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'http://localhost:%s/printers/fake' % PORT}} ) diff --git a/tests/test_form.py b/tests/test_form.py index c796f5b..e0431bd 100644 --- a/tests/test_form.py +++ b/tests/test_form.py @@ -6,7 +6,7 @@ def test_cancel_job_form(): from pyipptool.forms import cancel_job_form request = cancel_job_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/classes/PIY', 'job_id': 8, 'job_uri': 'https://localhost:631/jobs/8', @@ -22,7 +22,7 @@ def test_cancel_job_form(): def test_release_job_form_with_job_id(): from pyipptool.forms import release_job_form request = release_job_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/classes/PIY', 'job_id': 7}}) assert 'NAME "Release Job"' in request @@ -34,7 +34,7 @@ def test_release_job_form_with_job_id(): def test_release_job_form_with_job_uri(): from pyipptool.forms import release_job_form request = release_job_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'job_uri': 'https://localhost:631/jobs/7'}}) assert 'NAME "Release Job"' in request assert 'OPERATION "Release-Job"' in request @@ -44,16 +44,17 @@ def test_release_job_form_with_job_uri(): def test_create_printer_subscription_form(): from pyipptool.forms import create_printer_subscription_form request = create_printer_subscription_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/classes/PIY', 'requesting_user_name': 'admin'}, - 'notify_recipient_uri': 'rss://', - 'notify_events': 'all', - 'notify_attributes': 'notify-subscriber-user-name', - 'notify_charset': 'utf-8', - 'notify_natural_language': 'de', - 'notify_lease_duration': 128, - 'notify_time_interval': 1}) + 'subscription_attributes_tag': + {'notify_recipient_uri': 'rss://', + 'notify_events': 'all', + 'notify_attributes': 'notify-subscriber-user-name', + 'notify_charset': 'utf-8', + 'notify_natural_language': 'de', + 'notify_lease_duration': 128, + 'notify_time_interval': 1}}) assert 'NAME "Create Printer Subscription"' in request, request assert 'OPERATION "Create-Printer-Subscription"' in request, request assert 'ATTR charset attributes-charset utf-8' in request, request @@ -72,24 +73,28 @@ def test_create_printer_subscription_form(): def test_create_job_subscription_form_for_pull_delivery_method(): from pyipptool.forms import create_job_subscription_form request = create_job_subscription_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printer/p', 'requesting_user_name': 'admin'}, - 'notify_recipient_uri': 'rss://', - 'notify_job_id': 12, - 'notify_events': ('job-completed', 'job-created', 'job-progress'), - 'notify_attributes': 'notify-subscriber-user-name', - 'notify_charset': 'utf-8', - 'notify_natural_language': 'de', - 'notify_time_interval': 1}) + 'subscription_attributes_tag': + {'notify_recipient_uri': 'rss://', + 'notify_job_id': 12, + 'notify_events': ('job-completed', 'job-created', 'job-progress'), + 'notify_attributes': 'notify-subscriber-user-name', + 'notify_charset': 'utf-8', + 'notify_natural_language': 'de', + 'notify_time_interval': 1}}) assert 'NAME "Create Job Subscription"' in request assert 'OPERATION "Create-Job-Subscription"' in request + + assert 'GROUP operation-attributes-tag' in request, request assert 'ATTR charset attributes-charset utf-8' in request assert 'ATTR language attributes-natural-language en' in request assert 'ATTR name requesting-user-name admin' in request + assert 'GROUP subscription-attributes-tag' in request assert 'ATTR uri printer-uri https://localhost:631/printer/p' in request - assert 'ATTR integer notify-job-id 12' in request + assert 'ATTR integer notify-job-id 12' in request, request assert 'ATTR uri notify-recipient-uri rss://' in request assert ('ATTR keyword notify-events job-completed,job-created,job-progress' in request), request @@ -103,19 +108,25 @@ def test_cups_add_modify_class_form(): m_uri_0 = 'ipp://localhost:631/printers/p0' m_uri_1 = 'ipp://localhost:631/classes/c0' request = cups_add_modify_class_form.render( - {'auth_info_required': 'john', - 'member_uris': (m_uri_0, m_uri_1), - 'printer_is_accepting_jobs': True, - 'printer_info': 'multiline\ntext', - 'printer_location': 'The Office', - 'printer_more_info': 'http://example.com', - 'printer_op_policy': 'brain', - 'printer_state': '3', - 'printer_state_message': 'Ready to print', - 'requesting_user_name_allowed': 'me', - 'printer_is_shared': False}) + {'operation_attributes_tag': + {'printer_uri': 'https://localhost:631/printers/p0'}, + 'printer_attributes_tag': + {'auth_info_required': 'john', + 'member_uris': (m_uri_0, m_uri_1), + 'printer_is_accepting_jobs': True, + 'printer_info': 'multiline\ntext', + 'printer_location': 'The Office', + 'printer_more_info': 'http://example.com', + 'printer_op_policy': 'brain', + 'printer_state': '3', + 'printer_state_message': 'Ready to print', + 'requesting_user_name_allowed': 'me', + 'printer_is_shared': False}}) assert 'NAME "CUPS Add Modify Class"' assert 'OPERATION "CUPS-Add-Modify-Class"' in request + assert 'GROUP operation-attributes-tag' in request, request + assert 'ATTR uri printer-uri https://localhost:631/printers/p0' in request + assert 'GROUP printer-attributes-tag' in request assert 'ATTR uri member-uris %s,%s' % (m_uri_0, m_uri_1) in request assert 'ATTR keyword auth-info-required john' in request, request @@ -133,25 +144,28 @@ def test_cups_add_modify_class_form(): def test_cups_add_modify_printer_form(): from pyipptool.forms import cups_add_modify_printer_form request = cups_add_modify_printer_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0'}, - 'device_uri': 'cups-pdf:/', - 'auth_info_required': 'john', - 'job_sheets_default': 'none', - 'port_monitor': 'port', - 'ppd_name': 'printer.ppd', - 'printer_is_accepting_jobs': True, - 'printer_info': 'multiline\ntext', - 'printer_location': 'The Office', - 'printer_more_info': 'http://example.com', - 'printer_op_policy': 'pinky', - 'printer_state': '3', - 'printer_state_message': 'Ready to print', - 'requesting_user_name_allowed': 'me', - 'printer_is_shared': True}) + 'printer_attributes_tag': + {'device_uri': 'cups-pdf:/', + 'auth_info_required': 'john', + 'job_sheets_default': 'none', + 'port_monitor': 'port', + 'ppd_name': 'printer.ppd', + 'printer_is_accepting_jobs': True, + 'printer_info': 'multiline\ntext', + 'printer_location': 'The Office', + 'printer_more_info': 'http://example.com', + 'printer_op_policy': 'pinky', + 'printer_state': '3', + 'printer_state_message': 'Ready to print', + 'requesting_user_name_allowed': 'me', + 'printer_is_shared': True}}) assert 'NAME "CUPS Add Modify Printer"' assert 'OPERATION "CUPS-Add-Modify-Printer"' in request + assert 'GROUP operation-attributes-tag' in request assert 'ATTR uri printer-uri https://localhost:631/printers/p0' in request + assert 'GROUP printer-attributes-tag' in request assert 'ATTR uri device-uri cups-pdf:/' in request assert 'ATTR keyword auth-info-required john' in request @@ -173,9 +187,9 @@ def test_cups_add_modify_printer_form_with_None(): from pyipptool.forms import cups_add_modify_printer_form with pytest.raises(ValueError) as exec_info: cups_add_modify_printer_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0'}, - 'printer_state_message': None}) + 'printer_attributes_tag': {'printer_state_message': None}}) assert exec_info.value.message == ("None value provided for" " 'printer_state_message'") @@ -183,7 +197,7 @@ def test_cups_add_modify_printer_form_with_None(): def test_cups_delete_printer_form(): from pyipptool.forms import cups_delete_printer_form request = cups_delete_printer_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0'}}) assert 'NAME "CUPS Delete Printer"' in request assert 'OPERATION "CUPS-Delete-Printer"' in request @@ -194,7 +208,7 @@ def test_cups_delete_printer_form(): def test_cups_delete_class_form(): from pyipptool.forms import cups_delete_class_form request = cups_delete_class_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/classes/p0'}}) assert 'NAME "CUPS Delete Class"' in request assert 'OPERATION "CUPS-Delete-Class"' in request @@ -205,7 +219,7 @@ def test_cups_delete_class_form(): def test_cups_get_classes_form(): from pyipptool.forms import cups_get_classes_form request = cups_get_classes_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'first_printer_name': 'DA-Printer', 'limit': 2, 'printer_location': 'The Office', @@ -228,7 +242,7 @@ def test_cups_get_classes_form(): def test_cups_get_devices_form(): from pyipptool.forms import cups_get_devices_form request = cups_get_devices_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'device_class': 'fermionic', 'exclude_schemes': 'foo', 'include_schemes': 'bar', @@ -248,7 +262,7 @@ def test_cups_get_devices_form(): def test_cups_get_ppds_form(): from pyipptool.forms import cups_get_ppds_form request = cups_get_ppds_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'exclude_schemes': 'foo', 'include_schemes': 'bar', 'limit': 3, @@ -278,7 +292,7 @@ def test_cups_get_ppds_form(): def test_cups_get_printers_form(): from pyipptool.forms import cups_get_printers_form request = cups_get_printers_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'first_printer_name': 'DA-Printer', 'limit': 2, 'printer_location': 'The Office', @@ -301,20 +315,25 @@ def test_cups_get_printers_form(): def test_cups_reject_jobs_form(): from pyipptool.forms import cups_reject_jobs_form request = cups_reject_jobs_form.render( - {'operation_attributes': - {'printer_uri': ('https://localhost:631/' - 'printers/DA-PRINTER'), + {'operation_attributes_tag': + {'printer_uri': 'ipp://cups:631/printers/p', 'requesting_user_name': 'admin'}, - 'printer_state_message': 'You shall not pass'}) + 'printer_attributes_tag': + {'printer_state_message': 'You shall not pass'}}) assert 'NAME "CUPS Reject Jobs"' in request, request assert 'OPERATION "CUPS-Reject-Jobs"' in request, request + + assert 'GROUP operation-attributes-tag' in request + assert 'ATTR uri printer-uri ipp://cups:631/printers/p' in request + + assert 'GROUP printer-attributes-tag' in request assert 'ATTR text printer-state-message "You shall not pass"' in request def test_get_job_attributes_form(): from pyipptool.forms import get_job_attributes_form request = get_job_attributes_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/DA-PRINTER', 'job_id': 2, @@ -332,7 +351,7 @@ def test_get_job_attributes_form(): def test_get_jobs_form(): from pyipptool.forms import get_jobs_form request = get_jobs_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0', 'requesting_user_name': 'yoda', 'limit': 1, @@ -352,7 +371,7 @@ def test_get_jobs_form(): def test_get_printer_attributes_form(): from pyipptool.forms import get_printer_attributes_form request = get_printer_attributes_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0', 'requesting_user_name': 'yoda', @@ -368,7 +387,7 @@ def test_get_printer_attributes_form(): def test_get_subscriptions_form(): from pyipptool.forms import get_subscriptions_form request = get_subscriptions_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0', 'requesting_user_name': 'yoda', 'notify_job_id': 3, @@ -388,7 +407,7 @@ def test_get_subscriptions_form(): def test_get_notifications_form_for_one_notification(): from pyipptool.forms import get_notifications_form request = get_notifications_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0', 'requesting_user_name': 'yoda', 'notify_subscription_ids': 3, @@ -406,7 +425,7 @@ def test_get_notifications_form_for_one_notification(): def test_get_notifications_form_for_multiple_notifications(): from pyipptool.forms import get_notifications_form request = get_notifications_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'https://localhost:631/printers/p0', 'requesting_user_name': 'yoda', 'notify_subscription_ids': (3, 4, 5), @@ -424,7 +443,7 @@ def test_get_notifications_form_for_multiple_notifications(): def test_pause_printer_form(): from pyipptool.forms import pause_printer_form request = pause_printer_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'ipp://server:port/printers/name', 'requesting_user_name': 'yoda'}}) assert 'NAME "Pause Printer"' in request @@ -436,7 +455,7 @@ def test_pause_printer_form(): def test_resume_printer_form(): from pyipptool.forms import resume_printer_form request = resume_printer_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'ipp://server:port/printers/name', 'requesting_user_name': 'yoda'}}) assert 'NAME "Resume Printer"' in request @@ -448,7 +467,7 @@ def test_resume_printer_form(): def test_hold_new_jobs_form(): from pyipptool.forms import hold_new_jobs_form request = hold_new_jobs_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'ipp://server:port/printers/name', 'requesting_user_name': 'yoda', 'printer_message_from_operator': 'freeze jobs'}}) @@ -462,7 +481,7 @@ def test_hold_new_jobs_form(): def test_release_held_new_jobs_form(): from pyipptool.forms import release_held_new_jobs_form request = release_held_new_jobs_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'ipp://server:port/printers/name', 'requesting_user_name': 'yoda', 'printer_message_from_operator': 'melt jobs'}}) @@ -476,7 +495,7 @@ def test_release_held_new_jobs_form(): def test_cancel_subscription_form(): from pyipptool.forms import cancel_subscription_form request = cancel_subscription_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'ipp://server:port/printers/name', 'requesting_user_name': 'yoda', 'notify_subscription_id': 5}}) @@ -488,23 +507,25 @@ def test_cancel_subscription_form(): def test_create_job_form(): + """ + http://www.cups.org/documentation.php/spec-ipp.html#CREATE_JOB + """ from pyipptool.forms import create_job_form request = create_job_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'ipp://server:port/printers/name', 'job_name': 'foo', 'ipp_attribute_fidelity': True, - 'document_name': 'foo.txt', - 'compression': 'gzip', - 'document_format': 'text/plain', - 'document_natural_language': 'en', 'job_k_octets': 1024, 'job_impressions': 2048, 'job_media_sheets': 2}, - 'sub_operation_attributes': + 'job_attributes_tag': {'job_priority': 1, 'job_hold_until': 'indefinite', 'job_sheets': 'standard', + 'media': 'iso-a4-white', + 'auth_info': 'michael', + 'job_billing': 'no-idea', 'multiple_document_handling': 'single-document', 'copies': 2, 'finishings': 'punch', @@ -512,30 +533,19 @@ def test_create_job_form(): 'sides': 'two-sided-short-edge', 'number_up': 4, 'orientation_requested': 'reverse-landscape', - 'media': 'iso-a4-white', 'printer_resolution': '600dpi', - 'print_quality': 5}, - 'auth_info': 'michael', - 'job_billing': 'no-idea', - 'job_sheets': 'none', - 'media': 'media-default'}) + 'print_quality': 5}}) assert 'NAME "Create Job"' in request assert 'OPERATION "Create-Job"' in request assert ('ATTR uri printer-uri ipp://server:port/printers/name' in request), request assert 'ATTR name job-name foo' in request assert 'ATTR boolean ipp-attribute-fidelity 1' in request, request - assert 'ATTR name document-name foo.txt' in request - assert 'ATTR keyword compression gzip' in request - assert 'ATTR mimeMediaType document-format text/plain' in request - assert 'ATTR naturalLanguage document-natural-language en' in request assert 'ATTR integer job-k-octets 1024' in request assert 'ATTR integer job-impressions 2048' in request assert 'ATTR integer job-media-sheets 2' in request assert 'ATTR text auth-info "michael"' in request, request assert 'ATTR text job-billing "no-idea"' in request, request - assert 'ATTR keyword job-sheets none' in request, request - assert 'ATTR keyword media media-default' in request assert 'GROUP job-attributes-tag' in request assert 'ATTR integer job-priority 1' in request @@ -556,7 +566,7 @@ def test_create_job_form(): def test_print_job_form(): from pyipptool.forms import print_job_form request = print_job_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'printer_uri': 'ipp://server:port/printers/name', 'job_name': 'foo', 'ipp_attribute_fidelity': True, @@ -567,10 +577,13 @@ def test_print_job_form(): 'job_k_octets': 1024, 'job_impressions': 2048, 'job_media_sheets': 2}, - 'sub_operation_attributes': + 'job_attributes_tag': {'job_priority': 1, 'job_hold_until': 'indefinite', 'job_sheets': 'standard', + 'auth_info': 'michael', + 'job_billing': 'no-idea', + 'media': 'media-default', 'multiple_document_handling': 'single-document', 'copies': 2, 'finishings': 'punch', @@ -578,14 +591,15 @@ def test_print_job_form(): 'sides': 'two-sided-short-edge', 'number_up': 4, 'orientation_requested': 'reverse-landscape', - 'media': 'iso-a4-white', 'printer_resolution': '600dpi', 'print_quality': 5}, - 'auth_info': 'michael', - 'job_billing': 'no-idea', - 'job_sheets': 'none', - 'media': 'media-default', - 'file': '/path/to/file.txt'}) + 'subscription_attributes_tag': + {'notify_recipient_uri': 'rss://', + 'notify_events': ['all']}, + 'document_attributes_tag': + {'file': '/path/to/file.txt'}}) + + print request assert 'NAME "Print Job"' in request assert 'OPERATION "Print-Job"' in request assert ('ATTR uri printer-uri ipp://server:port/printers/name' in @@ -605,6 +619,10 @@ def test_print_job_form(): assert 'ATTR integer job-priority 1' in request assert 'ATTR keyword job-hold-until indefinite' in request assert 'ATTR keyword job-sheets standard' in request + assert 'ATTR text auth-info "michael"' in request, request + assert 'ATTR text job-billing "no-idea"' in request, request + assert 'ATTR keyword job-sheets standard' in request, request + assert 'ATTR keyword media media-default' in request assert 'ATTR keyword multiple-document-handling single-document' in request assert 'ATTR integer copies 2' in request assert 'ATTR enum finishings punch' in request @@ -612,19 +630,19 @@ def test_print_job_form(): assert 'ATTR keyword sides two-sided-short-edge' in request assert 'ATTR integer number-up 4' in request assert 'ATTR enum orientation-requested reverse-landscape' in request - assert 'ATTR keyword media iso-a4-white' in request assert 'ATTR resolution printer-resolution 600dpi' in request assert 'ATTR enum print-quality 5' in request + assert 'GROUP subscription-attributes-tag' in request + assert 'ATTR uri notify-recipient-uri rss://' in request + assert 'ATTR keyword notify-events all' in request + assert 'GROUP document-attributes-tag' in request - assert 'ATTR text auth-info "michael"' in request, request - assert 'ATTR text job-billing "no-idea"' in request, request - assert 'ATTR keyword job-sheets none' in request, request - assert 'ATTR keyword media media-default' in request assert 'FILE /path/to/file.txt' in request assert (request.index('GROUP operation-attributes-tag') < request.index('GROUP job-attributes-tag') < + request.index('GROUP subscription-attributes-tag') < request.index('GROUP document-attributes-tag') ) @@ -633,7 +651,7 @@ def test_send_document_form(): from pyipptool.forms import send_document_form request = send_document_form.render( - {'operation_attributes': + {'operation_attributes_tag': {'job_uri': 'http://cups:631/jobs/2', 'requesting_user_name': 'sweet', 'document_name': 'python.pdf', @@ -641,9 +659,12 @@ def test_send_document_form(): 'document_format': 'application/pdf', 'document_natural_language': 'en', 'last_document': True}, - 'file': '/path/to/a/file.pdf'}) + 'document_attributes_tag': + {'file': '/path/to/a/file.pdf'}}) assert 'NAME "Send Document"' in request assert 'OPERATION "Send-Document"' in request + + assert 'GROUP operation-attributes-tag' in request assert 'ATTR uri job-uri http://cups:631/jobs/2' in request assert 'ATTR name requesting-user-name sweet' in request assert 'ATTR name document-name python.pdf' in request @@ -651,6 +672,8 @@ def test_send_document_form(): assert 'ATTR mimeMediaType document-format application/pdf' in request assert 'ATTR naturalLanguage document-natural-language en' in request assert 'ATTR boolean last-document 1' in request, request + + assert 'GROUP document-attributes-tag' assert 'FILE /path/to/a/file.pdf' in request, request diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py index 09b3a50..613f7e0 100644 --- a/tests/test_highlevel.py +++ b/tests/test_highlevel.py @@ -3,6 +3,7 @@ import SocketServer import socket import tempfile +import textwrap import threading import time @@ -25,18 +26,26 @@ def test_ipptool_create_job_subscription_pull_delivery_method(_call_ipptool): notify_charset='utf-8', notify_natural_language='de', notify_time_interval=1) - request = _call_ipptool._mock_mock_calls[0][1][0] - assert 'ATTR uri printer-uri https://localhost:631/printer/p' in request - assert 'ATTR name requesting-user-name admin' in request - assert 'ATTR integer notify-job-id 108' in request - assert 'ATTR uri notify-recipient-uri rss://' in request - assert ('ATTR keyword notify-events job-completed,job-created,job-progress' - in request) - assert ('ATTR keyword notify-attributes notify-subscriber-user-name' - in request) - assert 'ATTR charset notify-charset utf-8' in request - assert 'ATTR language notify-natural-language de' in request - assert 'ATTR integer notify-time-interval 1' in request + request = _call_ipptool._mock_mock_calls[0][1][-1] + expected_request = textwrap.dedent(""" + { + NAME "Create Job Subscription" + OPERATION "Create-Job-Subscription" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri https://localhost:631/printer/p + ATTR name requesting-user-name admin + GROUP subscription-attributes-tag + ATTR uri notify-recipient-uri rss:// + ATTR keyword notify-events job-completed,job-created,job-progress + ATTR integer notify-job-id 108 + ATTR keyword notify-attributes notify-subscriber-user-name + ATTR charset notify-charset utf-8 + ATTR language notify-natural-language de + ATTR integer notify-time-interval 1 + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -52,16 +61,26 @@ def test_ipptool_create_printer_subscription(_call_ipptool): notify_natural_language='de', notify_lease_duration=0, notify_time_interval=1) - request = _call_ipptool._mock_mock_calls[0][1][0] - assert 'requesting-user-name admin' in request, request - assert 'printer-uri https://localhost:631/classes/PUBLIC-PDF' in request - assert 'notify-recipient-uri rss://' in request - assert 'notify-events all' in request - assert 'notify-attributes notify-subscriber-user-name' in request - assert 'notify-charset utf-8' in request - assert 'notify-natural-language de' in request - assert 'notify-lease-duration 0' in request - assert 'notify-time-interval 1' in request + request = _call_ipptool._mock_mock_calls[0][1][-1] + expected_request = textwrap.dedent(""" + { + NAME "Create Printer Subscription" + OPERATION "Create-Printer-Subscription" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri https://localhost:631/classes/PUBLIC-PDF + ATTR name requesting-user-name admin + GROUP subscription-attributes-tag + ATTR uri notify-recipient-uri rss:// + ATTR keyword notify-events all + ATTR keyword notify-attributes notify-subscriber-user-name + ATTR charset notify-charset utf-8 + ATTR language notify-natural-language de + ATTR integer notify-time-interval 1 + ATTR integer notify-lease-duration 0 + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -73,10 +92,20 @@ def test_cups_add_modify_printer(_call_ipptool): device_uri='cups-pdf:/', printer_is_shared=False, ) - request = _call_ipptool._mock_mock_calls[0][1][0] - assert 'printer-uri https://localhost:631/classes/PUBLIC-PDF' in request - assert 'device-uri cups-pdf:/' in request - assert 'ATTR boolean printer-is-shared 0' in request + request = _call_ipptool._mock_mock_calls[0][1][-1] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Add Modify Printer" + OPERATION "CUPS-Add-Modify-Printer" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri https://localhost:631/classes/PUBLIC-PDF + GROUP printer-attributes-tag + ATTR boolean printer-is-shared 0 + ATTR uri device-uri cups-pdf:/ + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -86,9 +115,17 @@ def test_get_job_attributes_with_job_id(_call_ipptool): printer_uri='https://localhost:631/classes/PUBLIC-PDF', job_id=2) request = _call_ipptool._mock_mock_calls[0][1][0] - assert 'printer-uri https://localhost:631/classes/PUBLIC-PDF' in request - assert 'job-id 2' in request - assert 'job-uri' not in request + expected_request = textwrap.dedent(""" + { + NAME "Get Job Attributes" + OPERATION "Get-Job-Attributes" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri https://localhost:631/classes/PUBLIC-PDF + ATTR integer job-id 2 + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -97,8 +134,16 @@ def test_get_job_attributes_with_job_uri(_call_ipptool): get_job_attributes( job_uri='https://localhost:631/jobs/2') request = _call_ipptool._mock_mock_calls[0][1][0] - assert 'job-uri https://localhost:631/jobs/2' in request - assert 'printer-uri' not in request + expected_request = textwrap.dedent(""" + { + NAME "Get Job Attributes" + OPERATION "Get-Job-Attributes" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri job-uri https://localhost:631/jobs/2 + }""").strip() + assert request == expected_request, request def test_timeout(): @@ -165,38 +210,93 @@ def test_authentication(): def test_release_job(_call_ipptool): from pyipptool import release_job _call_ipptool.return_value = {'Tests': [{}]} - release_job('https://localhost:631/') + release_job(job_uri='ipp://cups:631/jobs/3') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Release Job" + OPERATION "Release-Job" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri job-uri ipp://cups:631/jobs/3 + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cancel_job(_call_ipptool): from pyipptool import cancel_job _call_ipptool.return_value = {'Tests': [{}]} - cancel_job(job_uri='https://localhost:631/jobs/12') + cancel_job(job_uri='ipp://cups:631/jobs/12') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Cancel Job" + OPERATION "Cancel-Job" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri job-uri ipp://cups:631/jobs/12 + }""").strip() + assert request == expected_request, request @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': [{}]} - cups_add_modify_class(printer_uri='', + cups_add_modify_class(printer_uri='ipp://cups:631/classes/p', printer_is_shared=True) request = _call_ipptool._mock_mock_calls[0][1][0] - assert 'ATTR boolean printer-is-shared 1' in request + expected_request = textwrap.dedent(""" + { + NAME "CUPS Add Modify Class" + OPERATION "CUPS-Add-Modify-Class" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/classes/p + GROUP printer-attributes-tag + ATTR boolean printer-is-shared 1 + }""").strip() + assert request == expected_request, request @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': [{}]} - cups_delete_printer() + cups_delete_printer(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Delete Printer" + OPERATION "CUPS-Delete-Printer" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @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': [{}]} - cups_delete_class() + cups_delete_class(printer_uri='ipp://cups:631/classes/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Delete Class" + OPERATION "CUPS-Delete-Class" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/classes/p + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -204,6 +304,16 @@ def test_cups_get_classes(_call_ipptool): from pyipptool import cups_get_classes _call_ipptool.return_value = {'Tests': [{}]} cups_get_classes() + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Get Classes" + OPERATION "CUPS-Get-Classes" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -211,6 +321,16 @@ def test_cups_get_printers(_call_ipptool): from pyipptool import cups_get_printers _call_ipptool.return_value = {'Tests': [{}]} cups_get_printers() + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Get Printers" + OPERATION "CUPS-Get-Printers" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -218,6 +338,16 @@ def test_cups_get_devices(_call_ipptool): from pyipptool import cups_get_devices _call_ipptool.return_value = {'Tests': [{}]} cups_get_devices() + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Get Devices" + OPERATION "CUPS-Get-Devices" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -225,92 +355,231 @@ def test_cups_get_ppds(_call_ipptool): from pyipptool import cups_get_ppds _call_ipptool.return_value = {'Tests': [{}]} cups_get_ppds() + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Get PPDs" + OPERATION "CUPS-Get-PPDs" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + }""").strip() + assert request == expected_request, request @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': [{}]} - cups_move_job() + cups_move_job(job_uri='ipp://cups:631/jobs/12', + job_printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Move Job" + OPERATION "CUPS-Move-Job" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri job-uri ipp://cups:631/jobs/12 + GROUP job-attributes-tag + ATTR uri job-printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @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': [{}]} - cups_reject_jobs() + cups_reject_jobs(printer_uri='ipp://cups:631/printers/p', + requesting_user_name='boby') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "CUPS Reject Jobs" + OPERATION "CUPS-Reject-Jobs" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + ATTR name requesting-user-name boby + GROUP printer-attributes-tag + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_jobs(_call_ipptool): from pyipptool import get_jobs _call_ipptool.return_value = {'Tests': [{}]} - get_jobs(printer_uri='') + get_jobs(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Get Jobs" + OPERATION "Get-Jobs" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @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': [{}]} - get_printer_attributes(printer_uri='') + get_printer_attributes(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Get Printer Attributes" + OPERATION "Get-Printer-Attributes" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_subscriptions(_call_ipptool): from pyipptool import get_subscriptions _call_ipptool.return_value = {'Tests': [{}]} - get_subscriptions(printer_uri='') + get_subscriptions(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Get Subscriptions" + OPERATION "Get-Subscriptions" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_get_notifications(_call_ipptool): from pyipptool import get_notifications _call_ipptool.return_value = {'Tests': [{}]} - get_notifications(printer_uri='', + get_notifications(printer_uri='ipp://cups:631/printers/p', notify_subscription_ids=3) + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Get Notifications" + OPERATION "Get-Notifications" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + ATTR integer notify-subscription-ids 3 + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_pause_printer(_call_ipptool): from pyipptool import pause_printer _call_ipptool.return_value = {'Tests': [{}]} - pause_printer(printer_uri='') + pause_printer(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Pause Printer" + OPERATION "Pause-Printer" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @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': [{}]} - hold_new_jobs(printer_uri='') + hold_new_jobs(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Hold New Jobs" + OPERATION "Hold-New-Jobs" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @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': [{}]} - release_held_new_jobs(printer_uri='') + release_held_new_jobs(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Release Held New Jobs" + OPERATION "Release-Held-New-Jobs" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_resume_printer(_call_ipptool): from pyipptool import resume_printer _call_ipptool.return_value = {'Tests': [{}]} - resume_printer(printer_uri='') + resume_printer(printer_uri='ipp://cups:631/printers/p') + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Resume Printer" + OPERATION "Resume-Printer" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_cancel_subscription(_call_ipptool): from pyipptool import cancel_subscription _call_ipptool.return_value = {'Tests': [{}]} - cancel_subscription(printer_uri='', + cancel_subscription(printer_uri='ipp://cups:631/printers/p', notify_subscription_id=3) + request = _call_ipptool._mock_mock_calls[0][1][0] + expected_request = textwrap.dedent(""" + { + NAME "Cancel Subscription" + OPERATION "Cancel-Subscription" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + ATTR integer notify-subscription-id 3 + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_create_job(_call_ipptool): from pyipptool import create_job _call_ipptool.return_value = {'Tests': [{}]} - create_job(printer_uri='', + create_job(printer_uri='ipp://cups:631/classes/p', job_name='foo', job_priority=1, job_hold_until='indefinite', @@ -329,16 +598,50 @@ def test_create_job(_call_ipptool): job_k_octets=1024, job_impressions=2048, job_media_sheets=2, + auth_info='michael', + job_billing='no-idea', ) + request = _call_ipptool._mock_mock_calls[0][1][-1] + expected_request = textwrap.dedent(""" + { + NAME "Create Job" + OPERATION "Create-Job" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/classes/p + ATTR name job-name foo + ATTR boolean ipp-attribute-fidelity 0 + ATTR integer job-k-octets 1024 + ATTR integer job-impressions 2048 + ATTR integer job-media-sheets 2 + GROUP job-attributes-tag + ATTR integer job-priority 1 + ATTR keyword job-hold-until indefinite + ATTR keyword job-sheets standard + ATTR text auth-info "michael" + ATTR text job-billing "no-idea" + ATTR keyword multiple-document-handling single-document + ATTR integer copies 2 + ATTR enum finishings punch + ATTR rangeOfInteger page-ranges 1-6 + ATTR keyword sides two-sided-short-edge + ATTR integer number-up 4 + ATTR enum orientation-requested reverse-landscape + ATTR keyword media iso-a4-white + ATTR resolution printer-resolution 600dpi + ATTR enum print-quality 5 + }""").strip() + assert request == expected_request, request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') def test_print_job(_call_ipptool): from pyipptool import print_job _call_ipptool.return_value = {'Tests': [{}]} - with open(os.path.join(os.path.dirname(__file__), - 'hello.pdf'), 'rb') as tmp: - print_job(printer_uri='', + filename = os.path.join(os.path.dirname(__file__), 'hello.pdf') + with open(filename, 'rb') as tmp: + print_job(printer_uri='ipp://cups:631/classes/p', job_name='foo', ipp_attribute_fidelity=False, document_name='foo.txt', @@ -351,6 +654,8 @@ def test_print_job(_call_ipptool): job_priority=1, job_hold_until='indefinite', job_sheets='standard', + auth_info='michael', + job_billing='no-idea', multiple_document_handling='single-document', copies=2, finishings='punch', @@ -362,7 +667,47 @@ def test_print_job(_call_ipptool): printer_resolution='600dpi', print_quality='5', document_content=tmp.read()) - assert 'FILE /tmp/' in _call_ipptool._mock_mock_calls[0][1][-1] + request = _call_ipptool._mock_mock_calls[0][1][-1] + expected_request = textwrap.dedent(""" + { + NAME "Print Job" + OPERATION "Print-Job" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/classes/p + ATTR name job-name foo + ATTR boolean ipp-attribute-fidelity 0 + ATTR integer job-k-octets 1024 + ATTR integer job-impressions 2048 + ATTR integer job-media-sheets 2 + ATTR name document-name foo.txt + ATTR keyword compression gzip + ATTR mimeMediaType document-format text/plain + ATTR naturalLanguage document-natural-language en + GROUP job-attributes-tag + ATTR integer job-priority 1 + ATTR keyword job-hold-until indefinite + ATTR keyword job-sheets standard + ATTR text auth-info "michael" + ATTR text job-billing "no-idea" + ATTR keyword multiple-document-handling single-document + ATTR integer copies 2 + ATTR enum finishings punch + ATTR rangeOfInteger page-ranges 1-6 + ATTR keyword sides two-sided-short-edge + ATTR integer number-up 4 + ATTR enum orientation-requested reverse-landscape + ATTR keyword media iso-a4-white + ATTR resolution printer-resolution 600dpi + ATTR enum print-quality 5 + GROUP subscription-attributes-tag + GROUP document-attributes-tag + FILE /tmp/ + }""").strip() + assert ('\n'.join(request.splitlines()[:-2]) + == '\n'.join(expected_request.splitlines()[:-2])), request + assert expected_request.splitlines()[-2].startswith('FILE /tmp/') @mock.patch.object(pyipptool.wrapper, '_call_ipptool') @@ -371,9 +716,25 @@ def test_send_document_with_file(_call_ipptool): _call_ipptool.return_value = {'Tests': [{}]} with tempfile.NamedTemporaryFile('rb') as tmp: - send_document(document_content=tmp) - assert ('FILE {}'.format(tmp.name) - in _call_ipptool._mock_mock_calls[0][1][-1]) + send_document(printer_uri='ipp://cups:631/printers/p', + requesting_user_name='you', + document_content=tmp) + request = _call_ipptool._mock_mock_calls[0][1][-1] + expected_request = textwrap.dedent(""" + { + NAME "Send Document" + OPERATION "Send-Document" + GROUP operation-attributes-tag + ATTR charset attributes-charset utf-8 + ATTR language attributes-natural-language en + ATTR uri printer-uri ipp://cups:631/printers/p + ATTR name requesting-user-name you + ATTR mimeMediaType document-format application/pdf + ATTR boolean last-document 1 + GROUP document-attributes-tag + FILE %s + }""" % tmp.name).strip() + assert request == expected_request @mock.patch.object(pyipptool.wrapper, '_call_ipptool') diff --git a/tests/test_widget.py b/tests/test_widget.py index 7097a0c..6ba6ed8 100644 --- a/tests/test_widget.py +++ b/tests/test_widget.py @@ -68,21 +68,6 @@ def test_ipp_name_widget(): assert response == 'NAME "FOO"' -def test_ipp_group_widget(): - from pyipptool.widgets import IPPGroupWidget - widget = IPPGroupWidget() - rendrer = DummyRenderer() - schema = DummySchema(name='FOO') - - field = DummyField(schema, renderer=rendrer) - field.parent = None - sub_field = DummyField(None, renderer=rendrer) - sub_field.parent = field - - response = widget.serialize(sub_field, 'world') - assert response == 'GROUP FOO' - - def test_ipp_file_widget(): from pyipptool.widgets import IPPFileWidget