PKZLL'apytest_docker_tools/__init__.py''' An opionated set of helpers for defining Docker integration test environments with py.test fixtures. ''' from .factories import build, container, fetch, network, volume __version__ = '0.0.3' __all__ = [ 'build', 'container', 'fetch', 'network', 'volume', ] PKqL!pytest_docker_tools/plugin.pyimport docker import pytest @pytest.fixture(scope='session') def docker_client(request): return docker.from_env() @pytest.hookimpl(tryfirst=True, hookwrapper=True) def pytest_runtest_makereport(item, call): ''' This hook allows Docker containers to contribute their logs to the py.test report. ''' outcome = yield rep = outcome.get_result() if rep.when != 'call': return if not rep.failed: return for name, fixture in item.funcargs.items(): if isinstance(fixture, dict) and 'container' in fixture: container = fixture['container'] rep.sections.append(( name + ': ' + container.name, container.logs().decode('utf-8'), )) PK(LL`͵pytest_docker_tools/utils.pyimport sys import time def wait_for_callable(message, callable, timeout=30): ''' Runs a callable once a second until it returns True or we hit the timeout. ''' sys.stdout.write(message) try: for i in range(timeout): sys.stdout.write('.') sys.stdout.flush() if callable(): return time.sleep(1) finally: sys.stdout.write('\n') raise RuntimeError('Timeout exceeded') PK\L)pytest_docker_tools/factories/__init__.pyfrom .build import build from .container import container from .fetch import fetch from .network import network from .volume import volume __all__ = [ 'build', 'container', 'fetch', 'network', 'volume', ] PKm_L|P?&pytest_docker_tools/factories/build.pyimport sys import pytest def build(*, path, scope='session'): ''' Fixture factory for creating container images from a Dockerfile. For example in your conftest.py you can: from pytest_docker_tools import build test_image = build(path='path/to/buildcontext') Where the path is a folder containing a Dockerfile. By default the fixture has a session scope. ''' def build(request, docker_client): sys.stdout.write(f'Building {path}') try: image, logs = docker_client.images.build( path=path ) for line in logs: sys.stdout.write('.') sys.stdout.flush() finally: sys.stdout.write('\n') # request.addfinalizer(lambda: docker_client.images.remove(image.id)) return image pytest.fixture(scope=scope)(build) return build PKTL 8vv*pytest_docker_tools/factories/container.pyimport inspect from string import Formatter import pytest from pytest_docker_tools.utils import wait_for_callable from pytest_docker_tools.wrappers import Container def create_container(request, docker_client, *args, **kwargs): kwargs.update({'detach': True}) container = docker_client.containers.run(*args, **kwargs) request.addfinalizer(lambda: container.remove(force=True) and container.wait(timeout=10)) return Container(container) class FixtureFormatter(Formatter): def __init__(self, request): self.request = request def get_value(self, key, args, kwargs): return self.request.getfixturevalue(key) def _process_val(request, val): if isinstance(val, str): return FixtureFormatter(request).format(val) elif callable(val): return val(*[request.getfixturevalue(f) for f in inspect.getargspec(val)[0]]) return val def _process_list(request, val): return [_process(request, v) for v in val] def _process_dict(request, mapping): return {_process(request, k): _process(request, v) for (k, v) in mapping.items()} def _process(request, val): if isinstance(val, dict): return _process_dict(request, val) elif isinstance(val, list): return _process_list(request, val) else: return _process_val(request, val) def container(*, scope='function', **kwargs): ''' Fixture factory for creating containers. For example in your conftest.py you can: from pytest_docker_tools import container_fixture test_container = container_fixture('test_container', 'redis') This will create a container called 'test_container' from the 'redis' image. ''' def container(request, docker_client): local_kwargs = dict(kwargs) container = create_container( request, docker_client, **_process_dict(request, local_kwargs) ) wait_for_callable( f'Waiting for container to be ready', lambda: container.reload() or container.ready(), ) return container pytest.fixture(scope=scope)(container) return container PKi_Lūuj&pytest_docker_tools/factories/fetch.pyimport sys import pytest def fetch(tag, scope='session'): ''' Fixture factory for fetching container images from a repository. For example in your conftest.py you can: from pytest_docker_tools import factories factories.repository_image('test_image', 'redis:latest') By default the fixture has a session scope. ''' if ':' not in tag: tag += ':latest' def fetch(request, docker_client): sys.stdout.write(f'Fetching {tag}\n') image = docker_client.images.pull(tag) # request.addfinalizer(lambda: docker_client.images.remove(image.id)) return image pytest.fixture(scope=scope)(fetch) return fetch PKg_LlHH(pytest_docker_tools/factories/network.pyimport uuid import pytest def network(scope='function'): ''' Fixture factory for creating networks. For example in your conftest.py you can: from pytest_docker_tools import network_fixture test_storage = network_fixture('test_storage') Then you can reference that network from your test: def test_a_docker_network(test_storage): print(test_storage.id) The fixture has a function scope - it will be destroyed after your test exits. ''' def network(request, docker_client): network_id = 'pytest-' + str(uuid.uuid4()) print(f'Creating network {network_id}') network = docker_client.networks.create(network_id) request.addfinalizer(lambda: network.remove()) return network pytest.fixture(scope=scope)(network) return network PKd_Lʐq22'pytest_docker_tools/factories/volume.pyimport uuid import pytest def volume(scope='function'): ''' Fixture factory for creating volumes. For example in your conftest.py you can: from pytest_docker_tools import volume_fixture test_storage = volume_fixture('test_storage') Then you can reference that volume from your test: def test_a_docker_volume(test_storage): print(test_storage.id) The fixture has a function scope - it will be destroyed after your test exits. ''' def volume(request, docker_client): vol_id = 'pytest-' + str(uuid.uuid4()) print(f'Creating volume {vol_id}') volume = docker_client.volumes.create(vol_id) request.addfinalizer(lambda: volume.remove(True)) return volume pytest.fixture(scope=scope)(volume) return volume PK(LL'kBB(pytest_docker_tools/wrappers/__init__.pyfrom .container import Container __all__ = [ 'Container', ] PK;SLtt)pytest_docker_tools/wrappers/container.py''' This module contains a wrapper that adds some helpers to a Docker Container object that are useful for integration testing. ''' import io import tarfile class _Map(object): def __init__(self, container): self._container = container def values(self): return [self[k] for k in self.keys()] def items(self): return [(k, self[k]) for k in self.keys()] def __iter__(self): return iter(self.keys()) class IpMap(_Map): @property def primary(self): return next(iter(self.values())) def keys(self): return self._container.attrs['NetworkSettings']['Networks'].keys() def __getitem__(self, key): if not isinstance(key, str): key = key.name networks = self._container.attrs['NetworkSettings']['Networks'] if key not in networks: raise KeyError(f'Unknown network: {key}') return networks[key]['IPAddress'] class PortMap(_Map): def __init__(self, container): self._container = container def keys(self): return self._container.attrs['NetworkSettings']['Ports'].keys() def __getitem__(self, key): ports = self._container.attrs['NetworkSettings']['Ports'] if key not in ports: raise KeyError(f'Unknown port: {key}') if not ports[key]: return [] return [p['HostPort'] for p in ports[key]] class Container(object): def __init__(self, container): self._container = container self.ips = IpMap(container) self.ports = PortMap(container) def ready(self): if self.status == 'exited': raise RuntimeError(f'Container {self.name} has already exited before we noticed it was ready') if self.status != 'running': return False networks = self._container.attrs['NetworkSettings']['Networks'] for name, network in networks.items(): if not network['IPAddress']: return False # If a user has exposed a port then wait for LISTEN socket to show up in netstat ports = self._container.attrs['NetworkSettings']['Ports'] for port, listeners in ports.items(): if not listeners: continue port, proto = port.split('/') assert proto in ('tcp', 'udp') if proto == 'tcp' and port not in self.get_open_tcp_ports(): return False if proto == 'udp' and port not in self.get_open_udp_ports(): return False return True @property def attrs(self): return self._container.attrs @property def id(self): return self._container.id @property def env(self): kv_pairs = map(lambda v: v.split('=', 1), self._container.attrs['Config']['Env']) return {k: v for k, v in kv_pairs} @property def status(self): return self._container.status def reload(self): return self._container.reload() def kill(self, signal=None): return self._container.kill(signal) def remove(self, *args, **kwargs): raise RuntimeError('Do not remove this container manually. It will be removed automatically by py.test after the test finishes.') def logs(self): return self._container.logs().decode('utf-8') def get_files(self, path): ''' Retrieve files from a container at a given path. This is meant for extracting log files from a container where it is not using the docker logging capabilities. ''' archive_iter, _ = self._container.get_archive(path) archive_stream = io.BytesIO() [archive_stream.write(chunk) for chunk in archive_iter] archive_stream.seek(0) archive = tarfile.TarFile(fileobj=archive_stream) files = {} for info in archive.getmembers(): if not info.isfile(): continue reader = archive.extractfile(info.name) files[info.name] = reader.read().decode('utf-8') return files def get_open_tcp_ports(self): ''' Gets all TCP sockets in the LISTEN state ''' netstat = self._container.exec_run('cat /proc/net/tcp')[1].decode('utf-8').strip() ports = [] for line in netstat.split('\n'): # Not interested in empty lines if not line: continue line = line.split() # Only interested in listen sockets if line[3] != '0A': continue ports.append(str(int(line[1].split(':', 1)[1], 16))) return ports def get_open_udp_ports(self): ''' Gets all UDP sockets in the LISTEN state ''' netstat = self._container.exec_run('cat /proc/net/udp')[1].decode('utf-8').strip() ports = [] for line in netstat.split('\n'): # Not interested in empty lines if not line: continue line = line.split() # Only interested in listen sockets if line[3] != '0A': continue ports.append(str(int(line[1].split(':', 1)[1], 16))) return ports PK!HG^(44pytest_docker_tools-0.0.3.dist-info/entry_points.txt.,I-.14JON-/)#䔦gqqPK!HNO)pytest_docker_tools-0.0.3.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,zd&Y)r$[)T&UrPK!H!r{>,pytest_docker_tools-0.0.3.dist-info/METADATA[koǒ>cA*GYZ\Y׺7K0Lh^d~LIr]șz. $KdOhU(~]&<fUz+i[U0~]wE4cqZƋвBd^cXTB\LM&i1E*BwB]a,ԇkU!'uo#yVӣ<1<RYjLo~NvU5ժgI'Y$*??yߺ:l mV)2Zc<'oRi3 ;+ͺƊ@BYu_FgyZ(_;W5W\Iiq|,G2zTUQOe ``+UskCV8릺Éi+)E{4 L4؏ 7Xlzt4)]WՉUr'}ZHO$BUje\ a23I&hgⲺkr϶nQw*oXAʊP-%`լ0):Q%\Bɥhd.MT l8W~/A*C&z\hUؙykfBfGl ̈́3~JZ:W ̺-O IBۥE+ Pu'[)c5gg2قҝ:>^&+y̵<cTY9CzLG V&( xA4sh rjrVĈWihNltJbsI;61rđBl෥meatk?˔%$k[eg"efSiRaK3ɟ-x3"AŒYJRXgX4UA왱{ﵥMoy3LA.] ()b^]p)ZZ<1HJUKvԆ]R}>Lֲd* zkWP[!!Cq`Xhgja _G# %6P8Ɩ.Åfų_EQz"MW:3h1=fd :eQd* 96B63.OFub6W> h%s,~q0:KGE,|zQxz^ޜ8>^\ѲIxUnA+G2C6~Ц5\V#UP8٤JKmT>ܟ=tbWPy0yYMZB2Cl~Qe = B> r1]*1~.J㼬} l^)]wT+UöW `L˱ (;XeFO{- b6>ؚ8@9Ses*?Q=Knjncz~dMb0103t7.i[B;fׯ OD$ +?a.1kw*Q )@,퐒8n3$?PPeCdj8n V$*3kU! GͧNá@0H+g 褧e±8u,0q&Zs\\F. #*yځY,L81a&ɲ\'OĕE_JdY=ܡ=  ,k}+bЪBm.Q-ck{m=9,lձ؍Jz@G56-:X`Z9b6k\kFM tJic45A)V*wB&w/@dm ul̢赍 t˩+bHIXF~gjO!?&%aCX&X;Ƅ o-2ԁř3v):ކr}*{7.$s`X (H C11Q!%_/ Sv ;cHjoo`hämv،{k/~i")wd SR3`Ť4+k>Sé9$̉2'1mgH*5$" %.D4X2,6f}3'x}/i(,us)uylڿN|woğC{O+y sH;L5{2%\i'WNx.4UyU~OmJ1i' aH: V0(,-2X:Guٙ>yQ3`a"!\(\ g`µeǮ.d<бdy0vЗU[O̔yu',Y21߳“m\Z%:ggu2hhq&q(qHbA<_X|ݴK#g?2Q6w)'g]Wje&^5ew6Pn@}餮ʇp4^w8겭}jKI>jF33#խYsYxP5,pL4c蕡KDMd- XINZ_J@i4VdR'Wn?F>B$2put@PP L߼iAE9S`xM`4Tj2|WZʖA,&(g"ݒJ@nJ>jNNR՜*tV_Fyd>쩪=-jx6xVG^,}yu5Nd)MdF`̗p:um^-Eƕ]?rYmŹ$i!K̫jFOԬ^]'*|UQW+r5=M_=;zMP([@f74x;wgȍ<՜uqJ8mxpm88cr1`dTmU^ NWr0+p#ksXfȆUsFߛ@*iȢMqF՛7 e`\F,|?$2n:Fqөb%04 76 /^ 'h}RGŻQR9qmq:ؗI/H<ܱ9Q P¹W)ܭֶA"l|-L) v}{Qmژ%^Tθ+4WԒ`0_%_x5|0Sm`ojw<9 ܠ@1w0z8FN۷0\^,^ErCL/H/Yx+.% ]ф{4e.7}> ZFɲ ۨeUNh^}dV|;r񝐍Q؆8Ӧ%h {[:a:I=tuחx츌'AаY4*i y2BDډxF8;zAK'gjǴCM ܇/v !F^ޜ_LޞPc'^h4(g$u}0WFdB''qx|tdk %2m~w܋(,zH0n#ewF3j%~pH%=392<;_zye0$W:9'WZ}rQ5!^rvP6VÇqae1s׍}N}6%CaG5VlUU:_][{v7;>lY}TT2PmUJ*_mAFYU.aljHn>G_}&e4 RlzdG'ul޶–Кz[ z,S_&P3ֆ7c* eL 6z pjo0zL@R+BT_|ez8l) ?GW# qؓq18C\R=DH;Yif>ɣn.B;PԌd}/N J?\hnLR#_\?I-7zyROxvL ˾ڔz8V÷iNxK[30d%#m[ɣeY\|gJ4+!'{nܦč?o *ەaٽ5]@ܜɃ.cOy}~ysqvzsr/ym ƁkFn#^alBЛMJ+,/ݠp)c4nf[hFO/j%Tndcn?T?"/<Nc.6;Qu#l@^/S#},w٪怠J6(P3ذ&;jA[Q65Wsdpl/9 <ˑ[`'F[-ޛ&e0Ƕ<1hf.}tM+tySf~Y[NzL3=' 1f mPK!Hw9[^*pytest_docker_tools-0.0.3.dist-info/RECORDɖ@} `" &'DK6Rq(~dӉ r_V5s1!gexR \}M$Y& E΂|fn/퓥z*tJOwIPQգzݭ MwTF4j5igLiʬ~8SamXL}u+i&eVB_FRrY `Tu<'#F僬zqrg :;s|i7HMsU mG|tQb;ud쫵V]g&Ai"BHɒ(Rg[^bNεfG C$Q?#搈ZH$8oi\AuC| ~u|1մ,%r 6յGLY=F^^<3:3۵[j΃AϙHRAq|&UV$" /cjmڒ\7X.(/?v]' Cs͋1oyc.c*i`{Ps4-DL'yQ3('.%'gMZ=d[><`>uSy.kbJ ^riZEw" aх,~.j]閎τ4@xYJW'LAHa99Q*mn}~PKZLL'apytest_docker_tools/__init__.pyPKqL![pytest_docker_tools/plugin.pyPK(LL`͵pytest_docker_tools/utils.pyPK\L)pytest_docker_tools/factories/__init__.pyPKm_L|P?&pytest_docker_tools/factories/build.pyPKTL 8vv* pytest_docker_tools/factories/container.pyPKi_Lūuj&cpytest_docker_tools/factories/fetch.pyPKg_LlHH(apytest_docker_tools/factories/network.pyPKd_Lʐq22'pytest_docker_tools/factories/volume.pyPK(LL'kBB(fpytest_docker_tools/wrappers/__init__.pyPK;SLtt)pytest_docker_tools/wrappers/container.pyPK!HG^(443pytest_docker_tools-0.0.3.dist-info/entry_points.txtPK!HNO)#4pytest_docker_tools-0.0.3.dist-info/WHEELPK!H!r{>,4pytest_docker_tools-0.0.3.dist-info/METADATAPK!Hw9[^*}Ipytest_docker_tools-0.0.3.dist-info/RECORDPKL