Jimmy
f9bf119af6
This introduces the FastAPI editor implementation and related project setup so the app can be run and validated locally. Made-with: Cursor
238 lines
7.0 KiB
Python
238 lines
7.0 KiB
Python
import builtins
|
|
import importlib
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
|
|
def test_load_env_file_sets_missing_keys_only(tmp_path, monkeypatch):
|
|
import editor_app.config as config
|
|
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text(
|
|
"# comment\nFOO=bar\nBAZ='quoted'\nEXISTING=from_file\n", encoding="utf-8"
|
|
)
|
|
|
|
monkeypatch.setenv("EXISTING", "kept")
|
|
monkeypatch.delenv("FOO", raising=False)
|
|
monkeypatch.delenv("BAZ", raising=False)
|
|
|
|
config.load_env_file(env_file)
|
|
|
|
assert __import__("os").environ["FOO"] == "bar"
|
|
assert __import__("os").environ["BAZ"] == "quoted"
|
|
assert __import__("os").environ["EXISTING"] == "kept"
|
|
|
|
|
|
def test_load_env_file_ignores_missing_file(tmp_path):
|
|
import editor_app.config as config
|
|
|
|
config.load_env_file(tmp_path / "missing.env")
|
|
|
|
|
|
class _ProcNoStdout:
|
|
stdout = None
|
|
|
|
|
|
class _ProcWithLines:
|
|
def __init__(self):
|
|
self.stdout = iter(["line1\n", "line2\n"])
|
|
|
|
|
|
class _ProcWait:
|
|
def __init__(self, code=0):
|
|
self.code = code
|
|
|
|
def wait(self, timeout=None):
|
|
return self.code
|
|
|
|
|
|
class _ProcStop:
|
|
def __init__(self):
|
|
self.returncode = 0
|
|
self.terminated = False
|
|
self.killed = False
|
|
self.wait_calls = 0
|
|
|
|
def terminate(self):
|
|
self.terminated = True
|
|
|
|
def wait(self, timeout=None):
|
|
self.wait_calls += 1
|
|
if self.wait_calls == 1:
|
|
raise subprocess.TimeoutExpired("cmd", timeout)
|
|
return self.returncode
|
|
|
|
def kill(self):
|
|
self.killed = True
|
|
|
|
|
|
def test_python_runner_stream_helpers_and_wait():
|
|
import editor_app.services.python_runner as runner
|
|
|
|
with runner.python_runner.lock:
|
|
runner.python_runner.output_lines = []
|
|
|
|
runner.stream_process_output(_ProcNoStdout())
|
|
runner.stream_process_output(_ProcWithLines())
|
|
|
|
with runner.python_runner.lock:
|
|
assert runner.python_runner.output_lines[-2:] == ["line1\n", "line2\n"]
|
|
|
|
proc = _ProcWait(code=7)
|
|
runner.wait_for_process(proc)
|
|
with runner.python_runner.lock:
|
|
assert runner.python_runner.return_code == 7
|
|
assert runner.python_runner.running is False
|
|
assert runner.python_runner.process is None
|
|
|
|
|
|
def test_python_runner_run_failure_raises_http(monkeypatch, tmp_path):
|
|
import editor_app.config as config
|
|
import editor_app.services.python_runner as runner
|
|
|
|
config.WORKSPACE_ROOT = tmp_path
|
|
|
|
def _boom(*args, **kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr(runner.subprocess, "Popen", _boom)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
runner.run_python_file(tmp_path / "x.py", "x.py")
|
|
|
|
assert exc.value.status_code == 500
|
|
|
|
|
|
def test_python_runner_stop_kill_path():
|
|
import editor_app.services.python_runner as runner
|
|
|
|
proc = _ProcStop()
|
|
with runner.python_runner.lock:
|
|
runner.python_runner.process = proc
|
|
runner.python_runner.output_lines = []
|
|
runner.python_runner.running = True
|
|
|
|
message = runner.stop_python_process()
|
|
assert message == "Python process stopped"
|
|
assert proc.terminated is True
|
|
assert proc.killed is True
|
|
|
|
|
|
def test_completions_import_error_path(tmp_path, monkeypatch):
|
|
import editor_app.config as config
|
|
import editor_app.main as main
|
|
from fastapi.testclient import TestClient
|
|
|
|
config.WORKSPACE_ROOT = tmp_path
|
|
importlib.reload(main)
|
|
client = TestClient(main.app)
|
|
|
|
real_import = builtins.__import__
|
|
|
|
def fake_import(name, *args, **kwargs):
|
|
if name == "jedi":
|
|
raise ImportError("forced")
|
|
return real_import(name, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(builtins, "__import__", fake_import)
|
|
|
|
response = client.post(
|
|
"/api/python/completions",
|
|
json={
|
|
"file_path": "scratch.py",
|
|
"content": "x = 1",
|
|
"line": 1,
|
|
"column": 1,
|
|
"max_results": 10,
|
|
},
|
|
)
|
|
assert response.status_code == 500
|
|
|
|
|
|
def test_websocket_output_status_message(tmp_path):
|
|
import editor_app.config as config
|
|
import editor_app.main as main
|
|
from fastapi.testclient import TestClient
|
|
|
|
config.WORKSPACE_ROOT = tmp_path
|
|
importlib.reload(main)
|
|
|
|
with TestClient(main.app) as client:
|
|
with client.websocket_connect("/api/python/ws/output") as ws:
|
|
payload = ws.receive_json()
|
|
assert "lines" in payload
|
|
assert "running" in payload
|
|
assert payload["running"] is False
|
|
|
|
|
|
def test_create_app_startup_creates_lib(tmp_path):
|
|
import editor_app.config as config
|
|
import editor_app.main as main
|
|
from fastapi.testclient import TestClient
|
|
|
|
config.WORKSPACE_ROOT = tmp_path
|
|
importlib.reload(main)
|
|
|
|
assert not (tmp_path / "lib").exists()
|
|
with TestClient(main.app):
|
|
pass
|
|
assert (tmp_path / "lib").is_dir()
|
|
|
|
|
|
def test_python_runner_helpers_and_startup_paths(tmp_path, monkeypatch):
|
|
import editor_app.config as config
|
|
import editor_app.services.python_runner as runner
|
|
|
|
config.WORKSPACE_ROOT = tmp_path
|
|
runner.STARTUP_SCRIPT_FILE = tmp_path / ".connectionmachine_startup_script"
|
|
(tmp_path / "lib").mkdir(exist_ok=True)
|
|
|
|
# list_python_scripts should skip hidden paths.
|
|
(tmp_path / "code").mkdir(exist_ok=True)
|
|
(tmp_path / "code" / "a.py").write_text("print('x')\n", encoding="utf-8")
|
|
(tmp_path / ".hidden.py").write_text("print('x')\n", encoding="utf-8")
|
|
scripts = runner.list_python_scripts()
|
|
assert "code/a.py" in scripts
|
|
assert ".hidden.py" not in scripts
|
|
|
|
# startup script getters/setters.
|
|
assert runner.get_startup_script() is None
|
|
runner.STARTUP_SCRIPT_FILE.write_text("", encoding="utf-8")
|
|
assert runner.get_startup_script() is None
|
|
|
|
missing = tmp_path / "code" / "missing.py"
|
|
with pytest.raises(HTTPException) as missing_exc:
|
|
runner.set_startup_script(str(missing.relative_to(tmp_path)))
|
|
assert missing_exc.value.status_code == 404
|
|
|
|
(tmp_path / "code" / "note.txt").write_text("x", encoding="utf-8")
|
|
with pytest.raises(HTTPException) as nonpy_exc:
|
|
runner.set_startup_script("code/note.txt")
|
|
assert nonpy_exc.value.status_code == 400
|
|
|
|
selected = runner.set_startup_script("code/a.py")
|
|
assert selected == "code/a.py"
|
|
assert runner.get_startup_script() == "code/a.py"
|
|
|
|
# run_startup_script_if_configured no-op branches.
|
|
runner.STARTUP_SCRIPT_FILE.write_text("code/missing.py", encoding="utf-8")
|
|
runner.run_startup_script_if_configured()
|
|
runner.STARTUP_SCRIPT_FILE.write_text("code/note.txt", encoding="utf-8")
|
|
runner.run_startup_script_if_configured()
|
|
|
|
# run_startup_script_if_configured should call run_python_file on valid startup script.
|
|
called = {"count": 0}
|
|
|
|
def fake_run(target_path, requested_path):
|
|
called["count"] += 1
|
|
assert requested_path == "code/a.py"
|
|
assert str(target_path).endswith("code/a.py")
|
|
|
|
monkeypatch.setattr(runner, "run_python_file", fake_run)
|
|
runner.STARTUP_SCRIPT_FILE.write_text("code/a.py", encoding="utf-8")
|
|
runner.run_startup_script_if_configured()
|
|
assert called["count"] == 1
|