conftest.py 1.66 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45
import random
import socket

import pytest

from kresd import Kresd


def is_port_free(port, ip=None, ip6=None):
    def check(family, type_, dest):
        sock = socket.socket(family, type_)
        sock.bind(dest)
        sock.close()

    try:
        if ip is not None:
            check(socket.AF_INET, socket.SOCK_STREAM, (ip, port))
            check(socket.AF_INET, socket.SOCK_DGRAM, (ip, port))
        if ip6 is not None:
            check(socket.AF_INET6, socket.SOCK_STREAM, (ip6, port, 0, 0))
            check(socket.AF_INET6, socket.SOCK_DGRAM, (ip6, port, 0, 0))
    except OSError as exc:
        if exc.errno == 98:  # address alrady in use
            return False
        else:
            raise
    return True


@pytest.fixture
def kresd(tmpdir):
    ip = '127.0.0.1'
    ip6 = '::1'

    def make_port():
        for _ in range(10):  # max attempts
            port = random.randint(1024, 65535)
            if is_port_free(port, ip, ip6):
                return port
        raise RuntimeError("No available port found!")

    port = make_port()
    tls_port = make_port()
    with Kresd(tmpdir, port, tls_port, ip, ip6) as kresd:
        yield kresd
46 47 48
        # TODO: add verbose option?
        # with open(kresd.logfile_path) as log:
        #     print(log.read())  # display log for debugging purposes
49 50 51 52 53 54 55 56


@pytest.fixture(params=[
    'ip_tcp_socket',
    'ip6_tcp_socket',
    'ip_tls_socket',
    'ip6_tls_socket',
])
57 58 59 60 61 62 63 64 65 66 67 68
def make_kresd_sock(request, kresd):
    sock_func = getattr(kresd, request.param)

    def _make_kresd_sock():
        return sock_func()

    return _make_kresd_sock


@pytest.fixture
def kresd_sock(make_kresd_sock):
    return make_kresd_sock()