Enforce two-day limit on lease requests

appengine/machine_provider/handlers_endpoints_test.py

 #!/usr/bin/env python
 # Copyright 2015 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Unit tests for handlers_endpoints.py."""
 
 import datetime
 import json
 import unittest
@@ skipping 836 matching lines @@
 
     lease_response = jsonish_dict_to_rpc(
         self.call_api('lease', lease_request).json,
         rpc_messages.LeaseResponse,
     )
     self.assertEqual(
         lease_response.error,
         rpc_messages.LeaseRequestError.NONPOSITIVE_DEADLINE,
     )
 
+  def test_lease_duration_too_long(self):
+    def is_group_member(group):
+      return group == 'machine-provider-users'
+    self.mock(acl.auth, 'is_group_member', is_group_member)
+    lease_request = rpc_to_json(rpc_messages.LeaseRequest(
+        dimensions=rpc_messages.Dimensions(
+            os_family=rpc_messages.OSFamily.LINUX,
+        ),
+        duration=9999999999,
+        request_id='abc',
+    ))
+
+    lease_response = jsonish_dict_to_rpc(
+        self.call_api('lease', lease_request).json,
+        rpc_messages.LeaseResponse,
+    )
+    self.assertEqual(
+        lease_response.error,
+        rpc_messages.LeaseRequestError.LEASE_TOO_LONG,
+    )
+
   def test_lease_duration_and_lease_expiration_ts(self):
     def is_group_member(group):
       return group == 'machine-provider-users'
     self.mock(acl.auth, 'is_group_member', is_group_member)
     lease_request = rpc_to_json(rpc_messages.LeaseRequest(
         dimensions=rpc_messages.Dimensions(
             os_family=rpc_messages.OSFamily.LINUX,
         ),
         duration=1,
-        lease_expiration_ts=9999999999,
+        lease_expiration_ts=int(utils.time_time()) + 3600,
         request_id='abc',
     ))
 
     lease_response = jsonish_dict_to_rpc(
         self.call_api('lease', lease_request).json,
         rpc_messages.LeaseResponse,
     )
     self.assertEqual(
         lease_response.error,
         rpc_messages.LeaseRequestError.MUTUAL_EXCLUSION_ERROR,
     )
 
   def test_lease_timestamp(self):
     def is_group_member(group):
       return group == 'machine-provider-users'
     self.mock(acl.auth, 'is_group_member', is_group_member)
     lease_request = rpc_to_json(rpc_messages.LeaseRequest(
         dimensions=rpc_messages.Dimensions(
             os_family=rpc_messages.OSFamily.LINUX,
         ),
-        lease_expiration_ts=9999999999,
+        lease_expiration_ts=int(utils.time_time()) + 3600,
         request_id='abc',
     ))
 
     lease_response = jsonish_dict_to_rpc(
         self.call_api('lease', lease_request).json,
         rpc_messages.LeaseResponse,
     )
     self.assertFalse(lease_response.error)
 
   def test_lease_timestamp_passed(self):
@@ skipping 10 matching lines @@
 
     lease_response = jsonish_dict_to_rpc(
         self.call_api('lease', lease_request).json,
         rpc_messages.LeaseResponse,
     )
     self.assertEqual(
         lease_response.error,
         rpc_messages.LeaseRequestError.LEASE_EXPIRATION_TS_ERROR,
     )
 
+  def test_lease_timestamp_too_far(self):
+    def is_group_member(group):
+      return group == 'machine-provider-users'
+    self.mock(acl.auth, 'is_group_member', is_group_member)
+    lease_request = rpc_to_json(rpc_messages.LeaseRequest(
+        dimensions=rpc_messages.Dimensions(
+            os_family=rpc_messages.OSFamily.LINUX,
+        ),
+        lease_expiration_ts=9999999999,
+        request_id='abc',
+    ))
+
+    lease_response = jsonish_dict_to_rpc(
+        self.call_api('lease', lease_request).json,
+        rpc_messages.LeaseResponse,
+    )
+    self.assertEqual(
+        lease_response.error,
+        rpc_messages.LeaseRequestError.LEASE_TOO_LONG,
+    )
+
   def test_duplicate(self):
     def is_group_member(group):
       return group == 'machine-provider-users'
     self.mock(acl.auth, 'is_group_member', is_group_member)
     lease_request = rpc_to_json(rpc_messages.LeaseRequest(
         dimensions=rpc_messages.Dimensions(
             os_family=rpc_messages.OSFamily.OSX,
         ),
         duration=3,
         request_id='asdf',
@@ skipping 114 matching lines @@
         rpc_messages.LeaseResponse,
     )
     self.assertEqual(
         lease_response.error,
         rpc_messages.LeaseRequestError.UNSPECIFIED_TOPIC,
     )
 
 
 if __name__ == '__main__':
   unittest.main()