Merge pull request #73 from sbellem/get-notifications

Implemented Get-Notifications operation - closes #70

pyipptool/__init__.py

@@ -24,6 +24,7 @@
 get_jobs = wrapper.get_jobs
 get_printer_attributes = wrapper.get_printer_attributes
 get_subscriptions = wrapper.get_subscriptions
+get_notifications = wrapper.get_notifications
 pause_printer = wrapper.pause_printer
 resume_printer = wrapper.resume_printer
 hold_new_jobs = wrapper.hold_new_jobs

pyipptool/core.py

@@ -26,6 +26,7 @@
                     get_jobs_form,
                     get_printer_attributes_form,
                     get_subscriptions_form,
+                    get_notifications_form,
                     pause_printer_form,
                     resume_printer_form,
                     hold_new_jobs_form,
@@ -439,6 +440,24 @@ def get_subscriptions(self, uri,
         response = self._call_ipptool(uri, request)
         return response['Tests'][0]
 
+    def get_notifications(self,
+                          uri,
+                          printer_uri=None,
+                          notify_subscription_ids=None,
+                          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)
+        response = self._call_ipptool(uri, request)
+        return response['Tests'][0]
+
     def cancel_subscription(self, uri,
                             printer_uri=None,
                             requesting_user_name=colander.null,

pyipptool/forms.py

@@ -20,6 +20,7 @@
                       get_jobs_schema,
                       get_printer_attributes_schema,
                       get_subscriptions_schema,
+                      get_notifications_schema,
                       pause_printer_schema,
                       resume_printer_schema,
                       hold_new_jobs_schema,
@@ -53,6 +54,7 @@
 get_jobs_form = deform.Form(get_jobs_schema)
 get_printer_attributes_form = deform.Form(get_printer_attributes_schema)
 get_subscriptions_form = deform.Form(get_subscriptions_schema)
+get_notifications_form = deform.Form(get_notifications_schema)
 pause_printer_form = deform.Form(pause_printer_schema)
 resume_printer_form = deform.Form(resume_printer_schema)
 hold_new_jobs_form = deform.Form(hold_new_jobs_schema)

pyipptool/schemas.py

@@ -3,6 +3,15 @@
                       IPPNameWidget, IPPTupleWidget, IPPConstantTupleWidget)
 
 
+class IntegerOrTuple(colander.List):
+    def serialize(self, node, appstruct):
+        if isinstance(appstruct, (tuple, list)):
+            return super(IntegerOrTuple, self).serialize(
+                node,
+                ','.join([str(i) for i in appstruct]))
+        return super(IntegerOrTuple, self).serialize(node, appstruct)
+
+
 class StringOrTuple(colander.String):
     def serialize(self, node, appstruct):
         if isinstance(appstruct, (tuple, list)):
@@ -19,6 +28,10 @@ class Enum(colander.String):
     pass
 
 
+class Integer(IntegerOrTuple):
+    pass
+
+
 class Keyword(StringOrTuple):
     pass
 
@@ -148,6 +161,21 @@ class GetSubscriptionsAttributes(OperationAttributesWithPrinterUri):
         widget=IPPAttributeWidget())
 
 
+class GetNotificationsAttributes(OperationAttributesWithPrinterUri):
+    requesting_user_name = colander.SchemaNode(
+        Name(),
+        widget=IPPAttributeWidget())
+    notify_subscription_ids = colander.SchemaNode(
+        Integer(),
+        widget=IPPAttributeWidget())
+    notify_sequence_numbers = colander.SchemaNode(
+        Integer(),
+        widget=IPPAttributeWidget())
+    notify_wait = colander.SchemaNode(
+        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(
@@ -424,6 +452,15 @@ class GetSubscriptionsSchema(BaseIPPSchema):
     object_attributes_tag = colander.null
 
 
+class GetNotificationsSchema(BaseIPPSchema):
+    name = 'Get Notifications'
+    operation = 'Get-Notifications'
+    header = HeaderIPPSchema(widget=IPPConstantTupleWidget())
+    header['operation_attributes'] = GetNotificationsAttributes(
+        widget=IPPTupleWidget())
+    object_attributes_tag = colander.null
+
+
 class PausePrinterSchema(BaseIPPSchema):
     name = 'Pause Printer'
     operation = 'Pause-Printer'
@@ -507,6 +544,8 @@ class CancelSubscriptionSchema(BaseIPPSchema):
 
 get_subscriptions_schema = GetSubscriptionsSchema(widget=IPPBodyWidget())
 
+get_notifications_schema = GetNotificationsSchema(widget=IPPBodyWidget())
+
 pause_printer_schema = PausePrinterSchema(widget=IPPBodyWidget())
 
 resume_printer_schema = ResumePrinterSchema(widget=IPPBodyWidget())

tests/test_form.py

@@ -385,6 +385,44 @@ def test_get_subscriptions_form():
     assert 'ATTR boolean my-subscriptions 1' in request
 
 
+def test_get_notifications_form_for_one_notification():
+    from pyipptool.forms import get_notifications_form
+    request = get_notifications_form.render(
+        {'header':
+         {'operation_attributes':
+          {'printer_uri': 'https://localhost:631/printers/p0',
+           'requesting_user_name': 'yoda',
+           'notify_subscription_ids': 3,
+           'notify_sequence_numbers': 1,
+           'notify_wait': True}}})
+    assert 'NAME "Get Notifications"' in request
+    assert 'OPERATION "Get-Notifications"' in request
+    assert 'ATTR uri printer-uri https://localhost:631/printers/p0' in request
+    assert 'ATTR name requesting-user-name yoda' in request
+    assert 'ATTR integer notify-subscription-ids 3' in request
+    assert 'ATTR integer notify-sequence-numbers 1' in request
+    assert 'ATTR boolean notify-wait 1' in request
+
+
+def test_get_notifications_form_for_multiple_notifications():
+    from pyipptool.forms import get_notifications_form
+    request = get_notifications_form.render(
+        {'header':
+         {'operation_attributes':
+          {'printer_uri': 'https://localhost:631/printers/p0',
+           'requesting_user_name': 'yoda',
+           'notify_subscription_ids': (3, 4, 5),
+           'notify_sequence_numbers': (2, 9, 29),
+           'notify_wait': True}}})
+    assert 'NAME "Get Notifications"' in request
+    assert 'OPERATION "Get-Notifications"' in request
+    assert 'ATTR uri printer-uri https://localhost:631/printers/p0' in request
+    assert 'ATTR name requesting-user-name yoda' in request
+    assert 'ATTR integer notify-subscription-ids 3,4,5' in request
+    assert 'ATTR integer notify-sequence-numbers 2,9,29' in request
+    assert 'ATTR boolean notify-wait 1' in request
+
+
 def test_pause_printer_form():
     from pyipptool.forms import pause_printer_form
     request = pause_printer_form.render(

tests/test_highlevel.py

@@ -275,6 +275,16 @@ def test_get_subscriptions(_call_ipptool):
     assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/'
 
 
+@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('https://localhost:631/',
+                      printer_uri='',
+                      notify_subscription_ids=3)
+    assert _call_ipptool._mock_mock_calls[0][1][0] == 'https://localhost:631/'
+
+
 @mock.patch.object(pyipptool.wrapper, '_call_ipptool')
 def test_pause_printer(_call_ipptool):
     from pyipptool import pause_printer