--- a/fruitbar/crud.py Thu Jun 30 21:47:54 2016 +0800
+++ b/fruitbar/crud.py Mon Jul 18 21:16:47 2016 +0800
class ResourceList(Resource):
def get(self, workspace):
return [project['doc'] for project in g.db.get_many(self.db_index, workspace, with_doc=True)]
def post(self, workspace):
doc = request.json.copy()
doc.update(self.doc_stub)
doc.update({'workspace': workspace})
response = g.db.insert(doc)
return g.db.get('id', response['_id'], with_doc=True)
class ResourceCombiner(Resource):
def get(self, workspace):
for key, resource_class in self.combine.items():
resource = resource_class()
result[key] = resource.get(workspace)
class CRUDResource(Resource):
def get(self, workspace, resource_id):
doc = g.db.get('id', resource_id, with_doc=True)
if doc['workspace'] != workspace:
except (RecordNotFound, RecordDeleted):
def put(self, workspace, resource_id):
doc = g.db.get('id', resource_id, with_doc=True)
if doc['workspace'] != workspace:
except (RecordNotFound, RecordDeleted):
userdata = dict((k, v) for (k, v) in request.json.items() if k in self.safe_fields)
response = g.db.update(doc)
return self.get(workspace, response['_id'])
def delete(self, workspace, resource_id):
doc = g.db.get('id', resource_id, with_doc=True)
if doc['workspace'] != workspace:
except (RecordNotFound, RecordDeleted):
--- a/fruitbar/indexes.py Thu Jun 30 21:47:54 2016 +0800
+++ b/fruitbar/indexes.py Mon Jul 18 21:16:47 2016 +0800
def __init__(self, *args, **kwargs):
kwargs['key_format'] = '16s'
super(ProjectIndex, self).__init__(*args, **kwargs)
def make_key_value(self, data):
if data['_t'] == 'project':
return md5(data['workspace'].encode('utf-8')).digest(), None
def __init__(self, *args, **kwargs):
kwargs['key_format'] = '16s'
super(TaskIndex, self).__init__(*args, **kwargs)
def make_key_value(self, data):
return md5(data['workspace'].encode('utf-8')).digest(), None
def run_delete_for_project(self, db, workspace, project_id):
for task in db.get_many('task', workspace, with_doc=True):
if task['doc']['project_id'] == project_id:
--- a/test.py Thu Jun 30 21:47:54 2016 +0800
+++ b/test.py Mon Jul 18 21:16:47 2016 +0800
class BaseTestCase(TestCase):
self.c = app.test_client()
class IndexTestCase(BaseTestCase):
def create_resource(self):
url = self.list_url % self.workspace
response = self.c.post(url, content_type='application/json', data=json.dumps(self.new_resource))
data = json.loads(response.data)
self.assertIn('_t', data)
self.assertEqual(data.pop('_t'), self.resource_type)
self.assertIn('workspace', data)
self.assertEqual(data.pop('workspace'), self.workspace)
resource_id = data.pop('_id')
self.assertDictEqual(self.new_resource, data)
def read_resource(self, resource_id):
url = self.item_url % (self.workspace, resource_id)
response = self.c.get(url)
data = json.loads(response.data)
self.assertEqual(data['_id'], resource_id)
url = self.list_url % self.workspace
response = self.c.get(url)
data = json.loads(response.data)
self.assertTrue(any(resource.get('_id', None) == resource_id for resource in data))
url = self.all_url % self.workspace
response = self.c.get(url)
data = json.loads(response.data)
self.assertTrue(any(resource.get('_id', None) == resource_id for resource in chain(*data.values())))
def update_resource(self, resource_id):
url = self.item_url % (self.workspace, resource_id)
response = self.c.put(url, content_type='application/json', data=json.dumps(self.resource_update))
data = json.loads(response.data)
self.assertEqual(data['_id'], resource_id)
for key, value in self.resource_update.items():
self.assertEqual(data[key], value)
def delete_resource(self, resource_id):
url = self.item_url % (self.workspace, resource_id)
response = self.c.delete(url)
data = json.loads(response.data)
self.assertEqual(data, '')
response = self.c.get(url)
self.assertEqual(response.status_code, 404)
resource_type = 'project'
list_url = '/%s/projects/'
item_url = '/%s/projects/%s/'
project_id = self.create_resource()
self.read_resource(project_id)
item_url = '/%s/tasks/%s/'
task_id = self.create_resource()
self.read_resource(task_id)