Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use inspect_container method in localstack
Best Python code snippet using localstack_python
test_functional_shipy.py
Source:test_functional_shipy.py 
...26def test_docker_run_args(client, shipy):27 cargs = 'ping -v -c 100 127.0.0.1'28 container = run_template(client, shipy, cargs=cargs)29 assert cargs == '{} {}'.format(30 client.inspect_container(container)['Path'],31 ' '.join(client.inspect_container(container)['Args']))32def test_docker_run_entrypoint(client, shipy):33 cargs = '-c 100 127.0.0.1'34 farg = '--entrypoint'35 fval = ('ping',)36 container = run_template(client, shipy, farg=farg, fval=fval, cargs=cargs)37 assert fval[0] == \38 client.inspect_container(container)['Config']['Entrypoint'][0]39def test_docker_run_hostname(client, shipy):40 farg = '--hostname'41 fval = ('batman',)42 container = run_template(client, shipy, farg, fval=fval)43 assert fval[0] == \44 client.inspect_container(container)['Config']['Hostname']45def test_docker_run_user(client, shipy):46 argument = ('-u', '--user')47 fval = ('root',)48 for farg in argument:49 container = run_template(client, shipy, farg=farg, fval=fval)50 assert fval[0] == \51 client.inspect_container(container)['Config']['User']52def test_docker_run_env(client, shipy):53 argument = ('-e', '--env')54 fval = ('bat=man', 'bruce=wayne')55 for farg in argument:56 container = run_template(client, shipy,57 farg=farg, fval=fval)58 for env in fval:59 assert env in \60 client.inspect_container(container)['Config']['Env']61def test_docker_run_cpu_shares(client, shipy):62 farg = '--cpu-shares'63 fval = (5,)64 container = run_template(client, shipy, farg=farg, fval=fval)65 assert fval[0] == \66 client.inspect_container(container)['HostConfig']['CpuShares']67def test_docker_run_workdir(client, shipy):68 argument = ('-w', '--workdir')69 fval = ('/tmp',)70 for farg in argument:71 container = run_template(client, shipy, farg=farg, fval=fval)72 assert fval[0] == \73 client.inspect_container(container)['Config']['WorkingDir']74def test_docker_run_mac_address(client, shipy):75 farg = '--mac-address'76 fval = ('aa:aa:aa:aa:aa:aa',)77 container = run_template(client, shipy, farg=farg, fval=fval)78 assert fval[0] == \79 client.inspect_container(container)['NetworkSettings']['MacAddress']80def test_docker_run_labels(client, shipy):81 argument = ('-l', '--label')82 fval = ('batman=begins', 'thedarkknight=rises')83 for farg in argument:84 container = run_template(client, shipy,85 farg=farg, fval=fval)86 for label in fval:87 l = label.split('=')88 assert l[1] == \89 client.inspect_container(container)['Config']['Labels'][l[0]]90def test_docker_run_stop_signal(client, shipy):91 farg = '--stop-signal'92 fval = ('SIGKILL',)93 container = run_template(client, shipy, farg=farg, fval=fval)94 assert fval[0] == \95 client.inspect_container(container)['Config']['StopSignal']96def test_docker_run_volumes(client, shipy):97 argument = ('-v', '--volume')98 fval = ('/test_dir1:/test_dir_mount1:ro', '/test_dir2:/tmp')99 for farg in argument:100 container = run_template(client, shipy,101 farg=farg, fval=fval)102 binding_list = []103 for binding in fval:104 if len(binding) > 1:105 binding_list.append(binding)106 assert binding_list == \107 client.inspect_container(container)['HostConfig']['Binds']108 # this is asserting if the volumes are exposed on the container or not109 # a bit different from docker run, but can be called ideal110 assert_dict = {}111 for value in fval:112 assert_dict.update({value.split(':')[0]: {}})113 assert assert_dict == \114 client.inspect_container(container)['Config']['Volumes']115def test_docker_run_publish(client, shipy):116 argument = ('-p', '--publish')117 fval = ('13370:23370/udp', '13371:23371', '127.0.0.1:13372:23372',118 '23373', '127.0.0.1::23374')119 ext_fval = []120 only_ports = []121 for value in fval:122 container_port = value.split(':')[-1]123 host_params = value.split(':')[:-1]124 if container_port[-3:] not in ('tcp', 'udp'):125 ext_fval.append('{}:{}/tcp'.format(host_params, container_port))126 else:127 ext_fval.append('{}:{}'.format(host_params, container_port))128 for value in ext_fval:129 only_ports.append(value.split(':')[-1])130 for farg in argument:131 container = run_template(client, shipy,132 farg=farg, fval=fval)133 exposed_ports = \134 client.inspect_container(135 container)['Config']['ExposedPorts'].keys()136 exposed_ports.sort()137 # print ext_fval, exposed_ports138 assert only_ports == exposed_ports139 for cport, ncport in \140 client.inspect_container(141 container)['HostConfig']['PortBindings'].items():142 if ncport[0]['HostIp']:143 host_conf = '[\'{}\', \'{}\']'.format(144 ncport[0]['HostIp'], ncport[0]['HostPort'])145 elif ncport[0]['HostPort']:146 host_conf = '[\'{}\']'.format(ncport[0]['HostPort'])147 else:148 host_conf = []149 assert '{}:{}'.format(host_conf, cport) in ext_fval150 client.remove_container(container, force=True)151def test_docker_run_publish_all(client, shipy):152 argument = ('-P', '--publish-all')153 for farg in argument:154 container = run_template(client, shipy,155 farg=farg)156 assert client.inspect_container(157 container)['HostConfig']['PublishAllPorts']158def test_docker_run_links(client, shipy):159 farg = '--link'160 fval = []161 for _ in range(2):162 fval.append(cn())163 fval[1] += ':alias'164 fval = tuple(fval)165 ext_fval = []166 for linked_to in fval:167 run_template(client, shipy, cn=linked_to.split(':')[0])168 if len(linked_to.split(':')) == 1:169 linked_to += ':{}'.format(linked_to)170 ext_fval.append(linked_to)171 container = run_template(client, shipy, farg=farg, fval=fval)172 links = client.inspect_container(container)['HostConfig']['Links']173 assert len(fval) == len(links)174 for link in links:175 name, _, alias = ''.join(link.split(':')).split('/')[1:]176 assert '{}:{}'.format(name, alias) in ext_fval177def test_docker_run_privileged(client, shipy):178 farg = '--privileged'179 container = run_template(client, shipy, farg=farg)180 assert client.inspect_container(container)['HostConfig']['Privileged']181def test_docker_run_dns(client, shipy):182 farg = '--dns'183 fval = ('8.8.8.8', '8.8.4.4')184 container = run_template(client, shipy, farg=farg, fval=fval)185 assert [fval[0], fval[1]] == \186 client.inspect_container(container)['HostConfig']['Dns']187def test_docker_run_dns_search(client, shipy):188 farg = '--dns-search'189 fval = ('batman.com', 'brucewayne.com')190 container = run_template(client, shipy, farg=farg, fval=fval)191 assert [fval[0], fval[1]] == \192 client.inspect_container(container)['HostConfig']['DnsSearch']193def test_docker_run_volumes_from(client, shipy):194 farg = '--volumes-from'195 fval = []196 for _ in range(2):197 fval.append(cn())198 for volume_container in fval:199 run_template(client, shipy, cn=volume_container)200 container = run_template(client, shipy, farg=farg, fval=fval)201 assert fval == \202 client.inspect_container(container)['HostConfig']['VolumesFrom']203def test_docker_run_network_mode(client, shipy):204 reuse_network_container = cn()205 run_template(client, shipy, cn=reuse_network_container)206 for mode in ['bridge', 'none', 'container:{}'.format(207 reuse_network_container), 'host']:208 farg = '--net'209 fval = (mode,)210 container = run_template(client, shipy, farg=farg, fval=fval)211 assert fval[0] == \212 client.inspect_container(container)['HostConfig']['NetworkMode']213def test_docker_run_detach(client, shipy):214 argument = ('-d', '--detach')215 for farg in argument:216 container = run_template(client, shipy,217 farg=farg)218 assert not client.inspect_container(219 container)['Config']['AttachStdout']220def test_docker_run_restart_policy_on_failure(client, shipy):221 farg = '--restart'222 for fval in (('no',), ('on-failure',), ('on-failure:4',), ('always',), ('unless-stopped',)):223 container = run_template(client, shipy, farg=farg, fval=fval)224 inspect_output = client.inspect_container(container)['HostConfig']['RestartPolicy']225 fval_split = fval[0].split(':')226 assert fval_split[0] == inspect_output['Name']227 if fval_split[0] == 'on-failure' and len(fval_split) == 2:228 assert int(fval_split[1]) == inspect_output['MaximumRetryCount']229def test_docker_run_cap_add(client, shipy):230 farg = '--cap-add'231 fval = ('SYS_ADMIN', 'SYS_TIME')232 container = run_template(client, shipy, farg=farg, fval=fval)233 assert [fval[0], fval[1]] == \234 client.inspect_container(container)['HostConfig']['CapAdd']235def test_docker_run_cap_drop(client, shipy):236 farg = '--cap-drop'237 fval = ('SYS_ADMIN', 'SYS_TIME')238 container = run_template(client, shipy, farg=farg, fval=fval)239 assert [fval[0], fval[1]] == \240 client.inspect_container(container)['HostConfig']['CapDrop']241def test_docker_run_add_host(client, shipy):242 farg = '--add-host'243 fval = ('batman.example:10.0.0.1', 'wayne.manor:10.0.0.2')244 container = run_template(client, shipy, farg=farg, fval=fval)245 assert [fval[0], fval[1]] == \246 client.inspect_container(container)['HostConfig']['ExtraHosts']247def test_docker_run_read_only(client, shipy):248 farg = '--read-only'249 container = run_template(client, shipy, farg=farg)250 assert client.inspect_container(container)['HostConfig']['ReadonlyRootfs']251def test_docker_run_pid(client, shipy):252 farg = '--pid'253 fval = ('host',)254 container = run_template(client, shipy, farg=farg, fval=fval)255 assert fval[0] == \256 client.inspect_container(container)['HostConfig']['PidMode']257def test_docker_run_ipc(client, shipy):258 farg = '--ipc'259 ipc_cn = cn()260 for fval in (('host',),261 ('container:{}'.format(ipc_cn),)262 ):263 if fval[0].split(':')[0] == 'container':264 run_template(client, shipy, cn=ipc_cn)265 container = run_template(client, shipy, farg=farg, fval=fval)266 assert fval[0] == \267 client.inspect_container(container)['HostConfig']['IpcMode']268def test_docker_run_security_opt(client, shipy):269 farg = '--security-opt'270 fval = ('label=type:svirt_apache_t', 'label=level:s0:c100,c200')271 container = run_template(client, shipy, farg=farg, fval=fval)272 assert [fval[0], fval[1]] == \273 client.inspect_container(container)['HostConfig']['SecurityOpt']274def test_docker_run_ulimit(client, shipy):275 farg = '--ulimit'276 fval = ('nproc=3:4', 'nofile=1024')277 container = run_template(client, shipy, farg=farg, fval=fval)278 for v, u in zip(fval,279 client.inspect_container(container)['HostConfig']['Ulimits']):280 name, limit = v.split('=')281 limits = limit.split(':')282 soft = int(limits[0])283 hard = int(limits[1]) if len(limits) == 2 else soft284 assert u['Name'] == name285 assert u['Soft'] == soft286 assert u['Hard'] == hard287def test_docker_run_log_driver(client, shipy):288 farg = '--log-driver'289 fval = ('syslog',)290 container = run_template(client, shipy, farg=farg, fval=fval)291 assert fval[0] == \292 client.inspect_container(container)['HostConfig']['LogConfig']['Type']293def test_docker_run_log_opt(client, shipy):294 farg = '--log-opt'295 fval = ('tag=batman', 'syslog-format=rfc3164')296 sarg = '--log-driver'297 sval = 'syslog'298 container = run_template(client, shipy, farg=farg, fval=fval,299 sarg=sarg, sval=sval)300 assert sval == \301 client.inspect_container(container)[302 'HostConfig']['LogConfig']['Type']303 for opt in fval:304 k, v = opt.split('=')305 assert v == \306 client.inspect_container(container)[307 'HostConfig']['LogConfig']['Config'][k]308def test_docker_run_memory(client, shipy):309 farg = '--memory'310 fval = ('100M',)311 container = run_template(client, shipy, farg=farg, fval=fval)312 assert fval[0] == \313 size(client.inspect_container(container)['HostConfig']['Memory'])314def test_docker_run_memory_swap(client, shipy):315 farg = '--memory-swap'316 fval = ('110M',)317 sarg = '--memory'318 sval = '100M'319 container = run_template(client, shipy, farg=farg, fval=fval,320 sarg=sarg, sval=sval)321 assert sval == \322 size(client.inspect_container(container)['HostConfig']['Memory'])323 assert fval[0] == \324 size(client.inspect_container(container)['HostConfig']['MemorySwap'])325def test_docker_run_memory_swappiness(client, shipy):326 farg = '--memory-swappiness'327 fval = (42,)328 container = run_template(client, shipy, farg=farg, fval=fval)329 assert fval[0] == \330 client.inspect_container(container)['HostConfig']['MemorySwappiness']331def test_docker_run_shm_size(client, shipy):332 farg = '--shm-size'333 fval = ('100M',)334 container = run_template(client, shipy, farg=farg, fval=fval)335 assert fval[0] == \336 size(client.inspect_container(container)['HostConfig']['ShmSize'])337def test_docker_run_cpu_period(client, shipy):338 farg = '--cpu-period'339 fval = (1000,)340 container = run_template(client, shipy, farg=farg, fval=fval)341 assert fval[0] == \342 client.inspect_container(container)['HostConfig']['CpuPeriod']343def test_docker_run_group_add(client, shipy):344 farg = '--group-add'345 fval = ('wheel',)346 container = run_template(client, shipy, farg=farg, fval=fval)347 assert fval[0] == \348 client.inspect_container(container)['HostConfig']['GroupAdd'][0]349def test_docker_run_device(client, shipy):350 farg = '--device'351 fval = ('/dev/null:/dev/sda2:rwm', '/dev/zero:/dev/sda3:m')352 container = run_template(client, shipy, farg=farg, fval=fval)353 devices = client.inspect_container(container)['HostConfig']['Devices']354 chost_path, ccontainer_path, cperms = [], [], []355 for device in devices:356 chost_path.append(device['PathOnHost'])357 ccontainer_path.append(device['PathInContainer'])358 cperms.append(device['CgroupPermissions'])359 for dev in fval:360 host_path, container_path, perms = dev.split(':')361 assert host_path in chost_path362 assert container_path in ccontainer_path363 assert perms in cperms364def test_docker_run_tmpfs(client, shipy):365 farg = '--tmpfs'366 fval = ('/run/:rw,noexec,nosuid,size=65536k', '/mnt/:size=3G,uid=1000')367 container = run_template(client, shipy, farg=farg, fval=fval)368 tmpfs = client.inspect_container(container)['HostConfig']['Tmpfs']369 for val in fval:370 mount, args = val.split(':')371 assert tmpfs[mount] == args372def test_docker_run_interactive(client, shipy):373 argument = ('-i', '--interactive')374 for farg in argument:375 container = run_template(client, shipy, farg=farg)376 assert client.inspect_container(container)[377 'Config']['AttachStdin']378 assert client.inspect_container(container)[379 'Config']['OpenStdin']380 assert client.inspect_container(container)[381 'Config']['StdinOnce']382def test_docker_run_tty(client, shipy):383 argument = ('-t', '--tty')384 for farg in argument:385 container = run_template(client, shipy, farg=farg)386 assert client.inspect_container(container)['Config']['Tty']387def test_docker_run_volume_driver(client, shipy):388 farg = '--volume-driver'389 fval = ('local',)390 container = run_template(client, shipy, farg=farg, fval=fval)391 assert fval[0] == \392 client.inspect_container(container)['HostConfig']['VolumeDriver']393def test_docker_run_oom_kill_disable(client, shipy):394 farg = '--oom-kill-disable'395 container = run_template(client, shipy, farg=farg)396 assert client.inspect_container(container)['HostConfig']['OomKillDisable']397def test_docker_oom_score_adj(client, shipy):398 farg = '--oom-score-adj'399 fval = (42,)400 container = run_template(client, shipy, farg=farg, fval=fval)401 assert fval[0] == \402 client.inspect_container(container)['HostConfig']['OomScoreAdj']403# def test_docker_blkio_weight(client, shipy):404# farg = '--blkio-weight'405# fval = (42,)406# container = run_template(client, shipy, farg=farg, fval=fval)407#408# assert fval[0] == \409# client.inspect_container(container)['HostConfig']['BlkioWeight']410#411#412# def test_docker_blkio_weight_device(client, shipy):413# farg = '--blkio-weight-device'414# fval = ('/dev/null:42', '/dev/sdb:84')415# container = run_template(client, shipy, farg=farg, fval=fval)416#417# bk = client.inspect_container(container)['HostConfig']['BlkioWeightDevice']418#419# cpath, cweight = [], []420# for blkio in bk:421# cpath.append(blkio['Path'])422# cweight.append(blkio['Weight'])423#424# for val in fval:425# path, weight = val.split(':')426#427# assert path in cpath428# assert weight in cweight429#430#431# def test_docker_device_read_bps(client, shipy):432# farg = '--device-read-bps'433# fval = ('/dev/sda:1mb', '/dev/sdb:2mb')434# container = run_template(client, shipy, farg=farg, fval=fval)435#436# bk = client.inspect_container(container)['HostConfig']['BlkioDeviceReadBps']437#438# cpath, crate = [], []439# for blkio in bk:440# cpath.append(blkio['Path'])441# crate.append(blkio['Rate'])442#443# for val in fval:444# path, rate = val.split(':')445#446# assert path in cpath447# assert rate in crate448#449#450# def test_docker_device_write_bps(client, shipy):451# farg = '--device-write-bps'452# fval = ('/dev/sda:1mb', '/dev/sdb:2mb')453# container = run_template(client, shipy, farg=farg, fval=fval)454#455# bk = client.inspect_container(container)['HostConfig']['BlkioDeviceWriteBps']456#457# cpath, crate = [], []458# for blkio in bk:459# cpath.append(blkio['Path'])460# crate.append(blkio['Rate'])461#462# for val in fval:463# path, rate = val.split(':')464#465# assert path in cpath466# assert rate in crate467#468#469# def test_docker_device_read_iops(client, shipy):470# farg = '--device-read-iops'471# fval = ('/dev/sda:1000', '/dev/sdb:2000')472# container = run_template(client, shipy, farg=farg, fval=fval)473#474# bk = client.inspect_container(container)['HostConfig'][475# 'BlkioDeviceReadIOps']476#477# cpath, crate = [], []478# for blkio in bk:479# cpath.append(blkio['Path'])480# crate.append(blkio['Rate'])481#482# for val in fval:483# path, rate = val.split(':')484#485# assert path in cpath486# assert rate in crate487#488#489# def test_docker_device_write_iops(client, shipy):490# farg = '--device-write-iops'491# fval = ('/dev/sda:1000', '/dev/sdb:2000')492# container = run_template(client, shipy, farg=farg, fval=fval)493#494# bk = client.inspect_container(container)['HostConfig'][495# 'BlkioDeviceWriteIOps']496#497# cpath, crate = [], []498# for blkio in bk:499# cpath.append(blkio['Path'])500# crate.append(blkio['Rate'])501#502# for val in fval:503# path, rate = val.split(':')504#505# assert path in cpath...
open_nti_input_syslog_lib.py
Source:open_nti_input_syslog_lib.py
...77 global OPENNTI_CID78 global OPENNTI_INTERNAL_IP79 # Force Stop and delete existing container if exist80 try:81 old_container_id = c.inspect_container(OPENNTI_C_NAME)['Id']82 c.stop(container=old_container_id)83 c.remove_container(container=old_container_id)84 except:85 print "Container do not exit"86 container = c.create_container(87 image=OPENNTI_IMAGE_NAME,88 command='/sbin/my_init',89 name=OPENNTI_C_NAME,90 detach=True,91 ports=[92 3000,93 8083,94 808695 ],96 host_config=c.create_host_config(97 port_bindings={98 '8086': OPENNTI_API_PORT,99 '8083': OPENNTI_INFLUX_PORT,100 },101 binds=[102 CURRENT_DIR + ':/root/fluent',103 ]104 )105 )106 c.start(container)107 OPENNTI_CID = c.inspect_container(OPENNTI_C_NAME)['Id']108 OPENNTI_INTERNAL_IP = c.inspect_container(OPENNTI_C_NAME)['NetworkSettings']['IPAddress']109 # Wait few sec for the container to start110 time.sleep(5)111 return OPENNTI_CID112def stop_open_nti():113 global c114 # Force Stop and delete existing container if exist115 try:116 old_container_id = c.inspect_container(OPENNTI_C_NAME)['Id']117 c.stop(container=old_container_id)118 c.remove_container(container=old_container_id)119 except:120 print "Container do not exit"121def get_influxdb_handle():122 global INFLUXDB_HANDLE123 if INFLUXDB_HANDLE == '':124 INFLUXDB_HANDLE = influxdb.InfluxDBClient(125 host=OPENNTI_IP,126 port=OPENNTI_API_PORT,127 database=OPENNTI_DATABASE_NAME,128 username="juniper",129 password="juniper"130 )131 return INFLUXDB_HANDLE132def get_external_ip():133 return [(s.connect(('8.8.8.8', 53)), s.getsockname()[0], s.close()) for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]134def start_kafka():135 global c136 global KAFKA_CID137 global KAFKA_INTERNAL_IP138 global KAFKA_BROKER_PORT139 global KAFKA_API_PORT140 # Force Stop and delete existing container if exist141 try:142 old_container_id = c.inspect_container(KAFKA_C_NAME)['Id']143 c.stop(container=old_container_id)144 c.remove_container(container=old_container_id)145 except:146 print "Container do not exit"147 container = c.create_container(148 image=KAFKA_IMAGE_NAME,149 name=KAFKA_C_NAME,150 detach=True,151 environment=[152 "ADVERTISED_PORT="+str(KAFKA_BROKER_PORT),153 "ADVERTISED_HOST="+str(get_external_ip())154 ],155 ports=[156 2181,157 9092158 ],159 host_config=c.create_host_config(160 port_bindings={161 '2181': KAFKA_API_PORT,162 '9092': KAFKA_BROKER_PORT,163 }164 )165 )166 c.start(container)167 KAFKA_CID = c.inspect_container(KAFKA_C_NAME)['Id']168 KAFKA_INTERNAL_IP = c.inspect_container(KAFKA_C_NAME)['NetworkSettings']['IPAddress']169 # Wait few sec for the container to start170 time.sleep(5)171 return KAFKA_CID172def stop_kafka():173 global c174 # Force Stop and delete existing container if exist175 try:176 old_container_id = c.inspect_container(KAFKA_C_NAME)['Id']177 c.stop(container=old_container_id)178 c.remove_container(container=old_container_id)179 except:180 print "Container do not exit"181@timeout(5)182def check_kafka_msg(topic='events', nbr_msg=100):183 ## Collect Messages from Bus184 consumer = KafkaConsumer(185 bootstrap_servers=get_external_ip()+':'+str(KAFKA_BROKER_PORT),186 auto_offset_reset='earliest')187 consumer.subscribe([topic])188 counter = 0189 for message in consumer:190 counter = counter + 1191 if counter == nbr_msg:192 break193 return counter194def check_influxdb_running_database_exist():195 # Verify we can connect to InfluxDB and DB with a name juniper exists196 db = get_influxdb_handle()197 db_list = db.get_list_database()198 found_db = 0199 for db_entry in db_list:200 if db_entry['name'] == OPENNTI_DATABASE_NAME:201 found_db = 1202 else:203 continue204 return found_db205def check_kafka_is_running():206 # Verify we can connect to InfluxDB and DB with a name juniper exists207 consumer = KafkaConsumer(bootstrap_servers=get_external_ip()+':'+str(KAFKA_BROKER_PORT),208 auto_offset_reset='earliest')209 mytopic = consumer.topics()210 return 1211def start_fluentd_syslog(output_kafka='false', output_influx='false'):212 global c213 global CONTAINER_ID214 # Force Stop and delete existing container if exist215 try:216 old_container_id = c.inspect_container(CONTAINER_NAME)['Id']217 c.stop(container=old_container_id)218 c.remove_container(container=old_container_id)219 except:220 print "Container do not exit"221 container = c.create_container(222 image=IMAGE_NAME,223 name=CONTAINER_NAME,224 detach=True,225 environment=[226 "PORT_SYSLOG=16000",227 "INFLUXDB_PORT="+str(8086),228 "INFLUXDB_DB="+str(OPENNTI_DATABASE_NAME),229 "INFLUXDB_USER=juniper",230 "INFLUXDB_PWD=juniper",231 "INFLUXDB_ADDR="+OPENNTI_INTERNAL_IP,232 "INFLUXDB_FLUSH_INTERVAL=1",233 "KAFKA_ADDR="+str(get_external_ip()),234 "KAFKA_PORT="+str(KAFKA_BROKER_PORT),235 "KAFKA_TOPIC=events",236 "OUTPUT_KAFKA="+output_kafka,237 "OUTPUT_INFLUXDB="+output_influx,238 "OUTPUT_STDOUT=true"239 ],240 ports=[241 (16000, 'udp'),242 ],243 host_config=c.create_host_config(244 port_bindings={245 '16000/udp': TEST_PORT_SYSLOG,246 }247 )248 )249 c.start(container)250 time.sleep(1)251 CONTAINER_ID = c.inspect_container(CONTAINER_NAME)['Id']252 return CONTAINER_ID253def stop_fluentd():254 global c255 # Force Stop and delete existing container if exist256 try:257 old_container_id = c.inspect_container(CONTAINER_NAME)['Id']258 c.stop(container=old_container_id)259 c.remove_container(container=old_container_id)260 except:261 print "Container do not exit"262def replay_file(file_name):263 global c264 try:265 old_container_id = c.inspect_container(TCP_RELAY_CONTAINER_NAME)['Id']266 c.stop(container=old_container_id)267 c.remove_container(container=old_container_id)268 except:269 print "Container do not exit"270 container = c.create_container(271 image='dgarros/tcpreplay',272 command='/usr/bin/tcpreplay --pps=100 --intf1=eth0 /data/' + file_name,273 name=TCP_RELAY_CONTAINER_NAME,274 volumes=[275 '/data'276 ],277 host_config=c.create_host_config(278 binds=[279 TESTS_FIXTURES_DIR + ':/data',...
test_docker_interface.py
Source:test_docker_interface.py
...25 docker.remove_all_containers()26 def test_create_container(self):27 docker.create_container(self._container, force=False)28 self.assertIsNotNone(29 docker.inspect_container(self._container)30 )31 self.assertRaises(32 Exception,33 lambda: docker.create_container(self._container, force=False)34 )35 self.assertIsNotNone(36 docker.inspect_container(self._container)37 )38 def test_start_container(self):39 docker.create_container(self._container)40 self.assertIsNotNone(41 docker.inspect_container(self._container)42 )43 docker.start_container(self._container)44 stat = docker.inspect_container(self._container)45 self.assertIsNotNone(stat)46 self.assertTrue(stat['State']['Running'])47 def test_start_container_error(self):48 self.assertRaises(49 Exception,50 lambda: docker.start_container(self._container)51 )52 self.assertIsNone(53 docker.inspect_container(self._container)54 )55 def test_stop_container(self):56 docker.create_container(self._container)57 self.assertIsNotNone(58 docker.inspect_container(self._container)59 )60 docker.start_container(self._container)61 stat = docker.inspect_container(self._container)62 self.assertIsNotNone(stat)63 self.assertTrue(stat['State']['Running'])64 docker.stop_container(self._container)65 stat = docker.inspect_container(self._container)66 self.assertIsNotNone(stat)67 self.assertFalse(stat['State']['Running'])68 def test_stop_container_error(self):69 self.assertRaises(70 Exception,71 lambda: docker.stop_container(self._container)72 )73 self.assertIsNone(74 docker.inspect_container(self._container)75 )76 def test_container_delete(self):77 docker.create_container(self._container)78 self.assertIsNotNone(79 docker.inspect_container(self._container)80 )81 docker.delete_container(self._container)82 self.assertIsNone(83 docker.inspect_container(self._container)84 )85 def test_container_delete_error(self):86 self.assertRaises(87 Exception,88 lambda: docker.delete_container(self._container)89 )90 self.assertIsNone(91 docker.inspect_container(self._container)92 )93 def test_container_exec(self):94 docker.create_container(self._container)95 self.assertIsNotNone(96 docker.inspect_container(self._container)97 )98 docker.start_container(self._container)99 docker.exec_cmd(self._container, 'echo hello!')100 def test_container_exec_error(self):101 docker.create_container(self._container)102 self.assertIsNotNone(103 docker.inspect_container(self._container)104 )105 self.assertRaises(106 Exception,107 lambda: docker.exec_cmd(self._container, 'echo hello!')108 )109 docker.delete_container(self._container)110 self.assertIsNone(111 docker.inspect_container(self._container)112 )113 self.assertRaises(114 Exception,115 lambda: docker.exec_cmd(self._container, 'echo hello!')116 )117 # def test_network(self):118 # docker.delete_network()119 # docker.delete_network()120 # # TODO: assert that the network is correctlly created121 # docker.create_network()...
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?
