114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from .didactyl_client import DidactylClient
|
|
|
|
|
|
@dataclass
|
|
class AgentProcess:
|
|
binary_path: str
|
|
config_path: str
|
|
api_port: int = 8485
|
|
api_bind: str = "127.0.0.1"
|
|
debug_level: int = 5
|
|
log_file: Optional[str] = None
|
|
base_url: Optional[str] = None
|
|
|
|
def __post_init__(self) -> None:
|
|
self.binary_path = str(Path(self.binary_path))
|
|
self.config_path = str(Path(self.config_path))
|
|
self.log_file = self.log_file or "tests/results/agent_debug.log"
|
|
scheme = "https"
|
|
self.base_url = self.base_url or f"{scheme}://{self.api_bind}:{self.api_port}"
|
|
self.process: Optional[subprocess.Popen[str]] = None
|
|
|
|
def _command(self) -> list[str]:
|
|
return [
|
|
self.binary_path,
|
|
"--config",
|
|
self.config_path,
|
|
"--debug",
|
|
str(self.debug_level),
|
|
"--api-port",
|
|
str(self.api_port),
|
|
"--api-bind",
|
|
self.api_bind,
|
|
]
|
|
|
|
def start(self, timeout: float = 30.0) -> bool:
|
|
if self.is_alive():
|
|
return True
|
|
|
|
env = os.environ.copy()
|
|
env["DIDACTYL_LOG_FILE"] = str(self.log_file)
|
|
|
|
Path(self.log_file).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
self.process = subprocess.Popen(
|
|
self._command(),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
env=env,
|
|
)
|
|
|
|
client = DidactylClient(base_url=self.base_url, timeout=2.0, verify_tls=False)
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if self.process and self.process.poll() is not None:
|
|
return False
|
|
try:
|
|
data = client.status()
|
|
if data.get("success"):
|
|
return True
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.5)
|
|
return False
|
|
|
|
def stop(self, timeout: float = 10.0) -> bool:
|
|
if not self.process:
|
|
return True
|
|
if self.process.poll() is not None:
|
|
return True
|
|
|
|
try:
|
|
self.process.send_signal(signal.SIGTERM)
|
|
self.process.wait(timeout=timeout)
|
|
return True
|
|
except subprocess.TimeoutExpired:
|
|
self.process.kill()
|
|
self.process.wait(timeout=5)
|
|
return False
|
|
|
|
def restart(self, timeout: float = 30.0) -> bool:
|
|
self.stop()
|
|
return self.start(timeout=timeout)
|
|
|
|
def is_alive(self) -> bool:
|
|
return self.process is not None and self.process.poll() is None
|
|
|
|
def pid(self) -> Optional[int]:
|
|
return None if not self.process else self.process.pid
|
|
|
|
def return_code(self) -> Optional[int]:
|
|
return None if not self.process else self.process.poll()
|
|
|
|
def read_pipes(self) -> tuple[str, str]:
|
|
if not self.process:
|
|
return "", ""
|
|
out = ""
|
|
err = ""
|
|
if self.process.stdout:
|
|
out = self.process.stdout.read() or ""
|
|
if self.process.stderr:
|
|
err = self.process.stderr.read() or ""
|
|
return out, err
|