KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use run_script method in localstack
Best Python code snippet using localstack_python
test_algorithms.py
Source:test_algorithms.py 
...27 sites = np.all(tables.sites.position == np.floor(tables.sites.position))28 return edges_left and edges_right and migrations_left and migrations_right and sites29@pytest.mark.skipif(IS_WINDOWS, reason="Bintrees isn't availble on windows")30class TestAlgorithms:31 def run_script(self, cmd):32 # print("RUN", cmd)33 with tempfile.TemporaryDirectory() as tmpdir:34 outfile = pathlib.Path(tmpdir) / "out.trees"35 # Avoid dairquiri mucking up the logging setup for unittest.36 with unittest.mock.patch("daiquiri.setup"):37 algorithms.main(cmd.split() + [str(outfile)])38 return tskit.load(outfile)39 def test_defaults(self):40 ts = self.run_script("10")41 assert ts.num_samples == 1042 assert ts.num_trees > 143 assert not has_discrete_genome(ts)44 assert ts.sequence_length == 10045 def test_discrete(self):46 ts = self.run_script("10 -d")47 assert ts.num_trees > 148 assert has_discrete_genome(ts)49 def test_dtwf(self):50 ts = self.run_script("10 --model=dtwf")51 assert ts.num_trees > 152 assert not has_discrete_genome(ts)53 assert ts.sequence_length == 10054 def test_dtwf_migration(self):55 ts = self.run_script("10 -r 0 --model=dtwf -p 2 -g 0.1")56 assert ts.num_trees == 157 assert ts.sequence_length == 10058 assert ts.num_populations == 259 def test_dtwf_discrete(self):60 ts = self.run_script("10 -d --model=dtwf")61 assert ts.num_trees > 162 assert has_discrete_genome(ts)63 def test_full_arg(self):64 ts = self.run_script("30 -L 200 --full-arg")65 assert ts.num_trees > 166 node_flags = ts.tables.nodes.flags67 assert np.sum(node_flags == msprime.NODE_IS_RE_EVENT) > 068 assert np.sum(node_flags == msprime.NODE_IS_CA_EVENT) > 069 def test_migration_full_arg(self):70 ts = self.run_script("10 -p 3 -g 0.1 --full-arg")71 assert ts.num_trees > 172 node_flags = ts.tables.nodes.flags73 assert np.sum(node_flags == msprime.NODE_IS_MIG_EVENT) > 074 def test_gc(self):75 ts = self.run_script("10 -c 0.4 2 -d")76 assert ts.num_trees > 177 assert has_discrete_genome(ts)78 assert ts.sequence_length == 10079 def test_recomb_map(self):80 # Need the final -- to break the end of the recomb_rates option81 ts = self.run_script(82 "10 -d --recomb-positions 0 10 100 --recomb-rates 0 0.1 0 --"83 )84 assert ts.num_trees > 185 assert has_discrete_genome(ts)86 assert ts.sequence_length == 10087 def test_census_event(self):88 ts = self.run_script("10 --census-time 0.01")89 assert ts.num_trees > 190 node_time = ts.tables.nodes.time91 assert np.sum(node_time == 0.01) > 092 def test_single_sweep(self):93 ts = self.run_script(94 "10 --model=single_sweep --trajectory 0.1 0.9 0.01 --time-slice=0.1"95 )96 assert ts.num_trees > 197 def test_single_sweep_shorter(self):98 # Try to catch some situtations not covered in other test.99 ts = self.run_script(100 "10 --model=single_sweep --trajectory 0.1 0.5 0.1 --time-slice=0.01"101 )102 assert ts.num_trees > 1103 def test_bottleneck(self):104 ts = self.run_script("10 -r 0 --bottleneck 0.1 0 2")105 assert ts.num_trees == 1106 tree = ts.first()107 found = False108 for node in tree.nodes():109 if len(tree.children(node)) > 2:110 found = True111 assert found112 def test_pop_size_change_event(self):113 ts = self.run_script("10 -r 0 --population-size-change 1 0 1")114 assert ts.num_trees == 1115 def test_pop_growth_rate_change_event(self):116 ts = self.run_script("10 -r 0 --population-growth-rate-change 1 0 0.1")117 assert ts.num_trees == 1118 def test_migration_rate_change_event(self):119 ts = self.run_script("10 -r 0 -p 2 --migration-matrix-element-change 1 0 1 0.1")120 assert ts.num_trees == 1121 def test_verbose(self):122 stdout, stderr = test_cli.capture_output(123 self.run_script, "10 -c 0.4 2 -v --end-time=0.1"124 )125 assert len(stdout) > 0126 assert len(stderr) == 0127 def test_from_ts(self):128 from_ts = msprime.sim_ancestry(129 10,130 sequence_length=100,131 discrete_genome=True,132 recombination_rate=2,133 random_seed=2,134 end_time=0.1,135 )136 tree = from_ts.first()137 assert tree.num_roots > 1138 with tempfile.TemporaryDirectory() as tmpdir:139 from_ts_path = pathlib.Path(tmpdir) / "from.trees"140 from_ts.dump(from_ts_path)141 ts = self.run_script(f"0 --from-ts {from_ts_path}")142 assert ts.num_trees > 1143 assert ts.sequence_length == 100144 @pytest.mark.parametrize(145 ["num_founders", "num_generations", "r"],146 [147 (4, 5, 0),148 (4, 5, 0.1),149 (4, 5, 1),150 (4, 10, 0.1),151 (10, 5, 0.1),152 (100, 2, 0.1),153 ],154 )155 def test_pedigree(self, num_founders, num_generations, r):156 tables = simulate_pedigree(157 num_founders=num_founders, num_generations=num_generations158 )159 with tempfile.TemporaryDirectory() as tmpdir:160 ts_path = pathlib.Path(tmpdir) / "pedigree.trees"161 tables.dump(ts_path)162 ts = self.run_script(f"0 --from-ts {ts_path} -r {r} --model=fixed_pedigree")163 for node in ts.nodes():164 assert node.individual != tskit.NULL165 assert ts.num_samples == 2 * num_founders166 # Make the set of ancestors for each individual167 ancestors = [set() for _ in ts.individuals()]168 for individual in ts.individuals():169 for parent_id in individual.parents:170 stack = [parent_id]171 while len(stack) > 0:172 parent_id = stack.pop()173 if parent_id != tskit.NULL:174 ancestors[individual.id].add(parent_id)175 for u in ts.individual(parent_id).parents:176 stack.append(u)177 # founder nodes have correct time178 founder_ids = {179 ind.id for ind in ts.individuals() if len(ancestors[ind.id]) == 0180 }181 for founder in founder_ids:182 for node_id in ts.individual(founder).nodes:183 node = ts.node(node_id)184 assert node.time == num_generations - 1185 for tree in ts.trees():186 for root in tree.roots:187 node = ts.node(root)188 # If this is a unary root it must be from a founder.189 if tree.num_children(root) == 1:190 assert node.individual in founder_ids191 for node_id in tree.nodes():192 node = ts.node(node_id)193 individual = ts.individual(node.individual)194 if tree.parent(node_id) != tskit.NULL:195 parent_node = ts.node(tree.parent(node_id))196 assert parent_node.individual in ancestors[individual.id] | {197 tskit.NULL198 }199 else:200 assert node.individual in founder_ids201 @pytest.mark.parametrize("r", [0, 0.1, 1])202 def test_pedigree_trio(self, r):203 input_tables = simulate_pedigree(204 num_founders=2, num_children_prob=(0, 1), num_generations=2205 )206 with tempfile.TemporaryDirectory() as tmpdir:207 ts_path = pathlib.Path(tmpdir) / "pedigree.trees"208 input_tables.dump(ts_path)209 ts = self.run_script(f"0 --from-ts {ts_path} -r {r} --model=fixed_pedigree")210 output_tables = ts.dump_tables()211 input_tables.individuals.assert_equals(output_tables.individuals)212 input_tables.nodes.assert_equals(output_tables.nodes[: len(input_tables.nodes)])213 assert len(output_tables.edges) >= 2214 @pytest.mark.parametrize("num_founders", [1, 2, 20])215 def test_one_gen_pedigree(self, num_founders):216 tables = simulate_pedigree(num_founders=num_founders, num_generations=1)217 with tempfile.TemporaryDirectory() as tmpdir:218 ts_path = pathlib.Path(tmpdir) / "pedigree.trees"219 tables.dump(ts_path)220 ts = self.run_script(f"0 --from-ts {ts_path} -r 1 --model=fixed_pedigree")...
test_minigraph_case.py
Source:test_minigraph_case.py
...8 self.test_dir = os.path.dirname(os.path.realpath(__file__))9 self.script_file = utils.PYTHON_INTERPRETTER + ' ' + os.path.join(self.test_dir, '..', 'sonic-cfggen')10 self.sample_graph = os.path.join(self.test_dir, 'simple-sample-graph-case.xml')11 self.port_config = os.path.join(self.test_dir, 't0-sample-port-config.ini')12 def run_script(self, argument, check_stderr=False):13 print('\n Running sonic-cfggen ' + argument)14 if check_stderr:15 output = subprocess.check_output(self.script_file + ' ' + argument, stderr=subprocess.STDOUT, shell=True)16 else:17 output = subprocess.check_output(self.script_file + ' ' + argument, shell=True)18 if utils.PY3x:19 output = output.decode()20 linecount = output.strip().count('\n')21 if linecount <= 0:22 print(' Output: ' + output.strip())23 else:24 print(' Output: ({0} lines, {1} bytes)'.format(linecount + 1, len(output)))25 return output26 def test_dummy_run(self):27 argument = ''28 output = self.run_script(argument)29 self.assertEqual(output, '')30 def test_minigraph_sku(self):31 argument = '-v "DEVICE_METADATA[\'localhost\'][\'hwsku\']" -m "' + self.sample_graph + '"'32 output = self.run_script(argument)33 self.assertEqual(output.strip(), 'Force10-S6000')34 def test_print_data(self):35 argument = '-m "' + self.sample_graph + '" --print-data'36 output = self.run_script(argument)37 self.assertTrue(len(output.strip()) > 0)38 def test_jinja_expression(self):39 argument = '-m "' + self.sample_graph + '" -v "DEVICE_METADATA[\'localhost\'][\'type\']"'40 output = self.run_script(argument)41 self.assertEqual(output.strip(), 'ToRRouter')42 def test_additional_json_data(self):43 argument = '-a \'{"key1":"value1"}\' -v key1'44 output = self.run_script(argument)45 self.assertEqual(output.strip(), 'value1')46 def test_read_yaml(self):47 argument = '-v yml_item -y ' + os.path.join(self.test_dir, 'test.yml')48 output = self.run_script(argument)49 self.assertEqual(output.strip(), '[\'value1\', \'value2\']')50 def test_render_template(self):51 argument = '-y ' + os.path.join(self.test_dir, 'test.yml') + ' -t ' + os.path.join(self.test_dir, 'test.j2')52 output = self.run_script(argument)53 self.assertEqual(output.strip(), 'value1\nvalue2')54# everflow portion is not used55# def test_minigraph_everflow(self):56# argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v MIRROR_SESSION'57# output = self.run_script(argument)58# self.assertEqual(output.strip(), "{'everflow0': {'src_ip': '10.1.0.32', 'dst_ip': '10.0.100.1'}}")59 def test_minigraph_interfaces(self):60 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v \'INTERFACE.keys()|list\''61 output = self.run_script(argument)62 self.assertEqual(output.strip(), "[('Ethernet0', '10.0.0.58/31'), 'Ethernet0', ('Ethernet0', 'FC00::75/126')]")63 def test_minigraph_vlans(self):64 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v VLAN'65 output = self.run_script(argument)66 self.assertEqual(67 utils.to_dict(output.strip()),68 utils.to_dict("{'Vlan1000': {'alias': 'ab1', 'dhcp_servers': ['192.0.0.1', '192.0.0.2'], 'vlanid': '1000', 'mac': '00:aa:bb:cc:dd:ee' }}")69 )70 def test_minigraph_vlan_members(self):71 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v VLAN_MEMBER'72 output = self.run_script(argument)73 self.assertEqual(output.strip(), "{('Vlan1000', 'Ethernet8'): {'tagging_mode': 'untagged'}}")74 def test_minigraph_vlan_interfaces(self):75 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VLAN_INTERFACE.keys()|list"'76 output = self.run_script(argument)77 self.assertEqual(output.strip(), "[('Vlan1000', '192.168.0.1/27'), 'Vlan1000']")78 def test_minigraph_portchannels(self):79 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v PORTCHANNEL'80 output = self.run_script(argument)81 self.assertEqual(82 utils.to_dict(output.strip()),83 utils.to_dict("{'PortChannel01': {'admin_status': 'up', 'min_links': '1', 'members': ['Ethernet4'], 'mtu': '9100'}}")84 )85 def test_minigraph_console_port(self):86 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v CONSOLE_PORT'87 output = self.run_script(argument)88 self.assertEqual(89 utils.to_dict(output.strip()),90 utils.to_dict("{'1': {'baud_rate': '9600', 'remote_device': 'managed_device', 'flow_control': 1}}"))91 def test_minigraph_deployment_id(self):92 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_METADATA[\'localhost\'][\'deployment_id\']"'93 output = self.run_script(argument)94 self.assertEqual(output.strip(), "1")95 def test_minigraph_neighbor_metadata(self):96 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "DEVICE_NEIGHBOR_METADATA"'97 output = self.run_script(argument)98 self.assertEqual(99 utils.to_dict(output.strip()),100 utils.to_dict("{'switch-01t1': {'lo_addr': '10.1.0.186/32', 'mgmt_addr': '10.7.0.196/26', 'hwsku': 'Force10-S6000', 'type': 'LeafRouter', 'deployment_id': '2'}}")101 )102# everflow portion is not used103# def test_metadata_everflow(self):104# argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "MIRROR_SESSION"'105# output = self.run_script(argument)106# self.assertEqual(output.strip(), "{'everflow0': {'src_ip': '10.1.0.32', 'dst_ip': '10.0.100.1'}}")107 def test_metadata_tacacs(self):108 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "TACPLUS_SERVER"'109 output = self.run_script(argument)110 self.assertEqual(output.strip(), "{'10.0.10.7': {'priority': '1', 'tcp_port': '49'}, '10.0.10.8': {'priority': '1', 'tcp_port': '49'}}")111 def test_minigraph_mgmt_port(self):112 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "MGMT_PORT"'113 output = self.run_script(argument)114 self.assertEqual(output.strip(), "{'eth0': {'alias': 'eth0', 'admin_status': 'up', 'speed': '1000'}}")115 def test_metadata_ntp(self):116 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "NTP_SERVER"'117 output = self.run_script(argument)118 self.assertEqual(output.strip(), "{'10.0.10.1': {}, '10.0.10.2': {}}")119 def test_minigraph_vnet(self):120 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VNET"'121 output = self.run_script(argument)122 self.assertEqual(output.strip(), "")123 def test_minigraph_vxlan(self):124 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "VXLAN_TUNNEL"'125 output = self.run_script(argument)126 self.assertEqual(output.strip(), "")127 def test_minigraph_bgp_mon(self):128 argument = '-m "' + self.sample_graph + '" -p "' + self.port_config + '" -v "BGP_MONITORS"'129 output = self.run_script(argument)130 self.assertEqual(output.strip(), "{}")131 def test_mux_cable_parsing(self):132 result = minigraph.parse_xml(self.sample_graph, port_config_file=self.port_config)133 134 expected_mux_cable_ports = ["Ethernet4", "Ethernet8"]135 port_table = result['PORT']136 for port_name, port in port_table.items():137 if port_name in expected_mux_cable_ports:138 self.assertTrue(port["mux_cable"])139 else:...
EventScript.py
Source:EventScript.py
...56 del self.OnIdleShutdownCallback57 del self.OnMachineStartupCallback58 del self.OnThermalShutdownCallback59 del self.OnMachineRestartCallback60 def run_script(self, *args):61 try:62 job = args[1]63 script = job.GetJobExtraInfoKeyValueWithDefault("EventScript", "")64 # Make arguments available to script with sys.argv65 sys.argv = args66 # Add script directory to sys.path67 sys.path.append(os.path.dirname(script))68 # Execute script69 execfile(script)70 except:71 print(traceback.format_exc())72 def OnJobSubmitted(self, job):73 self.run_script("OnJobSubmitted", job)74 def OnJobStarted(self, job):75 self.run_script("OnJobStarted", job)76 def OnJobFinished(self, job):77 self.run_script("OnJobFinished", job)78 def OnJobRequeued(self, job):79 self.run_script("OnJobRequeued", job)80 def OnJobFailed(self, job):81 self.run_script("OnJobFailed", job)82 def OnJobSuspended(self, job):83 self.run_script("OnJobSuspended", job)84 def OnJobResumed(self, job):85 self.run_script("OnJobResumed", job)86 def OnJobPended(self, job):87 self.run_script("OnJobPended", job)88 def OnJobReleased(self, job):89 self.run_script("OnJobReleased", job)90 def OnJobDeleted(self, job):91 self.run_script("OnJobDeleted", job)92 def OnJobError(self, job, task, report):93 self.run_script("OnJobError", job, task, report)94 def OnJobPurged(self, job):95 self.run_script("OnJobPurged", job)96 def OnHouseCleaning(self):97 self.run_script("OnHouseCleaning")98 def OnRepositoryRepair(self, job):99 self.run_script("OnRepositoryRepair", job)100 def OnSlaveStarted(self, job):101 self.run_script("OnSlaveStarted", job)102 def OnSlaveStopped(self, job):103 self.run_script("OnSlaveStopped", job)104 def OnSlaveIdle(self, job):105 self.run_script("OnSlaveIdle", job)106 def OnSlaveRendering(self, slaveName, job):107 self.run_script("OnSlaveRendering", job, slaveName)108 def OnSlaveStartingJob(self, slaveName, job):109 self.run_script("OnSlaveStartingJob", job, slaveName)110 def OnSlaveStalled(self, slaveName, job):111 self.run_script("OnSlaveStalled", job, slaveName)112 def OnIdleShutdown(self, job):113 self.run_script("OnIdleShutdown", job)114 def OnMachineStartup(self, job):115 self.run_script("OnMachineStartup", job)116 def OnThermalShutdown(self, job):117 self.run_script("OnThermalShutdown", job)118 def OnMachineRestart(self, job):...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
