launchpad-reviewers team mailing list archive

[Julian Edwards <julian.edwards@xxxxxxxxxxxxx>]

Julian Edwards has proposed merging lp:~julian-edwards/maas/celery-power-job into lp:maas.

Requested reviews:
  Launchpad code reviewers (launchpad-reviewers)

For more details, see:
https://code.launchpad.net/~julian-edwards/maas/celery-power-job/+merge/106944
-- 
https://code.launchpad.net/~julian-edwards/maas/celery-power-job/+merge/106944
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~julian-edwards/maas/celery-power-job into lp:maas.

=== modified file 'src/provisioningserver/tasks.py'
--- src/provisioningserver/tasks.py	2012-05-18 06:47:12 +0000
+++ src/provisioningserver/tasks.py	2012-05-23 06:59:27 +0000
@@ -11,3 +11,27 @@
 
 __metaclass__ = type
 __all__ = []
+
+
+from celery.decorators import task
+
+from provisioningserver.enum import POWER_TYPE
+from provisioningserver.power.poweraction import (
+    PowerAction,
+    PowerActionFail,
+    )
+
+
+@task
+def power_on_ether_wake(**kwargs):
+    try:
+        pa = PowerAction(POWER_TYPE.WAKE_ON_LAN)
+        pa.execute(**kwargs)
+    except PowerActionFail:
+        # TODO: signal to webapp that it failed
+
+        # Re-raise, so the job is marked as failed.  Only currently
+        # useful for tests.
+        raise
+
+    # TODO: signal to webapp that it worked.

=== added file 'src/provisioningserver/tests/test_tasks.py'
--- src/provisioningserver/tests/test_tasks.py	1970-01-01 00:00:00 +0000
+++ src/provisioningserver/tests/test_tasks.py	2012-05-23 06:59:27 +0000
@@ -0,0 +1,44 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Tests for Celery tasks."""
+
+from __future__ import (
+    absolute_import,
+    print_function,
+    unicode_literals,
+    )
+
+__metaclass__ = type
+__all__ = []
+
+from testresources import FixtureResource
+
+from maastesting.celery import CeleryFixture
+from maastesting.testcase import TestCase
+from provisioningserver.power.poweraction import PowerActionFail
+from provisioningserver.tasks import power_on_ether_wake
+
+
+class TaskTestCase(TestCase):
+
+    def assertSuccess(self, task_result):
+        self.assertEqual("SUCCESS", task_result.status)
+
+
+class TestPowerTasks(TaskTestCase):
+
+    resources = (
+        ("celery", FixtureResource(CeleryFixture())),
+        )
+
+    def test_ether_wake_power_on_with_not_enough_template_args(self):
+        # In eager test mode the assertion is raised immediately rather
+        # than being stored in the AsyncResult, so we need to test for
+        # that instead of using result.get().
+        self.assertRaises(PowerActionFail, power_on_ether_wake.delay)
+
+    def test_ether_wake_power_on(self):
+        mac = "AA:BB:CC:DD:EE:FF"
+        result = power_on_ether_wake.delay(mac=mac)
+        self.assertSuccess(result)