test_examples.py

import collections
import re
import subprocess
import time
from typing import Any, Optional, Sequence

import pytest

from kopf.testing import KopfRunner


def test_all_examples_are_runnable(mocker, with_crd, exampledir, caplog):

    # If the example has its own opinion on the timing, try to respect it.
    # See e.g. /examples/99-all-at-once/example.py.
    example_py = exampledir / 'example.py'
    e2e_startup_time_limit = _parse_e2e_value(str(example_py), 'E2E_STARTUP_TIME_LIMIT')
    e2e_startup_stop_words = _parse_e2e_value(str(example_py), 'E2E_STARTUP_STOP_WORDS')
    e2e_cleanup_time_limit = _parse_e2e_value(str(example_py), 'E2E_CLEANUP_TIME_LIMIT')
    e2e_cleanup_stop_words = _parse_e2e_value(str(example_py), 'E2E_CLEANUP_STOP_WORDS')
    e2e_creation_time_limit = _parse_e2e_value(str(example_py), 'E2E_CREATION_TIME_LIMIT')
    e2e_creation_stop_words = _parse_e2e_value(str(example_py), 'E2E_CREATION_STOP_WORDS')
    e2e_deletion_time_limit = _parse_e2e_value(str(example_py), 'E2E_DELETION_TIME_LIMIT')
    e2e_deletion_stop_words = _parse_e2e_value(str(example_py), 'E2E_DELETION_STOP_WORDS')
    e2e_tracebacks = _parse_e2e_value(str(example_py), 'E2E_TRACEBACKS')
    e2e_success_counts = _parse_e2e_value(str(example_py), 'E2E_SUCCESS_COUNTS')
    e2e_failure_counts = _parse_e2e_value(str(example_py), 'E2E_FAILURE_COUNTS')
    e2e_test_creation = _parse_e2e_presence(str(example_py), r'@kopf.on.create\(')
    e2e_test_highlevel = _parse_e2e_presence(str(example_py), r'@kopf.on.(create|update|delete)\(')

    # check whether there are mandatory deletion handlers or not
    m = re.search(r'@kopf\.on\.delete\((\s|.*)?(optional=(\w+))?\)', example_py.read_text(), re.M)
    requires_finalizer = False
    if m:
        requires_finalizer = True
        if m.group(2):
            requires_finalizer = not eval(m.group(3))

    # Skip the e2e test if the framework-optional but test-required library is missing.
    m = re.search(r'import kubernetes', example_py.read_text(), re.M)
    if m:
        pytest.importorskip('kubernetes')

    # To prevent lengthy sleeps on the simulated retries.
    mocker.patch('kopf.reactor.handling.DEFAULT_RETRY_DELAY', 1)

    # To prevent lengthy threads in the loop executor when the process exits.
    mocker.patch('kopf.config.WatchersConfig.default_stream_timeout', 10)

    # Run an operator and simulate some activity with the operated resource.
    with KopfRunner(['run', '--standalone', '--verbose', str(example_py)], timeout=60) as runner:

        # Give it some time to start.
        _sleep_till_stopword(caplog=caplog,
                             delay=e2e_startup_time_limit,
                             patterns=e2e_startup_stop_words or ['Client is configured'])

        # Trigger the reaction. Give it some time to react and to sleep and to retry.
        subprocess.run("kubectl apply -f examples/obj.yaml", shell=True, check=True)
        _sleep_till_stopword(caplog=caplog,
                             delay=e2e_creation_time_limit,
                             patterns=e2e_creation_stop_words)

        # Trigger the reaction. Give it some time to react.
        subprocess.run("kubectl delete -f examples/obj.yaml", shell=True, check=True)
        _sleep_till_stopword(caplog=caplog,
                             delay=e2e_deletion_time_limit,
                             patterns=e2e_deletion_stop_words)

    # Give it some time to finish.
    _sleep_till_stopword(caplog=caplog,
                         delay=e2e_cleanup_time_limit,
                         patterns=e2e_cleanup_stop_words or ['Hung tasks', 'Root tasks'])

    # Verify that the operator did not die on start, or during the operation.
    assert runner.exception is None
    assert runner.exit_code == 0

    # There are usually more than these messages, but we only check for the certain ones.
    # This just shows us that the operator is doing something, it is alive.
    if requires_finalizer:
        assert '[default/kopf-example-1] Adding the finalizer' in runner.stdout
    if e2e_test_creation:
        assert '[default/kopf-example-1] Creation event:' in runner.stdout
    if requires_finalizer:
        assert '[default/kopf-example-1] Deletion event:' in runner.stdout
    if e2e_test_highlevel:
        assert '[default/kopf-example-1] Deleted, really deleted' in runner.stdout
    if not e2e_tracebacks:
        assert 'Traceback (most recent call last):' not in runner.stdout

    # Verify that once a handler succeeds, it is never re-executed again.
    handler_names = re.findall(r"Handler '(.+?)' succeeded", runner.stdout)
    if e2e_success_counts is not None:
        checked_names = [name for name in handler_names if name in e2e_success_counts]
        name_counts = collections.Counter(checked_names)
        assert name_counts == e2e_success_counts
    else:
        name_counts = collections.Counter(handler_names)
        assert set(name_counts.values()) == {1}

    # Verify that once a handler fails, it is never re-executed again.
    handler_names = re.findall(r"Handler '(.+?)' failed (?:permanently|with an exception. Will stop.)", runner.stdout)
    if e2e_failure_counts is not None:
        checked_names = [name for name in handler_names if name in e2e_failure_counts]
        name_counts = collections.Counter(checked_names)
        assert name_counts == e2e_failure_counts
    else:
        name_counts = collections.Counter(handler_names)
        assert not name_counts


def _parse_e2e_value(path: str, name: str) -> Any:
    with open(path, 'rt', encoding='utf-8') as f:
        name = re.escape(name)
        text = f.read()
        m = re.search(fr'^{name}\s*=\s*(.+)$', text, re.M)
        return eval(m.group(1)) if m else None


def _parse_e2e_presence(path: str, pattern: str) -> bool:
    with open(path, 'rt', encoding='utf-8') as f:
        text = f.read()
        m = re.search(pattern, text, re.M)
        return bool(m)


def _sleep_till_stopword(
        caplog,
        delay: float,
        patterns: Sequence[str] = (),
        *,
        interval: Optional[float] = None,
) -> bool:
    patterns = list(patterns or [])
    delay = delay or (10.0 if patterns else 1.0)
    interval = interval or min(1.0, max(0.1, delay / 10.))
    started = time.perf_counter()
    found = False
    while not found and time.perf_counter() - started < delay:
        for message in list(caplog.messages):
            if any(re.search(pattern, message) for pattern in patterns):
                found = True
                break
        else:
            time.sleep(interval)
    return found