ottertune/server/website/tests/test_upload.py

104 lines
4.3 KiB
Python

#
# OtterTune - test_upload.py
#
# Copyright (c) 2017-18, Carnegie Mellon University Database Group
#
import os
from django.core.urlresolvers import reverse
from django.test import TestCase
from website.models import Result, Workload
from website.settings import PROJECT_ROOT
from .utils import (TEST_BASIC_SESSION_ID, TEST_BASIC_SESSION_UPLOAD_CODE,
TEST_PASSWORD, TEST_TUNING_SESSION_ID, TEST_TUNING_SESSION_UPLOAD_CODE,
TEST_USERNAME, TEST_WORKLOAD_ID)
class UploadResultsTests(TestCase):
fixtures = ['test_website.json']
def setUp(self):
self.client.login(username=TEST_USERNAME, password=TEST_PASSWORD)
test_files_dir = os.path.join(PROJECT_ROOT, 'tests', 'test_files')
self.upload_files = {
'metrics_before': os.path.join(test_files_dir, 'sample_metrics_start.json'),
'metrics_after': os.path.join(test_files_dir, 'sample_metrics_end.json'),
'knobs': os.path.join(test_files_dir, 'sample_knobs.json'),
'summary': os.path.join(test_files_dir, 'sample_summary.json')
}
@staticmethod
def open_files(file_info):
files = {}
for name, path in list(file_info.items()):
files[name] = open(path)
return files
@staticmethod
def close_files(files):
for name, fp in list(files.items()):
if name != 'upload_code':
fp.close()
def upload_to_session_ok(self, session_id, upload_code):
num_initial_results = Result.objects.filter(session__id=session_id).count()
form_addr = reverse('new_result')
post_data = self.open_files(self.upload_files)
post_data['upload_code'] = upload_code
response = self.client.post(form_addr, post_data)
self.close_files(post_data)
self.assertEqual(response.status_code, 200)
num_final_results = Result.objects.filter(session__id=session_id).count()
self.assertEqual(num_final_results - num_initial_results, 1)
def upload_to_session_fail_invalidation(self, session_id, upload_code):
form_addr = reverse('new_result')
post_data = {'upload_code': upload_code}
response = self.client.post(form_addr, post_data)
self.assertContains(response, "New result form is not valid:", status_code=400)
self.assertContains(response, "This field is required", count=4, status_code=400)
def upload_to_session_invalid_upload_code(self, session_id):
form_addr = reverse('new_result')
post_data = self.open_files(self.upload_files)
post_data['upload_code'] = "invalid_upload_code"
response = self.client.post(form_addr, post_data)
self.close_files(post_data)
self.assertContains(response, "Invalid upload code", status_code=400)
def test_upload_form_not_post(self):
form_addr = reverse('new_result')
response = self.client.get(form_addr)
self.assertContains(response, "Request type was not POST", status_code=400)
def test_set_modified_workload_on_upload(self):
workload0 = Workload.objects.get(pk=TEST_WORKLOAD_ID)
workload0.status = 3
workload0.save()
self.upload_to_session_ok(TEST_BASIC_SESSION_ID, TEST_BASIC_SESSION_UPLOAD_CODE)
status = Workload.objects.get(pk=TEST_WORKLOAD_ID).status
self.assertEqual(status, 1)
def test_upload_to_basic_session_ok(self):
self.upload_to_session_ok(TEST_BASIC_SESSION_ID, TEST_BASIC_SESSION_UPLOAD_CODE)
def test_upload_to_tuning_session_ok(self):
self.upload_to_session_ok(TEST_TUNING_SESSION_ID, TEST_TUNING_SESSION_UPLOAD_CODE)
def test_upload_to_basic_session_fail_invalidation(self):
self.upload_to_session_fail_invalidation(TEST_BASIC_SESSION_ID,
TEST_BASIC_SESSION_UPLOAD_CODE)
def test_upload_to_tuning_session_fail_invalidation(self):
self.upload_to_session_fail_invalidation(TEST_TUNING_SESSION_ID,
TEST_TUNING_SESSION_UPLOAD_CODE)
def test_upload_to_basic_session_invalid_upload_code(self):
self.upload_to_session_invalid_upload_code(TEST_BASIC_SESSION_ID)
def test_upload_to_tuning_session_invalid_upload_code(self):
self.upload_to_session_invalid_upload_code(TEST_TUNING_SESSION_ID)