Расширенное тестирование (kks test 2.0)

[vvd170501]

Идея ещё с прошлого года, но до её реализации я так и не дошёл.
Возможно, заинтересует кого-нибудь из слушателей АКОСа этого года.

Во второй половине курса было много задач, в которых использовались аргументы командной строки и/или предполагалась работа с файловой системой.
kks test умеет генерировать только входные данные для программы (в теории, с флагом -v генератор может изменять ФС, но это изначально не предполагалось).

Есть предложение расширить функционал тестирования, чтобы можно было:

• Использовать для запуска решения аргументы командной строки
• Настраивать внешние условия для тестов
  (например, создавать файлы/директории, с которыми должна работать программа, а после тестирования удалять их)
• Настраивать переменные окружения для тестируемой прогаммы (в том году было было 2 или 3 задачи, в которых они использовались, но это достаточно просто реализуется)

В такой реализации gen.py может выглядеть примерно так:

import os
import sys
import random
imoort typing as t


"""
Sample problem:

1) The program receives a file name `f` in argv[1].
2) An IP is passed via `HOST` environment variable.
3) Input contains a number `n`.
The program must:
1) Read 32-bit integers from `f` (in text mode).
2) Myltiply each by `n`.
3) Send the multiplied integers to `HOST:9999` in binary form (little-endian).
4) Write to stdout the number of processed integers.
"""


def _filename(test: int) -> str:
    random.seed(test)
    return random.choice(['./a.txt', './b.txt'])


def _multiplier(test: int) -> int:
    random.seed(test)
    return random.randint(1, 100)


def _data(test: int) -> t.List[int]:
    random.seed(test)
    return [randint(1, 1000) for i in range(10)]


def gen_input(test: int) -> None:
    """Generates test input.

    Test input should be written to stdout.
    It will be passed to stdin of your program.
    """
    random.seed(test)
    print(_multiplier(test))


def args(test: int) -> t.List[str]:
    """Generates command-line arguments for the tested program.

    If this function returns ['a', 'b'],
    your program will be launched like this: `./a.out a b`. 
    """
    return [_filename(test)]


def env(test):
    """Generates test environment variables."""
    random.seed(test)
    env = os.environ.copy()
    env['HOST'] = f'127.0.0.{random.randint(1, 254)}'
    return env


def external(test):
    """Sets up and cleans external conditions for tests.

    Possible examples:
    - changing the FS (creating/modifying files),
    - changing the working directory
    - setting limits (setrlimit, cgroups?)
    - rtc.
    """
    # ========== Setup ==========
    filename = _filename(test)
    data = _data(test)
    with open(filename, 'w') as f:
        f.write(' '.join(data) + '\n')
    # ========== Testing ==========
    yield
    # program is being tested
    # checks are performed
    # ========== Cleanup ==========
    os.unlink(filename)


def check(test):
    """Checks external changes.

    If results are incorrect, should raise an error.
    Output from the program and solve.py will be compared separately.
    """
    # Assuming a server is listening at `HOST:9999`
    # and is writing received integers to `/tmp/server_log` in text form. 
    with open('/tmp/server_log', 'r') as f:
        # NOTE Not the most efficient way
        last_line = f.readlines()[-1]
    data = _data(test)
    modified_data = [int(x) for x in last_line.split()]
    multiplier = _multiplier(test)
    assert modified_data == [x * multiplier for x in data]


# Backwards compatibility
if __name__ == '__main__':
    t = int(sys.argv[1])
    gen_input(t)

[vvd170501]

Что стоит учесть в реализации

• Запуск отдельного процесса для генерации каждого теста - дорого (см. сравнение производительности ниже).
  Тем более, если каждый раз запускается новый интерпретатор и заново выполняет все import'ы.
  Поэтому есть смысл сделать gen.py импортируемым модулем с функциями, как в примере.
• Новые версии kks должны поддерживать старый формат gen.py. Т.е. нужно корректно обрабатывать случай, когда в gen.py нет нужных функций.
  Также можно сделать опциональными все функции, кроме gen_input, чтобы шаблон генератора оставался небольшим.
• Желательно, ятобы gen.py нового формата работал со всеми версиями kks (для тестов, в которых используется только stdin).
  В коде из примера это решается вызовом gen_input при запуске файла.
• В примере gen_input использует вывод в stdout. Почему так сделано:
  • Часто проще выводить тестовые данные по частям, чем собирать полную строку.
    Например, для числовой последовательности можно вызывать print в цикле вместо return ' '.join(map(str, data)).
  • gen.py старого формата при необходимости легко конвертируется в новый. Весь код достаточно перенести в gen_input.
• Если использовать stdout для gen_input (как в примере), придётся найти способ перехватывать вывод функции.
  Возможное решение - запустить отдельный процесс, который будет по запросу вызывать gen_input, и читать его вывод (см. пример ниже).

Сравнение производительности методов запуска генератора

Простой генератор:

# gen.py
import sys
import random

def gen_input(test):
    random.seed(test)
    print(random.randint(1, 1000))

if __name__ == '__main__':
    gen_input(int(sys.argv[1]))

Отдельный процесс для каждого теста (kks test -V запускает генератор примерно так):

# old.py
import sys
import subprocess

DBG = bool(sys.argv[2]) if len(sys.argv) > 2 else False

for test in range(int(sys.argv[1])):
    p = subprocess.run(['python3', './gen.py', str(test)], capture_output=True)
    if DBG:
        print(p.stdout.decode())

Вызов функции в цикле. Во время выполнения функции sys.stdout перенаправляется в буфер

# new.py
import sys
from contextlib import redirect_stdout
from io import BytesIO, TextIOWrapper

from gen import gen_input

DBG = bool(sys.argv[2]) if len(sys.argv) > 2 else False

for test in range(int(sys.argv[1])):
    with redirect_stdout(TextIOWrapper(BytesIO())) as out:
        gen_input(test)
    out.flush()
    test_input = out.buffer.getvalue()
    if DBG:
        print(test_input.decode())

Такая реализация может некорректно работать в многопоточном коде (какой-то вывод из других потоков может попасть в out вместо настоящего sys.stdout). Пока kks использует только один поток, проблем быть не должно.

Альтернативный вариант реализации (запуск цикла в отдельном процессе, перенаправление в pipe)

Тестовый код, может содержать баги

# new.py
import fcntl
import os
import sys
from signal import SIGINT

from gen import gen_input

DBG = bool(sys.argv[2]) if len(sys.argv) > 2 else False

# Probably UNIX-only

p_control_gen, p_control_main = os.pipe()
p_feedback_main, p_feedback_gen = os.pipe()
p_result_main, p_result_gen = os.pipe()

pid = os.fork()
if not pid:
    os.close(p_control_main)
    os.close(p_feedback_main)
    os.close(p_result_main)
    # possible flush of old stdout?
    sys.stdin = open(p_control_gen, 'r')
    sys.stdout = open(p_result_gen, 'w')
    feedback = open(p_feedback_gen, 'w')
    try:
        while True:
            test = int(input())  # Use binary?
            gen_input(test)
            sys.stdout.flush()
            feedback.write('1')
            feedback.flush()
    except KeyboardInterrupt: 
        pass
else:
    os.close(p_control_gen)
    os.close(p_feedback_gen)
    os.close(p_result_gen)
    control = open(p_control_main, 'w')
    result = p_result_main
    feedback = p_feedback_main
    flg = fcntl.fcntl(feedback, fcntl.F_GETFL)
    fcntl.fcntl(feedback, fcntl.F_SETFL, flg | os.O_NONBLOCK)
    flg = fcntl.fcntl(result, fcntl.F_GETFL)
    fcntl.fcntl(result, fcntl.F_SETFL, flg | os.O_NONBLOCK)

    for test in range(int(sys.argv[1])):
        control.write(str(test)+'\n')
        control.flush()
        data = []
        BUFSIZE = 4096
        while True:
            #chunk = result.read(BUFSIZE)
            chunk = None
            try:
                chunk = os.read(result, BUFSIZE)
                if not chunk:  # EOF
                    break
                data.append(chunk)
            except BlockingIOError:
                pass  # empty input / sync issues?
            if chunk is None or len(chunk) < BUFSIZE:
                try:
                    chk = os.read(feedback, 1)
                    if chk:
                        break
                except BlockingIOError:
                    pass  # more data available
        test_input= b''.join(data)
        if DBG:
            print(test_input.decode())
    os.close(result)
    os.close(feedback)
    os.kill(pid, SIGINT)
    os.wait()

Сравнение:

$ time python old.py 100
real    0m6.199s
user    0m4.387s
sys     0m1.593s

$ time python new.py 100
real    0m0.075s
user    0m0.057s
sys     0m0.017s

Результат - ускорение в ~50-80 раз.
При реальном тестировании можно рассчитывать на максимальное ускорение в ~2 раза, т.к. тестируемое решение и solve.py всё равно перезапускаются на каждом тесте.