Source code for glare.tests.unit.db.test_quota_functions

# Copyright 2017 - Nokia Networks
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.

from six import BytesIO

from glare.db.sqlalchemy import api
from glare.tests.unit import base


[docs]class TestQuotaFunctions(base.BaseTestArtifactAPI): """Test quota db functions."""
[docs] def setUp(self): super(TestQuotaFunctions, self).setUp() self.session = api.get_session()
[docs] def test_count_artifact_number(self): # initially there are no artifacts self.assertEqual(0, api.count_artifact_number( self.req.context, self.session)) # create 5 images, 3 heat templates, 2 murano packages and 7 samples amount = { 'images': 5, 'heat_templates': 3, 'murano_packages': 2, 'sample_artifact': 7 } for type_name in amount: for num in range(amount[type_name]): self.controller.create( self.req, type_name, {'name': type_name + str(num)}) # create 1 artifact of each type from different user req = self.get_fake_request(self.users['user2']) for type_name in amount: self.controller.create(req, type_name, {'name': type_name}) # count numbers for each type for type_name in amount: num = api.count_artifact_number( self.req.context, self.session, type_name) self.assertEqual(amount[type_name], num) # count the whole amount of artifacts self.assertEqual(17, api.count_artifact_number( self.req.context, self.session))
[docs] def test_calculate_uploaded_data(self): # initially there is no data self.assertEqual(0, api.calculate_uploaded_data( self.req.context, self.session)) # create a sample artifact art1 = self.controller.create( self.req, 'sample_artifact', {'name': 'art1'}) # upload 10 bytes to 'blob' self.controller.upload_blob( self.req, 'sample_artifact', art1['id'], 'blob', BytesIO(b'a' * 10), 'application/octet-stream') self.assertEqual(10, api.calculate_uploaded_data( self.req.context, self.session)) # upload 3 blobs to dict_of_blobs with 25, 35 and 45 bytes respectively self.controller.upload_blob( self.req, 'sample_artifact', art1['id'], 'dict_of_blobs/blob1', BytesIO(b'a' * 25), 'application/octet-stream') self.controller.upload_blob( self.req, 'sample_artifact', art1['id'], 'dict_of_blobs/blob2', BytesIO(b'a' * 35), 'application/octet-stream') self.controller.upload_blob( self.req, 'sample_artifact', art1['id'], 'dict_of_blobs/blob3', BytesIO(b'a' * 45), 'application/octet-stream') self.assertEqual(115, api.calculate_uploaded_data( self.req.context, self.session)) # create another sample artifact and upload 100 bytes there art2 = self.controller.create( self.req, 'sample_artifact', {'name': 'art2'}) self.controller.upload_blob( self.req, 'sample_artifact', art2['id'], 'blob', BytesIO(b'a' * 100), 'application/octet-stream') self.assertEqual(215, api.calculate_uploaded_data( self.req.context, self.session)) # create image and upload 150 bytes there img1 = self.controller.create( self.req, 'images', {'name': 'img1'}) self.controller.upload_blob( self.req, 'images', img1['id'], 'image', BytesIO(b'a' * 150), 'application/octet-stream') # the whole amount of uploaded data is 365 bytes self.assertEqual(365, api.calculate_uploaded_data( self.req.context, self.session)) # 215 bytes for sample_artifact self.assertEqual(215, api.calculate_uploaded_data( self.req.context, self.session, 'sample_artifact')) # 150 bytes for images self.assertEqual(150, api.calculate_uploaded_data( self.req.context, self.session, 'images')) # create an artifact from another user and check that it's not included # for the original user req = self.get_fake_request(self.users['user2']) another_art = self.controller.create( req, 'sample_artifact', {'name': 'another'}) # upload 1000 bytes to 'blob' self.controller.upload_blob( req, 'sample_artifact', another_art['id'], 'blob', BytesIO(b'a' * 1000), 'application/octet-stream') # original user still has 365 bytes self.assertEqual(365, api.calculate_uploaded_data( self.req.context, self.session)) # user2 has 1000 self.assertEqual( 1000, api.calculate_uploaded_data(req.context, self.session))
[docs] def test_quota_operations(self): # create several quotas values = { "project1": { "max_uploaded_data": 1000, "max_uploaded_data:images": 500, "max_artifact_number": 10 }, "project2": { "max_uploaded_data": 1000, "max_uploaded_data:sample_artifact": 500, "max_artifact_number": 20 }, "project3": { "max_uploaded_data": 1000 } } api.set_quotas(values, self.session) res = api.get_all_quotas(self.session) self.assertEqual(values, res) # Redefine quotas new_values = { "project1": { "max_uploaded_data": 200, "max_uploaded_data:images": 1000, "max_artifact_number": 30, "max_artifact_number:images": 20 }, "project2": {}, } api.set_quotas(new_values, self.session) # project3 should remain unchanged new_values['project3'] = {"max_uploaded_data": 1000} # project 2 quotas removed new_values.pop('project2') res = api.get_all_quotas(self.session) self.assertEqual(new_values, res)