test scheduling features of authoring API

splice/ingest.py

@@ -285,6 +285,8 @@ def distribute(data, channel_id, deploy, scheduled_dt=None):
         now = datetime.utcnow()
         if now > scheduled_dt:
             raise ScheduleError("scheduled date needs to be in the future")
+        elif deploy:
+            raise ScheduleError("cannot specify deploy and schedule at the same time")
 
     channel = (
         env.db.session

splice/queries.py

@@ -424,6 +424,51 @@ def get_upcoming_distributions(limit=100, include_past=False):
     return channels
 
 
+def get_scheduled_distributions(minutes, dt_query=None):
+    """
+    Returns distributions scheduled as from a point in time, up to a given number of minutes
+    As a regular task, it is intended to run at least once hourly.
+    :minutes: sets the closest fraction of an hour to select scheduled distributions
+    :dt_query: optionally set the date time to find schedules for
+    """
+    from splice.environment import Environment
+
+    env = Environment.instance()
+
+    if not minutes or 1 < minutes >= 60:
+        raise ValueError("minutes needs to be a number between 1..59 inclusive")
+
+    if dt_query is None:
+        dt_query = datetime.utcnow()
+
+    # getting around PEP8 E712 warning. This is necessary for SQLAlchemy
+    false_value = False
+
+    stmt = (
+        env.db.session
+        .query(Distribution)
+        .filter(Distribution.deployed == false_value)
+    )
+
+    # triggers a match, with the minimum time given by the closest
+    # chunk of the hour given by minute
+    minute_start = (dt_query.minute // minutes) * minutes
+    min_query_dt = datetime(dt_query.year, dt_query.month, dt_query.day, dt_query.hour, minute_start)
+
+    if (minute_start + minutes) >= 60:
+        # if the next chunk goes beyond the hour, limit the query to the next hour
+        max_query_dt = datetime(dt_query.year, dt_query.month, dt_query.day, dt_query.hour + 1, 0)
+    else:
+        max_query_dt = datetime(dt_query.year, dt_query.month, dt_query.day, dt_query.hour, minute_start + minutes)
+
+    stmt = stmt.filter(
+        Distribution.scheduled_start_date.between(min_query_dt, max_query_dt))
+
+    dists = stmt.all()
+
+    return dists
+
+
 def unschedule_distribution(dist_id):
     """
     Remove a distribution id if it is scheduled but not deployed yet

splice/web/api/authoring.py

@@ -33,7 +33,6 @@ def all_tiles():
         if scheduled_ts:
             # scheduled_ts assumed to be in seconds
             scheduled_dt = datetime.utcfromtimestamp(int(scheduled_ts))
-            deploy = False
         urls = distribute(new_data, channel_id, deploy, scheduled_dt)
     except NoResultFound, e:
         msg = "channel_id {0} does not exist".format(channel_id)
@@ -54,6 +53,9 @@ def all_tiles():
     except BadRequest, e:
         env.log("PAYLOAD_ERROR msg:{0}".format(e.message), level=logging.ERROR, name="client_error")
         return jsonify({"err": [{"msg": e.message}]}), 400
+    except ValueError, e:
+        env.log(e.message, level=logging.ERROR, exc_info=sys.exc_info(), name="application")
+        return jsonify({"err": [{"msg": e.message}]}), 400
     except Exception, e:
         env.log(e.message, level=logging.ERROR, exc_info=sys.exc_info(), name="application")
         return jsonify({"err": [{"msg": e.message}]}), 500

tests/api/test_authoring.py

@@ -1,8 +1,11 @@
+import calendar
+from datetime import datetime, timedelta
 from nose.tools import assert_equal
 from flask import url_for
 from mock import Mock, PropertyMock
 from tests.base import BaseTestCase
 import splice.ingest
+from splice.queries import get_scheduled_distributions
 from splice.environment import Environment
 
 env = Environment.instance()
@@ -51,3 +54,45 @@ def test_publish_corrupt_payload(self):
     def test_publish_no_payload(self):
         response = self.client.post(url_for('api.authoring.all_tiles'))
         assert_equal(response.status_code, 400)
+
+    def test_schedule_future(self):
+        now = datetime.utcnow() + timedelta(hours=5)
+        now_ts = calendar.timegm(now.timetuple())
+
+        url = "{0}?deploy=0&channelId=1&scheduledTS={1}".format(url_for('api.authoring.all_tiles'), now_ts)
+
+        # no scheduled distros
+        dists = get_scheduled_distributions(1, now)
+        assert_equal(0, len(dists))
+
+        response = self.client.post(url, data=self.sample_tile_data)
+        assert_equal(response.status_code, 200)
+        # count is 5: 4 images and one distribution file
+        assert_equal(5, self.key_mock.set_contents_from_string.call_count)
+
+        # scheduled distros now
+        dists = get_scheduled_distributions(1, now)
+        assert_equal(1, len(dists))
+
+    def test_schedule_past(self):
+        now = datetime.utcnow() - timedelta(hours=5)
+        now_ts = calendar.timegm(now.timetuple())
+
+        url = "{0}?deploy=0&channelId=1&scheduledTS={1}".format(url_for('api.authoring.all_tiles'), now_ts)
+
+        response = self.client.post(url, data=self.sample_tile_data)
+        assert_equal(response.status_code, 400)
+
+    def test_schedule_bad_data(self):
+        url = "{0}?deploy=0&channelId=1&scheduledTS={1}".format(url_for('api.authoring.all_tiles'), "bad_data")
+
+        response = self.client.post(url, data=self.sample_tile_data)
+        assert_equal(response.status_code, 400)
+
+    def test_schedule_and_deploy(self):
+        now = datetime.utcnow() + timedelta(hours=5)
+        now_ts = calendar.timegm(now.timetuple())
+        url = "{0}?deploy=1&channelId=1&scheduledTS={1}".format(url_for('api.authoring.all_tiles'), now_ts)
+
+        response = self.client.post(url, data=self.sample_tile_data)
+        assert_equal(response.status_code, 400)