Fix delete function with running job associated

User should not delete a function with running jobs associated.

Change-Id: I627ac7ef057768e7c2d99f56e17da9eacbf4ad76
This commit is contained in:
Jenkins 2017-08-21 23:41:05 +00:00 committed by Lingxian Kong
parent f3df0936d5
commit 55a06cfb77
3 changed files with 43 additions and 4 deletions

View File

@ -174,8 +174,12 @@ class FunctionsController(rest.RestController):
with db_api.transaction():
func_db = db_api.get_function(id)
source = func_db.code['source']
if len(func_db.jobs) > 0:
raise exc.NotAllowedException(
'The function is still associated with running job(s).'
)
source = func_db.code['source']
if source == 'package':
self.storage_provider.delete(context.get_ctx().projectid, id)

View File

@ -106,7 +106,20 @@ class Job(model_base.QinlingSecureModelBase):
# Delete service mapping automatically when deleting function.
Function.service = relationship("FunctionServiceMapping", uselist=False,
cascade="all, delete-orphan")
Function.service = relationship(
"FunctionServiceMapping",
uselist=False,
cascade="all, delete-orphan"
)
Runtime.functions = relationship("Function", back_populates="runtime")
Function.jobs = relationship("Job", back_populates="function")
# Only get jobs
Function.jobs = relationship(
"Job",
back_populates="function",
primaryjoin=(
"and_(Function.id==Job.function_id, "
"~Job.status.in_(['done', 'cancelled']))"
)
)

View File

@ -12,11 +12,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import datetime
import json
import tempfile
import mock
from qinling import status
from qinling.tests.unit.api import base
from qinling.tests.unit import base as unit_base
@ -118,3 +120,23 @@ class TestFunctionController(base.APITest):
unit_base.DEFAULT_PROJECT_ID, db_func.id
)
mock_delete_func.assert_called_once_with(db_func.id)
def test_delete_with_running_job(self):
db_func = self.create_function(
runtime_id=self.runtime_id, prefix=TEST_CASE_NAME
)
self.create_job(
function_id=db_func.id,
prefix=TEST_CASE_NAME,
status=status.AVAILABLE,
first_execution_time=datetime.utcnow(),
next_execution_time=datetime.utcnow(),
count=1
)
resp = self.app.delete(
'/v1/functions/%s' % db_func.id,
expect_errors=True
)
self.assertEqual(403, resp.status_int)