PKYLpytest_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.6' __all__ = [ 'build', 'container', 'fetch', 'network', 'volume', ] PKlL&pytest_docker_tools/builder.pyimport textwrap import pytest from .templates import find_fixtures_in_params, resolve_fixtures_in_params def build_fixture_function(callable, kwargs): name = callable.__name__ docstring = getattr(callable, '__doc__', '').format(**kwargs) fixtures = find_fixtures_in_params(kwargs).union(set(('request', 'docker_client'))) fixtures_str = ','.join(fixtures) template = textwrap.dedent(f''' def {name}({fixtures_str}): \'\'\' {docstring} \'\'\' real_kwargs = resolve_fixtures_in_params(request, kwargs) return _{name}(request, docker_client, **real_kwargs) ''') globals = { 'resolve_fixtures_in_params': resolve_fixtures_in_params, f'_{name}': callable, 'kwargs': kwargs, } exec(template, globals) return globals[name] def fixture_factory(scope='function'): def inner(callable): def factory(*, scope='function', **kwargs): fixture_factory = build_fixture_function(callable, kwargs) pytest.fixture(scope=scope)(fixture_factory) return fixture_factory factory.__name__ = callable.__name__ factory.__doc__ = getattr(callable, '__doc__', '') return factory return inner PKZYL]%pytest_docker_tools/plugin.pyimport docker import pytest from .wrappers import Container @pytest.fixture(scope='session') def docker_client(request): ''' A Docker client configured from environment variables ''' 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 not rep.failed: return if 'request' not in item.funcargs: return for name, fixturedef in item.funcargs['request']._fixture_defs.items(): if not hasattr(fixturedef, 'cached_result'): continue fixture = fixturedef.cached_result[0] if isinstance(fixture, Container): rep.sections.append(( name + ': ' + fixture.name, fixture.logs(), )) PKZLs~33 pytest_docker_tools/templates.py''' # Template handling Fixture factories can take static strings: ```python redis = container( image='redis:latest', ) ``` But that is not useful when building multiple containers that need to reference one another or you need to parameterize the fixture. This module provides two facilities: The ability to reference fixtures using python string template notation and the ability to know what fixtures this will fetch in at test collection (and generation) time. For example: ``` def test_simple_resolve(request): # These are parameters declared at import time - they can be reevaluated in the context of multiple tests kwargs = { 'somekey': ['{pytestconfig.getoption("verbose")}'], } # This can be used in a fixture factory to fill in the templates resolved = resolve_fixtures_in_params(request, kwargs) # And then the test can access them pytestconfig = request.getfixturevalue('pytestconfig') assert resolved['somekey'][0] == str(pytestconfig.getoption("verbose")) } ``` In order to make fixtures generated by a fixture factory more seamless we need to know a fixtures dependencies at collection time. We have a helper to find them: def test_simple_find(): # These are parameters declared at import time - they can be reevaluated in the context of multiple tests kwargs = { 'somekey': ['{pytestconfig.getoption("verbose")}'], } dependencies = find_fixtures_in_params(kwargs) assert dependencies = set('pytestconfig') ''' import inspect from string import Formatter __all__ = [ 'find_fixtures_in_params', 'resolve_fixtures_in_params', ] class FixtureFormatter(Formatter): def __init__(self, request): self.request = request def get_value(self, key, args, kwargs): return self.request.getfixturevalue(key) class Renderer(object): def __init__(self, request): self.request = request def visit_value(self, val): if isinstance(val, str): return FixtureFormatter(self.request).format(val) elif callable(val): return val(*[self.request.getfixturevalue(f) for f in inspect.getargspec(val)[0]]) return val def visit_list(self, val): return [self.visit(v) for v in val] def visit_dict(self, mapping): return {self.visit(k): self.visit(v) for (k, v) in mapping.items()} def visit(self, value): if isinstance(value, dict): return self.visit_dict(value) elif isinstance(value, list): return self.visit_list(value) elif value: return self.visit_value(value) class FixtureFinder(object): def visit_value(self, val): if isinstance(val, str): for literal_text, format_spec, conversion, _ in Formatter().parse(val): if format_spec: yield format_spec.split('.')[0].split('[')[0] elif callable(val): yield from inspect.getargspec(val)[0] def visit_list(self, val): for v in val: yield from self.visit(v) def visit_dict(self, mapping): for k, v in mapping.items(): yield from self.visit(k) yield from self.visit(v) def visit(self, value): if isinstance(value, dict): yield from self.visit_dict(value) elif isinstance(value, list): yield from self.visit_list(value) elif value: yield from self.visit_value(value) def find_fixtures_in_params(value): ''' Walk an object and identify fixtures references in templates in strings. ''' finder = FixtureFinder() return set(finder.visit(value)) def resolve_fixtures_in_params(request, value): ''' Walk an object and resolve fixture values referenced in template strings. ''' renderer = Renderer(request) return renderer.visit(value) 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', ] PK nL""&pytest_docker_tools/factories/build.pyimport sys from pytest_docker_tools.builder import fixture_factory @fixture_factory(scope='session') def build(request, docker_client, **kwargs): ''' Docker image: built from "{path}" ''' sys.stdout.write(f'Building {kwargs["path"]}') try: image, logs = docker_client.images.build(**kwargs) 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 PKlL_pס*pytest_docker_tools/factories/container.pyfrom pytest_docker_tools.builder import fixture_factory from pytest_docker_tools.utils import wait_for_callable from pytest_docker_tools.wrappers import Container @fixture_factory() def container(request, docker_client, **kwargs): ''' Docker container: image={image} ''' kwargs.update({'detach': True}) raw_container = docker_client.containers.run(**kwargs) request.addfinalizer(lambda: raw_container.remove(force=True) and raw_container.wait(timeout=10)) container = Container(raw_container) wait_for_callable( f'Waiting for container to be ready', lambda: container.reload() or container.ready(), ) return container PK{tLRS&pytest_docker_tools/factories/fetch.pyimport sys from pytest_docker_tools.builder import fixture_factory @fixture_factory(scope='session') def fetch(request, docker_client, **kwargs): ''' Docker image: Fetched from {repository} ''' sys.stdout.write(f'Fetching {kwargs["repository"]}\n') image = docker_client.images.pull(**kwargs) # request.addfinalizer(lambda: docker_client.images.remove(image.id)) return image PKWoLtgd(pytest_docker_tools/factories/network.pyimport uuid from pytest_docker_tools.builder import fixture_factory @fixture_factory() def network(request, docker_client, **kwargs): ''' Docker network ''' name = kwargs.pop('name', 'pytest-{uuid}').format(uuid=str(uuid.uuid4())) print(f'Creating network {name}') network = docker_client.networks.create(name, **kwargs) request.addfinalizer(lambda: network.remove()) return network PKHoLBS'pytest_docker_tools/factories/volume.pyimport uuid from pytest_docker_tools.builder import fixture_factory @fixture_factory() def volume(request, docker_client, **kwargs): ''' Docker volume ''' name = kwargs.pop('name', 'pytest-{uuid}').format(uuid=str(uuid.uuid4())) print(f'Creating volume {name}') volume = docker_client.volumes.create(name, **kwargs) request.addfinalizer(lambda: volume.remove(True)) return volume PK(LL'kBB(pytest_docker_tools/wrappers/__init__.pyfrom .container import Container __all__ = [ 'Container', ] PKL*)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 [int(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 name(self): return self._container.name @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.6.dist-info/entry_points.txt.,I-.14JON-/)#䔦gqqPK!HNO)pytest_docker_tools-0.0.6.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,zd&Y)r$[)T&UrPK!HHκN,pytest_docker_tools-0.0.6.dist-info/METADATA\oGE'ƁT dtX9[4Ɏ3 |}UCR~I6gə~TWj~gF0UmT$OʜrӘNٍAhv'_ 6#WFe5JO,OREMj ozZ5o U#؀fU$h]*5hO DߨdF6²NYK 4"u)W]LW財ZM eQd&ݘ==`Y+P=YQ3JJJзWTiȇM Yi? ֤f*3ŧǧ%1 (f<.& kZx+kg- &{Dh,F2J'L#dPKb>)ȍID]4 ɼ1V;53-F5*GajZݲ J\ц n*/<9O+.ETsB–[f1UpC)zQf-1Q!&Bֱah ch2(oq\5 [GңiUNI"ê;ea,A>}.f7Q@'3w2,8xըo.;s`sp=xf;ۋyIZ J,φA[dz{nA2z67IbG#M?~_]>ytwS5uBh֥^, r&Ac9 iG. IlY'ee $RKpV`88 8aj^t4|q[V[3nYhL}_H(VldWaZ4>V-fadIcM xjO[رY+5ZW"+`O^h[XsV{9w¦,Fϊ-;eAbUTEN_^ȄY&Im}*&ai(Sj|ISKXY"k"mD1mNᒋ ;hFwn]u;ӽNaDVc!&rnBl n$P h98:] C..t3Bը[U=q+F$KyE0 S•@/Bz*9ŻN:Λ`[mF3e+1p@ƬUP RxVE1A)ḀMnix V $w^yJȽڅ{j 2C]6@?h\Z0Pn!Q(!ZAѧS ff&C{O# :ޒz5jǩJح9'p,)Hز!DwF2](; Iwy6(R֘AUMjS?rX%r:yOEp1IrrH99&[v2X Ku/_F>-56,'0ËA[ꛓ{O~?Z#)I9~ "Ly {3Y)ԃ`*:Ppe09\ !ΰ%c{02Y-S˼XatU=+JqH#_09]c ߀)OA͞L-9(HJ㉗9 L@•^Pfȟ$ 8G&?>RɲYeK*RI68mHg8p@E&9څ&Ohc̉asɯv~!cM0>QBaNw]. ƴLe0,Mݹ&Jfr<I}?OZ#"oR:h|< 2L8HĄpl㣐ޡ~V/.\ƥA%QQ+h:#I߀.[<Hyԁf{o9"-(!{4!I@ ;pyՅ`Xdy L^A02(ǚ-U (ǍMqʪtQ[s&gݯ H(ӠCTx5nBщ^N-J\\2g=V9q>bl˵w[=gmMjZ P! 4*b& .ϑ}Q|O{G0oKۉt#L9FPoCvPx^l}ȊD١ZӻنK9`%բy-9cu1SˋDHYZiko;co ~yCmnNqVҥ NKj?c 4uSM,%Ɩ}p`RcgHcJB;y@;攲䌗/I;tp2{c)~SLH z CQLc`_N8<Vׅl I?(l 8!;2(;'yv IB7  (0YB>^cK?l$ZiҝCKl9&2=s "=j Jq30!Ce Cm̟ 0j/~:p50iUD1tp35{(3dSm-֜uW=;M`?EuG=$mWvEJTgGv)H3S= E3D?̾hOPg)'‡U(7^ Gj%RzSƯᲙq̽ݹ2+.2܊ZW/S?jljf]DPL^ ~ TbTi.B/Owg+$/?av#6>XQPRNˋ}*DyDǀpPٹmftcEf;@^&B0Yc~qឺHѯ]n]rh{u˿i]4kVW+D˩޳nWȆs:W E\^έ{ԛQLB̽Vq"G}qvU \̀)d rM9W !.x(c o+M7 k<=TT% N@?5y:oKj`7LPg[[5?>`A =xj^fhu8sfFjU46bg<[덜oL4qoZ˕mO62xHzZXqrOn{xsr~Y$/|L'ă&1s}R% a'%r[KoT  \&ZgTylR ̴FW2 ٍ3J*t,H) &SsgWN,vp[;H-!mK]l]FG{tx=" p2S^ mނ/utxWTQ*J>~ڞ3 ɻ}kt-A~M0*ضB%IC2圮fC*ۄ?#VY'Nu:ulV(C?[;1wTx{f0xP~ջXήw4ŵBA-G\ٙZT\l>sAk.7󪑊7w9nF7r.e*`{#x(D΂ 5 <^}rO9,yBOFcZRGGuT^9Eouuf+\_b㥥{Qt5_V"2>_cYTlVӅT'.uy_ ]qܣpoh[wR?$&q*/ +"uk#/.t֜rȭv.;J7-Bqc㞨a@c.DǺ;z=7(>iyt:O dFjZB'߃TU?{[uwj9!yaE#{w-Z7GAm~ ʻ~ڄD>@Blt+_eᣛ0zV7vmF*1h,wIXGUB]S)>'o]<=l#@! uFlq}}%n[*8G깍?ܪ7|F4_< CtnӜ9+݈[]SwoOP b? [bk_%\k@0k:}K9ʬ(~qю__PK!HB/ *pytest_docker_tools-0.0.6.dist-info/RECORD˒H}? Y }1 B!A H=TLDM;1Dtosv&'qu1q$?ڙW h~~K( uo\&ZgCeD VNNAI.RԽ"NȖ,5afgЬ&?}KWfUV%@[djȋM׏&a|Y[U0ޫ)rFYTUmz9P87wj5%#ѕ\dfHQHl3l\^ɴ,#'NQ(.5!+bsFr:_h"Fٺn1hpiBA:|wS9v7P٢l_sPJ˓ꔇ f{/]ܐkg#QXm r^5ؙuTQd| II <1}-/F 坱}Fave^ÈD&ܕo,]oK1 q?q)y>P<_FaOjm쩇Ot[x8b+`&E~mKT@ <"ZnwC|9tCkI%,/T^L"A*én`{tWB{үAiP4qEöQb[k(~߃^"Y0pgQ0o߶-Elt; <rʊW:$ RGkwp Jc+7$p>Ka$בW>:ki])|~ PKYLpytest_docker_tools/__init__.pyPKlL&[pytest_docker_tools/builder.pyPKZYL]%}pytest_docker_tools/plugin.pyPKZLs~33 \ pytest_docker_tools/templates.pyPK(LL`͵pytest_docker_tools/utils.pyPK\L)pytest_docker_tools/factories/__init__.pyPK nL""&pytest_docker_tools/factories/build.pyPKlL_pס*tpytest_docker_tools/factories/container.pyPK{tLRS&]"pytest_docker_tools/factories/fetch.pyPKWoLtgd(3$pytest_docker_tools/factories/network.pyPKHoLBS'&pytest_docker_tools/factories/volume.pyPK(LL'kBB('pytest_docker_tools/wrappers/__init__.pyPKL*)y(pytest_docker_tools/wrappers/container.pyPK!HG^(44=pytest_docker_tools-0.0.6.dist-info/entry_points.txtPK!HNO)=pytest_docker_tools-0.0.6.dist-info/WHEELPK!HHκN,>pytest_docker_tools-0.0.6.dist-info/METADATAPK!HB/ *Tpytest_docker_tools-0.0.6.dist-info/RECORDPK X