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())