refactoring stage 3
- run command
This commit is contained in:
136
run.py
136
run.py
@@ -1,16 +1,9 @@
|
||||
#!/usr/bin/env python3
|
||||
import base64
|
||||
import os
|
||||
import pathlib
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum, auto, unique
|
||||
from html import escape, unescape
|
||||
from enum import Enum, unique
|
||||
|
||||
import click
|
||||
import yaml
|
||||
@@ -68,7 +61,7 @@ def _parse_range_string(range_str: str) -> list[int]:
|
||||
raise ValueError("Invalid range format")
|
||||
|
||||
|
||||
def _parse_range_string_list(str_list) -> list[int]:
|
||||
def _parse_range_string_list(str_list: list[str]) -> list[int]:
|
||||
"""
|
||||
여러 범위 문자열을 정수 리스트로 변환하는 함수; 이때 중복은 제거함.
|
||||
* e.g. ["1..3", "5", "7..9"] -> [1, 2, 3, 5, 7, 8, 9]
|
||||
@@ -241,7 +234,7 @@ class StateManager:
|
||||
cls._instance = super().__new__(cls)
|
||||
return cls._instance
|
||||
|
||||
def load(self) -> dict:
|
||||
def _load(self) -> dict:
|
||||
if self._state is not None:
|
||||
return self._state
|
||||
try:
|
||||
@@ -260,18 +253,18 @@ class StateManager:
|
||||
yaml.safe_dump(self._state, f)
|
||||
|
||||
def get_space(self, lang: str) -> dict | None:
|
||||
state = self.load()
|
||||
state = self._load()
|
||||
return state.get("space", {}).get(lang)
|
||||
|
||||
def set_space(self, lang: str, data: dict) -> None:
|
||||
state = self.load()
|
||||
state = self._load()
|
||||
if "space" not in state:
|
||||
state["space"] = {}
|
||||
state["space"][lang] = data
|
||||
self._state = state
|
||||
|
||||
def clear_space(self, lang: str) -> None:
|
||||
state = self.load()
|
||||
state = self._load()
|
||||
if "space" in state and lang in state["space"]:
|
||||
del state["space"][lang]
|
||||
self._state = state
|
||||
@@ -396,17 +389,22 @@ def register(location: str):
|
||||
"""
|
||||
storage_manager = StorageManager()
|
||||
if storage_manager.check_location(location):
|
||||
click.echo(f"Location '{location}' is already registered.")
|
||||
click.secho(
|
||||
f"Location '{location}' is already registered.", fg="yellow", bold=True
|
||||
)
|
||||
else:
|
||||
storage_manager.register_location(location)
|
||||
click.echo(f"Location '{location}' registered successfully.")
|
||||
click.secho(
|
||||
f"Location '{location}' registered successfully.", fg="cyan", bold=True
|
||||
)
|
||||
|
||||
|
||||
@click.command(name="create")
|
||||
@click.argument("target", type=str, nargs=1, required=True)
|
||||
@click.option("--completed/--no-completed", "-c/-nc", default=False, help="Mark as completed")
|
||||
def create(target: str, completed: bool):
|
||||
"""
|
||||
지정된 target을 storage에 생성하는 명령어
|
||||
지정된 location을 storage에 생성하는 명령어
|
||||
"""
|
||||
loc, prob, lang = _dispatch_target(target)
|
||||
storage_manager = StorageManager()
|
||||
@@ -423,62 +421,47 @@ def create(target: str, completed: bool):
|
||||
|
||||
|
||||
@click.command(name="run")
|
||||
@click.argument("target", type=str, nargs=1, required=True)
|
||||
@click.argument("lang", type=str, nargs=1, required=True)
|
||||
@click.option("--testcase", "-t", default=["1"], multiple=True)
|
||||
@click.option("--verbose/--no-verbose", "-v/-nv", default=True)
|
||||
def run(target: str, testcase: str, verbose: bool):
|
||||
def run(lang: str, testcase: list[str], verbose: bool):
|
||||
"""
|
||||
지정된 언어로 빌드 및 실행하는 명령어
|
||||
"""
|
||||
loc, prob, from_language = _dispatch_target(target)
|
||||
|
||||
if from_language is Language.UNDEFINED:
|
||||
# Language 확인
|
||||
target_language: Language = Language.convert_name(lang)
|
||||
|
||||
if target_language is Language.UNDEFINED:
|
||||
raise click.ClickException("Undefined Language Exception")
|
||||
try:
|
||||
with open("state.yml", "r", encoding="utf-8") as f:
|
||||
state = yaml.safe_load(f)
|
||||
except FileNotFoundError as e:
|
||||
raise click.ClickException(e)
|
||||
except yaml.YAMLError as e:
|
||||
raise click.ClickException(e)
|
||||
|
||||
if "space" not in state:
|
||||
raise click.ClickException("State(state.yml) Exception")
|
||||
elif not state["space"]:
|
||||
click.secho("state.space has no member", fg="yellow", bold=True)
|
||||
state["space"] = {}
|
||||
# load된 state가 있는지 확인
|
||||
space = StateManager().get_space(target_language.value)
|
||||
|
||||
lang_space_state: dict | None
|
||||
if from_language.value not in state["space"]:
|
||||
lang_space_state = None
|
||||
else:
|
||||
lang_space_state = state["space"][from_language.value]
|
||||
|
||||
if lang_space_state is None or not lang_space_state:
|
||||
raise click.ClickException("Run Exception: There are no pre-defined state")
|
||||
if space is None:
|
||||
raise click.ClickException(
|
||||
f"No loaded state for language '{target_language.value}'"
|
||||
)
|
||||
|
||||
# 테케 분석
|
||||
try:
|
||||
tcs: list[int] = _parse_range_string_list(target)
|
||||
except ValueError as e:
|
||||
raise click.ClickException(e)
|
||||
tcs: list[int] = _parse_range_string_list(testcase)
|
||||
|
||||
# build
|
||||
|
||||
try:
|
||||
s = subprocess.run(
|
||||
from_language.build_command,
|
||||
target_language.build_command,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=from_language.working_dir,
|
||||
cwd=target_language.working_dir,
|
||||
)
|
||||
print(s.stderr)
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
click.echo(">>>>>> [Error while BUILD process] >>>>>>")
|
||||
click.secho(">>>>>> [BUILD TIME ERROR] >>>>>>", fg="red", bold=True)
|
||||
raise click.ClickException(e.stderr)
|
||||
|
||||
# execution for each test case
|
||||
for tc in tcs:
|
||||
if not (
|
||||
os.path.isfile(f"{TC_DIR}/{tc}.in") and os.path.isfile(f"{TC_DIR}/{tc}.out")
|
||||
@@ -487,30 +470,43 @@ def run(target: str, testcase: str, verbose: bool):
|
||||
else:
|
||||
f_in_path = f"{TC_DIR}/{tc}.in"
|
||||
f_out_path = f"{TC_DIR}/{tc}.out"
|
||||
with open(f_in_path, "r", encoding="utf-8") as f_in:
|
||||
try:
|
||||
stdout = subprocess.run(
|
||||
from_language.executable_command,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
|
||||
with open(f_in_path, "r", encoding="utf-8") as f_in, \
|
||||
open(f_out_path, "r", encoding="utf-8") as f_out:
|
||||
expected = f_out.read().strip()
|
||||
|
||||
proc = subprocess.Popen(
|
||||
target_language.executable_command,
|
||||
stdin=f_in,
|
||||
).stdout
|
||||
except subprocess.CalledProcessError as e:
|
||||
click.echo(">>>>>> [Error while RUNNING process] >>>>>>")
|
||||
click.echo(f"{e.stderr}")
|
||||
click.echo(f"returncode: {e.returncode}")
|
||||
stdout = ""
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
|
||||
if not verbose:
|
||||
for line in proc.stdout: # type: ignore
|
||||
print(line, end="")
|
||||
|
||||
stdout, stderr = proc.communicate()
|
||||
|
||||
if proc.returncode != 0:
|
||||
click.secho(">>>>>> [RUN TIME ERROR] >>>>>>", fg="red", bold=True)
|
||||
click.secho(f"{stderr}", fg="red")
|
||||
click.secho(f"returncode: {proc.returncode}", fg="red")
|
||||
continue
|
||||
flag: bool = False
|
||||
with open(f_out_path, "r", encoding="utf-8") as f_out:
|
||||
flag = stdout.strip() == f_out.read().strip()
|
||||
click.echo(f"{tc:02d}: {flag}")
|
||||
|
||||
flag = stdout.strip() == expected
|
||||
symbol = click.style("✓", fg="green") if flag else click.style("✗", fg="red")
|
||||
click.echo(f"{tc:02d}: {symbol}")
|
||||
|
||||
if verbose:
|
||||
click.echo("===" * 3)
|
||||
click.echo(stdout)
|
||||
click.echo("===" * 3)
|
||||
click.secho("===" * 3, fg="bright_black")
|
||||
click.secho(f"[INPUT]", fg="cyan", bold=True)
|
||||
with open(f_in_path, "r") as f:
|
||||
click.echo(f.read())
|
||||
click.secho(f"[ACTUAL]", fg="cyan", bold=True)
|
||||
click.echo(stdout.strip() if stdout else "(empty)")
|
||||
click.secho("===" * 3, fg="bright_black")
|
||||
|
||||
|
||||
@click.command(name="load")
|
||||
@@ -687,9 +683,7 @@ def export(from_: str, completed: bool, copy: bool):
|
||||
if completed:
|
||||
s_is_completed = True
|
||||
|
||||
dest_path: str = (
|
||||
f"{STORAGE_DIR}/{s_loc}/{from_language.value}{'/completed' if s_is_completed else ''}/{s_file}.{from_language.value}"
|
||||
)
|
||||
dest_path: str = f"{STORAGE_DIR}/{s_loc}/{from_language.value}{'/completed' if s_is_completed else ''}/{s_file}.{from_language.value}"
|
||||
source_path: str = (
|
||||
f"{SRC_SPACE_DIR}/src-{from_language.value}/src/main.{from_language.value}"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user