89 lines
3.7 KiB
Python
89 lines
3.7 KiB
Python
"""Opt-in read-only app discovery against an already open, named application."""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import jsonschema
|
|
from artemis.plugins.models import ServerConfig
|
|
from artemis.plugins.session import PluginSession
|
|
|
|
|
|
async def run(executable: Path, app: str, output: Path):
|
|
session = await PluginSession(
|
|
ServerConfig(command=str(executable)), executable.parent
|
|
).start()
|
|
try:
|
|
schema = next(t.inputSchema for t in session.tools if t.name == "list_windows")
|
|
query = {"app": app, "on_screen_only": True}
|
|
bad_query = {**query, "pid": 0}
|
|
try:
|
|
jsonschema.validate(bad_query, schema)
|
|
except jsonschema.ValidationError:
|
|
pass
|
|
else:
|
|
raise AssertionError("Host schema still accepts PID 0")
|
|
bad = await session.call("list_windows", bad_query)
|
|
assert bad.isError
|
|
assert bad.structuredContent["artemis"]["execution"] == "not_started"
|
|
assert "omit pid" in "\n".join(b.text for b in bad.content if b.type == "text")
|
|
|
|
jsonschema.validate(query, schema)
|
|
discovered = await session.call("list_windows", query)
|
|
assert not discovered.isError
|
|
windows = discovered.structuredContent["windows"]
|
|
assert windows, "Open the intended app before running this opt-in probe"
|
|
|
|
nullable = {**query, "pid": None}
|
|
jsonschema.validate(nullable, schema)
|
|
found = await session.call("list_windows", nullable)
|
|
assert not found.isError and found.structuredContent["windows"]
|
|
assert "pid" not in found.structuredContent["lookup"]["filters"]
|
|
|
|
# This Python test process is not the requested application. A conflict
|
|
# must stay empty and suggest a new app-only query, without a guessed PID.
|
|
conflict = await session.call("list_windows", {**query, "pid": os.getpid()})
|
|
assert not conflict.isError and conflict.structuredContent["windows"] == []
|
|
next_query = conflict.structuredContent["lookup"]["suggested_arguments"]
|
|
assert next_query == query
|
|
assert "frontmost" not in "\n".join(
|
|
b.text for b in conflict.content if b.type == "text"
|
|
)
|
|
recovered = await session.call("list_windows", next_query)
|
|
assert not recovered.isError and recovered.structuredContent["windows"]
|
|
|
|
window = recovered.structuredContent["windows"][0]
|
|
target = {k: window[k] for k in ("pid", "window_id")}
|
|
state = await session.call("get_window_state", {**target, "max_elements": 100})
|
|
assert not state.isError, state
|
|
assert any(b.type == "image" for b in state.content)
|
|
assert state.structuredContent["snapshot_id"]
|
|
assert state.structuredContent["tree_markdown"]
|
|
assert state.structuredContent["artemis"]["source"]["target_id"] == str(
|
|
target["window_id"]
|
|
)
|
|
report = {
|
|
"app": app,
|
|
"zero_pid_rejected_by_host_and_adapter": True,
|
|
"app_only_window_count": len(windows),
|
|
"null_pid_omitted": True,
|
|
"conflicting_pid_recovered_without_substitution": True,
|
|
"screenshot_and_tree": True,
|
|
}
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(json.dumps(report, ensure_ascii=False, indent=2), "utf-8")
|
|
print(json.dumps(report, ensure_ascii=False))
|
|
finally:
|
|
await session.aclose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(__doc__)
|
|
parser.add_argument("executable", type=Path)
|
|
parser.add_argument("--app", required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
asyncio.run(run(args.executable.resolve(), args.app, args.output))
|