diff --git a/datacenterlight/tasks.py b/datacenterlight/tasks.py
index b897c54a..a09fb7c7 100644
--- a/datacenterlight/tasks.py
+++ b/datacenterlight/tasks.py
@@ -1,15 +1,17 @@
-from dynamicweb.celery import app
+from datetime import datetime
+
+from celery.exceptions import MaxRetriesExceededError
from celery.utils.log import get_task_logger
from django.conf import settings
+from django.core.mail import EmailMessage
+
+from dynamicweb.celery import app
+from hosting.models import HostingOrder, HostingBill
+from membership.models import StripeCustomer
from opennebula_api.models import OpenNebulaManager
from opennebula_api.serializers import VirtualMachineSerializer
-from hosting.models import HostingOrder, HostingBill
from utils.forms import UserBillingAddressForm
-from datetime import datetime
-from membership.models import StripeCustomer
-from django.core.mail import EmailMessage
from utils.models import BillingAddress
-from celery.exceptions import MaxRetriesExceededError
logger = get_task_logger(__name__)
@@ -41,13 +43,15 @@ def retry_task(task, exception=None):
@app.task(bind=True, max_retries=settings.CELERY_MAX_RETRIES)
-def create_vm_task(self, vm_template_id, user, specs, template, stripe_customer_id, billing_address_data,
+def create_vm_task(self, vm_template_id, user, specs, template,
+ stripe_customer_id, billing_address_data,
billing_address_id,
charge):
vm_id = None
try:
final_price = specs.get('price')
- billing_address = BillingAddress.objects.filter(id=billing_address_id).first()
+ billing_address = BillingAddress.objects.filter(
+ id=billing_address_id).first()
customer = StripeCustomer.objects.filter(id=stripe_customer_id).first()
# Create OpenNebulaManager
manager = OpenNebulaManager(email=settings.OPENNEBULA_USERNAME,
@@ -114,7 +118,8 @@ def create_vm_task(self, vm_template_id, user, specs, template, stripe_customer_
'subject': settings.DCL_TEXT + " Order from %s" % context['email'],
'from_email': settings.DCL_SUPPORT_FROM_ADDRESS,
'to': ['info@ungleich.ch'],
- 'body': "\n".join(["%s=%s" % (k, v) for (k, v) in context.items()]),
+ 'body': "\n".join(
+ ["%s=%s" % (k, v) for (k, v) in context.items()]),
'reply_to': [context['email']],
}
email = EmailMessage(**email_data)
@@ -124,11 +129,13 @@ def create_vm_task(self, vm_template_id, user, specs, template, stripe_customer_
try:
retry_task(self)
except MaxRetriesExceededError:
- msg_text = 'Finished {} retries for create_vm_task'.format(self.request.retries)
+ msg_text = 'Finished {} retries for create_vm_task'.format(
+ self.request.retries)
logger.error(msg_text)
# Try sending email and stop
email_data = {
- 'subject': '{} CELERY TASK ERROR: {}'.format(settings.DCL_TEXT, msg_text),
+ 'subject': '{} CELERY TASK ERROR: {}'.format(settings.DCL_TEXT,
+ msg_text),
'from_email': settings.DCL_SUPPORT_FROM_ADDRESS,
'to': ['info@ungleich.ch'],
'body': ',\n'.join(str(i) for i in self.request.args)
diff --git a/datacenterlight/tests.py b/datacenterlight/tests.py
index a79ca8be..0d95be79 100644
--- a/datacenterlight/tests.py
+++ b/datacenterlight/tests.py
@@ -1,3 +1,120 @@
# from django.test import TestCase
+from time import sleep
+
+import stripe
+from celery.result import AsyncResult
+from django.conf import settings
+from django.core.management import call_command
# Create your tests here.
+from django.test import TestCase, override_settings
+from model_mommy import mommy
+
+from datacenterlight.models import VMTemplate
+from datacenterlight.tasks import create_vm_task
+from membership.models import StripeCustomer
+from opennebula_api.serializers import VMTemplateSerializer
+from utils.models import BillingAddress
+from utils.stripe_utils import StripeUtils
+
+
+class CeleryTaskTestCase(TestCase):
+ @override_settings(
+ task_eager_propagates=True,
+ task_always_eager=True,
+ )
+ def setUp(self):
+ self.customer_password = 'test_password'
+ self.customer_email = 'celery-createvm-task-test@ungleich.ch'
+ self.customer_name = "Monty Python"
+ self.user = {
+ 'email': self.customer_email,
+ 'name': self.customer_name
+ }
+ self.customer = mommy.make('membership.CustomUser')
+ self.customer.set_password(self.customer_password)
+ self.customer.email = self.customer_email
+ self.customer.save()
+ self.stripe_utils = StripeUtils()
+ stripe.api_key = settings.STRIPE_API_PRIVATE_KEY_TEST
+ self.token = stripe.Token.create(
+ card={
+ "number": '4111111111111111',
+ "exp_month": 12,
+ "exp_year": 2022,
+ "cvc": '123'
+ },
+ )
+ # Run fetchvmtemplates so that we have the VM templates from
+ # OpenNebula
+ call_command('fetchvmtemplates')
+
+ def test_create_vm_task(self):
+ """Tests the create vm task."""
+
+ # We create a VM from the first template available to DCL
+ vm_template = VMTemplate.objects.all().first()
+ template_data = VMTemplateSerializer(vm_template).data
+
+ # The specs of VM that we want to create
+ specs = {
+ 'cpu': 1,
+ 'memory': 2,
+ 'disk_size': 10,
+ 'price': 15,
+ }
+
+ stripe_customer = StripeCustomer.get_or_create(
+ email=self.customer_email,
+ token=self.token)
+ billing_address = BillingAddress(
+ cardholder_name=self.customer_name,
+ postal_code='1232',
+ country='CH',
+ street_address='Monty\'s Street',
+ city='Hollywood')
+ billing_address.save()
+ billing_address_data = {'cardholder_name': self.customer_name,
+ 'postal_code': '1231',
+ 'country': 'CH',
+ 'token': self.token,
+ 'street_address': 'Monty\'s Street',
+ 'city': 'Hollywood'}
+
+ billing_address_id = billing_address.id
+ vm_template_id = template_data.get('id', 1)
+ final_price = specs.get('price')
+
+ # Make stripe charge to a customer
+ stripe_utils = StripeUtils()
+ charge_response = stripe_utils.make_charge(
+ amount=final_price,
+ customer=stripe_customer.stripe_id)
+
+ # Check if the payment was approved
+ if not charge_response.get(
+ 'response_object') and not charge_response.get('paid'):
+ msg = charge_response.get('error')
+ raise Exception("make_charge failed: {}".format(msg))
+
+ charge = charge_response.get('response_object')
+ async_task = create_vm_task.delay(vm_template_id, self.user,
+ specs,
+ template_data,
+ stripe_customer.id,
+ billing_address_data,
+ billing_address_id,
+ charge)
+ new_vm_id = 0
+ res = None
+ for i in range(0, 10):
+ sleep(5)
+ res = AsyncResult(async_task.task_id)
+ if res.result is not None and res.result > 0:
+ new_vm_id = res.result
+ break
+
+ # We expect a VM to be created within 50 seconds
+ self.assertGreater(new_vm_id, 0,
+ "VM could not be created. res._get_task_meta() = {}"
+ .format(res._get_task_meta()))
diff --git a/dynamicweb/settings/base.py b/dynamicweb/settings/base.py
index a724b38e..383ff9ff 100644
--- a/dynamicweb/settings/base.py
+++ b/dynamicweb/settings/base.py
@@ -2,15 +2,14 @@
Copyright 2015 ungleich.
"""
-# -*- coding: utf-8 -*-
-# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
-import os
-
-from django.utils.translation import ugettext_lazy as _
+import logging
# dotenv
import dotenv
-import logging
+# -*- coding: utf-8 -*-
+# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
+import os
+from django.utils.translation import ugettext_lazy as _
logger = logging.getLogger(__name__)
@@ -493,6 +492,7 @@ REGISTRATION_MESSAGE = {'subject': "Validation mail",
}
STRIPE_API_PRIVATE_KEY = env('STRIPE_API_PRIVATE_KEY')
STRIPE_API_PUBLIC_KEY = env('STRIPE_API_PUBLIC_KEY')
+STRIPE_API_PRIVATE_KEY_TEST = env('STRIPE_API_PRIVATE_KEY_TEST')
ANONYMOUS_USER_NAME = 'anonymous@ungleich.ch'
GUARDIAN_GET_INIT_ANONYMOUS_USER = 'membership.models.get_anonymous_user_instance'