# 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 delivery with a basic server """ # Import Local Modules from marvin.cloudstackTestCase import cloudstackTestCase 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 from http.server import BaseHTTPRequestHandler, HTTPServer import logging # Import System modules import time import json import socket import _thread _multiprocess_shared_ = True deliveries_received = [] class WebhookReceiver(BaseHTTPRequestHandler): """ WebhookReceiver class to receive webhook events """ def _set_response(self): self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() def do_POST(self): content_length = int(self.headers['Content-Length']) post_data = self.rfile.read(content_length) post_data = post_data.decode('utf-8') event_id = self.headers.get('X-CS-Event-ID') print("POST request,\nPath: %s\nHeaders:\n%s\n\nBody:\n%s\n" % (str(self.path), str(self.headers), post_data)) self._set_response() global deliveries_received if deliveries_received is None: deliveries_received = [] deliveries_received.append({'event': event_id, 'payload': post_data}) if event_id != None: self.wfile.write("Event with ID: {} successfully processed!".format(str(event_id)).encode('utf-8')) else: self.wfile.write("POST request for {}".format(self.path).encode('utf-8')) class TestWebhookDelivery(cloudstackTestCase): @classmethod def setUpClass(cls): testClient = super(TestWebhookDelivery, cls).getClsTestClient() cls.apiclient = testClient.getApiClient() cls.services = testClient.getParsedTestDataConfig() cls.mgtSvrDetails = cls.config.__dict__["mgtSvr"][0].__dict__ # Get Zone, Domain and templates cls.domain = get_domain(cls.apiclient) cls.zone = get_zone(cls.apiclient, testClient.getZoneForTests()) cls.logger = logging.getLogger('TestWebhookDelivery') cls.logger.setLevel(logging.DEBUG) s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect((cls.mgtSvrDetails["mgtSvrIp"], cls.mgtSvrDetails["port"])) cls.server_ip = s.getsockname()[0] s.close() if cls.server_ip == "127.0.0.1": s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) cls.server_ip = s.getsockname()[0] s.close() # use random port for webhookreceiver server s = socket.socket() s.bind(('', 0)) cls.server_port = s.getsockname()[1] s.close() cls.webhook_receiver_url = "http://" + cls.server_ip + ":" + str(cls.server_port) cls.logger.debug("Running Webhook receiver @ %s" % cls.webhook_receiver_url) def startMgmtServer(tname, server): cls.logger.debug("Starting WebhookReceiver") try: server.serve_forever() except Exception: pass cls.server = HTTPServer(('0.0.0.0', cls.server_port), WebhookReceiver) _thread.start_new_thread(startMgmtServer, ("webhook-receiver", cls.server,)) cls._cleanup = [] @classmethod def tearDownClass(cls): if cls.server: cls.server.socket.close() global deliveries_received deliveries_received = [] super(TestWebhookDelivery, 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(TestWebhookDelivery, 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 createWebhook(self, apiclient, scope=None, domainid=None, account=None, payloadurl=None, description=None, sslverification=None, secretkey=None, state=None): name = "Test-" + random_gen() if payloadurl is None: payloadurl = self.webhook_receiver_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) @attr(tags=["devcloud", "advanced", "advancedns", "smoke", "basic", "sg"], required_hardware="false") def test_01_webhook_deliveries(self): global deliveries_received self.createDomainAccount() self.createWebhook(self.userapiclient) 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__)) time.sleep(2) 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" ) for delivery in list_deliveries: self.assertEqual( delivery.success, True, "Check webhook delivery success" ) self.assertEqual( delivery.response, ("Event with ID: %s successfully processed!" % delivery.eventid), "Check webhook delivery response" ) delivery_matched = False for received in deliveries_received: if received['event'] == delivery.eventid: self.assertEqual( delivery.payload, received['payload'], "Check webhook delivery payload" ) delivery_matched = True self.assertTrue( delivery_matched, "Delivery for %s did not match with server" % delivery.id )