Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use test_server method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

test_server.py

Source:test_server.py

1# Licensed under the Apache License, Version 2.0 (the "License"); you may2# not use this file except in compliance with the License. You may obtain3# a copy of the License at4#5#      http://www.apache.org/licenses/LICENSE-2.06#7# Unless required by applicable law or agreed to in writing, software8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the10# License for the specific language governing permissions and limitations11# under the License.12import uuid13import sys14from openstack.tests.functional import base15from openstack import utils16utils.enable_logging(debug=True, stream=sys.stdout)17class TestServer(base.BaseFunctionalTest):18    NAME = uuid.uuid4().hex19    server = None20    network = None21    subnet = None22    cidr = '10.99.99.0/16'23    @classmethod24    def setUpClass(cls):25        super(TestServer, cls).setUpClass()26        # flavor = cls.conn.compute.find_flavor(base.FLAVOR_NAME,27        #                                       ignore_missing=False)28        # image = cls.conn.compute.find_image(base.IMAGE_NAME,29        #                                     ignore_missing=False)30        # cls.network, cls.subnet = test_network.create_network(cls.conn,31        #                                                       cls.NAME,32        #                                                       cls.cidr)33        # if not cls.network:34        #     # We can't call TestCase.fail from within the setUpClass35        #     # classmethod, but we need to raise some exception in order36        #     # to get this setup to fail and thusly fail the entire class.37        #     raise Exception("Unable to create network for TestServer")38        #39        # sot = cls.conn.compute.create_server(40        #     name=cls.NAME, flavor_id=flavor.id, image_id=image.id,41        #     networks=[{"uuid": cls.network.id}])42        # cls.conn.compute.wait_for_server(sot)43        # assert isinstance(sot, server.Server)44        # cls.assertIs(cls.NAME, sot.name)45        # cls.server = sot46        cls.server = cls.conn.compute.get_server("1658753f-8ab2-4650-be05-5ac8d353c56a")47    @classmethod48    def tearDownClass(cls):49        super(TestServer, cls).tearDownClass()50        # sot = cls.conn.compute.delete_server(cls.server.id)51        # cls.assertIs(None, sot)52        # # Need to wait for the stack to go away before network delete53        # cls.conn.compute.wait_for_delete(cls.server)54        # cls.linger_for_delete()55        # test_network.delete_network(cls.conn, cls.network, cls.subnet)56    # def test_find(self):57    #     sot = self.conn.compute.find_server(self.NAME)58    #     self.assertEqual(self.server.id, sot.id)59    #60    # def test_get(self):61    #     sot = self.conn.compute.get_server(self.server.id)62    #     self.assertEqual(self.NAME, sot.name)63    #     self.assertEqual(self.server.id, sot.id)64    #65    # def test_list(self):66    #     names = [o.name for o in self.conn.compute.servers()]67    #     self.assertIn(self.NAME, names)68    def test_server_metadata(self):69        test_server = self.conn.compute.get_server(self.server.id)70        # get metadata71        test_server = self.conn.compute.get_server_metadata(test_server)72        self.assertFalse(test_server.metadata)73        # set no metadata74        self.conn.compute.set_server_metadata(test_server)75        test_server = self.conn.compute.get_server_metadata(test_server)76        self.assertFalse(test_server.metadata)77        # set empty metadata78        self.conn.compute.set_server_metadata(test_server, k0='')79        server = self.conn.compute.get_server_metadata(test_server)80        self.assertTrue(server.metadata)81        # set metadata82        res = self.conn.compute.set_server_metadata(test_server, k1='v1')83        test_server = self.conn.compute.get_server_metadata(test_server)84        self.assertTrue(test_server.metadata)85        self.assertEqual(2, len(test_server.metadata))86        self.assertIn('k0', test_server.metadata)87        self.assertEqual('', test_server.metadata['k0'])88        self.assertIn('k1', test_server.metadata)89        self.assertEqual('v1', test_server.metadata['k1'])90        # set more metadata91        self.conn.compute.set_server_metadata(test_server, k2='v2')92        test_server = self.conn.compute.get_server_metadata(test_server)93        self.assertTrue(test_server.metadata)94        self.assertEqual(3, len(test_server.metadata))95        self.assertIn('k0', test_server.metadata)96        self.assertEqual('', test_server.metadata['k0'])97        self.assertIn('k1', test_server.metadata)98        self.assertEqual('v1', test_server.metadata['k1'])99        self.assertIn('k2', test_server.metadata)100        self.assertEqual('v2', test_server.metadata['k2'])101        # update metadata102        self.conn.compute.update_server_metadata(test_server, "k0", 'updatek0')103        test_server = self.conn.compute.get_server_metadata(test_server)104        self.assertTrue(test_server.metadata)105        self.assertEqual(3, len(test_server.metadata))106        self.assertEqual('updatek0', test_server.metadata['k0'])107        # delete metadata108        self.conn.compute.delete_server_metadata(109            test_server, test_server.metadata.keys())110        test_server = self.conn.compute.get_server_metadata(test_server)111        self.assertFalse(test_server.metadata)112    def test_get_server(self):113        test_server = self.conn.compute.get_server(self.server)...

test_main.py

Source:test_main.py

1from controllers import config, login, function_admission, function_oasis, servers, patient_dashboard, function_create_task, function_complete_task, function_mdo,  function_authorization2import random, time3from datetime import date4import admission, oasis, create_task, complete_task, create_mdo, complete_authorization, other_process, commMain, complete_woundprocess5import pymsgbox6import plottingTask7def test_main(servertest):8    9    test_server = servertest # Change the value to qa or live10           11    admission.admission_medicare(test_server) #PATIENT ADMISSION12    #oasis.oasispart() #COMPLETE OASIS SOC13    time.sleep(8)14    plottingTask.plot()  15    function_mdo.testinfo()16    config.driver.close()17    18def preadmitpatient_medicare(servertest):19    20    test_server = servertest 21    admission.preadmission_medicare(test_server) #PATIENT ADMISSION22    23    if servertest == "qa":24        config.driver.get("https://qado.medisource.com/patients/pre-admitted")25    elif servertest == "live": 26        config.driver.get("https://app.medisource.com/patients/pre-admitted")27    28    pymsgbox.alert('Test Success!', 'Success')29    print('Test success!')   30    config.driver.close()31def preadmitpatient_nonmedicare(servertest):32    33    test_server = servertest 34    admission.preadmission_nonmedicare(test_server) #PATIENT ADMISSION35    36    if servertest == "qa":37        config.driver.get("https://qado.medisource.com/patients/pre-admitted")38    elif servertest == "live": 39        config.driver.get("https://app.medisource.com/patients/pre-admitted")40        41    pymsgbox.alert('Test Success!', 'Success')42    print('Test success!')   43    config.driver.close()44 45def communiMain(servertest):46    47    test_server = servertest 48    49    50    if servertest == "qa":51        config.driver.get("https://qado.medisource.com/patients/pre-admitted")52    elif servertest == "live": 53        config.driver.get("https://app.medisource.com/patients/pre-admitted")54    55    admission.preadmission_medicare(test_server) #PATIENT ADMISSION56        57    commMain.communicationmain()58 59    60def wound(servertest, searchpatient):61    test_server = servertest 62    complete_woundprocess.complete_woundprocess(servertest, searchpatient)63    64    print('Test success!')   65    pymsgbox.alert('Test Success!', 'Success')66    config.driver.close()67def authorization(servertest, dayrange):68    test_server = servertest 69    complete_authorization.authorization(test_server, dayrange)70    71    print('Test success!')   72    pymsgbox.alert('Test Success!', 'Success')73    config.driver.close()    74    75def usermanagement(servertest):76    test_server = servertest 77    other_process.userprocess(test_server)78    79    print('Test success!')   80    pymsgbox.alert('Test Success!', 'Success')81    config.driver.close()        82    83    84def hospital(servertest):85    test_server = servertest 86    other_process.medres_hospitals(test_server)87    88    print('Test success!')   89    pymsgbox.alert('Test Success!', 'Success')90    config.driver.close()        91       92       93def physicians(servertest, addtype):94    test_server = servertest 95    add_type = addtype96    print('Physician')    97    other_process.medres_physicians(test_server, add_type)98    99    print('Test success!')   100    pymsgbox.alert('Test Success!', 'Success')101    config.driver.close()   102     103           104def referralsources(servertest):105    test_server = servertest 106    print('referral sources')    107     108            109def insurancecompanies(servertest):110    test_server = servertest 111    print('insurance companies')    112     113             114def healthcarevendors(servertest):115    test_server = servertest 116    print('healthcare vendors')    117     118                 119def emergencyservices(servertest):120    test_server = servertest 121    print('emergency services')    122     123  124    125   126  127  128  129  130  131  132  133  ...

httpd_unittest.py

Source:httpd_unittest.py

1#!/usr/bin/python2# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.3# Use of this source code is governed by a BSD-style license that can be4# found in the LICENSE file.5"""HTTPlistener unittest."""6import logging, os, sys, threading, urllib, unittest7from httpd import HTTPListener, SecureHTTPListener8GET_TEST_PATH = '/get_test'9def run_get_test(test_server, url):10    err = 111    get_done = test_server.add_wait_url(GET_TEST_PATH)12    get_resp = ''13    try:14        get_resp = urllib.urlopen(url).read()15    except IOError, e:16        pass17    if not (get_done.is_set() and get_resp):18        print 'FAILED'19    else:20        print 'PASSED'21        err = 022    return err23def test():24    test_server = HTTPListener(8000, docroot='/tmp')25    post_done = test_server.add_wait_url("/post_test",26                                         matchParams={'test': 'passed'})27    def _Spam():28        while not post_done.is_set():29            print 'TEST: server running'30            post_done.wait()31        return32    test_server.run()33    t = threading.Thread(target=_Spam).start()34    params = urllib.urlencode({'test': 'passed'})35    err = 136    # TODO(seano): This test doesn't seem to pass.37    post_resp = ''38    try:39        post_resp = urllib.urlopen('http://localhost:8000/post_test',40                                   params).read()41    except IOError, e:42        pass43    if not (post_done.is_set() and44            test_server.get_form_entries()['test'] != 'passed'):45        print 'FAILED'46    else:47        print 'PASSED'48        err = 049    err = run_get_test(test_server, 'http://localhost:8000' + GET_TEST_PATH)50    test_server.stop()51    if err != 0:52        return err53    creds_path = (os.path.dirname(os.path.realpath( __file__)) +54                  '/httpd_unittest_server')55    ssl_port=5000056    test_server = SecureHTTPListener(port=ssl_port,57                                     cert_path=(creds_path+'.pem'),58                                     key_path=(creds_path+'.key'))59    test_server.run()60    err = run_get_test(test_server,61                       'https://localhost:%d%s' % (ssl_port, GET_TEST_PATH))62    test_server.stop()63    return err64def run_server():65    """Example method showing how to start a HTTPListener."""66    test_server = HTTPListener(8000, docroot='/tmp')67    latch = test_server.add_wait_url('/quitquitquit')68    test_server.run()69    logging.info('server started')70    while not latch.is_set():71        try:72            latch.wait(1)73        except KeyboardInterrupt:74            sys.exit()75    test_server.stop()76    return77if __name__ == '__main__':78    if len(sys.argv) > 1:79        run_server()80    else:81        test()82if __name__ == '__main__':...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.