|
1 | | -import asyncio |
2 | | -import atexit |
3 | 1 | import importlib |
4 | 2 | import os |
5 | | -import platform |
6 | | -import sys |
7 | | -import tempfile |
8 | | -from collections.abc import Callable, Generator |
| 3 | +from collections.abc import Callable |
9 | 4 | from dataclasses import dataclass |
10 | 5 | from pathlib import Path |
11 | 6 | from types import ModuleType |
|
14 | 9 |
|
15 | 10 | root_path = Path(__file__).resolve().parent |
16 | 11 | inputs_path = root_path.joinpath("inputs") |
17 | | -output_path_reference = root_path.joinpath("output_reference") |
18 | | -output_path_betterproto = root_path.joinpath("output_betterproto") |
19 | | -output_path_betterproto_pydantic = root_path.joinpath("output_betterproto_pydantic") |
20 | | - |
21 | | - |
22 | | -def get_files(path, suffix: str) -> Generator[str, None, None]: |
23 | | - for r, dirs, files in os.walk(path): |
24 | | - for filename in [f for f in files if f.endswith(suffix)]: |
25 | | - yield os.path.join(r, filename) |
26 | 12 |
|
27 | 13 |
|
28 | 14 | def get_directories(path): |
29 | 15 | for root, directories, files in os.walk(path): |
30 | 16 | yield from directories |
31 | 17 |
|
32 | 18 |
|
33 | | -async def protoc( |
34 | | - path: str | Path, |
35 | | - output_dir: str | Path, |
36 | | - reference: bool = False, |
37 | | - pydantic_dataclasses: bool = False, |
38 | | -): |
39 | | - path: Path = Path(path).resolve() |
40 | | - output_dir: Path = Path(output_dir).resolve() |
41 | | - python_out_option: str = "python_betterproto_out" if not reference else "python_out" |
42 | | - |
43 | | - if pydantic_dataclasses: |
44 | | - plugin_path = Path("src/betterproto2/plugin/main.py") |
45 | | - |
46 | | - if "Win" in platform.system(): |
47 | | - with tempfile.NamedTemporaryFile("w", encoding="UTF-8", suffix=".bat", delete=False) as tf: |
48 | | - # See https://stackoverflow.com/a/42622705 |
49 | | - tf.writelines( |
50 | | - [ |
51 | | - "@echo off", |
52 | | - f"\nchdir {os.getcwd()}", |
53 | | - f"\n{sys.executable} -u {plugin_path.as_posix()}", |
54 | | - ] |
55 | | - ) |
56 | | - |
57 | | - tf.flush() |
58 | | - |
59 | | - plugin_path = Path(tf.name) |
60 | | - atexit.register(os.remove, plugin_path) |
61 | | - |
62 | | - command = [ |
63 | | - sys.executable, |
64 | | - "-m", |
65 | | - "grpc.tools.protoc", |
66 | | - f"--plugin=protoc-gen-custom={plugin_path.as_posix()}", |
67 | | - "--experimental_allow_proto3_optional", |
68 | | - "--custom_opt=pydantic_dataclasses", |
69 | | - f"--proto_path={path.as_posix()}", |
70 | | - f"--custom_out={output_dir.as_posix()}", |
71 | | - *[p.as_posix() for p in path.glob("*.proto")], |
72 | | - ] |
73 | | - else: |
74 | | - command = [ |
75 | | - sys.executable, |
76 | | - "-m", |
77 | | - "grpc.tools.protoc", |
78 | | - f"--proto_path={path.as_posix()}", |
79 | | - f"--{python_out_option}={output_dir.as_posix()}", |
80 | | - *[p.as_posix() for p in path.glob("*.proto")], |
81 | | - ] |
82 | | - proc = await asyncio.create_subprocess_exec( |
83 | | - *command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE |
84 | | - ) |
85 | | - stdout, stderr = await proc.communicate() |
86 | | - return stdout, stderr, proc.returncode |
87 | | - |
88 | | - |
89 | 19 | @dataclass |
90 | 20 | class TestCaseJsonFile: |
91 | 21 | json: str |
|
0 commit comments