Various changes to support further high-level automation efforts.
* added a RESTful interface for TKO. right now there's only a single, simple resource for accessing test attributes.
* extended the REST server library in a few ways, most notably to support
* querying on keyvals, with something like ?has_keyval=mykey=myvalue&...
* operators, delimited by a colon, like ?hostname:in=host1,host2,host3
* loading relationships over many items efficiently (see InstanceEntry.prepare_for_full_representation()). this is used to fill in keyvals when requesting a job listing, but it can (and should) be used in other places, such as listing labels for a host collection.
* loading a collection with inlined full representations, by passing full_representations=true
* added various features to the AFE RESTful interface as necessary.
* various fixes to the rest_client library, most notably
* changed HTTP client in rest_client.py to use DI rather than singleton, easing testability. the same should be done for _get_request_headers(), to be honest.
* better support for query params, including accepting a MultiValueDict and supporting URIs that already have query args
* basic support for redirects
* builtin support for requesting a full collection (get_full()), when clients explicitly expect the result not to be paged. i'm still considering alternative approaches to this -- it may make sense to have something like this be the default, and have clients set a default page size limit rather than passing it every time.
* minor change to mock.py to provide better debugging output.
Signed-off-by: Steve Howard <showard@google.com>
git-svn-id: http://test.kernel.org/svn/autotest/trunk@4438 592f7852-d20e-0410-864c-8624ca9c26a4
diff --git a/frontend/shared/resource_test_utils.py b/frontend/shared/resource_test_utils.py
new file mode 100644
index 0000000..8cb742f
--- /dev/null
+++ b/frontend/shared/resource_test_utils.py
@@ -0,0 +1,136 @@
+import operator, unittest
+import simplejson
+from django.test import client
+from autotest_lib.frontend.afe import frontend_test_utils, models as afe_models
+
+class ResourceTestCase(unittest.TestCase,
+ frontend_test_utils.FrontendTestMixin):
+ URI_PREFIX = None # subclasses may override this to use partial URIs
+
+ def setUp(self):
+ super(ResourceTestCase, self).setUp()
+ self._frontend_common_setup()
+ self._setup_debug_user()
+ self.client = client.Client()
+
+
+ def tearDown(self):
+ super(ResourceTestCase, self).tearDown()
+ self._frontend_common_teardown()
+
+
+ def _setup_debug_user(self):
+ user = afe_models.User.objects.create(login='debug_user')
+ acl = afe_models.AclGroup.objects.get(name='my_acl')
+ user.aclgroup_set.add(acl)
+
+
+ def _expected_status(self, method):
+ if method == 'post':
+ return 201
+ if method == 'delete':
+ return 204
+ return 200
+
+
+ def raw_request(self, method, uri, **kwargs):
+ method = method.lower()
+ if method == 'put':
+ # the put() implementation in Django's test client is poorly
+ # implemented and only supports url-encoded keyvals for the data.
+ # the post() implementation is correct, though, so use that, with a
+ # trick to override the method.
+ method = 'post'
+ kwargs['REQUEST_METHOD'] = 'PUT'
+
+ client_method = getattr(self.client, method)
+ return client_method(uri, **kwargs)
+
+
+ def request(self, method, uri, encode_body=True, **kwargs):
+ expected_status = self._expected_status(method)
+
+ if 'data' in kwargs:
+ kwargs.setdefault('content_type', 'application/json')
+ if kwargs['content_type'] == 'application/json':
+ kwargs['data'] = simplejson.dumps(kwargs['data'])
+
+ if uri.startswith('http://'):
+ full_uri = uri
+ else:
+ assert self.URI_PREFIX
+ full_uri = self.URI_PREFIX + '/' + uri
+
+ response = self.raw_request(method, full_uri, **kwargs)
+ self.assertEquals(
+ response.status_code, expected_status,
+ 'Requesting %s\nExpected %s, got %s: %s (headers: %s)'
+ % (full_uri, expected_status, response.status_code,
+ response.content, response._headers))
+
+ if response['content-type'] != 'application/json':
+ return response.content
+
+ try:
+ return simplejson.loads(response.content)
+ except ValueError:
+ self.fail('Invalid reponse body: %s' % response.content)
+
+
+ def sorted_by(self, collection, attribute):
+ return sorted(collection, key=operator.itemgetter(attribute))
+
+
+ def _read_attribute(self, item, attribute_or_list):
+ if isinstance(attribute_or_list, basestring):
+ attribute_or_list = [attribute_or_list]
+ for attribute in attribute_or_list:
+ item = item[attribute]
+ return item
+
+
+ def check_collection(self, collection, attribute_or_list, expected_list,
+ length=None, check_number=None):
+ """Check the members of a collection of dicts.
+
+ @param collection: an iterable of dicts
+ @param attribute_or_list: an attribute or list of attributes to read.
+ the results will be sorted and compared with expected_list. if
+ a list of attributes is given, the attributes will be read
+ hierarchically, i.e. item[attribute1][attribute2]...
+ @param expected_list: list of expected values
+ @param check_number: if given, only check this number of entries
+ @param length: expected length of list, only necessary if check_number
+ is given
+ """
+ actual_list = sorted(self._read_attribute(item, attribute_or_list)
+ for item in collection['members'])
+ if length is None and check_number is None:
+ length = len(expected_list)
+ if length is not None:
+ self.assertEquals(len(actual_list), length,
+ 'Expected %s, got %s: %s'
+ % (length, len(actual_list),
+ ', '.join(str(item) for item in actual_list)))
+ if check_number:
+ actual_list = actual_list[:check_number]
+ self.assertEquals(actual_list, expected_list)
+
+
+ def check_relationship(self, resource_uri, relationship_name,
+ other_entry_name, field, expected_values,
+ length=None, check_number=None):
+ """Check the members of a relationship collection.
+
+ @param resource_uri: URI of base resource
+ @param relationship_name: name of relationship attribute on base
+ resource
+ @param other_entry_name: name of other entry in relationship
+ @param field: name of field to grab on other entry
+ @param expected values: list of expected values for the given field
+ """
+ response = self.request('get', resource_uri)
+ relationship_uri = response[relationship_name]['href']
+ relationships = self.request('get', relationship_uri)
+ self.check_collection(relationships, [other_entry_name, field],
+ expected_values, length, check_number)