1
# Copyright 2012 Canonical Ltd. This software is licensed under the
2
# GNU Affero General Public License version 3 (see the file LICENSE).
4
"""Tests for Celery tasks."""
6
from __future__ import (
17
from maasserver.enum import ARCHITECTURE
18
from maastesting.celery import CeleryFixture
19
from maastesting.factory import factory
20
from maastesting.matchers import ContainsAll
21
from maastesting.testcase import TestCase
22
from provisioningserver.enum import POWER_TYPE
23
from provisioningserver.power.poweraction import PowerActionFail
24
from provisioningserver.tasks import (
27
write_tftp_config_for_node,
29
from testresources import FixtureResource
30
from testtools.matchers import FileContains
32
# An arbitrary MAC address. Not using a properly random one here since
33
# we might accidentally affect real machines on the network.
34
arbitrary_mac = "AA:BB:CC:DD:EE:FF"
37
class TestPowerTasks(TestCase):
40
("celery", FixtureResource(CeleryFixture())),
43
def test_ether_wake_power_on_with_not_enough_template_args(self):
44
# In eager test mode the assertion is raised immediately rather
45
# than being stored in the AsyncResult, so we need to test for
46
# that instead of using result.get().
48
PowerActionFail, power_on.delay, POWER_TYPE.WAKE_ON_LAN)
50
def test_ether_wake_power_on(self):
51
result = power_on.delay(POWER_TYPE.WAKE_ON_LAN, mac=arbitrary_mac)
52
self.assertTrue(result.successful())
54
def test_ether_wake_does_not_support_power_off(self):
56
PowerActionFail, power_off.delay,
57
POWER_TYPE.WAKE_ON_LAN, mac=arbitrary_mac)
60
class TestTFTPTasks(TestCase):
63
("celery", FixtureResource(CeleryFixture())),
66
def test_write_tftp_config_for_node_writes_files(self):
67
arch = ARCHITECTURE.i386
68
mac = factory.getRandomMACAddress()
69
mac2 = factory.getRandomMACAddress()
70
tftproot = self.make_dir()
71
kernel = factory.getRandomString()
72
menutitle = factory.getRandomString()
73
append = factory.getRandomString()
75
result = write_tftp_config_for_node.delay(
76
arch, (mac, mac2), tftproot=tftproot, menutitle=menutitle,
77
kernelimage=kernel, append=append)
79
self.assertTrue(result.successful(), result)
80
expected_file1 = os.path.join(
81
tftproot, 'maas', arch, "generic", "pxelinux.cfg",
82
mac.replace(":", "-"))
83
expected_file2 = os.path.join(
84
tftproot, 'maas', arch, "generic", "pxelinux.cfg",
85
mac2.replace(":", "-"))
88
FileContains(matcher=ContainsAll((kernel, menutitle, append))))
91
FileContains(matcher=ContainsAll((kernel, menutitle, append))))