PKeMpytest_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.11' __all__ = [ 'build', 'container', 'fetch', 'network', 'volume', ] PK`MK__pytest_docker_tools/builder.pyimport textwrap from .templates import find_fixtures_in_params, resolve_fixtures_in_params def build_fixture_function(callable, scope, wrapper_class, 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''' import pytest @pytest.fixture(scope=scope) def {name}({fixtures_str}): \'\'\' {docstring} \'\'\' real_kwargs = resolve_fixtures_in_params(request, kwargs) return _{name}(request, docker_client, wrapper_class=wrapper_class, **real_kwargs) ''') globals = { 'resolve_fixtures_in_params': resolve_fixtures_in_params, f'_{name}': callable, 'kwargs': kwargs, 'scope': scope, 'wrapper_class': wrapper_class, } exec(template, globals) return globals[name] def fixture_factory(scope='function'): wrapper_class = None def inner(callable): def factory(*, scope=scope, wrapper_class=wrapper_class, **kwargs): return build_fixture_function(callable, scope, wrapper_class, kwargs) factory.__name__ = callable.__name__ factory.__doc__ = getattr(callable, '__doc__', '') return factory return inner PKbM+44!pytest_docker_tools/exceptions.pyclass TimeoutError(Exception): pass class ContainerError(Exception): def __init__(self, container, *args, **kwargs): self._container = container super().__init__(*args, **kwargs) class ContainerFailed(ContainerError): pass class ContainerNotReady(ContainerError): pass PKD^M^݆pytest_docker_tools/plugin.pyimport docker import pytest from .exceptions import ContainerError 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 call.excinfo and isinstance(call.excinfo.value, ContainerError): container = call.excinfo.value._container rep.sections.append((container.name, container.logs())) 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(), )) PKnLs~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) PKu^Mkpytest_docker_tools/utils.pyimport sys import time from .exceptions import TimeoutError 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 TimeoutError(f'Timeout of {timeout}s exceeded') PKnL./pytest_docker_tools/contexts/scratch/Dockerfile# For volume seeding we want a near-empty volune, You can't 'docker create scratch' # and you can have an iamge with just `FROM scratch`. FROM scratch CMD ["/bin/sh"] PKnL)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`MOm&pytest_docker_tools/factories/build.pyimport sys from pytest_docker_tools.builder import fixture_factory @fixture_factory(scope='session') def build(request, docker_client, wrapper_class, **kwargs): ''' Docker image: built from "{path}" ''' # The docker build command now defaults to --rm=true, but docker-py doesnt # Let's do what docker build does by default kwargs.setdefault('rm', True) 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)) wrapper_class = wrapper_class or (lambda image: image) return wrapper_class(image) PK^M+zz*pytest_docker_tools/factories/container.pyfrom pytest_docker_tools.builder import fixture_factory from pytest_docker_tools.exceptions import ContainerNotReady, TimeoutError from pytest_docker_tools.utils import wait_for_callable from pytest_docker_tools.wrappers import Container @fixture_factory() def container(request, docker_client, wrapper_class, **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)) wrapper_class = wrapper_class or Container container = wrapper_class(raw_container) try: wait_for_callable('Waiting for container to be ready', container.ready) except TimeoutError: raise ContainerNotReady(container, 'Timeout while waiting for container to be ready') return container PK`ML"&pytest_docker_tools/factories/fetch.pyimport sys from pytest_docker_tools.builder import fixture_factory @fixture_factory(scope='session') def fetch(request, docker_client, wrapper_class, **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)) wrapper_class = wrapper_class or (lambda image: image) return wrapper_class(image) PK`M{u(pytest_docker_tools/factories/network.pyimport uuid from pytest_docker_tools.builder import fixture_factory @fixture_factory() def network(request, docker_client, wrapper_class, **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()) wrapper_class = wrapper_class or (lambda network: network) return wrapper_class(network) PK`Mc'pytest_docker_tools/factories/volume.pyimport io import os import tarfile import uuid from pytest_docker_tools.builder import fixture_factory def _populate_volume(docker_client, volume, seeds): fp = io.BytesIO() tf = tarfile.open(mode="w:gz", fileobj=fp) for path, contents in seeds.items(): ti = tarfile.TarInfo(path) if contents is None: ti.type = tarfile.DIRTYPE tf.addfile(ti) else: ti.size = len(contents) tf.addfile(ti, io.BytesIO(contents)) tf.close() fp.seek(0) image, logs = docker_client.images.build( path=os.path.join(os.path.dirname(__file__), '..', 'contexts/scratch'), rm=True, ) list(logs) container = docker_client.containers.create( image=image.id, volumes={ f'{volume.name}': {'bind': '/data'}, }, ) try: container.put_archive('/data', fp) finally: container.remove(force=True) @fixture_factory() def volume(request, docker_client, wrapper_class, **kwargs): ''' Docker volume ''' name = kwargs.pop('name', 'pytest-{uuid}').format(uuid=str(uuid.uuid4())) seeds = kwargs.pop('initial_content', {}) print(f'Creating volume {name}') volume = docker_client.volumes.create(name, **kwargs) request.addfinalizer(lambda: volume.remove(True)) if seeds: _populate_volume(docker_client, volume, seeds) wrapper_class = wrapper_class or (lambda volume: volume) return wrapper_class(volume) PKnL'kBB(pytest_docker_tools/wrappers/__init__.pyfrom .container import Container __all__ = [ 'Container', ] PK^M\j(L2R)D~A|Fmz},N QVF6*F5*/e-R5ׅ.hԢ LPŵ"WEcJ7KX?7sike2W{\r˦F8)&GW#H?։* Lӿ=}me},Z. Pֵ}rc <x綨 Ugkd2I+vV4{ Vlԫf(t"zIc\+XVYY~E#ֈc™>K U#bp0#SŹ8O+,(J M3Npn_‚fGG4Ovje+ZU¸eeUe:akxUEfI<tORC pܗX<-Wb $`ߔbvlVg 0puXaKT@Ual8~^(9NCtPsat3YÔ5ΐbjI|_F[3am 6mF p=V{W'mk<Yݞ-*C"+ Oɜ#*A$eO 0T@iRm6W 8)>Snq5kYDkK U4J1=} $LoruJa 5|9 !3xixSTg~r'}3C@u픧[,8ءtc⻼['/ȡzB%'#7$jxf7p_ ;h7'# &gp7"zx{@ j)K"›Rʑ ъ$m03/ d)fLyV f tM(>RxX qيj7BYimofǞp<r{u<ܞlalq? LQtVg7 {Fidї:gO￷m`Y|30V~/[MNrJ:Tā'ӠZj5n<PK߂ XD]fu1>cU4_C LKPֻVyլ;8oXGoWC|nsuI\4!.Xw0Yڅ cYIFdtME e^/Jo7؜I]PnaJ9qV؞ÖץNM S^Pf(?YSEQ f|0G؀`hAL'")Ŭl+YsN/:_A r}o3S-ׅ;1.L#Rw"n8sLA .@!Ϋf=OQUYE/Pz47y=!nqLv-niuKVcdB-jk@)/Da]疑?g95ftI+#+dPNcV8\s7xmrq?o/:v.RY`=[cWNǿ/wx~e٤> Ә:13.::vQT p*ԣh:/|_}X~2_x: _Qk/7G2\@|G7P8(5P)m6jjT>'*q]\VԜlHQLuykMF(rO\R1TOV3 Tmsf6aCJTԌ*| 3,ޒ!Jx[H L[{Ayn V"rj|V@aMAL/8JX;{9v&~(`2<0.Uf/)uVthBw¬>a1b0DT" !Iܴ"2׵ir7eu`:NVP2CeA S*LKeJLR df^ @vV.pL8E\J^4!/mzp wz{E3L(2{8"w)w.I;Na'+q Pi/- s|ެP uQ*?"P@.0I ~4XŖj01: Y^v8HnKo'Lwڹv7^5(?'_/bZHeI+K@X&l'D'GꬮA0U!f쀻5b]M 5hm(3^ܽ)U9h9 Xs4t}}$bh[e˽bn[!Ew))tCf~@P8V2ֿ a@(hra!7( k7~ >U3'A>%]-< [?ztQn%}1PDhcH,UJ>͘W,ʙQ4Ix [CA}adm*\R]ާX|K1鼵|4xDģ6uji$2B]?359^&2,b(=FQLcOKZŨz CmĜwpղk"͔7j!ī]vJʪG'Fb?*|p5zk;O ~w]s^ ,,XGfg)ir ѓ] %Pnܸ*= ,G&rX>|LPw]CG=z?|畂O9?]W]`]_5FFYֶ2P{pAd^3BcA k؍kp-@ P ;lp n-C *9[ P9Vf V:IK!/y̕饤 ~υ$} i.kkVeMj5](Β~5åd%^w'2En:Ы ßg(H7`]fQR8ҝB͉3u3'0mMTׅZMt=OE*@XcފDǪ^[dž%X7A_k_ ,a}=?@;5䇒k;>G|nGԢq_(9 ʛ5m +5'B iIR擱v$z0[aP\.YnC/=x kw=|SmX|_+0|8bGANF^Uܢ*7uxh ϤCm1}+&n0u^p@m#JJ.ss~Bľbd<d(, ¤hEVԃ?)޽<:rI\R<#¸~ATk܃|vsTI~ 35R-j(*A.Xu7ߖhԮow tĔ*bWF/+,ܓ|EmbR߸!oθ/+kY½_5Y3ij{6"}kV+O5*B b'ς_׋)6+$Ҵ^,%d*Ֆ bH }5Y8nz?KKt{t!9!¾38Gq*oCIϼWNΡ?RάxXj*l.̦de~ \ËMS=1>Y5x@\o0{H ӜoFdlۋm}t!!IF{fuWLd'`CSzVw>zH`Ȃ,9o΋kEwrG۶u 7m1jgsJV]jפː;mO%SnMn.pNK6ֈ4m"aJv{d.fޘ7}KE5&qSԤ{#k\A*.UЪ h]w?mID6  cc čk!ŢUV!bMM!k~`ꞺT=+r\-]MՃfa΋; R8?|kgIQ?PKeMpytest_docker_tools/__init__.pyPK`MK__\pytest_docker_tools/builder.pyPKbM+44!pytest_docker_tools/exceptions.pyPKD^M^݆jpytest_docker_tools/plugin.pyPKnLs~33 + pytest_docker_tools/templates.pyPKu^Mkpytest_docker_tools/utils.pyPKnL./pytest_docker_tools/contexts/scratch/DockerfilePKnL)pytest_docker_tools/factories/__init__.pyPK`MOm&!pytest_docker_tools/factories/build.pyPK^M+zz*i$pytest_docker_tools/factories/container.pyPK`ML"&+(pytest_docker_tools/factories/fetch.pyPK`M{u(Z*pytest_docker_tools/factories/network.pyPK`Mc',pytest_docker_tools/factories/volume.pyPKnL'kBB(2pytest_docker_tools/wrappers/__init__.pyPK^M