Расширенное тестирование (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)

**Тестовый код, может содержать баги** ```python3 # 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 всё равно перезапускаются на каждом тесте.