+ | {{ job.time_terminated }} |
{{ job.kitodo_details.task_id }} |
{{ job.kitodo_details.process_id }} |
- {% if job.workflow is defined %}{{ job.workflow }}{% endif %} |
+ {{ job.workflow }} |
{{ job.return_code }} {% if job.return_code == 0 %}(SUCCESS){% else %}(FAILURE){% endif %} |
{{ job.kitodo_details.processdir.name }} |
diff --git a/ocrdmonitor/sshps.py b/ocrdmonitor/sshps.py
deleted file mode 100644
index b37610c..0000000
--- a/ocrdmonitor/sshps.py
+++ /dev/null
@@ -1,45 +0,0 @@
-from __future__ import annotations
-
-import subprocess
-from pathlib import Path
-from typing import Protocol
-
-from ocrdmonitor.processstatus import PS_CMD, ProcessStatus
-
-
-class SSHConfig(Protocol):
- host: str
- port: int
- user: str
- keyfile: Path
-
-
-_SSH = (
- "ssh -o StrictHostKeyChecking=no -i '{keyfile}' -p {port} {user}@{host} '{ps_cmd}'"
-)
-
-
-def process_status(config: SSHConfig, process_group: int) -> list[ProcessStatus]:
- ssh_cmd = _build_ssh_command(config, process_group)
-
- result = subprocess.run(
- ssh_cmd,
- shell=True,
- universal_newlines=True,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- encoding="utf-8",
- )
-
- return ProcessStatus.from_ps_output(result.stdout)
-
-
-def _build_ssh_command(config: SSHConfig, process_group: int) -> str:
- ps_cmd = PS_CMD.format(process_group or "")
- return _SSH.format(
- port=config.port,
- keyfile=config.keyfile,
- user=config.user,
- host=config.host,
- ps_cmd=ps_cmd,
- )
diff --git a/ocrdmonitor/sshremote.py b/ocrdmonitor/sshremote.py
new file mode 100644
index 0000000..598255a
--- /dev/null
+++ b/ocrdmonitor/sshremote.py
@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import shlex
+from pathlib import Path
+from typing import Protocol
+
+from ocrdmonitor.processstatus import ProcessStatus
+
+
+class SSHConfig(Protocol):
+ host: str
+ port: int
+ user: str
+ keyfile: Path
+
+
+class SSHRemote:
+ def __init__(self, config: SSHConfig) -> None:
+ self._config = config
+
+ async def read_file(self, path: str) -> str:
+ result = await asyncio.create_subprocess_shell(
+ _ssh(self._config, f"cat {path}"),
+ stdout=asyncio.subprocess.PIPE,
+ )
+ await result.wait()
+
+ if not result.stdout:
+ return ""
+
+ return (await result.stdout.read()).decode()
+
+ async def process_status(self, process_group: int) -> list[ProcessStatus]:
+ pid_cmd = ProcessStatus.shell_command(process_group)
+ result = await asyncio.create_subprocess_shell(
+ _ssh(self._config, pid_cmd),
+ stdout=asyncio.subprocess.PIPE,
+ )
+
+ if await result.wait() > 0:
+ logging.error(
+ f"checking status of process {process_group} failed: {result.stderr}"
+ )
+ return []
+
+ if not result.stdout:
+ return []
+
+ output = (await result.stdout.read()).decode()
+ return ProcessStatus.from_shell_output(output)
+
+
+def _ssh(config: SSHConfig, cmd: str) -> str:
+ return shlex.join(
+ (
+ "ssh",
+ "-o",
+ "StrictHostKeyChecking=no",
+ "-i",
+ str(config.keyfile),
+ "-p",
+ str(config.port),
+ f"{config.user}@{config.host}",
+ *shlex.split(cmd),
+ )
+ )
diff --git a/tests/ocrdmonitor/server/test_job_endpoint.py b/tests/ocrdmonitor/server/test_job_endpoint.py
index 5715da9..58fa120 100644
--- a/tests/ocrdmonitor/server/test_job_endpoint.py
+++ b/tests/ocrdmonitor/server/test_job_endpoint.py
@@ -8,7 +8,7 @@
import pytest
from fastapi.testclient import TestClient
from httpx import Response
-from ocrdmonitor.ocrdcontroller import ProcessQuery
+from ocrdmonitor.ocrdcontroller import RemoteServer
from ocrdmonitor.ocrdjob import OcrdJob
from ocrdmonitor.processstatus import ProcessState, ProcessStatus
from ocrdmonitor.server.settings import OcrdControllerSettings
@@ -37,7 +37,7 @@ def running_ocrd_job(
expected_status = make_status(pid)
running_job = replace(JOB_TEMPLATE, pid=pid)
jobfile = write_job_file_for(running_job)
- patch_process_query(monkeypatch, expected_status)
+ patch_controller_remote(monkeypatch, expected_status)
yield running_job, expected_status
@@ -86,18 +86,20 @@ def make_status(pid: int) -> ProcessStatus:
return expected_status
-def patch_process_query(
+def patch_controller_remote(
monkeypatch: pytest.MonkeyPatch, expected_status: ProcessStatus
) -> None:
- def make_process_query(self: OcrdControllerSettings) -> ProcessQuery:
- def process_query_stub(process_group: int) -> list[ProcessStatus]:
- if process_group != expected_status.pid:
- raise ValueError(f"Unexpected process group {process_group}")
- return [expected_status]
+ def make_remote_stub(self: OcrdControllerSettings) -> RemoteServer:
+ class RemoteStub:
+ async def read_file(self, path: str) -> str:
+ return str(expected_status.pid)
- return process_query_stub
+ async def process_status(self, process_group: int) -> list[ProcessStatus]:
+ return [expected_status]
- monkeypatch.setattr(OcrdControllerSettings, "process_query", make_process_query)
+ return RemoteStub()
+
+ monkeypatch.setattr(OcrdControllerSettings, "controller_remote", make_remote_stub)
def assert_lists_completed_job(
@@ -106,6 +108,7 @@ def assert_lists_completed_job(
texts = collect_texts_from_job_table(response.content, "completed-jobs")
assert texts == [
+ str(completed_job.time_terminated),
str(completed_job.kitodo_details.task_id),
str(completed_job.kitodo_details.process_id),
completed_job.workflow_file.name,
@@ -123,6 +126,7 @@ def assert_lists_running_job(
texts = collect_texts_from_job_table(response.content, "running-jobs")
assert texts == [
+ str(running_job.time_created),
str(running_job.kitodo_details.task_id),
str(running_job.kitodo_details.process_id),
running_job.workflow_file.name,
@@ -135,7 +139,7 @@ def assert_lists_running_job(
def collect_texts_from_job_table(content: bytes, table_id: str) -> list[str]:
- selector = f"#{table_id} td:not(:has(a)), #{table_id} td > a"
+ selector = f"#{table_id} td:not(:has(a)):not(:has(button)), #{table_id} td > a"
return scraping.parse_texts(content, selector)
diff --git a/tests/ocrdmonitor/test_jobs.py b/tests/ocrdmonitor/test_jobs.py
index dd51c54..c47fb97 100644
--- a/tests/ocrdmonitor/test_jobs.py
+++ b/tests/ocrdmonitor/test_jobs.py
@@ -1,4 +1,5 @@
from dataclasses import replace
+from datetime import datetime, timedelta
from pathlib import Path
@@ -16,19 +17,25 @@
REMOTEDIR={remotedir}
WORKFLOW={workflow}
CONTROLLER={controller_address}
+TIME_CREATED={created_at}
+TIME_TERMINATED={terminated_at}
"""
+created_at = datetime(2023, 4, 12, hour=13, minute=0, second=0)
+terminated_at = created_at + timedelta(hours=1)
JOB_TEMPLATE = OcrdJob(
kitodo_details=KitodoProcessDetails(
- process_id=5432,
- task_id=45989,
+ process_id="5432",
+ task_id="45989",
processdir=Path("/data/5432"),
),
workdir=Path("ocr-d/data/5432"),
workflow_file=Path("ocr-workflow-default.sh"),
remotedir="/remote/job/dir",
controller_address="controller.ocrdhost.com",
+ time_created=created_at,
+ time_terminated=terminated_at,
)
@@ -41,6 +48,8 @@ def jobfile_content_for(job: OcrdJob) -> str:
workflow=job.workflow_file.as_posix(),
remotedir=job.remotedir,
controller_address=job.controller_address,
+ created_at=created_at,
+ terminated_at=terminated_at,
)
if job.pid is not None:
@@ -52,7 +61,9 @@ def jobfile_content_for(job: OcrdJob) -> str:
return out
-def test__parsing_a_ocrd_job_file_for_completed_job__returns_ocrdjob_with_a_return_code() -> None:
+def test__parsing_a_ocrd_job_file_for_completed_job__returns_ocrdjob_with_a_return_code() -> (
+ None
+):
expected = replace(JOB_TEMPLATE, return_code=0)
content = jobfile_content_for(expected)
@@ -61,7 +72,9 @@ def test__parsing_a_ocrd_job_file_for_completed_job__returns_ocrdjob_with_a_retu
assert actual == expected
-def test__parsing_a_ocrd_job_file_for_running_job__returns_ocrdjob_with_a_process_id() -> None:
+def test__parsing_a_ocrd_job_file_for_running_job__returns_ocrdjob_with_a_process_id() -> (
+ None
+):
expected = replace(JOB_TEMPLATE, pid=1)
content = jobfile_content_for(expected)
diff --git a/tests/ocrdmonitor/test_processstatus.py b/tests/ocrdmonitor/test_processstatus.py
index a0f1111..3347959 100644
--- a/tests/ocrdmonitor/test_processstatus.py
+++ b/tests/ocrdmonitor/test_processstatus.py
@@ -1,7 +1,7 @@
import datetime
import pytest
-from ocrdmonitor.processstatus import ProcessState, ProcessStatus, run
+from ocrdmonitor.processstatus import ProcessState, ProcessStatus
PS_OUTPUT = """
1 Ss 0.0 3872 01:12:46
@@ -16,7 +16,7 @@
def test__parsing_psoutput__returns_list_of_process_status() -> None:
- actual = ProcessStatus.from_ps_output(PS_OUTPUT)
+ actual = ProcessStatus.from_shell_output(PS_OUTPUT)
assert actual == [
ProcessStatus(
@@ -38,6 +38,6 @@ def test__parsing_psoutput__returns_list_of_process_status() -> None:
@pytest.mark.parametrize("output", FAILING_OUTPUTS)
def test__parsing_psoutput_with_error__returns_empty_list(output: str) -> None:
- actual = ProcessStatus.from_ps_output(output)
+ actual = ProcessStatus.from_shell_output(output)
assert actual == []
diff --git a/tests/ocrdmonitor/test_sshps.py b/tests/ocrdmonitor/test_sshremote.py
similarity index 51%
rename from tests/ocrdmonitor/test_sshps.py
rename to tests/ocrdmonitor/test_sshremote.py
index e008383..5f94add 100644
--- a/tests/ocrdmonitor/test_sshps.py
+++ b/tests/ocrdmonitor/test_sshremote.py
@@ -1,11 +1,11 @@
from pathlib import Path
-from typing import Any, Callable, TypeVar
-
import pytest
+from typing import Any, Awaitable, Callable, TypeVar
+
from testcontainers.general import DockerContainer
from ocrdmonitor.processstatus import ProcessState
-from ocrdmonitor.sshps import process_status
+from ocrdmonitor.sshremote import SSHRemote
from tests.ocrdmonitor.sshcontainer import (
get_process_group_from_container,
SSHConfig,
@@ -15,28 +15,30 @@
T = TypeVar("T")
-def run_until_truthy(fn: Callable[..., T], *args: Any) -> T:
- while not (result := fn(*args)):
- continue
-
- return result
-
-
+@pytest.mark.asyncio
@pytest.mark.integration
-def test_ps_over_ssh__returns_list_of_process_status(
+async def test_ps_over_ssh__returns_list_of_process_status(
openssh_server: DockerContainer,
) -> None:
process_group = get_process_group_from_container(openssh_server)
-
- config = SSHConfig(
- host="localhost",
- port=2222,
- user="testcontainer",
- keyfile=Path(KEYDIR) / "id.rsa",
+ sut = SSHRemote(
+ config=SSHConfig(
+ host="localhost",
+ port=2222,
+ user="testcontainer",
+ keyfile=Path(KEYDIR) / "id.rsa",
+ ),
)
- actual = run_until_truthy(process_status, config, process_group)
+ actual = await run_until_truthy(sut.process_status, process_group)
first_process = actual[0]
assert first_process.pid == process_group
assert first_process.state == ProcessState.SLEEPING
+
+
+async def run_until_truthy(fn: Callable[..., Awaitable[T]], *args: Any) -> T:
+ while not (result := await fn(*args)):
+ continue
+
+ return result
|