cloudstack/test/integration/smoke/test_webhook_lifecycle.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
""" BVT tests for webhooks lifecycle functionalities
"""
# Import Local Modules
from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.cloudstackAPI import (listEvents)
from marvin.lib.base import (Account,
                             Domain,
                             Webhook,
                             SSHKeyPair)
from marvin.lib.common import (get_domain,
                               get_zone)
from marvin.lib.utils import (random_gen)
from marvin.cloudstackException import CloudstackAPIException
from nose.plugins.attrib import attr
import logging
# Import System modules
import time
from datetime import datetime


_multiprocess_shared_ = True
HTTP_PAYLOAD_URL = "http://smee.io/C9LPa7Ei3iB6Qj2"
HTTPS_PAYLOAD_URL = "https://smee.io/C9LPa7Ei3iB6Qj2"

class TestWebhooks(cloudstackTestCase):

    @classmethod
    def setUpClass(cls):
        testClient = super(TestWebhooks, cls).getClsTestClient()
        cls.apiclient = testClient.getApiClient()
        cls.services = testClient.getParsedTestDataConfig()

        # Get Zone, Domain and templates
        cls.domain = get_domain(cls.apiclient)
        cls.zone = get_zone(cls.apiclient, testClient.getZoneForTests())

        cls._cleanup = []
        cls.logger = logging.getLogger('TestWebhooks')
        cls.logger.setLevel(logging.DEBUG)

    @classmethod
    def tearDownClass(cls):
        super(TestWebhooks, cls).tearDownClass()

    def setUp(self):
        self.cleanup = []
        self.domain1 = Domain.create(
            self.apiclient,
            self.services["domain"])
        self.cleanup.append(self.domain1)

    def tearDown(self):
        super(TestWebhooks, self).tearDown()

    def popItemFromCleanup(self, item_id):
        for idx, x in enumerate(self.cleanup):
            if x.id == item_id:
                self.cleanup.pop(idx)
                break

    def createDomainAccount(self, isDomainAdmin=False):
        self.account = Account.create(
            self.apiclient,
            self.services["account"],
            admin=isDomainAdmin,
            domainid=self.domain1.id)
        self.cleanup.append(self.account)
        self.userapiclient = self.testClient.getUserApiClient(
            UserName=self.account.name,
            DomainName=self.account.domain
        )

    def runWebhookLifecycleTest(self, apiclient, scope=None, domainid=None, account=None, normaluser=None, payloadurl=None, description=None, sslverification=None, secretkey=None, state=None, isdelete=True):
        name = "Test-" + random_gen()
        if payloadurl is None:
            payloadurl = HTTP_PAYLOAD_URL
        self.webhook = Webhook.create(
            apiclient,
            name=name,
            payloadurl=payloadurl,
            description=description,
            scope=scope,
            sslverification=sslverification,
            secretkey=secretkey,
            state=state,
            domainid=domainid,
            account=account
        )
        self.cleanup.append(self.webhook)
        self.assertNotEqual(
            self.webhook,
            None,
            "Check webhook created"
        )
        webhook_id = self.webhook.id
        self.logger.debug("Created webhook: %s" % str(self.webhook.__dict__))
        self.assertEqual(
            name,
            self.webhook.name,
            "Check webhook name"
        )
        self.assertEqual(
            payloadurl,
            self.webhook.payloadurl,
            "Check webhook payloadurl"
        )
        if state is None:
            state = 'Enabled'
        self.assertEqual(
            state,
            self.webhook.state,
            "Check webhook state"
        )
        if scope is None or normaluser is not None:
            scope = 'Local'
        self.assertEqual(
            scope,
            self.webhook.scope,
            "Check webhook scope"
        )
        if sslverification is None:
            sslverification = False
        self.assertEqual(
            sslverification,
            self.webhook.sslverification,
            "Check webhook sslverification"
        )
        if domainid is not None:
            if normaluser is not None:
                domainid = normaluser.domainid
            self.assertEqual(
                domainid,
                self.webhook.domainid,
                "Check webhook domainid"
            )
        if account is not None:
            self.assertEqual(
                account,
                self.webhook.account,
                "Check webhook account"
            )
        if description is not None:
            self.assertEqual(
                description,
                self.webhook.description,
                "Check webhook description"
            )
        if secretkey is not None:
            self.assertEqual(
                secretkey,
                self.webhook.secretkey,
                "Check webhook secretkey"
            )
        list_webhook = Webhook.list(
            apiclient,
            id=webhook_id
        )
        self.assertNotEqual(
            list_webhook,
            None,
            "Check webhook list"
        )
        self.assertEqual(
            len(list_webhook),
            1,
            "Check webhook list length"
        )
        self.assertEqual(
            list_webhook[0].id,
            webhook_id,
            "Check webhook list item"
        )
        if isdelete == False:
            return
        self.webhook.delete(apiclient)
        self.popItemFromCleanup(webhook_id)
        list_webhook = Webhook.list(
            apiclient,
            id=webhook_id
        )
        self.assertTrue(
            list_webhook is None or len(list_webhook) == 0,
            "Check webhook list after delete"
        )

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_01_create_webhook_admin_local(self):
        self.runWebhookLifecycleTest(self.apiclient)

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_02_create_webhook_admin_domain(self):
        self.runWebhookLifecycleTest(self.apiclient, 'Domain', self.domain1.id)

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_03_create_webhook_admin_global(self):
        self.runWebhookLifecycleTest(self.apiclient, 'Global')

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_04_create_webhook_domainadmin_local(self):
        self.createDomainAccount(True)
        self.runWebhookLifecycleTest(self.userapiclient)

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_05_create_webhook_domainadmin_subdomain(self):
        self.createDomainAccount(True)
        self.domain11 = Domain.create(
            self.apiclient,
            self.services["domain"],
            parentdomainid=self.domain1.id)
        self.cleanup.append(self.domain11)
        self.runWebhookLifecycleTest(self.userapiclient, 'Domain', self.domain11.id)

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_06_create_webhook_domainadmin_global_negative(self):
        self.createDomainAccount(True)
        try:
            self.runWebhookLifecycleTest(self.userapiclient, 'Global')
        except CloudstackAPIException as e:
            self.assertTrue(
                "errorText:Scope Global can not be specified for owner" in str(e),
                "Check Global scope error check"
            )

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_07_create_webhook_user_local(self):
        self.createDomainAccount()
        self.runWebhookLifecycleTest(self.userapiclient)

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_08_create_webhook_user_domain(self):
        """For normal user scope will always be Local irrespective of the passed value
        """
        self.createDomainAccount()
        self.runWebhookLifecycleTest(self.userapiclient, 'Domain', self.domain1.id, normaluser=self.account)

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_09_create_webhook_user_gloabl(self):
        """For normal user scope will always be Local irrespective of the passed value
        """
        self.createDomainAccount()
        self.runWebhookLifecycleTest(self.userapiclient, 'Global', normaluser=self.account)

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_10_create_webhook_admin_advanced(self):
        self.createDomainAccount()
        self.runWebhookLifecycleTest(
            self.apiclient,
            payloadurl=HTTPS_PAYLOAD_URL,
            scope="Local",
            description="Webhook",
            sslverification=True,
            secretkey="webhook",
            state="Disabled",
            domainid=self.domain1.id,
            account=self.account.name
        )

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_11_update_webhook(self):
        self.createDomainAccount()
        self.runWebhookLifecycleTest(self.userapiclient, isdelete=False)
        description = "Desc-" + random_gen()
        secretkey = random_gen()
        state = 'Disabled'
        updated_webhook = self.webhook.update(
            self.userapiclient,
            description=description,
            secretkey=secretkey,
            state=state
        )['webhook']
        self.assertNotEqual(
            updated_webhook,
            None,
            "Check updated webhook"
        )
        self.assertEqual(
            description,
            updated_webhook.description,
            "Check webhook description"
        )
        self.assertEqual(
            secretkey,
            updated_webhook.secretkey,
            "Check webhook secretkey"
        )
        self.assertEqual(
            state,
            updated_webhook.state,
            "Check webhook state"
        )

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_12_list_user_webhook_deliveries(self):
        self.createDomainAccount()
        self.runWebhookLifecycleTest(self.userapiclient, isdelete=False)
        now = datetime.now() # current date and time
        start_time = now.strftime("%Y-%m-%d %H:%M:%S")
        self.keypair = SSHKeyPair.register(
            self.userapiclient,
            name="Test-" + random_gen(),
            publickey="ssh-rsa: e6:9a:1e:b5:98:75:88:5d:56:bc:92:7b:43:48:05:b2"
        )
        self.logger.debug("Registered sshkeypair: %s" % str(self.keypair.__dict__))
        cmd = listEvents.listEventsCmd()
        cmd.startdate = start_time
        cmd.listall = True
        events = self.apiclient.listEvents(cmd)
        register_sshkeypair_event_count = 0
        if events is not None:
            for event in events:
                if event.type == "REGISTER.SSH.KEYPAIR":
                    register_sshkeypair_event_count = register_sshkeypair_event_count + 1
        time.sleep(5)
        list_deliveries = self.webhook.list_deliveries(
            self.userapiclient,
            page=1,
            pagesize=20
        )
        self.assertNotEqual(
            list_deliveries,
            None,
            "Check webhook deliveries list"
        )
        self.assertTrue(
            len(list_deliveries) > 0,
            "Check webhook deliveries list length"
        )
        register_sshkeypair_delivery_count = 0
        for delivery in list_deliveries:
            if delivery.eventtype == "REGISTER.SSH.KEYPAIR":
                register_sshkeypair_delivery_count = register_sshkeypair_delivery_count + 1
        self.assertEqual(
            register_sshkeypair_event_count,
            register_sshkeypair_delivery_count,
            "Check sshkeypair webhook deliveries count"
        )
        self.webhook.delete_deliveries(
            self.userapiclient
        )
        list_deliveries = self.webhook.list_deliveries(
            self.userapiclient
        )
        self.assertTrue(
            list_deliveries is None or len(list_deliveries) == 0,
            "Check webhook deliveries list after delete"
        )

    @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false")
    def test_13_webhook_execute_delivery(self):
        self.createDomainAccount()
        self.runWebhookLifecycleTest(self.userapiclient, isdelete=False)
        payload = "{ \"CloudStack\": \"Integration Test\" }"
        delivery = self.webhook.execute_delivery(
            self.userapiclient,
            payload=payload
        )
        self.assertNotEqual(
            delivery,
            None,
            "Check test webhook delivery"
        )
        self.assertEqual(
            self.webhook.id,
            delivery.webhookid,
            "Check test webhook delivery webhook"
        )
        self.assertEqual(
            payload,
            delivery.payload,
            "Check test webhook delivery payload"
        )
        self.assertEqual(
            self.webhook.id,
            delivery.webhookid,
            "Check test webhook delivery webhook"
        )