KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use start_container method in localstack
Best Python code snippet using localstack_python
service_test.py
Source:service_test.py 
...24 Service(name='foo', ports=['8000'])25 def test_containers(self):26 foo = self.create_service('foo')27 bar = self.create_service('bar')28 foo.start_container()29 self.assertEqual(len(foo.containers()), 1)30 self.assertEqual(foo.containers()[0].name, 'figtest_foo_1')31 self.assertEqual(len(bar.containers()), 0)32 bar.start_container()33 bar.start_container()34 self.assertEqual(len(foo.containers()), 1)35 self.assertEqual(len(bar.containers()), 2)36 names = [c.name for c in bar.containers()]37 self.assertIn('figtest_bar_1', names)38 self.assertIn('figtest_bar_2', names)39 def test_containers_one_off(self):40 db = self.create_service('db')41 container = db.create_container(one_off=True)42 self.assertEqual(db.containers(stopped=True), [])43 self.assertEqual(db.containers(one_off=True, stopped=True), [container])44 def test_project_is_added_to_container_name(self):45 service = self.create_service('web')46 service.start_container()47 self.assertEqual(service.containers()[0].name, 'figtest_web_1')48 def test_start_stop(self):49 service = self.create_service('scalingtest')50 self.assertEqual(len(service.containers(stopped=True)), 0)51 service.create_container()52 self.assertEqual(len(service.containers()), 0)53 self.assertEqual(len(service.containers(stopped=True)), 1)54 service.start()55 self.assertEqual(len(service.containers()), 1)56 self.assertEqual(len(service.containers(stopped=True)), 1)57 service.stop(timeout=1)58 self.assertEqual(len(service.containers()), 0)59 self.assertEqual(len(service.containers(stopped=True)), 1)60 service.stop(timeout=1)61 self.assertEqual(len(service.containers()), 0)62 self.assertEqual(len(service.containers(stopped=True)), 1)63 def test_kill_remove(self):64 service = self.create_service('scalingtest')65 service.start_container()66 self.assertEqual(len(service.containers()), 1)67 service.remove_stopped()68 self.assertEqual(len(service.containers()), 1)69 service.kill()70 self.assertEqual(len(service.containers()), 0)71 self.assertEqual(len(service.containers(stopped=True)), 1)72 service.remove_stopped()73 self.assertEqual(len(service.containers(stopped=True)), 0)74 def test_create_container_with_one_off(self):75 db = self.create_service('db')76 container = db.create_container(one_off=True)77 self.assertEqual(container.name, 'figtest_db_run_1')78 def test_create_container_with_one_off_when_existing_container_is_running(self):79 db = self.create_service('db')80 db.start()81 container = db.create_container(one_off=True)82 self.assertEqual(container.name, 'figtest_db_run_1')83 def test_create_container_with_unspecified_volume(self):84 service = self.create_service('db', volumes=['/var/db'])85 container = service.create_container()86 service.start_container(container)87 self.assertIn('/var/db', container.inspect()['Volumes'])88 def test_recreate_containers(self):89 service = self.create_service(90 'db',91 environment={'FOO': '1'},92 volumes=['/var/db'],93 entrypoint=['ps'],94 command=['ax']95 )96 old_container = service.create_container()97 self.assertEqual(old_container.dictionary['Config']['Entrypoint'], ['ps'])98 self.assertEqual(old_container.dictionary['Config']['Cmd'], ['ax'])99 self.assertIn('FOO=1', old_container.dictionary['Config']['Env'])100 self.assertEqual(old_container.name, 'figtest_db_1')101 service.start_container(old_container)102 volume_path = old_container.inspect()['Volumes']['/var/db']103 num_containers_before = len(self.client.containers(all=True))104 service.options['environment']['FOO'] = '2'105 (intermediate, new) = service.recreate_containers()106 self.assertEqual(len(intermediate), 1)107 self.assertEqual(len(new), 1)108 new_container = new[0]109 intermediate_container = intermediate[0]110 self.assertEqual(intermediate_container.dictionary['Config']['Entrypoint'], ['echo'])111 self.assertEqual(new_container.dictionary['Config']['Entrypoint'], ['ps'])112 self.assertEqual(new_container.dictionary['Config']['Cmd'], ['ax'])113 self.assertIn('FOO=2', new_container.dictionary['Config']['Env'])114 self.assertEqual(new_container.name, 'figtest_db_1')115 service.start_container(new_container)116 self.assertEqual(new_container.inspect()['Volumes']['/var/db'], volume_path)117 self.assertEqual(len(self.client.containers(all=True)), num_containers_before + 1)118 self.assertNotEqual(old_container.id, new_container.id)119 def test_start_container_passes_through_options(self):120 db = self.create_service('db')121 db.start_container(environment={'FOO': 'BAR'})122 self.assertEqual(db.containers()[0].environment['FOO'], 'BAR')123 def test_start_container_inherits_options_from_constructor(self):124 db = self.create_service('db', environment={'FOO': 'BAR'})125 db.start_container()126 self.assertEqual(db.containers()[0].environment['FOO'], 'BAR')127 def test_start_container_creates_links(self):128 db = self.create_service('db')129 web = self.create_service('web', links=[(db, None)])130 db.start_container()131 web.start_container()132 self.assertIn('figtest_db_1', web.containers()[0].links())133 self.assertIn('db_1', web.containers()[0].links())134 def test_start_container_creates_links_with_names(self):135 db = self.create_service('db')136 web = self.create_service('web', links=[(db, 'custom_link_name')])137 db.start_container()138 web.start_container()139 self.assertIn('custom_link_name', web.containers()[0].links())140 def test_start_normal_container_does_not_create_links_to_its_own_service(self):141 db = self.create_service('db')142 c1 = db.start_container()143 c2 = db.start_container()144 self.assertNotIn(c1.name, c2.links())145 def test_start_one_off_container_creates_links_to_its_own_service(self):146 db = self.create_service('db')147 c1 = db.start_container()148 c2 = db.start_container(one_off=True)149 self.assertIn(c1.name, c2.links())150 def test_start_container_builds_images(self):151 service = Service(152 name='test',153 client=self.client,154 build='tests/fixtures/simple-dockerfile',155 project='figtest',156 )157 container = service.start_container()158 container.wait()159 self.assertIn('success', container.logs())160 self.assertEqual(len(self.client.images(name='figtest_test')), 1)161 def test_start_container_uses_tagged_image_if_it_exists(self):162 self.client.build('tests/fixtures/simple-dockerfile', tag='figtest_test')163 service = Service(164 name='test',165 client=self.client,166 build='this/does/not/exist/and/will/throw/error',167 project='figtest',168 )169 container = service.start_container()170 container.wait()171 self.assertIn('success', container.logs())172 def test_start_container_creates_ports(self):173 service = self.create_service('web', ports=[8000])174 container = service.start_container().inspect()175 self.assertEqual(list(container['NetworkSettings']['Ports'].keys()), ['8000/tcp'])176 self.assertNotEqual(container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'], '8000')177 def test_start_container_stays_unpriviliged(self):178 service = self.create_service('web')179 container = service.start_container().inspect()180 self.assertEqual(container['HostConfig']['Privileged'], False)181 def test_start_container_becomes_priviliged(self):182 service = self.create_service('web', privileged = True)183 container = service.start_container().inspect()184 self.assertEqual(container['HostConfig']['Privileged'], True)185 def test_expose_does_not_publish_ports(self):186 service = self.create_service('web', expose=[8000])187 container = service.start_container().inspect()188 self.assertEqual(container['NetworkSettings']['Ports'], {'8000/tcp': None})189 def test_start_container_creates_port_with_explicit_protocol(self):190 service = self.create_service('web', ports=['8000/udp'])191 container = service.start_container().inspect()192 self.assertEqual(list(container['NetworkSettings']['Ports'].keys()), ['8000/udp'])193 def test_start_container_creates_fixed_external_ports(self):194 service = self.create_service('web', ports=['8000:8000'])195 container = service.start_container().inspect()196 self.assertIn('8000/tcp', container['NetworkSettings']['Ports'])197 self.assertEqual(container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'], '8000')198 def test_start_container_creates_fixed_external_ports_when_it_is_different_to_internal_port(self):199 service = self.create_service('web', ports=['8001:8000'])200 container = service.start_container().inspect()201 self.assertIn('8000/tcp', container['NetworkSettings']['Ports'])202 self.assertEqual(container['NetworkSettings']['Ports']['8000/tcp'][0]['HostPort'], '8001')203 def test_scale(self):204 service = self.create_service('web')205 service.scale(1)206 self.assertEqual(len(service.containers()), 1)207 service.scale(3)208 self.assertEqual(len(service.containers()), 3)209 service.scale(1)210 self.assertEqual(len(service.containers()), 1)211 service.scale(0)212 self.assertEqual(len(service.containers()), 0)213 def test_scale_on_service_that_cannot_be_scaled(self):214 service = self.create_service('web', ports=['8000:8000'])...
test_run_data_managers.py
Source:test_run_data_managers.py
1#!/usr/bin/env python2# Stop pylint complaining from things that are necessary3# for pytest to work.4# pylint: disable=no-self-use,unused-import5import sys6import time7import pytest8import yaml9from docker_for_galaxy import GALAXY_ADMIN_KEY, GALAXY_ADMIN_PASSWORD, GALAXY_ADMIN_USER, start_container # noqa: F401 prevent unused error10from ephemeris import run_data_managers11from ephemeris.run_data_managers import DataManagers12from ephemeris.shed_tools import InstallRepositoryManager13from ephemeris.sleep import galaxy_wait14AUTH_BY = "key"15class TestRunDataManagers(object):16 """This class tests run-data-managers"""17 def test_install_data_managers(self, start_container): # noqa: F811 Prevent start_container unused warning.18 """Install the data_managers on galaxy"""19 container = start_container20 data_managers = [21 dict(name="data_manager_fetch_genome_dbkeys_all_fasta",22 owner="devteam"),23 dict(name="data_manager_sam_fasta_index_builder",24 owner="devteam"),25 dict(name="data_manager_bwa_mem_index_builder",26 owner="devteam")27 ]28 irm = InstallRepositoryManager(container.gi)29 irm.install_repositories(data_managers)30 # Galaxy is restarted because otherwise data tables are not watched.31 container.container.exec_run("supervisorctl restart galaxy:")32 time.sleep(10) # give time for the services to go down33 galaxy_wait(container.url)34 def test_run_data_managers(self, start_container): # noqa: F811 Prevent start_container unused warning.35 """Tests an installation using the command line"""36 container = start_container37 argv = ["run-data-managers"]38 if AUTH_BY == "user":39 argv.extend([40 "--user", GALAXY_ADMIN_USER,41 "-p", GALAXY_ADMIN_PASSWORD,42 ])43 else:44 argv.extend(["-a", GALAXY_ADMIN_KEY])45 argv.extend([46 "-g", container.url,47 "--config", "tests/run_data_managers.yaml.test"48 ])49 sys.argv = argv50 run_data_managers.main()51 def test_run_data_managers_installation_skipped(self, start_container): # noqa: F811 Prevent start_container unused warning.52 container = start_container53 with open("tests/run_data_managers.yaml.test") as config_file:54 configuration = yaml.safe_load(config_file)55 dm = DataManagers(container.gi, configuration)56 install_results = dm.run()57 assert (len(install_results.successful_jobs) == 0)58 assert (len(install_results.skipped_jobs) == 9)59 assert (len(install_results.failed_jobs) == 0)60 def test_run_data_managers_installation_fail(self, start_container, caplog): # noqa: F811 Prevent start_container unused warning.61 container = start_container62 configuration = dict(63 data_managers=[64 dict(65 id="data_manager_fetch_genome_all_fasta_dbkey",66 params=[67 {'dbkey_source|dbkey_source_selector': 'new'},68 {'dbkey_source|dbkey': 'INVALID'},69 {'dbkey_source|dbkey_name': 'INVALID_KEY'},70 {'sequence_name': 'INVALID'},71 {'sequence_id': 'INVALID'},72 {'reference_source|reference_source_selector': 'ncbi'},73 {'reference_source|requested_identifier': 'INVALID0123'}74 ],75 data_table_reload=[76 "all_fasta",77 "__dbkeys__"78 ]79 )80 ]81 )82 dm = DataManagers(container.gi, configuration)83 with pytest.raises(RuntimeError):84 dm.run()85 assert ("HTTP Error 404" in caplog.text)86 assert ("Not all jobs successful! aborting..." in caplog.text)...
sort.py
Source:sort.py
1#! /usr/bin/python2'''Sorts bed/gff file with a respect to strandness'''3import argparse4import sys;5import os;6from collections import defaultdict;7from pybedtools import BedTool;8parser = argparse.ArgumentParser(description='Sorts bed/gff file with a respect to strandness');9parser.add_argument('path', metavar = 'N', nargs = '?', type = str, help = "Path to the bed/gff file to be sorted");10args = parser.parse_args();11def clear_dict(d):12 for v in d.values():13 v[:]=[];14 15 16def flush(chrom_container, start_container):17 for strand, intervals in start_container.items():18 chrom_container[strand].extend(list(sorted(intervals, key = lambda x: x.end)));19 20sortedbed = BedTool(args.path).sort();21if(len(sortedbed) == 0):22 sys.stderr.write("input file is empty\n");23 sys.exit();24chrom_container = defaultdict(list)25start_container = defaultdict(list)26start_container[sortedbed[0].strand].append(sortedbed[0])27chrom = sortedbed[0].chrom;28start = sortedbed[0].start29for interval in sortedbed[1:]:30 if(chrom == interval.chrom):31 if(start == interval.start):32 start_container[interval.strand].append(interval);33 else:34 flush(chrom_container, start_container);35 clear_dict(start_container)36 start = interval.start;37 start_container[interval.strand].append(interval);38 39 else:40 flush(chrom_container, start_container);41 for strand, intervals in chrom_container.items():42 for si in intervals:43 sys.stdout.write(str(si))44 45 chrom = interval.chrom;46 start = interval.start;47 clear_dict(start_container);48 clear_dict(chrom_container);49 start_container[interval.strand].append(interval);50 51else: 52 flush(chrom_container, start_container);53 for strand, intervals in chrom_container.items():54 for si in intervals:...
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?
