Skip to content

CIRCLE3 / cloud

Go to a project
Commit 8738a70e by Carpoon
Options

Added tests 

Test for import export functions
Exam functions
Periodc clean funtions
parent bc1550c0
Showing with 601 additions and 2 deletions
circle/dashboard/tests/test_views.py
...@@ -27,7 +27,7 @@ from django.contrib.auth import authenticate ...@@ -27,7 +27,7 @@ from django.contrib.auth import authenticate
from common.tests.celery_mock import MockCeleryMixin from common.tests.celery_mock import MockCeleryMixin
from dashboard.views import VmAddInterfaceView from dashboard.views import VmAddInterfaceView
from vm.models import Instance, InstanceTemplate, Lease, Node, Trait from vm.models import Instance, InstanceTemplate, Lease, Node, Trait, ExportedVM
from vm.operations import (WakeUpOperation, AddInterfaceOperation, from vm.operations import (WakeUpOperation, AddInterfaceOperation,
AddPortOperation, RemoveInterfaceOperation, AddPortOperation, RemoveInterfaceOperation,
DeployOperation, RenameOperation) DeployOperation, RenameOperation)
...@@ -35,8 +35,15 @@ from ..models import Profile ...@@ -35,8 +35,15 @@ from ..models import Profile
from firewall.models import Vlan, Host, VlanGroup from firewall.models import Vlan, Host, VlanGroup
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from simplesshkey.models import UserKey from simplesshkey.models import UserKey
from exam.models import Exam
from storage.models import DataStore, ExportedDisk, Disk
from storage.tasks.periodic_tasks import export_garbage_disk_collector
from datetime import datetime
from dateutil.relativedelta import relativedelta
import django.conf import django.conf
from django.conf import settings as settings_global
settings = django.conf.settings.FIREWALL_SETTINGS settings = django.conf.settings.FIREWALL_SETTINGS
...@@ -45,10 +52,392 @@ class LoginMixin(object): ...@@ -45,10 +52,392 @@ class LoginMixin(object):
response = client.post('/accounts/login/', {'username': username, response = client.post('/accounts/login/', {'username': username,
'password': password}, 'password': password},
follow=follow) follow=follow)
self.assertNotEqual(response.status_code, 403) self.assertNotEqual(response.status_code, 404)
return response return response
class ExamListTest(LoginMixin, MockCeleryMixin, TestCase):
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.set_password('password')
self.u1.user_permissions.add(Permission.objects.get(codename="start_exam"))
self.u1.save()
self.u2 = User.objects.create(username='user2')
self.u2.set_password('password')
self.u2.save()
self.u2.user_permissions.add(Permission.objects.get(codename="start_exam"))
self.u2.user_permissions.add(Permission.objects.get(codename="create_exam"))
self.u2.save()
self.u3 = User.objects.create(username='user3')
self.u3.set_password('password')
self.u3.save()
self.exam1 = Exam(name="test_exam1", owner=self.u1)
self.exam1.save()
self.exam2 = Exam(name="test_exam2", owner=self.u1)
self.exam2.save()
self.exam3 = Exam(name="test_exam3", owner=self.u2)
self.exam3.save()
self.exam3.set_level(self.u1, 'operator')
def test_anon_filter(self):
c = Client()
response = c.get('/dashboard/exam/list/?s="3"')
self.assertEqual(response.status_code, 302)
def test_not_permitteduser(self):
c = Client()
self.login(c, 'user3')
response = c.get('/dashboard/exam/list/"')
self.assertEqual(response.status_code, 404)
def test_permitteduser_filter(self):
c = Client()
self.login(c, 'user1')
response = c.get('/dashboard/exam/list/?s="3"')
self.assertEqual(response.status_code, 200)
def tearDown(self):
super(ExamListTest, self).tearDown()
self.u1.delete()
class ExamDetailTest(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json', 'node.json']
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.set_password('password')
self.u1.save()
self.u2 = User.objects.create(username='user2')
self.u2.set_password('password')
self.u2.save()
self.u2.user_permissions.add(Permission.objects.get(codename="create_exam"))
self.exam = Exam(name="test_exam", owner=self.u2)
self.exam1.save()
self.uu1 = User.objects.create(username='u_user1')
self.uu1.save()
self.uu2 = User.objects.create(username='u_user2')
self.uu2.save()
self.uu3 = User.objects.create(username='u_user3')
self.uu3.save()
def tearDown(self):
super(ExamDetailTest, self).tearDown()
self.u1.delete()
self.u2.delete()
self.us.delete()
def test_anon_vm_page(self):
c = Client()
response = c.get('/dashboard/exam/%s/' % self.exam1.pk)
self.assertRedirects(response, '/accounts/login/?next=/dashboard/exam/%s/' % self.exam1.pk)
def test_404_exam_page(self):
client = Client()
self.login(client, 'user1')
response = client.get('/dashboard/exam/404/')
self.assertEqual(response.status_code, 404)
def test_access(self):
client = Client()
self.login(client, 'user2')
response = client.get('/dashboard/exam/%s/' % self.exam1.pk)
self.assertEqual(response.status_code, 200)
def test_access_missing(self):
client = Client()
self.login(client, 'user1')
response = client.get('/dashboard/exam/%s/' % self.exam1.pk)
self.assertEqual(response.status_code, 302)
def test_create_missing_rights(self):
client = Client()
self.login(client, 'user1')
response = client.post('/dashboard/exam/create/',
{'name': 'test', })
self.assertEqual(response.status_code, 403)
def test_create(self):
client = Client()
self.login(client, 'user2')
response = client.post('/dashboard/exam/create/',
{'name': 'test_create', })
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/6/')
def test_unpermitted_set_name(self):
client = Client()
self.login(client, "user1")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_name': 'new_name'})
self.assertEqual(response.status_code, 403)
self.exam1.refresh_from_db()
self.assertEqual(self.exam1.name, "test_exam")
def test_permitted_set_name(self):
client = Client()
self.login(client, "user2")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_name': 'new_name'})
self.exam1.refresh_from_db()
self.assertEqual(response.status_code, 302)
self.assertEqual(self.exam1.name, "new_name")
def test_unpermitted_set_description(self):
client = Client()
self.login(client, "user1")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_description': 'new_description'})
self.assertEqual(response.status_code, 403)
self.exam1.refresh_from_db()
self.assertEqual(self.exam1.description, "")
def test_permitted_set_description(self):
client = Client()
self.login(client, "user2")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_description': 'new_description'})
self.assertEqual(response.status_code, 302)
self.exam1.refresh_from_db()
self.assertEqual(self.exam1.description, "new_description")
def test_unpermitted_set_template(self):
template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user1")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new-template': template.pk})
self.assertEqual(response.status_code, 403)
self.exam1.refresh_from_db()
self.assertEqual(self.exam1.template, None)
def test_permitted_set_template(self):
template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user2")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new-template': template.pk})
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
self.exam1.refresh_from_db()
self.assertEqual(self.exam1.template, template)
def test_unpermitted_add_member(self):
client = Client()
self.login(client, "user1")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_member': self.uu1.pk})
self.assertEqual(response.status_code, 403)
self.exam1.refresh_from_db()
self.assertNotIn(self.uu1, self.exam1.examinees.all())
def test_permitted_add_member(self):
client = Client()
self.login(client, "user2")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_member': self.uu1.username})
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
self.exam1.refresh_from_db()
self.assertIn(self.uu1, set(self.exam1.examinees.all()))
def test_unpermitted_remove_member(self):
client = Client()
self.login(client, "user1")
self.exam1.examinees.add(self.uu1)
response = client.post("/dashboard/exam/%s/remove/user/%s/" % (self.exam1.pk, self.uu1.pk))
self.assertEqual(response.status_code, 403)
self.exam1.refresh_from_db()
self.assertIn(self.uu1, set(self.exam1.examinees.all()))
def test_permitted_remove_member(self):
client = Client()
self.login(client, "user2")
self.exam1.examinees.add(self.uu1)
response = client.post("/dashboard/exam/%s/remove/user/%s/" % (self.exam1.pk, self.uu1.pk))
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
self.exam1.refresh_from_db()
self.assertNotIn(self.uu1, set(self.exam1.examinees.all()))
def test_unpermitted_add_members(self):
client = Client()
self.login(client, "user1")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_members': "%s\r\n%s" % (self.uu2.username, self.uu3.username)})
self.assertEqual(response.status_code, 403)
self.exam1.refresh_from_db()
self.assertNotIn(self.uu2, self.exam1.examinees.all())
self.assertNotIn(self.uu3, self.exam1.examinees.all())
def test_permitted_add_members(self):
client = Client()
self.login(client, "user2")
response = client.post("/dashboard/exam/%s/" % self.exam1.pk,
{'new_members': "%s\r\n%s" % (self.uu2.username, self.uu3.username)})
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
self.exam1.refresh_from_db()
self.assertIn(self.uu2, self.exam1.examinees.all())
self.assertIn(self.uu3, self.exam1.examinees.all())
def test_unpermitted_remove_members(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
client = Client()
self.login(client, "user1")
response = client.post("/dashboard/exam/%s/remove/user/all/" % self.exam1.pk)
self.assertEqual(response.status_code, 403)
self.exam1.refresh_from_db()
self.assertIn(self.uu2, self.exam1.examinees.all())
self.assertIn(self.uu3, self.exam1.examinees.all())
def test_permitted_remove_members(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
client = Client()
self.login(client, "user2")
response = client.post("/dashboard/exam/%s/remove/user/all/" % self.exam1.pk)
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
self.exam1.refresh_from_db()
self.assertNotIn(self.uu2, self.exam1.examinees.all())
self.assertNotIn(self.uu3, self.exam1.examinees.all())
def test_unpermitted_start_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user1")
response = client.get("/dashboard/exam/start/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 403)
def test_permitted_start_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user2")
response = client.get("/dashboard/exam/start/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
def test_unpermitted_stop_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user1")
response = client.get("/dashboard/exam/stop/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 403)
def test_permitted_stop_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user2")
response = client.get("/dashboard/exam/stop/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
def test_unpermitted_shutdown_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user1")
response = client.get("/dashboard/exam/shutdown/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 403)
def test_permitted_shutdown_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user2")
response = client.get("/dashboard/exam/shutdown/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
def test_unpermitted_destroy_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user1")
response = client.get("/dashboard/exam/destroy/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 403)
def test_permitted_destroy_vms(self):
self.exam1.examinees.add(self.uu2)
self.exam1.examinees.add(self.uu3)
self.exam1.template = InstanceTemplate.objects.get(id=1)
client = Client()
self.login(client, "user2")
response = client.get("/dashboard/exam/destroy/%s/" % self.exam1.pk)
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % self.exam1.pk)
def test_transfer_not_offered(self):
c2 = self.u2.notification_set.count()
c = Client()
self.login(c, 'user1')
response = c.post('/dashboard/exam/1/tx/')
self.assertEqual(response.status_code, 404)
self.assertEqual(self.u2.notification_set.count(), c2)
def test_transfer_not_owner(self):
exam_change = Exam(name="change_exam", owner=self.u1)
exam_change.save()
exam_change.set_level(self.u2, 'operator')
exam_change.save()
n_1 = self.u1.notification_set.count()
client = Client()
self.login(client, 'user2')
response = client.post('/dashboard/exam/%s/tx/' % exam_change.pk, {'name': 'user1'})
self.assertEqual(response.status_code, 403)
def test_transfer_owned_offer_and_accepted(self):
exam_change = Exam(name="change_exam", owner=self.u2)
exam_change.save()
n_1 = self.u1.notification_set.count()
client_1 = Client()
self.login(client_1, 'user1')
client_2 = Client()
self.login(client_2, 'user2')
response = client_2.post('/dashboard/exam/%s/tx/' % exam_change.pk, {'name': 'user1'})
self.assertEqual(response.status_code, 302)
self.assertEqual(self.u1.notification_set.count(), n_1 + 1)
message = str(list(self.u1.notification_set.all())[-1].message)
link = message.split('<a href="')[1].split('"')[0]
response = client_1.post(link)
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/%s/' % exam_change.pk)
self.assertEqual(exam_change.owner, self.u2)
def test_unpermitted_delete(self):
exam1_id = self.exam1.pk
client = Client()
self.login(client, "user1")
response = client.post('/dashboard/exam/delete/%s/' % self.exam1.pk)
self.assertEqual(response.status_code, 403)
self.assertIsInstance(Exam.objects.get(pk=exam1_id), Exam)
def test_permitted_delete(self):
exam1_id = self.exam1.pk
client = Client()
self.login(client, "user2")
response = client.post('/dashboard/exam/delete/%s/' % self.exam1.pk)
self.assertEqual(response.status_code, 302)
self.assertRedirects(response, '/dashboard/exam/list/')
try:
Exam.objects.get(pk=exam1_id)
except Exception as e:
self.assertIsInstance(e, Exam.DoesNotExist)
class VmDetailTest(LoginMixin, MockCeleryMixin, TestCase): class VmDetailTest(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json', 'node.json'] fixtures = ['test-vm-fixture.json', 'node.json']
...@@ -1387,6 +1776,152 @@ class GroupListTest(LoginMixin, MockCeleryMixin, TestCase): ...@@ -1387,6 +1776,152 @@ class GroupListTest(LoginMixin, MockCeleryMixin, TestCase):
self.g2.delete() self.g2.delete()
class DiskImport(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json']
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.set_password('password')
self.u1.save()
def test_not_permitted_disk_import(self):
client = Client()
self.login(client, 'user1')
response = client.get('/dashboard/vm/1/op/import_disk/')
self.assertEqual(response.status_code, 403)
def test_permitted_disk_import(self):
client = Client()
self.login(client, 'user1')
self.u1.user_permissions.add(Permission.objects.get(codename='import_disk'))
self.u1.save()
inst = Instance.objects.get(pk=1)
inst.set_level(self.u1, 'owner')
inst.save()
response = client.get('/dashboard/vm/1/op/import_disk/')
self.assertEqual(response.status_code, 200)
class DiskExport(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json']
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.set_password('password')
self.u1.save()
def test_not_permitted_disk_export(self):
client = Client()
self.login(client, 'user1')
response = client.get('/dashboard/vm/1/op/export_disk/')
self.assertEqual(response.status_code, 403)
def test_permitted_disk_export(self):
client = Client()
self.login(client, 'user1')
self.u1.user_permissions.add(Permission.objects.get(codename='export_disk'))
self.u1.save()
inst = Instance.objects.get(pk=1)
inst.set_level(self.u1, 'owner')
inst.save()
response = client.get('/dashboard/vm/1/op/export_disk/')
self.assertEqual(response.status_code, 200)
class TemplateImport(LoginMixin, MockCeleryMixin, TestCase):
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.set_password('password')
self.u1.save()
def test_permitted_template_import(self):
client = Client()
self.login(client, 'user1')
self.u1.user_permissions.add(Permission.objects.get(codename='import_template'))
self.u1.save()
response = client.get('/dashboard/template/import/')
self.assertEqual(response.status_code, 200)
def test_not_permitted_template_import(self):
client = Client()
self.login(client, 'user1')
response = client.get('/dashboard/template/import/')
self.assertEqual(response.status_code, 403)
class VmExport(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json', 'node.json']
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.set_password('password')
self.u1.save()
def test_permitted_vm_export(self):
client = Client()
self.login(client, 'user1')
inst = Instance.objects.get(pk=1)
inst.set_level(self.u1, 'operator')
inst.save()
self.u1.user_permissions.add(Permission.objects.get(codename='export_vm'))
self.u1.save()
response = client.get('/dashboard/vm/1/op/export_vm/')
self.assertEqual(response.status_code, 200)
def test_not_permitted_vm_export(self):
client = Client()
self.login(client, 'user1')
inst = Instance.objects.get(pk=1)
inst.set_level(self.u1, 'user')
inst.save()
self.u1.user_permissions.add(Permission.objects.get(codename='export_vm'))
response = client.get('/dashboard/vm/1/op/export_vm/')
self.assertEqual(response.status_code, 403)
class ExportedVMDelete(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json']
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.set_password('password')
self.u1.save()
self.u2 = User.objects.create(username='user2')
self.u2.set_password('password')
self.u2.save()
self.u2.save()
self.vm = Instance.objects.get(pk=1)
self.vm.owner = self.u1
self.vm.save()
datastore = DataStore.objects.filter(pk=1).get()
self.exportedvm = ExportedVM(name="test", filename="", vm=self.vm, datastore=datastore)
self.exportedvm.save()
def test_anon_filter(self):
c = Client()
response = c.get('/dashboard/exportedvm/delete/%s/' % self.exportedvm.pk)
self.assertEqual(response.status_code, 302)
def test_not_permitteduser(self):
c = Client()
self.login(c, 'user3')
response = c.get('/dashboard/exportedvm/delete/%s/' % self.exportedvm.pk)
self.assertEqual(response.status_code, 302)
def test_permitteduser_filter(self):
c = Client()
self.login(c, 'user1')
response = c.get('/dashboard/exportedvm/delete/%s/' % self.exportedvm.pk)
self.assertEqual(response.status_code, 200)
def tearDown(self):
super(ExportedVMDelete, self).tearDown()
self.u1.delete()
self.u2.delete()
class VmDetailVncTest(LoginMixin, MockCeleryMixin, TestCase): class VmDetailVncTest(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json', 'node.json'] fixtures = ['test-vm-fixture.json', 'node.json']
...@@ -1920,3 +2455,61 @@ class TwoFactorTest(LoginMixin, TestCase): ...@@ -1920,3 +2455,61 @@ class TwoFactorTest(LoginMixin, TestCase):
[('/', 302), [('/', 302),
('/dashboard/', 302)] ('/dashboard/', 302)]
) )
class ExportedVMCleanupTestCase(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json']
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.save()
self.vm = Instance.objects.get(pk=1)
self.vm.owner = self.u1
self.vm.save()
datastore = DataStore.objects.filter(pk=1).get()
self.exportedvm1 = ExportedVM(name="test1", filename="1", vm=self.vm, datastore=datastore)
self.exportedvm1.save()
self.exportedvm2 = ExportedVM(name="test2", filename="2", vm=self.vm, datastore=datastore, created=datetime.now() - relativedelta(hours=settings_global.EXPORT_KEEP_TIME_IN_HOURS - 1))
self.exportedvm2.save()
self.exportedvm3 = ExportedVM(name="test3", filename="3", vm=self.vm, datastore=datastore, created=datetime.now() - relativedelta(hours=settings_global.EXPORT_KEEP_TIME_IN_HOURS + 1))
self.exportedvm3.save()
def test_exam_cleanup(self,):
self.assertEqual(3, len(ExportedVM.objects.all()))
export_garbage_disk_collector()
self.assertEqual(2, len(ExportedVM.objects.all()))
def tearDown(self):
super(ExportedVMCleanupTestCase, self).tearDown()
self.u1.delete()
self.exportedvm1.delete()
self.exportedvm2.delete()
class ExportedDiskCleanupTestCase(LoginMixin, MockCeleryMixin, TestCase):
fixtures = ['test-vm-fixture.json']
def setUp(self):
self.u1 = User.objects.create(username='user1')
self.u1.save()
self.vm = Instance.objects.get(pk=1)
self.vm.owner = self.u1
self.vm.save()
datastore = DataStore.objects.filter(pk=1).get()
self.exporteddisk1 = ExportedDisk(name="disk1", format="vmdk", filename="1", datastore=datastore, disk=Disk.objects.filter(pk=1).get())
self.exporteddisk1.save()
self.exporteddisk2 = ExportedDisk(name="disk2", format="vmdk", filename="2", datastore=datastore, disk=Disk.objects.filter(pk=1).get(), created=datetime.now() - relativedelta(hours=settings_global.EXPORT_KEEP_TIME_IN_HOURS - 1))
self.exporteddisk2.save()
self.exporteddisk3 = ExportedDisk(name="disk3", format="vmdk", filename="3", datastore=datastore, disk=Disk.objects.filter(pk=1).get(), created=datetime.now() - relativedelta(hours=settings_global.EXPORT_KEEP_TIME_IN_HOURS + 1))
self.exporteddisk3.save()
def test_exam_cleanup(self,):
self.assertEqual(3, len(ExportedDisk.objects.all()))
export_garbage_disk_collector()
self.assertEqual(2, len(ExportedDisk.objects.all()))
def tearDown(self):
super(ExportedDiskCleanupTestCase, self).tearDown()
self.u1.delete()
self.exporteddisk1.delete()
self.exporteddisk2.delete()
circle/vm/tests/test_operations.py
...@@ -24,6 +24,7 @@ from vm.operations import ( ...@@ -24,6 +24,7 @@ from vm.operations import (
DeployOperation, DestroyOperation, FlushOperation, MigrateOperation, DeployOperation, DestroyOperation, FlushOperation, MigrateOperation,
RebootOperation, ResetOperation, SaveAsTemplateOperation, RebootOperation, ResetOperation, SaveAsTemplateOperation,
ShutdownOperation, ShutOffOperation, SleepOperation, WakeUpOperation, ShutdownOperation, ShutOffOperation, SleepOperation, WakeUpOperation,
ExportVmOperation
) )
...@@ -104,3 +105,8 @@ class SleepOperationTestCase(TestCase): ...@@ -104,3 +105,8 @@ class SleepOperationTestCase(TestCase):
class WakeUpOperationTestCase(TestCase): class WakeUpOperationTestCase(TestCase):
def test_operation_registered(self): def test_operation_registered(self):
assert WakeUpOperation.id in getattr(Instance, op_reg_name) assert WakeUpOperation.id in getattr(Instance, op_reg_name)
class ExportVmOperationTestCase(TestCase):
def test_operation_registered(self):
assert ExportVmOperation.id in getattr(Instance, op_reg_name)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment