autoupdate.py: Allow max_updates through update URL
Currently in order to put a ceiling on the number of updates that can be
performed in a devserver, we need to spawn a new devserver with flag
--max_updates. The problem is now for each run of an autotest, they
should spwan a new devserver alongside the lab devservers which is not
ideal.
We solve this problem by dynamically configuring the devserver
using a unique identifier. This is done by calling into 'session_id' API
of the devserver. Then clients can send their requests appending this
session_id to be responded likewise.
For maximum updates, we first configure the devserver to set a
'max_updates' data for a unique session ID. Then client can send
requests using the session ID as a query string be capped on the number
of updates they get.
BUG=chromium:1004489
TEST=autoupdate_unittest.py
TEST=devserver_integration_test.py
Change-Id: Ieef921b177ba0ec789d6471a34a4f8e44f5482af
Reviewed-on: https://chromium-review.googlesource.com/c/chromiumos/platform/dev-util/+/1996148
Tested-by: Amin Hassani <ahassani@chromium.org>
Commit-Queue: Amin Hassani <ahassani@chromium.org>
Reviewed-by: Allen Li <ayatane@chromium.org>
diff --git a/autoupdate_unittest.py b/autoupdate_unittest.py
index 648ef7c..94db2b4 100755
--- a/autoupdate_unittest.py
+++ b/autoupdate_unittest.py
@@ -111,5 +111,69 @@
self.assertIn('error-unknownApplication', au_mock.HandleUpdatePing(request))
+
+class MaxUpdatesTableTest(unittest.TestCase):
+ """Tests MaxUpdatesTable"""
+
+ def testSessionTable(self):
+ """Tests that SessionData() method correctly returns requested data."""
+ table = autoupdate.SessionTable()
+ g_data = {'foo': 0}
+
+ table.SetSessionData('id-1', g_data)
+ with table.SessionData('id-1') as data:
+ data.update({'foo': data.get('foo') + 1})
+ # Value of the global data should be increased by now.
+ self.assertTrue(g_data['foo'], 1)
+
+ # Increase again.
+ with table.SessionData('id-1') as data:
+ data.update({'foo': data.get('foo') + 1})
+ self.assertTrue(g_data['foo'], 2)
+
+ # Make sure multiple sessions can be set and used.
+ g_data2 = {'foo': 10}
+ table.SetSessionData('id-2', g_data2)
+ with table.SessionData('id-2') as data:
+ data.update({'foo': data.get('foo') + 1})
+ self.assertTrue(g_data2['foo'], 11)
+
+ def testNoneSession(self):
+ """Tests if a session is not set, it should be returned as None."""
+ table = autoupdate.SessionTable()
+ # A session ID that has never been set should not return anything.
+ with table.SessionData('foo-id') as data:
+ self.assertDictEqual(data, {})
+
+ def testOverrideSession(self):
+ """Tests that a session can be overriden.."""
+ table = autoupdate.SessionTable()
+
+ table.SetSessionData('id-1', {'foo': 0})
+ table.SetSessionData('id-1', {'bar': 1})
+ with table.SessionData('id-1') as data:
+ self.assertEqual(data.get('bar'), 1)
+
+ @mock.patch.object(autoupdate.SessionTable, '_IsSessionExpired',
+ side_effect=lambda s: 'foo' in s.data)
+ @mock.patch.object(autoupdate.SessionTable, '_ShouldPurge',
+ return_value=True)
+ # pylint: disable=unused-argument
+ def testPurge(self, p, ps):
+ """Tests that data is being purged correctly."""
+ table = autoupdate.SessionTable()
+
+ table.SetSessionData('id-1', {'foo': 1})
+ table.SetSessionData('id-2', {'bar': 2})
+
+ # Set a random session to make _Purge() be called.
+ table.SetSessionData('blah', {'blah': 1})
+ # Only id-1 should be purged by now.
+ with table.SessionData('id-1') as data:
+ self.assertDictEqual(data, {})
+ with table.SessionData('id-2') as data:
+ self.assertDictEqual(data, {'bar': 2})
+
+
if __name__ == '__main__':
unittest.main()