diff --git a/pyipptool/__init__.py b/pyipptool/__init__.py index 0565c70..87fe07b 100644 --- a/pyipptool/__init__.py +++ b/pyipptool/__init__.py @@ -31,6 +31,7 @@ resume_printer_form, hold_new_jobs_form, release_held_new_jobs_form, + cancel_subscription_form, ) @@ -440,6 +441,20 @@ def get_subscriptions(uri, 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': diff --git a/pyipptool/forms.py b/pyipptool/forms.py index 80a87d4..66fd65e 100644 --- a/pyipptool/forms.py +++ b/pyipptool/forms.py @@ -23,6 +23,7 @@ resume_printer_schema, hold_new_jobs_schema, release_held_new_jobs_schema, + cancel_subscription_schema, ) default_dir = resource_filename('pyipptool', 'templates/') @@ -53,3 +54,4 @@ resume_printer_form = deform.Form(resume_printer_schema) hold_new_jobs_form = deform.Form(hold_new_jobs_schema) release_held_new_jobs_form = deform.Form(release_held_new_jobs_schema) +cancel_subscription_form = deform.Form(cancel_subscription_schema) diff --git a/pyipptool/schemas.py b/pyipptool/schemas.py index a8c36a8..ef10c6a 100644 --- a/pyipptool/schemas.py +++ b/pyipptool/schemas.py @@ -86,6 +86,12 @@ class SubscriptionOperationAttributes(OperationAttributesWithPrinterUri): 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()) @@ -418,6 +424,15 @@ class ReleaseHeldNewJobsSchema(HoldNewJobsSchema): operation = 'Release-Held-New-Jobs' +class CancelSubscriptionSchema(BaseIPPSchema): + name = 'Cancel Subscription' + operation = 'Cancel-Subscription' + header = HeaderIPPSchema(widget=IPPConstantTupleWidget()) + header['operation_attributes'] = CancelSubscriptionOperationAttributes( + widget=IPPTupleWidget()) + object_attributes_tag = colander.null + + cancel_job_schema = CancelJobSchema(widget=IPPBodyWidget()) release_job_schema = ReleaseJobSchema(widget=IPPBodyWidget()) @@ -465,3 +480,5 @@ class ReleaseHeldNewJobsSchema(HoldNewJobsSchema): hold_new_jobs_schema = HoldNewJobsSchema(widget=IPPBodyWidget()) release_held_new_jobs_schema = ReleaseHeldNewJobsSchema(widget=IPPBodyWidget()) + +cancel_subscription_schema = CancelSubscriptionSchema(widget=IPPBodyWidget()) diff --git a/tests/test_form.py b/tests/test_form.py index f61335b..9001fe2 100644 --- a/tests/test_form.py +++ b/tests/test_form.py @@ -392,3 +392,18 @@ def test_release_held_new_jobs_form(): assert 'ATTR uri printer-uri ipp://server:port/printers/name' in request assert 'ATTR name requesting-user-name yoda' in request assert 'ATTR text printer-message-from-operator "melt jobs"' in request + + +def test_cancel_subscription_form(): + from pyipptool.forms import cancel_subscription_form + request = cancel_subscription_form.render( + {'header': + {'operation_attributes': + {'printer_uri': 'ipp://server:port/printers/name', + 'requesting_user_name': 'yoda', + 'notify_subscription_id': 5}}}) + assert 'NAME "Cancel Subscription"' in request + assert 'OPERATION "Cancel-Subscription"' in request + assert 'ATTR uri printer-uri ipp://server:port/printers/name' in request + assert 'ATTR name requesting-user-name yoda' in request + assert 'ATTR integer notify-subscription-id 5' in request, request diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py index 46a4aaa..ad63804 100644 --- a/tests/test_highlevel.py +++ b/tests/test_highlevel.py @@ -272,3 +272,13 @@ def test_resume_printer(_call_ipptool): _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} resume_printer('https://localhost:631/', printer_uri='') assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/' + + +@mock.patch.object(pyipptool, '_call_ipptool') +def test_cancel_subscription(_call_ipptool): + from pyipptool import cancel_subscription + _call_ipptool.return_value = {'Tests': [{'ResponseAttributes': ''}]} + cancel_subscription('https://localhost:631/', + printer_uri='', + notify_subscription_id=3) + assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/'