pyentrypoint/tests/pyentrypoint_test.py

# Tests using pytest
from __future__ import absolute_import
from __future__ import unicode_literals

import fnmatch
import os
from multiprocessing import Process

from yaml import load
from yaml import Loader

from pyentrypoint import DockerLinks
from pyentrypoint import Entrypoint

LINKS = [
    'test1',
    'test2',
    'test3',
    'test4',
]


def test_all_links():
    links = DockerLinks()
    all_links = links.links()

    assert len(all_links) == 4
    for _, item in all_links.items():
        assert len(set(item["names"]).intersection(LINKS))


def test_filtering():
    links = DockerLinks()
    test1 = links.links("test1")

    assert len(test1) == 1

    test2_and_3 = links.links("test2", "test3")

    assert len(test2_and_3) == 2

    test4_and_5 = links.links("test4", "notexist")
    assert len(test4_and_5) == 1

    test5 = links.links("notexist")
    assert len(test5) == 0


def test_env():
    links = DockerLinks()
    env = links.links("test1", "test3")

    for _, item in env.items():
        assert item["environment"]["FOO"] == "bar"


def test_ports():
    links = DockerLinks()

    ports = links.links('test1', 'test2')

    for _, item in ports.items():
        if 'test1' in item["names"]:
            assert item["ports"]["800"]['protocol'] == 'tcp'
            assert item["ports"]["8001"]['protocol'] == 'udp'
        else:
            assert item["ports"]["800"]['protocol'] == 'udp'
            assert item["ports"]["8001"]['protocol'] == 'tcp'


def test_entrypoint_links():
    entry = Entrypoint(conf='configs/base.yml')
    links = entry.config.links

    assert len(links.all) == 4

    assert len(links.test1) == 2

    assert links.test2_800.port == 800


def test_containers():
    links = DockerLinks()
    ctns = links.to_containers()

    assert len(ctns) == 4

    for ctn in ctns:
        if 'test1' in ctn.names:
            assert ctn.environ['FOO'] == 'bar'
            assert len(ctn.links) == 2
        if 'test2' in ctn.names:
            assert len(ctn.links) == 2
        if 'test3' in ctn.names:
            assert ctn.environ['FOO'] == 'bar'
            assert len(ctn.links) == 0
        if 'test4' in ctn.names:
            assert len(ctn.links) == 0

        # Test sorted names
        int(ctn.id, base=16)
        assert len(ctn.names[0]) <= len(ctn.names[1])


def test_templates():
    test_confs = ['configs/base.yml']
    for test_conf in test_confs:
        entry = Entrypoint(conf='configs/base.yml')

        conf = entry.config

        entry.apply_conf()

        for _, config_file in conf.get_templates():
            with open(config_file, mode='r') as r:
                test = load(stream=r, Loader=Loader)

            assert len(set(test['All links'])) == 4
            assert len(set(test['All links 1'])) == 2
            assert len(set(test['All links 2'])) == 2

            assert fnmatch.fnmatch(test['Links 2 800'][0], 'udp://*:800')

            # test environment
            assert test['All environ']['FOO'] == 'bar'
            assert test['All links 2 environ']['FOO'] == 'bar'

            test_names = [
                'test1',
                'test2',
                'test3',
                'test4',
            ]

            # test names
            for test_name in test_names:
                assert test_name in test['All names']

            # test id
            for id in test['ID']:
                int(id, base=16)

            # test env
            assert test['ENV']['SECRET'] == 'nothing'
            assert test['ENVIRON']['SECRET'] == 'nothing'


def test_conf_commands():
    entry = Entrypoint(conf='configs/base.yml')

    for cmd in entry.config.pre_conf_commands:
        entry.run_conf_cmd(cmd)
    for cmd in entry.config.post_conf_commands:
        entry.run_conf_cmd(cmd)

    with open('/tmp/OK') as f:
        assert f.readline().startswith('TEST')

    with open('/tmp/OKOK') as f:
        assert f.readline().startswith('TEST2')

    with open('/tmp/OKOKOK') as f:
        assert f.readline().startswith('TEST3')


def test_command():
    run = [
        #  ((Process instance), (file to check), (uid), (gid))
        (Process(target=Entrypoint(
            conf='configs/base.yml',
            args=['-c', 'echo OK > /tmp/CMD']).launch),
            '/tmp/CMD', 1000, 1000),
        (Process(target=Entrypoint(
            conf='configs/base.yml',
            args=['bash', '-c', 'echo ${SECRET}OK > /tmp/CMD2']).launch),
            '/tmp/CMD2', 1000, 1000),
        (Process(target=Entrypoint(
            conf='configs/usernames.yml',
            args=['bash', '-c', 'echo OK > /tmp/CMD3']).launch),
            '/tmp/CMD3', 33, 33),
        (Process(target=Entrypoint(
            conf='configs/unhandled.yml',
            args=['bash', '-c', 'echo OK > /tmp/CMD4']).launch),
            '/tmp/CMD4', 0, 0),
        (Process(target=Entrypoint(
            conf='/dontexist',
            args=['bash', '-c', 'echo OK > /tmp/CMD5']).launch),
            '/tmp/CMD5', 0, 0),
    ]

    for proc, test, uid, gid in run:
        proc.start()
        proc.join()
        with open(test) as f:
            assert f.readline().startswith('OK')
        assert os.stat(test).st_uid == uid
        assert os.stat(test).st_gid == gid


def test_config_file():
    os.environ['ENTRYPOINT_CONFIG'] = 'configs/base.yml'
    entry = Entrypoint(conf='configs/base.yml')

    assert entry.config.has_config

    del os.environ['ENTRYPOINT_CONFIG']


def test_force_config():
    entry = Entrypoint(conf='configs/base.yml')
    assert not entry.should_config

    os.environ['ENTRYPOINT_FORCE'] = 'True'
    assert entry.should_config

    del os.environ['ENTRYPOINT_FORCE']


def test_display_raw():
    entry = Entrypoint(conf='configs/base.yml')
    assert not entry.raw_output

    os.environ['ENTRYPOINT_RAW'] = 'True'
    assert entry.raw_output

    del os.environ['ENTRYPOINT_RAW']