测试转移到py中, 检查测试结果

run_test.py

@@ -9,39 +9,179 @@ import asyncio
 from watchfiles import awatch, watch
 from termcolor import colored
 from datetime import datetime
+import orjson
+import os
+
+
+async def assert_sql(sql, target):
+    p = await subprocess.create_subprocess_exec(
+        "xmake",
+        "run",
+        "sql-parser",
+        sql,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    stdout, stderr = await p.communicate()
+    if stderr:
+        print(colored(stderr, "yellow"))
+    if b"error" in stdout:
+        print(stdout.decode("utf-8"))
+        print(datetime.now(), "-" * 40)
+        assert False, "sql-parser error"
+    try:
+        output = orjson.loads(stdout)
+    except Exception as e:
+        output = {"error": e, "output": stdout.decode("utf-8")}
+
+    assert (
+        output == target
+    ), f"""{colored("sql-parser error", "red")}
+input: {colored(sql, "yellow")}
+expect: {colored(target, "green")}
+actual: {colored(output, "red")}
+"""
+
+
+async def assert_sqls():
+    await assert_sql(
+        "create table asd;", [{"type": "create_table", "table_name": "asd", "cols": []}]
+    )
+    await assert_sql(
+        "create table tb (col1 INT, col2 string, col3 FLOAT);",
+        [
+            {
+                "type": "create_table",
+                "table_name": "tb",
+                "cols": [
+                    {
+                        "type": "create_column",
+                        "column_name": "col1",
+                        "data_type": "INT",
+                        "primary_key": False,
+                    },
+                    {
+                        "type": "create_column",
+                        "column_name": "col2",
+                        "data_type": "string",
+                        "primary_key": False,
+                    },
+                    {
+                        "type": "create_column",
+                        "column_name": "col3",
+                        "data_type": "FLOAT",
+                        "primary_key": False,
+                    },
+                ],
+            }
+        ],
+    )
+    await assert_sql(
+        """
+    create table tb1 (
+        col1 int primary key, 
+        col2 FLOAT
+    );
+    """,
+        [
+            {
+                "type": "create_table",
+                "table_name": "tb1",
+                "cols": [
+                    {
+                        "type": "create_column",
+                        "column_name": "col1",
+                        "data_type": "int",
+                        "primary_key": True,
+                    },
+                    {
+                        "type": "create_column",
+                        "column_name": "col2",
+                        "data_type": "FLOAT",
+                        "primary_key": False,
+                    },
+                ],
+            }
+        ],
+    )
+    await assert_sql(
+        """
+        create table tb2 (
+            x float,
+            y int,
+            z int
+        );
+        """,
+        [
+            {
+                "type": "create_table",
+                "table_name": "tb2",
+                "cols": [
+                    {
+                        "type": "create_column",
+                        "column_name": "x",
+                        "data_type": "float",
+                        "primary_key": False,
+                    },
+                    {
+                        "type": "create_column",
+                        "column_name": "y",
+                        "data_type": "int",
+                        "primary_key": False,
+                    },
+                    {
+                        "type": "create_column",
+                        "column_name": "z",
+                        "data_type": "int",
+                        "primary_key": False,
+                    },
+                ],
+            }
+        ],
+    )
+    print(colored("all test right!", 'green'))
 
 
 async def on_modified(event):
-    sqls = open('./target.sql').read().split('\n\n')
-    p = await subprocess.create_subprocess_shell("xmake", stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    p = await subprocess.create_subprocess_shell(
+        "xmake", stdout=subprocess.PIPE, stderr=subprocess.PIPE
+    )
     stdout, _ = await p.communicate()
-    if b'error' in stdout:
-        print(stdout.decode('utf-8'))
-        print(datetime.now(), '-' * 40)
+    if b"error" in stdout:
+        print(stdout.decode("utf-8"))
+        print(datetime.now(), "-" * 40)
         return
 
-    for sql in sqls:
-        print(f'run sql: \n{colored(sql, "yellow")}\n')
-        p = await subprocess.create_subprocess_exec(
-            'xmake', 'run', 'sql-parser', sql, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+    try:
+        await assert_sqls()
+    except Exception as e:
+        print(e)
+    print(datetime.now(), "-" * 40)
+
+
+async def restart():
+    async for _ in awatch(__file__):
+        print("restart")
+        os.execl("/bin/python", Path(__file__).as_posix(), Path(__file__).as_posix())
+
+        open("/tmp/restart-watch.sh", "w").write(
+            f"kill {os.getpid()} && python {__file__}"
         )
+        os.system(f"bash /tmp/restart-watch.sh")
 
-        for t in ['stderr', 'stdout']:
-            content = (await eval(f'p.{t}').read()).decode('utf-8')
-            if content:
-                print(colored(t + ':', 'green'))
-                print(content)
-        return_code = await p.wait()
-        
-        if return_code != 0:
-            print(f'{return_code=}')
-            break
-    print(datetime.now(), '-' * 40)
 
-async def main():
-    async for changes in awatch('src'):
+async def watch_src():
+    async for changes in awatch("src"):
         await asyncio.wait_for(on_modified(changes), 20)
 
 
+async def main():
+    try:
+        await assert_sqls()
+    except Exception as e:
+        print(e)
+    await asyncio.gather(restart(), watch_src())
+
+
 if __name__ == "__main__":
     asyncio.run(main())