Get broadcasts by status

CHANGELOG.rst

@@ -1,5 +1,11 @@
 Changelog
 ---------
+2014-12-29: v1.4.0
+    Add a new endpoint to get broadcasts by list_id and status
+
+2014-11-17: v1.3.0
+   * Add a new endpoint to cancel scheduled broadcast
+
 2014-11-20 v1.2.1
   * Removed SSL Certificate workaround
   * Upgraded httplib2 >= 0.9.0

aweber_api/entry.py

@@ -126,9 +126,9 @@ def schedule_broadcast(self, bc_id, scheduled_for):
 
         * Note:
             This method only works on List Entry resources and
-            requires access to subscriber information. Please
+            requires send broadcast email permissions. Please
             refer to the AWeber API Reference Documentation at
-            https://labs.aweber.com/docs/reference/1.0#account
+            https://labs.aweber.com/docs/reference/1.0#Broadcast Scheduler
             for more details on how to call this method.
 
         """
@@ -137,11 +137,54 @@ def schedule_broadcast(self, bc_id, scheduled_for):
         url = '{0}/broadcasts/{1}/schedule'.format(self.url, bc_id)
         return self.adapter.request('POST', url, body, response='status')
 
+    def get_broadcasts(self, status, **kwargs):
+        """Invoke the API method to retrieve broadcasts by status.
+
+        * Note:
+            This method only works on List Entry resources. Please
+            refer to the AWeber API Reference Documentation at
+            https://labs.aweber.com/docs/reference/1.0#Get Broadcast
+            for more details on how to call this method.
+
+        """
+        self._method_for('list')
+        params = {'status': status}
+        params.update(kwargs)
+        query_string = urlencode(params)
+        url = '{0.url}/broadcasts?{1}'.format(self, query_string)
+
+        data = self.adapter.request('GET', url)
+        collection = aweber_api.AWeberCollection(url, data, self.adapter)
+        collection._data['total_size'] = self._get_broadcast_count(
+            query_string)
+        return collection
+
+    def cancel_broadcast(self, bc_id):
+        """Invoke the API method to cancel the given scheduled broadcast.
+
+        * Note:
+            This method only works on List Entry resources and
+            requires send broadcast email permissions. Please refer
+            to the AWeber API Reference Documentation at
+            https://labs.aweber.com/docs/reference/1.0#Cancel Broadcast
+            more details on how to call this method.
+
+        """
+        self._method_for('list')
+        url = '{0}/broadcasts/{1}/cancel'.format(self.url, bc_id)
+        return self.adapter.request('POST', url, data={}, response='status')
+
     def _get_total_size(self, uri, **kwargs):
         """Get actual total size number from total_size_link."""
         total_size_uri = '{0}&ws.show=total_size'.format(uri)
         return int(self.adapter.request('GET', total_size_uri))
 
+    def _get_broadcast_count(self, query_string):
+        """Get actual total size number from total_size_link."""
+        total_size_uri = '{0.url}/broadcasts/total?{1}'.format(
+            self, query_string)
+        return int(self.adapter.request('GET', total_size_uri)['total_size'])
+
     def get_parent_entry(self):
         """Return the parent entry of this entry
 

setup.py

@@ -8,7 +8,7 @@
 
 setup(
     name='aweber_api',
-    version='1.2.1',
+    version='1.4.0',
     author='AWeber Dev Team',
     author_email='api@aweber.com',
     maintainer='AWeber API Team',

tests/mock_adapter.py

@@ -48,6 +48,10 @@
                          'email=joe%40example.com': ({}, 'subscribers/find'),
         '/accounts/1/lists/303449/subscribers?ws.show=total_size&ws.op=find&' \
                          'email=joe%40example.com': ({}, 'subscribers/find_ts'),
+         '/accounts/1/lists/303449/broadcasts/total?status=sent': (
+             {'total_size': 10}, 'campaigns/303449'),
+         '/accounts/1/lists/303449/broadcasts?status=sent': (
+             {'total_size': 10}, 'campaigns/303449'),
     },
     'POST' : {
         '/accounts/1/lists/303449/any_collection':  ({
@@ -71,6 +75,16 @@
             'location': '/accounts/1/lists/303449/broadcasts/3/schedule'},
             'error'
         ),
+        '/accounts/1/lists/303449/broadcasts/2/cancel': ({
+            'status': '204',
+            'location': '/accounts/1/lists/303449/broadcasts/2/cancel'},
+            None
+        ),
+        '/accounts/1/lists/303449/broadcasts/3/cancel': ({
+            'status': '400',
+            'location': '/accounts/1/lists/303449/broadcasts/3/cancel'},
+            'error'
+        ),
 
     },
     'PATCH' : {

tests/test_aweber_entry.py

@@ -160,6 +160,65 @@ def test_should_raise_exception_when_failing(self):
         )
 
 
+class TestListCancelBroadcast(ListTestCase):
+
+    def setUp(self):
+        super(TestListCancelBroadcast, self).setUp()
+        self.aweber.adapter.requests = []
+        self.status = self.list_.cancel_broadcast(bc_id=2)
+        self.request = self.aweber.adapter.requests[0]
+
+    def test_should_return_status(self):
+        self.assertEqual(int(self.status), 204)
+
+    def test_should_make_post_request(self):
+        self.assertEqual(self.request['method'], 'POST')
+
+    def test_should_build_correct_url(self):
+            self.assertEqual(self.request['url'],
+            '/accounts/1/lists/303449/broadcasts/2/cancel'
+        )
+
+    def test_should_pass_empty_date(self):
+            self.assertEqual(self.request['data'], {})
+
+
+class TestListCancelBroadcastError(ListTestCase):
+
+    def setUp(self):
+        super(TestListCancelBroadcastError, self).setUp()
+        self.list_ = self.aweber.load_from_url('/accounts/1/lists/303449')
+        self.aweber.adapter.requests = []
+
+    def test_should_raise_exception_when_failing(self):
+        self.assertRaises(
+            APIException,
+            self.list_.cancel_broadcast,
+            bc_id=3,
+        )
+
+
+class TestListGetBroadcasts(ListTestCase):
+
+    def setUp(self):
+        super(TestListGetBroadcasts, self).setUp()
+        self.aweber.adapter.requests = []
+        self.broadcasts = self.list_.get_broadcasts(status='sent')
+        self.request = self.aweber.adapter.requests[0]
+
+    def test_should_return_collection(self):
+        self.assertEqual(type(self.broadcasts), AWeberCollection)
+
+    def test_should_make_get_request(self):
+        self.assertEqual(self.request['method'], 'GET')
+
+    def test_should_build_correct_url(self):
+        self.assertEqual(
+            self.request['url'],
+            '/accounts/1/lists/303449/broadcasts?status=sent'
+        )
+
+
 class SubscriberTestCase(TestCase):
 
     def setUp(self):