sql-parser/run_sql_parser_test.py

from pathlib import Path
import sys
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from watchdog.events import LoggingEventHandler
import asyncio.subprocess as subprocess
import asyncio
from watchfiles import Change, awatch, watch
from termcolor import colored
from datetime import datetime
import orjson
import os
import tempfile
from tests_config import sql_parser_tests


async def run_and_output(
    *args: str, timeout=5
) -> tuple[bytes, bytes]:
    p = await subprocess.create_subprocess_exec(
        *args,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=timeout)
    return stdout, stderr

async def rebuild() -> bool:
    stdout, _ = await run_and_output('xmake')
    if b"error" in stdout:
        print(stdout.decode("utf-8"))
        print(datetime.now(), "-" * 40)
        return False
    else:
        return True

async def assert_sql(sql: str, expected: dict):
    stdout, stderr = await run_and_output('xmake', 'run', "sql-parser", sql)

    if b"error" in stdout:
        print(stdout.decode("utf-8"))
        print(datetime.now(), "-" * 40)
        print(f'other: {colored(stderr.decode("utf-8"), "yellow")}')
        assert False, "sql-parser error"

    try:
        output = orjson.loads(stdout)
    except Exception as e:
        output = {"error": e, "output": stdout.decode("utf-8")}
    open("/tmp/temp/test.py", "wb").write(
        f'"{sql}"\n\n'.encode("utf-8")
        + orjson.dumps(output, option=orjson.OPT_INDENT_2)
        + (b"\n\n" + stderr).replace(b"\n", b"\n# ")
    )
    assert (
        output == expected
    ), f"""{colored("sql-parser error", "red")}
input: {colored(sql, "yellow")}
expect: {colored(expected, "green")}
actual: {colored(output, "red")}
other: {colored(stderr.decode("utf-8"), "yellow")}

"""


async def assert_sqls():
    for sql, excepted in sql_parser_tests:
        await assert_sql(sql, excepted)


async def on_parser_modified():
    print(datetime.now(), colored("run parser tests...", "yellow"))

    try:
        await assert_sqls()
    except Exception as e:
        print(e)
    else:
        print(datetime.now(), colored("all parser tests right!", "green"))


async def on_checker_modified():
    print(datetime.now(), colored("run checker tests...", "yellow"))
    stdout, stderr = await run_and_output('xmake', 'run', "sql-checker", "-f", "test.cpp", "-g", "2")
    print(stdout.decode("utf-8"))
    print(datetime.now(), colored("all checker tests right!", "green"))


async def restart():
    async for _ in awatch(__file__):
        print("restart")
        os.execl("/bin/python", Path(__file__).as_posix(), Path(__file__).as_posix())


async def watch_parser():
    async for changes in awatch("./src/parser.y", "./src/parser.l"):
        if await rebuild():
            await asyncio.wait_for(on_parser_modified(), 10)


async def watch_checker():
    async for changes in awatch("./src/checker.cpp", "./src/checker.h"):
        if await rebuild():
            await on_checker_modified()


async def main():
    await asyncio.gather(
        restart(),
        watch_parser(),
        watch_checker(),
        on_parser_modified(),
        on_checker_modified(),
    )


if __name__ == "__main__":
    asyncio.run(main())