KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use nottest method in Nose

Best Python code snippet using nose

test_basic.py

Source:test_basic.py

...22def nottest_unless(criteria):23    def _nottest_unless(f):24        if criteria:25            return f26        return nottest(f)27    return _nottest_unless28def nottest_if(criteria):29    def _nottest_if(f):30        if not criteria:31            return f32        return nottest(f)33    return _nottest_if34class EnumValuesMixin(object):35    @classmethod36    def setup_class(cls):37        cls.enum_values = set([v for v in Location])38        cls.app = QCoreApplication(sys.argv)39        cls.app.setApplicationName('Yksom')40        cls.app.setOrganizationName('uranusjr')41        configure(application_name='Yksom', organization_name='uranusjr')42    @classmethod43    def teardown_class(cls):44        # Make sure all locations are tested.45        eq_(cls.enum_values, set())46    def setUp(self):...

CordTestServer.py

Source:CordTestServer.py

1# Copyright 2017-present Open Networking Foundation2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14#15# Copyright 2016-present Ciena Corporation16#17# Licensed under the Apache License, Version 2.0 (the "License");18# you may not use this file except in compliance with the License.19# You may obtain a copy of the License at20#21# http://www.apache.org/licenses/LICENSE-2.022#23# Unless required by applicable law or agreed to in writing, software24# distributed under the License is distributed on an "AS IS" BASIS,25# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.26# See the License for the specific language governing permissions and27# limitations under the License.28#29from CordContainer import Container, Onos, OnosStopWrapper, OnosCord, OnosCordStopWrapper, Quagga, QuaggaStopWrapper, Radius, reinitContainerClients30from nose.tools import nottest31from SimpleXMLRPCServer import SimpleXMLRPCServer32from resource import getrlimit, RLIMIT_NOFILE33import daemon34import xmlrpclib35import os36import signal37import json38import time39import threading40##Server to handle container restart/stop requests from test container.41##Used now to restart ONOS from vrouter test container42CORD_TEST_HOST = '172.17.0.1'43CORD_TEST_PORT = 2500044class CordTestServer(object):45    onos_cord = None46    def __restart_onos(self, node = None, config = None, timeout = 10):47        if self.onos_cord:48            onos_config = '{}/network-cfg.json'.format(self.onos_cord.onos_config_dir)49        else:50            onos_config = '{}/network-cfg.json'.format(Onos.host_config_dir)51        if config is None:52            try:53                os.unlink(onos_config)54            except:55                pass56        print('Restarting ONOS')57        if self.onos_cord:58            self.onos_cord.start(restart = True, network_cfg = config)59        else:60            Onos.restart_node(node = node, network_cfg = config, timeout = timeout)61        return 'DONE'62    def restart_onos(self, kwargs):63        return self.__restart_onos(**kwargs)64    def __shutdown_onos(self, node = None):65        if node is None:66            node = Onos.NAME67        OnosStopWrapper(node)68        return 'DONE'69    def shutdown_onos(self, kwargs):70        return self.__shutdown_onos(**kwargs)71    def __restart_cluster(self, config = None, timeout = 10, setup = False):72        Onos.restart_cluster(network_cfg = config, timeout = timeout, setup = setup)73        return 'DONE'74    def restart_cluster(self, kwargs):75        return self.__restart_cluster(**kwargs)76    def __add_cluster_onos(self, count = 1, config = None):77        Onos.add_cluster(count = count, network_cfg = config)78        return 'DONE'79    def add_cluster_onos(self, kwargs):80        return self.__add_cluster_onos(**kwargs)81    def __restart_quagga(self, config = None, boot_delay = 30 ):82        config_file = Quagga.quagga_config_file83        if config is not None:84            quagga_config = '{}/testrib_gen.conf'.format(Quagga.host_quagga_config)85            config_file = '{}/testrib_gen.conf'.format(Quagga.guest_quagga_config)86            with open(quagga_config, 'w+') as fd:87                fd.write(str(config))88        print('Restarting QUAGGA with config file %s, delay %d' %(config_file, boot_delay))89        Quagga(prefix = Container.IMAGE_PREFIX, restart = True, config_file = config_file, boot_delay = boot_delay)90        return 'DONE'91    def restart_quagga(self, kwargs):92        return self.__restart_quagga(**kwargs)93    def stop_quagga(self):94        quaggaStop = QuaggaStopWrapper()95        time.sleep(5)96        try:97            quagga_config_gen = '{}/testrib_gen.conf'.format(Quagga.host_quagga_config)98            os.unlink(quagga_config_gen)99        except: pass100        return 'DONE'101    def __run_shell_quagga(self, cmd = None):102        ret = 0103        if cmd is not None:104            exec_cmd = 'docker exec {} {}'.format(Quagga.NAME, cmd)105            ret = os.system(exec_cmd)106        return ret107    def __run_shell(self, cmd = None):108        ret = 0109        if cmd is not None:110            ret = os.system(cmd)111        return ret112    def run_shell_quagga(self, kwargs):113        return self.__run_shell_quagga(**kwargs)114    def run_shell(self, kwargs):115        return self.__run_shell(**kwargs)116    def restart_radius(self):117        print('Restarting RADIUS Server')118        Radius(prefix = Container.IMAGE_PREFIX, restart = True)119        return 'DONE'120    def shutdown(self):121        print('Shutting down cord test server')122        os.kill(0, signal.SIGKILL)123        return 'DONE'124def find_files_by_path(*paths):125    wanted = []126    for p in paths:127        try:128            fd = os.open(p, os.O_RDONLY)129            wanted.append(os.fstat(fd)[1:3])130        finally:131            os.close(fd)132    def fd_wanted(fd):133        try:134            return os.fstat(fd)[1:3] in wanted135        except OSError:136            return False137    max_fd = getrlimit(RLIMIT_NOFILE)[1]138    return [ fd for fd in xrange(max_fd) if fd_wanted(fd) ]139@nottest140def cord_test_server_start(daemonize = True,141                           cord_test_host = CORD_TEST_HOST,142                           cord_test_port = CORD_TEST_PORT,143                           onos_cord = None,144                           foreground=False):145    server = SimpleXMLRPCServer( (cord_test_host, cord_test_port) )146    server.register_instance(CordTestServer())147    CordTestServer.onos_cord = onos_cord148    if daemonize is True:149        ##before daemonizing, preserve urandom needed by paramiko150        preserve_list = find_files_by_path('/dev/urandom')151        preserve_list.append(server)152        d = daemon.DaemonContext(files_preserve = preserve_list,153                                 detach_process = True)154        with d:155            reinitContainerClients()156            server.serve_forever()157    else:158        if foreground:159            try:160                server.serve_forever()161            except KeyboardInterrupt:162                return server163        else:164            task = threading.Thread(target = server.serve_forever)165            ##terminate when main thread exits166            task.daemon = True167            task.start()168    return server169@nottest170def cord_test_server_stop(server):171    server.shutdown()172    server.server_close()173@nottest174def get_cord_test_loc():175    host = os.getenv('CORD_TEST_HOST', CORD_TEST_HOST)176    port = int(os.getenv('CORD_TEST_PORT', CORD_TEST_PORT))177    return host, port178def rpc_server_instance():179    '''Stateless'''180    host, port = get_cord_test_loc()181    rpc_server = 'http://{}:{}'.format(host, port)182    return xmlrpclib.Server(rpc_server, allow_none = True)183@nottest184def __cord_test_onos_restart(**kwargs):185    return rpc_server_instance().restart_onos(kwargs)186@nottest187def cord_test_onos_restart(node = None, config = None, timeout = 10):188    '''Send ONOS restart to server'''189    for i in range(3):190        try:191            data = __cord_test_onos_restart(node = node, config = config, timeout = timeout)192            if data == 'DONE':193                return True194        except:195            time.sleep(2)196    return False197@nottest198def __cord_test_onos_shutdown(**kwargs):199    return rpc_server_instance().shutdown_onos(kwargs)200@nottest201def cord_test_onos_shutdown(node = None):202    data = __cord_test_onos_shutdown(node = node)203    if data == 'DONE':204        return True205    return False206@nottest207def __cord_test_restart_cluster(**kwargs):208    return rpc_server_instance().restart_cluster(kwargs)209@nottest210def cord_test_restart_cluster(config = None, timeout = 10, setup = False):211    for i in range(3):212        try:213            data = __cord_test_restart_cluster(config = config, timeout = timeout, setup = setup)214            if data == 'DONE':215                return True216        except:217            time.sleep(2)218    return False219@nottest220def __cord_test_onos_add_cluster(**kwargs):221    return rpc_server_instance().add_cluster_onos(kwargs)222@nottest223def cord_test_onos_add_cluster(count = 1, config = None):224    data = __cord_test_onos_add_cluster(count = count, config = config)225    if data == 'DONE':226        return True227    return False228@nottest229def __cord_test_quagga_restart(**kwargs):230    return rpc_server_instance().restart_quagga(kwargs)231@nottest232def cord_test_quagga_restart(config = None, boot_delay = 30):233    '''Send QUAGGA restart to server'''234    data = __cord_test_quagga_restart(config = config, boot_delay = boot_delay)235    if data == 'DONE':236        return True237    return False238@nottest239def __cord_test_quagga_shell(**kwargs):240    return rpc_server_instance().run_shell_quagga(kwargs)241@nottest242def cord_test_quagga_shell(cmd = None):243    '''Send QUAGGA shell cmd to server'''244    return __cord_test_quagga_shell(cmd = cmd)245@nottest246def __cord_test_shell(**kwargs):247    return rpc_server_instance().run_shell(kwargs)248@nottest249def cord_test_shell(cmd = None):250    '''Send shell cmd to run remotely'''251    return __cord_test_shell(cmd = cmd)252@nottest253def cord_test_quagga_stop():254    data = rpc_server_instance().stop_quagga()255    if data == 'DONE':256        return True257    return False258@nottest259def cord_test_radius_restart():260    '''Send Radius server restart to server'''261    data = rpc_server_instance().restart_radius()262    if data == 'DONE':263        return True264    return False265@nottest266def cord_test_server_shutdown(host, port):267    '''Shutdown the cord test server'''268    rpc_server = 'http://{}:{}'.format(host, port)269    try:270        xmlrpclib.Server(rpc_server, allow_none = True).shutdown()271    except: pass...

test_standard_stats.py

Source:test_standard_stats.py

1"""2Currently just running and making sure there's no uncaught exceptions.3TODO: check that the correct output is generated too4In future we should put this in a web-accessible place5Usage:  pytest test_standard_stats.py6"""7import tempfile8import pytest9import toml10from pathlib import Path11from lama.tests import (wt_registration_dir, mut_registration_dir, target_dir,12                        stats_output_dir)13from lama.stats.standard_stats import lama_stats_new14root_config = dict(15    stats_types=[16        'organ_volumes',17         'intensity',18        'jacobians'19    ],20    normalise_organ_vol_to_mask=True,21    use_log_jacobians = True,22    reg_folder='similarity',23    jac_folder='similarity',24    mask='mask_tight.nrrd',25    label_info='label_info.csv',26    label_map='labels.nrrd',27    blur_fwhm=100,28    voxel_size=14.0,29    invert_stats=True,30    normalise='mask',31    use_staging=True,32    memmap=True33)34@pytest.fixture()35def get_config() -> Path:36    """37    Fixture to get stats config and to write it to temporary file38    Returns39    -------40    location of stats config temporary file41    """42    def make_config(config_updates={}):43        # Create a nested function to allow the fixture to take arguments44        c = root_config.copy()45        c.update(config_updates)46        with tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.toml') as fh:47            fh.write(toml.dumps(c))48        return c, fh.name49    return make_config50def test_all(get_config):51    """52    Run the stats module. The data required for this to work must be initially made53    by running the test:  tests/test_data_generation.py54    """55    config, config_file = get_config()56    lama_stats_new.run(config_file, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir)57@pytest.mark.skip58def test_no_mask(get_config):59    config, config_file = get_config({'mask': None})60    # Should get value errors if mask not specified61    # with pytest.raises(ValueError):62    #     lama_stats_new.run(config_file, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir)63# # @nottest64# # def test_no_use():65# #     """66# #     For the organ volume test, if the meta data dataframe has a column of 'no_analysis', do not include the label in the67# #     analysis68# #     """69# #     pass70#71# @nottest72# def test_mutant_id_subset():73#     """74#     Thest whether specifying mutant id subset works75#     """76#77#     config = stats_config_dir / 'new_stats_config_test_mutant_id_subset.toml'78#     lama_stats_new.run(config, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir)79#80# @nottest81# def test_specifiy_line():82#     """83#     Run the stats module. The data requirted for this to work must be initially made84#     by running the test:  tests/test_lama.py:test_lama_job_runner()85#     """86#     line = 'mutant_1'  # Only run this line87#88#     # Remove any previous output89#     if stats_output_dir.is_dir():90#         shutil.rmtree(stats_output_dir)91#92#     config = stats_config_dir / 'new_stats_config.toml'93#     lama_stats_new.run(config, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir, lines_to_process=[line])94#95#     output_lines = [x.name for x in stats_output_dir.iterdir() if x.is_dir()]96#     assert_equal([line],  output_lines)97#98#99# @nottest100# def test_erroneuos_configs():101#     error_config_dir = stats_config_dir / 'erroneous_configs'102#103#     for config in error_config_dir.iterdir():104#         lama_stats_new.run(config, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir)105#106# # @nottest107# # def test_organ_vols():108# #     config = join(CONFIG_DIR, 'organ_vols.yaml')109# #     run_lama_stats.run(config)110# #111# # @nottest112# # def test_glcm():113# #     config = join(CONFIG_DIR, 'test_glcm.yaml')114# #     run_lama_stats.run(config)115# #116# # @nottest117# # def test_from_arbitrary_directory():118# #     config = join(CONFIG_DIR, 'all_specimens.yaml')119# #     os.chdir(os.path.expanduser('~'))120# #     run_lama_stats.run(config)121# #122# # @nottest123# # def test_missing_config():124# #     config = join(CONFIG_DIR, 'all_specimens.flaml')125# #     assert_raises(Exception, run_lama_stats.run, config)126# #127# # @nottest128# # def test_misc():129# #     config = join(CONFIG_DIR, 'misc_test.yaml')130# #     run_lama_stats.run(config)131# #132# #133# # @nottest134# # def test_incorrect_staging_file():135# #     pass136#...

test_lama_stats.py

Source:test_lama_stats.py

1"""2Currently just running and making sure there's no uncaught exceptions.3TODO: check that the correct output is generated too4To run these tests, the test data needs to be fechted from bit/dev/lama_stats_test_data5In future we should put this in a web-accessible place6Usage:  pytest -v -m "not notest"7"""8import shutil9import tempfile10from nose.tools import nottest, assert_equal11import toml12# import paths from the __init__13from . import (stats_config_dir, wt_registration_dir, mut_registration_dir, target_dir,14               stats_output_dir)15from lama.stats.standard_stats import lama_stats_new16root_config = dict(17    stats_types=[18        'organ_volumes'19        # 'intensity',20        # 'jacobians'21    ],22    reg_folder='similarity',23    jac_folder='similarity',24    mask='mask_tight.nrrd',25    label_info='label_info.csv',26    label_map='labels.nrrd',27    blur_fwhm=100,28    voxel_size=14.0,29    invert_stats=True,30    normalise='mask',31    use_staging=False,32)33# @nottest34def test_all():35    """36    Run the stats module. The data requirted for this to work must be initially made37    by running the test:  tests/test_lama.py:test_lama_job_runner()38    """39    with tempfile.NamedTemporaryFile(mode='w+', delete=False) as fh:40        # print(fh.name)41        fh.write(toml.dumps(root_config))42    lama_stats_new.run(fh.name, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir)43# @nottest44# def test_no_use():45#     """46#     For the organ volume test, if the meta data dataframe has a column of 'no_analysis', do not include the label in the47#     analysis48#     """49#     pass50@nottest51def test_mutant_id_subset():52    """53    Thest whether specifying mutant id subset works54    """55    config = stats_config_dir / 'new_stats_config_test_mutant_id_subset.toml'56    lama_stats_new.run(config, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir)57@nottest58def test_specifiy_line():59    """60    Run the stats module. The data requirted for this to work must be initially made61    by running the test:  tests/test_lama.py:test_lama_job_runner()62    """63    line = 'mutant_1'  # Only run this line64    # Remove any previous output65    if stats_output_dir.is_dir():66        shutil.rmtree(stats_output_dir)67    config = stats_config_dir / 'new_stats_config.toml'68    lama_stats_new.run(config, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir, lines_to_process=[line])69    output_lines = [x.name for x in stats_output_dir.iterdir() if x.is_dir()]70    assert_equal([line],  output_lines)71@nottest72def test_erroneuos_configs():73    error_config_dir = stats_config_dir / 'erroneous_configs'74    for config in error_config_dir.iterdir():75        lama_stats_new.run(config, wt_registration_dir, mut_registration_dir, stats_output_dir, target_dir)76# @nottest77# def test_organ_vols():78#     config = join(CONFIG_DIR, 'organ_vols.yaml')79#     run_lama_stats.run(config)80#81# @nottest82# def test_glcm():83#     config = join(CONFIG_DIR, 'test_glcm.yaml')84#     run_lama_stats.run(config)85#86# @nottest87# def test_from_arbitrary_directory():88#     config = join(CONFIG_DIR, 'all_specimens.yaml')89#     os.chdir(os.path.expanduser('~'))90#     run_lama_stats.run(config)91#92# @nottest93# def test_missing_config():94#     config = join(CONFIG_DIR, 'all_specimens.flaml')95#     assert_raises(Exception, run_lama_stats.run, config)96#97# @nottest98# def test_misc():99#     config = join(CONFIG_DIR, 'misc_test.yaml')100#     run_lama_stats.run(config)101#102#103# @nottest104# def test_incorrect_staging_file():105#     pass106if __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.