feat: convert joj1 test dir
This commit is contained in:
parent
782ee0af72
commit
75bb6d6667
|
@ -1,4 +1,4 @@
|
|||
from typing import Any, List, Optional
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import rtoml
|
||||
from pydantic import BaseModel, Field
|
||||
|
@ -30,8 +30,8 @@ class Cmd(BaseModel):
|
|||
proc_limit: int = Field(0, serialization_alias="procLimit")
|
||||
cpu_rate_limit: int = Field(0, serialization_alias="cpuRateLimit")
|
||||
cpu_set_limit: str = Field("", serialization_alias="cpuSetLimit")
|
||||
copy_in: dict[str, CmdFile] = Field({}, serialization_alias="copyIn")
|
||||
copy_in_cached: dict[str, str] = Field({}, serialization_alias="copyInCached")
|
||||
copy_in: Dict[str, CmdFile] = Field({}, serialization_alias="copyIn")
|
||||
copy_in_cached: Dict[str, str] = Field({}, serialization_alias="copyInCached")
|
||||
copy_in_dir: str = Field(".", serialization_alias="copyInDir")
|
||||
copy_out: List[str] = Field([], serialization_alias="copyOut")
|
||||
copy_out_cached: List[str] = Field([], serialization_alias="copyOutCached")
|
||||
|
@ -57,8 +57,8 @@ class OptionalCmd(BaseModel):
|
|||
proc_limit: Optional[int] = Field(None, serialization_alias="procLimit")
|
||||
cpu_rate_limit: Optional[int] = Field(None, serialization_alias="cpuRateLimit")
|
||||
cpu_set_limit: Optional[str] = Field(None, serialization_alias="cpuSetLimit")
|
||||
copy_in: Optional[dict[str, CmdFile]] = Field(None, serialization_alias="copyIn")
|
||||
copy_in_cached: Optional[dict[str, str]] = Field(
|
||||
copy_in: Optional[Dict[str, CmdFile]] = Field(None, serialization_alias="copyIn")
|
||||
copy_in_cached: Optional[Dict[str, str]] = Field(
|
||||
None, serialization_alias="copyInCached"
|
||||
)
|
||||
copy_in_dir: Optional[str] = Field(None, serialization_alias="copyInDir")
|
||||
|
@ -99,7 +99,7 @@ class ExecutorConfig(BaseModel):
|
|||
|
||||
class ParserConfig(BaseModel):
|
||||
name: str
|
||||
with_: dict[str, Any] = Field(..., serialization_alias="with")
|
||||
with_: Dict[str, Any] = Field(..., serialization_alias="with")
|
||||
|
||||
|
||||
class StageConfig(BaseModel):
|
||||
|
|
|
@ -1,13 +1,30 @@
|
|||
import json
|
||||
import os
|
||||
from typing import Any, List
|
||||
from typing import Any, Dict, List, Tuple
|
||||
|
||||
import pytest
|
||||
import rtoml
|
||||
|
||||
from joj3_config_generator.models import RepoConfig, TaskConfig
|
||||
from tests.convert.utils import read_convert_files
|
||||
from tests.utils import safe_id
|
||||
|
||||
|
||||
def get_test_cases() -> List[tuple[str, RepoConfig, TaskConfig, dict[str, Any]]]:
|
||||
def read_convert_files(root: str) -> Tuple[RepoConfig, TaskConfig, Dict[str, Any]]:
|
||||
repo_toml_path = os.path.join(root, "repo.toml")
|
||||
task_toml_path = os.path.join(root, "task.toml")
|
||||
result_json_path = os.path.join(root, "task.json")
|
||||
with open(repo_toml_path) as repo_file:
|
||||
repo_toml = repo_file.read()
|
||||
with open(task_toml_path) as task_file:
|
||||
task_toml = task_file.read()
|
||||
with open(result_json_path) as result_file:
|
||||
expected_result: Dict[str, Any] = json.load(result_file)
|
||||
repo_obj = rtoml.loads(repo_toml)
|
||||
task_obj = rtoml.loads(task_toml)
|
||||
return RepoConfig(**repo_obj), TaskConfig(**task_obj), expected_result
|
||||
|
||||
|
||||
def get_test_cases() -> List[Tuple[str, RepoConfig, TaskConfig, Dict[str, Any]]]:
|
||||
test_cases = []
|
||||
tests_dir = os.path.dirname(os.path.realpath(__file__))
|
||||
for dir_name in os.listdir(tests_dir):
|
||||
|
@ -18,8 +35,8 @@ def get_test_cases() -> List[tuple[str, RepoConfig, TaskConfig, dict[str, Any]]]
|
|||
return test_cases
|
||||
|
||||
|
||||
@pytest.fixture(params=get_test_cases(), ids=lambda x: x[0])
|
||||
@pytest.fixture(params=get_test_cases(), ids=safe_id)
|
||||
def test_case(
|
||||
request: pytest.FixtureRequest,
|
||||
) -> tuple[RepoConfig, TaskConfig, dict[str, Any]]:
|
||||
return request.param[1:] # return repo, task, expected_result
|
||||
) -> Tuple[RepoConfig, TaskConfig, Dict[str, Any]]:
|
||||
return request.param[1:]
|
||||
|
|
|
@ -1,10 +1,10 @@
|
|||
from typing import Any
|
||||
from typing import Any, Dict, Tuple
|
||||
|
||||
from joj3_config_generator.convert import convert
|
||||
from joj3_config_generator.models import RepoConfig, TaskConfig
|
||||
|
||||
|
||||
def test_convert(test_case: tuple[RepoConfig, TaskConfig, dict[str, Any]]) -> None:
|
||||
def test_convert(test_case: Tuple[RepoConfig, TaskConfig, Dict[str, Any]]) -> None:
|
||||
repo, task, expected_result = test_case
|
||||
result = convert(repo, task).model_dump(by_alias=True)
|
||||
assert result == expected_result
|
|
@ -1,22 +0,0 @@
|
|||
import json
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import rtoml
|
||||
|
||||
from joj3_config_generator.models import RepoConfig, TaskConfig
|
||||
|
||||
|
||||
def read_convert_files(root: str) -> tuple[RepoConfig, TaskConfig, dict[str, Any]]:
|
||||
repo_toml_path = os.path.join(root, "repo.toml")
|
||||
task_toml_path = os.path.join(root, "task.toml")
|
||||
result_json_path = os.path.join(root, "task.json")
|
||||
with open(repo_toml_path) as repo_file:
|
||||
repo_toml = repo_file.read()
|
||||
with open(task_toml_path) as task_file:
|
||||
task_toml = task_file.read()
|
||||
with open(result_json_path) as result_file:
|
||||
expected_result: dict[str, Any] = json.load(result_file)
|
||||
repo_obj = rtoml.loads(repo_toml)
|
||||
task_obj = rtoml.loads(task_toml)
|
||||
return RepoConfig(**repo_obj), TaskConfig(**task_obj), expected_result
|
42
tests/convert_joj1/conftest.py
Normal file
42
tests/convert_joj1/conftest.py
Normal file
|
@ -0,0 +1,42 @@
|
|||
import json
|
||||
import os
|
||||
from typing import Any, Dict, List, Tuple
|
||||
|
||||
import pytest
|
||||
import rtoml
|
||||
|
||||
from joj3_config_generator.models import JOJ1Config, TaskConfig
|
||||
from tests.utils import safe_id
|
||||
|
||||
|
||||
def read_convert_joj1_files(root: str) -> Tuple[JOJ1Config, TaskConfig, Dict[str, Any]]:
|
||||
task_yaml_path = os.path.join(root, "task.yaml")
|
||||
task_toml_path = os.path.join(root, "task.toml")
|
||||
expected_json_path = os.path.join(root, "task.json")
|
||||
with open(task_yaml_path) as repo_file:
|
||||
task_yaml = repo_file.read()
|
||||
with open(task_toml_path) as task_file:
|
||||
task_toml = task_file.read()
|
||||
with open(expected_json_path) as result_file:
|
||||
expected_result: Dict[str, Any] = json.load(result_file)
|
||||
joj1_obj = rtoml.loads(task_yaml)
|
||||
task_obj = rtoml.loads(task_toml)
|
||||
return JOJ1Config(**joj1_obj), TaskConfig(**task_obj), expected_result
|
||||
|
||||
|
||||
def get_test_cases() -> List[Tuple[str, JOJ1Config, TaskConfig, Dict[str, Any]]]:
|
||||
test_cases = []
|
||||
tests_dir = os.path.dirname(os.path.realpath(__file__))
|
||||
for dir_name in os.listdir(tests_dir):
|
||||
dir_path = os.path.join(tests_dir, dir_name)
|
||||
if os.path.isdir(dir_path) and dir_name != "__pycache__":
|
||||
joj1, task, expected_result = read_convert_joj1_files(dir_path)
|
||||
test_cases.append((dir_name, joj1, task, expected_result))
|
||||
return test_cases
|
||||
|
||||
|
||||
@pytest.fixture(params=get_test_cases(), ids=safe_id)
|
||||
def test_case(
|
||||
request: pytest.FixtureRequest,
|
||||
) -> Tuple[JOJ1Config, TaskConfig, Dict[str, Any]]:
|
||||
return request.param[1:]
|
9
tests/convert_joj1/test_convert_joj1_cases.py
Normal file
9
tests/convert_joj1/test_convert_joj1_cases.py
Normal file
|
@ -0,0 +1,9 @@
|
|||
from typing import Any, Dict, Tuple
|
||||
|
||||
from joj3_config_generator.models import JOJ1Config, TaskConfig
|
||||
|
||||
|
||||
def test_convert_joj1(test_case: Tuple[JOJ1Config, TaskConfig, Dict[str, Any]]) -> None:
|
||||
joj1, task, expected_result = test_case
|
||||
result: Dict[str, Any] = {}
|
||||
assert result == expected_result
|
7
tests/utils.py
Normal file
7
tests/utils.py
Normal file
|
@ -0,0 +1,7 @@
|
|||
from typing import Any
|
||||
|
||||
|
||||
def safe_id(x: Any) -> str:
|
||||
if not x or not isinstance(x, (tuple, list)) or len(x) == 0:
|
||||
return "no_test_cases"
|
||||
return str(x[0])
|
Loading…
Reference in New Issue
Block a user