feat: add user_id index to profiles table for faster lookups
The new index on the user_id column in the profiles table improves query performance for user-specific lookups, reducing full table scans during authentication and profile retrieval operations.
This commit is contained in:
@@ -0,0 +1,298 @@
|
||||
import asyncio
|
||||
from datetime import timedelta
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from tests.conftest import BASE_URL, login_user, run_async
|
||||
from devplacepy.database import db, get_table, init_db, refresh_snapshot
|
||||
from devplacepy import project_files
|
||||
from devplacepy.services.containers import api, store, runtime
|
||||
from devplacepy.services.containers.backend.base import Mount, PortMapping, RunSpec
|
||||
from devplacepy.services.containers.backend.docker_cli import build_run_argv, parse_size
|
||||
from devplacepy.services.containers.backend.fake import FakeBackend
|
||||
from devplacepy.services.containers.build_service import ContainerBuildService
|
||||
from devplacepy.services.containers.service import ContainerService
|
||||
from devplacepy.services.containers.api import INSTANCE_LABEL
|
||||
from devplacepy.services.devii.tasks.schedule import Schedule, now_utc, to_iso
|
||||
from devplacepy.services.jobs import queue
|
||||
|
||||
_CONTAINER_TABLES = ("dockerfiles", "dockerfile_versions", "builds", "instances",
|
||||
"instance_events", "instance_metrics", "instance_schedules")
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _init_db():
|
||||
init_db()
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(tmp_path, monkeypatch):
|
||||
fake = FakeBackend()
|
||||
runtime.set_backend(fake)
|
||||
monkeypatch.setattr("devplacepy.config.CONTAINER_WORKSPACES_DIR", tmp_path / "ws")
|
||||
monkeypatch.setattr("devplacepy.config.CONTAINER_BUILD_CONTEXTS_DIR", tmp_path / "ctx")
|
||||
pid = "ctest-p1"
|
||||
project = {"uid": pid, "slug": "ctest", "title": "C", "user_uid": "ctest-u1"}
|
||||
user = {"uid": "ctest-u1", "username": "ctestadmin"}
|
||||
project_files.write_text_file(pid, user, "app.py", "print(1)\n")
|
||||
yield {"fake": fake, "project": project, "user": user}
|
||||
runtime.set_backend(None)
|
||||
for table in _CONTAINER_TABLES + ("jobs",):
|
||||
if table in db.tables:
|
||||
rows = [r for r in get_table(table).find()]
|
||||
for row in rows:
|
||||
if str(row.get("project_uid", row.get("kind", ""))).startswith("ctest") or row.get("kind") == "container_build":
|
||||
get_table(table).delete(uid=row["uid"])
|
||||
for row in list(get_table("project_files").find()):
|
||||
if str(row.get("project_uid", "")).startswith("ctest"):
|
||||
get_table("project_files").delete(uid=row["uid"])
|
||||
|
||||
|
||||
def _drain_builds():
|
||||
async def drive():
|
||||
service = ContainerBuildService()
|
||||
for _ in range(80):
|
||||
await service.run_once()
|
||||
refresh_snapshot()
|
||||
pending = [j for j in queue.list_jobs(kind="container_build") if j["status"] in ("pending", "running")]
|
||||
if not pending and not service._inflight:
|
||||
return
|
||||
await asyncio.sleep(0.02)
|
||||
run_async(drive())
|
||||
|
||||
|
||||
# ---------------- backend argv ----------------
|
||||
|
||||
def test_build_run_argv_exact():
|
||||
spec = RunSpec(image="myapp:3", name="inst", labels={INSTANCE_LABEL: "u1"}, env={"A": "B"},
|
||||
cpu_limit="1.5", mem_limit="512m", ports=[PortMapping(8080, 80)],
|
||||
mounts=[Mount("/ws", "/app")], restart_policy="on-failure", command=["python", "app.py"])
|
||||
argv = build_run_argv(spec)
|
||||
assert argv[:5] == ["docker", "run", "-d", "--name", "inst"]
|
||||
assert "--label" in argv and f"{INSTANCE_LABEL}=u1" in argv
|
||||
assert "--cpus" in argv and "1.5" in argv
|
||||
assert "-p" in argv and "8080:80/tcp" in argv
|
||||
assert "-v" in argv and "/ws:/app:rw" in argv
|
||||
assert "--restart" in argv and "on-failure" in argv
|
||||
assert argv[-3:] == ["myapp:3", "python", "app.py"]
|
||||
|
||||
|
||||
def test_never_policy_not_passed_to_docker():
|
||||
spec = RunSpec(image="i:1", name="n", restart_policy="never")
|
||||
assert "--restart" not in build_run_argv(spec)
|
||||
|
||||
|
||||
def test_parse_size():
|
||||
assert parse_size("1.0GiB") == 1024 ** 3
|
||||
assert parse_size("512MB") == 512 * 1024 ** 2
|
||||
|
||||
|
||||
# ---------------- versioning + builds ----------------
|
||||
|
||||
def test_create_dockerfile_and_autobuild(env):
|
||||
result = run_async(api.create_dockerfile(env["project"], env["user"], name="web", description="d"))
|
||||
df = result["dockerfile"]
|
||||
assert df["current_version"] == 1
|
||||
assert result["build"]["build_number"] == 1
|
||||
_drain_builds()
|
||||
build = store.get_build(result["build"]["uid"])
|
||||
assert build["status"] == store.BUILD_SUCCESS
|
||||
assert env["fake"].built_images and "web:1" in env["fake"].built_images[0]
|
||||
|
||||
|
||||
def test_unchanged_content_skips_build(env):
|
||||
result = run_async(api.create_dockerfile(env["project"], env["user"], name="web", content="FROM scratch\n"))
|
||||
df = store.get_dockerfile(result["dockerfile"]["uid"])
|
||||
same = run_async(api.save_version(df, env["user"], "FROM scratch\n"))
|
||||
assert same["changed"] is False and same["build"] is None
|
||||
changed = run_async(api.save_version(df, env["user"], "FROM scratch\nRUN echo hi\n"))
|
||||
assert changed["changed"] is True
|
||||
assert store.get_dockerfile(df["uid"])["current_version"] == 2
|
||||
assert changed["build"]["build_number"] == 2
|
||||
|
||||
|
||||
def test_build_failure_recorded(env):
|
||||
env["fake"].fail_build = True
|
||||
result = run_async(api.create_dockerfile(env["project"], env["user"], name="web"))
|
||||
_drain_builds()
|
||||
assert store.get_build(result["build"]["uid"])["status"] == store.BUILD_FAILED
|
||||
|
||||
|
||||
def test_duplicate_name_rejected(env):
|
||||
run_async(api.create_dockerfile(env["project"], env["user"], name="web"))
|
||||
with pytest.raises(api.ContainerError):
|
||||
run_async(api.create_dockerfile(env["project"], env["user"], name="web"))
|
||||
|
||||
|
||||
# ---------------- reconcile ----------------
|
||||
|
||||
def _ready_instance(env, **kwargs):
|
||||
result = run_async(api.create_dockerfile(env["project"], env["user"], name="web"))
|
||||
_drain_builds()
|
||||
df = store.get_dockerfile(result["dockerfile"]["uid"])
|
||||
build = store.get_build(result["build"]["uid"])
|
||||
return run_async(api.create_instance(env["project"], df, build, name=kwargs.pop("name", "inst"), **kwargs))
|
||||
|
||||
|
||||
def test_reconcile_launches_and_stops(env):
|
||||
inst = _ready_instance(env, restart_policy="never", autostart=True)
|
||||
assert inst["desired_state"] == store.DESIRED_RUNNING
|
||||
service = ContainerService()
|
||||
run_async(service.run_once())
|
||||
refresh_snapshot()
|
||||
inst = store.get_instance(inst["uid"])
|
||||
assert inst["status"] == store.ST_RUNNING and inst["container_id"]
|
||||
assert [r.name for r in run_async(env["fake"].ps())] == [inst["slug"]]
|
||||
api.set_desired_state(inst, store.DESIRED_STOPPED)
|
||||
run_async(service.run_once())
|
||||
refresh_snapshot()
|
||||
assert store.get_instance(inst["uid"])["status"] == store.ST_STOPPED
|
||||
|
||||
|
||||
def test_reconcile_reaps_orphan(env):
|
||||
fake = env["fake"]
|
||||
run_async(fake.run(RunSpec(image="x:1", name="ghost", labels={INSTANCE_LABEL: "missing-uid"})))
|
||||
service = ContainerService()
|
||||
run_async(service.run_once())
|
||||
assert not run_async(fake.ps())
|
||||
assert fake.removed
|
||||
|
||||
|
||||
def test_reconcile_removes_marked_instance(env):
|
||||
inst = _ready_instance(env, autostart=True)
|
||||
service = ContainerService()
|
||||
run_async(service.run_once())
|
||||
refresh_snapshot()
|
||||
inst = store.get_instance(inst["uid"])
|
||||
api.mark_for_removal(inst)
|
||||
run_async(service.run_once())
|
||||
refresh_snapshot()
|
||||
assert store.get_instance(inst["uid"]) is None
|
||||
assert not run_async(env["fake"].ps())
|
||||
|
||||
|
||||
def test_build_status_not_flagged_as_error(env):
|
||||
from types import SimpleNamespace
|
||||
from devplacepy.services.devii.container import ContainerController
|
||||
from devplacepy.services.devii.agentic.loop import _is_error, _summary
|
||||
df = store.create_dockerfile("ctest-p1", env["user"], "svc", "", "")
|
||||
ver = store.create_version(df, "FROM scratch", env["user"])
|
||||
build = store.create_build(df, ver, 1, "svc:1", True)
|
||||
store.update_build(build["uid"], {"status": "building"})
|
||||
ctl = ContainerController(SimpleNamespace(username=None))
|
||||
result = run_async(ctl.dispatch("container_build_status", {"build_uid": build["uid"]}))
|
||||
assert _is_error(result) is False
|
||||
assert _summary(result) == "building"
|
||||
store.update_build(build["uid"], {"status": "failed", "error": "docker build exited 1"})
|
||||
failed = run_async(ctl.dispatch("container_build_status", {"build_uid": build["uid"]}))
|
||||
import json as _json
|
||||
assert _is_error(failed) is False
|
||||
assert "error" not in _json.loads(failed)
|
||||
assert _json.loads(failed)["build_error"] == "docker build exited 1"
|
||||
|
||||
|
||||
def test_schedule_fires(env):
|
||||
inst = _ready_instance(env, autostart=False)
|
||||
assert inst["desired_state"] == store.DESIRED_STOPPED
|
||||
past = Schedule(kind="once", run_at=now_utc() - timedelta(hours=1))
|
||||
api.add_schedule(inst, "start", past)
|
||||
service = ContainerService()
|
||||
run_async(service._fire_schedules())
|
||||
refresh_snapshot()
|
||||
assert store.get_instance(inst["uid"])["desired_state"] == store.DESIRED_RUNNING
|
||||
|
||||
|
||||
# ---------------- HTTP admin gate ----------------
|
||||
|
||||
def _promote_admin(username: str) -> None:
|
||||
users = get_table("users")
|
||||
user = users.find_one(username=username)
|
||||
if user:
|
||||
users.update({"uid": user["uid"], "role": "Admin"}, ["uid"])
|
||||
|
||||
|
||||
def _api_key(username: str) -> str:
|
||||
refresh_snapshot()
|
||||
return get_table("users").find_one(username=username)["api_key"]
|
||||
|
||||
|
||||
def test_http_non_admin_forbidden(app_server, page, seeded_db):
|
||||
project = requests.post(f"{BASE_URL}/projects/create",
|
||||
headers={"X-API-KEY": _api_key("bob_test"), "Accept": "application/json"},
|
||||
data={"title": "NoAdmin", "description": "x", "project_type": "software", "status": "s"})
|
||||
slug = project.json()["data"]["slug"] or project.json()["data"]["uid"]
|
||||
r = requests.get(f"{BASE_URL}/projects/{slug}/containers/data",
|
||||
headers={"X-API-KEY": _api_key("bob_test"), "Accept": "application/json"})
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_ingress_validation(env):
|
||||
from devplacepy.services.containers.backend.base import PortMapping
|
||||
assert api.validate_ingress("zwoeks", 8899, [PortMapping(8899, 8899)]) == ("zwoeks", 8899)
|
||||
assert api.validate_ingress("", None, []) == ("", 0)
|
||||
with pytest.raises(api.ContainerError):
|
||||
api.validate_ingress("BAD SLUG", None, [PortMapping(80, 80)])
|
||||
with pytest.raises(api.ContainerError):
|
||||
api.validate_ingress("x", 9999, [PortMapping(80, 80)])
|
||||
store.create_instance({"uid": "z", "project_uid": "ctest-p1", "name": "z",
|
||||
"ports_json": "[]", "ingress_slug": "taken"})
|
||||
with pytest.raises(api.ContainerError):
|
||||
api.validate_ingress("taken", None, [PortMapping(80, 80)])
|
||||
|
||||
|
||||
def test_http_ingress_proxy(app_server):
|
||||
import http.server
|
||||
import socket
|
||||
import socketserver
|
||||
import threading
|
||||
|
||||
sock = socket.socket()
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
port = sock.getsockname()[1]
|
||||
sock.close()
|
||||
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/plain")
|
||||
self.end_headers()
|
||||
self.wfile.write(b"hello from upstream " + self.path.encode())
|
||||
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
httpd = socketserver.TCPServer(("127.0.0.1", port), Handler)
|
||||
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
slug = f"ing{port}"
|
||||
uid = f"ingtest-{port}"
|
||||
get_table("instances").insert({
|
||||
"uid": uid, "name": "ingress", "project_uid": "ingtest", "status": "running",
|
||||
"ingress_slug": slug, "ingress_port": 8000,
|
||||
"ports_json": f'[{{"host": {port}, "container": 8000, "proto": "tcp"}}]',
|
||||
})
|
||||
try:
|
||||
r = requests.get(f"{BASE_URL}/p/{slug}/foo")
|
||||
assert r.status_code == 200, r.text
|
||||
assert "hello from upstream" in r.text and "/foo" in r.text
|
||||
assert requests.get(f"{BASE_URL}/p/does-not-exist-slug").status_code == 404
|
||||
finally:
|
||||
httpd.shutdown()
|
||||
get_table("instances").delete(uid=uid)
|
||||
|
||||
|
||||
def test_http_admin_create_dockerfile(app_server, page, seeded_db):
|
||||
_promote_admin("alice_test")
|
||||
key = _api_key("alice_test")
|
||||
headers = {"X-API-KEY": key, "Accept": "application/json"}
|
||||
project = requests.post(f"{BASE_URL}/projects/create", headers=headers,
|
||||
data={"title": "WithAdmin", "description": "x", "project_type": "software", "status": "s"})
|
||||
slug = project.json()["data"]["slug"] or project.json()["data"]["uid"]
|
||||
r = requests.post(f"{BASE_URL}/projects/{slug}/containers/dockerfiles", headers=headers,
|
||||
data={"name": "svc", "description": "", "tags": ""})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["data"]["dockerfile"]["name"] == "svc"
|
||||
data = requests.get(f"{BASE_URL}/projects/{slug}/containers/data", headers=headers).json()
|
||||
assert any(d["name"] == "svc" for d in data["dockerfiles"])
|
||||
@@ -0,0 +1,54 @@
|
||||
import json
|
||||
|
||||
from devplacepy.services.devii.agentic.loop import _run_tool_call
|
||||
from tests.conftest import run_async
|
||||
|
||||
|
||||
class _FakeDispatcher:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
async def dispatch(self, name, arguments):
|
||||
self.calls.append((name, arguments))
|
||||
return json.dumps({"status": "ok"})
|
||||
|
||||
|
||||
def _run(call, dispatcher=None):
|
||||
return json.loads(run_async(_run_tool_call(dispatcher, call)))
|
||||
|
||||
|
||||
def test_truncated_arguments_reported_not_dispatched():
|
||||
dispatcher = _FakeDispatcher()
|
||||
call = {"function": {"name": "project_write_file", "arguments": '{"path":"a.md","content":"# hi'}}
|
||||
out = _run(call, dispatcher)
|
||||
assert out["error"] == "tool_input_truncated"
|
||||
assert "one write tool call per turn" in out["message"]
|
||||
assert dispatcher.calls == []
|
||||
|
||||
|
||||
def test_non_object_arguments_rejected():
|
||||
dispatcher = _FakeDispatcher()
|
||||
out = _run({"function": {"name": "x", "arguments": '"a string"'}}, dispatcher)
|
||||
assert out["error"] == "tool_input_error"
|
||||
assert dispatcher.calls == []
|
||||
|
||||
|
||||
def test_valid_string_arguments_dispatched():
|
||||
dispatcher = _FakeDispatcher()
|
||||
out = _run({"function": {"name": "vote", "arguments": '{"value":1}'}}, dispatcher)
|
||||
assert out["status"] == "ok"
|
||||
assert dispatcher.calls == [("vote", {"value": 1})]
|
||||
|
||||
|
||||
def test_valid_dict_arguments_dispatched():
|
||||
dispatcher = _FakeDispatcher()
|
||||
out = _run({"function": {"name": "vote", "arguments": {"value": -1}}}, dispatcher)
|
||||
assert out["status"] == "ok"
|
||||
assert dispatcher.calls == [("vote", {"value": -1})]
|
||||
|
||||
|
||||
def test_missing_arguments_defaults_to_empty_object():
|
||||
dispatcher = _FakeDispatcher()
|
||||
out = _run({"function": {"name": "auth_status"}}, dispatcher)
|
||||
assert out["status"] == "ok"
|
||||
assert dispatcher.calls == [("auth_status", {})]
|
||||
@@ -0,0 +1,263 @@
|
||||
import asyncio
|
||||
import time
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import init_db, get_table
|
||||
from devplacepy import project_files as pf
|
||||
from devplacepy.project_files import ProjectFileError
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _init_db():
|
||||
init_db()
|
||||
yield
|
||||
|
||||
|
||||
_pid = [0]
|
||||
|
||||
|
||||
def _project():
|
||||
_pid[0] += 1
|
||||
pid = f"plines-{_pid[0]}"
|
||||
user = {"uid": f"plines-owner-{_pid[0]}"}
|
||||
return pid, user
|
||||
|
||||
|
||||
# ---------------- in-process unit tests ----------------
|
||||
|
||||
def test_read_lines_range_and_total():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "a\nb\nc\nd")
|
||||
out = pf.read_lines(pid, "f.txt", 2, 3)
|
||||
assert out["lines"] == ["b", "c"]
|
||||
assert out["total_lines"] == 4
|
||||
assert out["content"] == "b\nc"
|
||||
|
||||
|
||||
def test_read_lines_open_end_clamps():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "a\nb\nc")
|
||||
out = pf.read_lines(pid, "f.txt", 2, None)
|
||||
assert out["lines"] == ["b", "c"]
|
||||
assert out["end"] == 3
|
||||
|
||||
|
||||
def test_replace_lines_middle():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "a\nb\nc\nd")
|
||||
pf.replace_lines(pid, "f.txt", 2, 3, "X\nY\nZ")
|
||||
assert pf.read_file(pid, "f.txt")["content"] == "a\nX\nY\nZ\nd"
|
||||
|
||||
|
||||
def test_replace_lines_empty_content_deletes():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "a\nb\nc")
|
||||
pf.replace_lines(pid, "f.txt", 2, 2, "")
|
||||
assert pf.read_file(pid, "f.txt")["content"] == "a\nc"
|
||||
|
||||
|
||||
def test_insert_lines_prepend_and_append_positions():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "a\nb")
|
||||
pf.insert_lines(pid, "f.txt", 1, "TOP")
|
||||
pf.insert_lines(pid, "f.txt", 99, "BOTTOM")
|
||||
assert pf.read_file(pid, "f.txt")["content"] == "TOP\na\nb\nBOTTOM"
|
||||
|
||||
|
||||
def test_delete_lines_range():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "a\nb\nc\nd")
|
||||
pf.delete_lines(pid, "f.txt", 2, 3)
|
||||
assert pf.read_file(pid, "f.txt")["content"] == "a\nd"
|
||||
|
||||
|
||||
def test_append_preserves_trailing_newline():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "g.txt", "one\ntwo\n")
|
||||
pf.append_lines(pid, "g.txt", "three")
|
||||
assert pf.read_file(pid, "g.txt")["content"] == "one\ntwo\nthree\n"
|
||||
|
||||
|
||||
def test_append_without_trailing_adds_newline_between():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "g.txt", "one")
|
||||
pf.append_lines(pid, "g.txt", "two")
|
||||
assert pf.read_file(pid, "g.txt")["content"] == "one\ntwo"
|
||||
|
||||
|
||||
def test_line_ops_reject_missing_dir_and_binary():
|
||||
pid, u = _project()
|
||||
pf.make_dir(pid, u, "adir")
|
||||
pf.store_upload(pid, u, "", "blob.bin", bytes(range(64)))
|
||||
with pytest.raises(ProjectFileError):
|
||||
pf.read_lines(pid, "nope.txt")
|
||||
with pytest.raises(ProjectFileError):
|
||||
pf.replace_lines(pid, "adir", 1, 1, "x")
|
||||
with pytest.raises(ProjectFileError):
|
||||
pf.append_lines(pid, "blob.bin", "x")
|
||||
|
||||
|
||||
def test_replace_out_of_range_raises():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "a\nb")
|
||||
with pytest.raises(ProjectFileError):
|
||||
pf.replace_lines(pid, "f.txt", 9, 9, "x")
|
||||
|
||||
|
||||
def test_append_respects_max_chars():
|
||||
pid, u = _project()
|
||||
pf.write_text_file(pid, u, "f.txt", "x")
|
||||
with pytest.raises(ProjectFileError):
|
||||
pf.append_lines(pid, "f.txt", "y" * (pf.MAX_TEXT_CHARS + 10))
|
||||
|
||||
|
||||
def test_normalize_blocks_traversal_in_line_ops():
|
||||
pid, u = _project()
|
||||
with pytest.raises(ProjectFileError):
|
||||
pf.read_lines(pid, "../escape.txt")
|
||||
|
||||
|
||||
# ---------------- HTTP endpoint tests ----------------
|
||||
|
||||
_counter = [0]
|
||||
|
||||
|
||||
def _signup():
|
||||
_counter[0] += 1
|
||||
name = f"pl{int(time.time() * 1000)}{_counter[0]}"
|
||||
requests.post(f"{BASE_URL}/auth/signup", data={
|
||||
"username": name, "email": f"{name}@t.dev",
|
||||
"password": "secret123", "confirm_password": "secret123",
|
||||
}, allow_redirects=True)
|
||||
return name, get_table("users").find_one(username=name)["api_key"]
|
||||
|
||||
|
||||
def _h(key):
|
||||
return {"X-API-KEY": key, "Accept": "application/json"}
|
||||
|
||||
|
||||
def _create_project(key, title):
|
||||
r = requests.post(f"{BASE_URL}/projects/create", headers=_h(key), data={
|
||||
"title": title, "description": "lines test", "project_type": "software", "status": "In Development",
|
||||
})
|
||||
assert r.status_code == 200, r.text
|
||||
return r.json()["data"]
|
||||
|
||||
|
||||
def _write(key, slug, path, content):
|
||||
r = requests.post(f"{BASE_URL}/projects/{slug}/files/write", headers=_h(key),
|
||||
data={"path": path, "content": content}, allow_redirects=False)
|
||||
assert r.status_code in (200, 302), r.text
|
||||
|
||||
|
||||
def _raw(key, slug, path):
|
||||
return requests.get(f"{BASE_URL}/projects/{slug}/files/raw", headers=_h(key), params={"path": path}).json()
|
||||
|
||||
|
||||
def test_http_read_lines(app_server):
|
||||
_, key = _signup()
|
||||
proj = _create_project(key, "HTTP Read Lines")
|
||||
slug = proj["slug"] or proj["uid"]
|
||||
_write(key, slug, "f.txt", "a\nb\nc\nd")
|
||||
r = requests.get(f"{BASE_URL}/projects/{slug}/files/lines", headers=_h(key), params={"path": "f.txt", "start": 2, "end": 3})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["lines"] == ["b", "c"]
|
||||
assert body["total_lines"] == 4
|
||||
|
||||
|
||||
def test_http_replace_insert_delete_append_roundtrip(app_server):
|
||||
_, key = _signup()
|
||||
proj = _create_project(key, "HTTP Edit Roundtrip")
|
||||
slug = proj["slug"] or proj["uid"]
|
||||
_write(key, slug, "f.txt", "a\nb\nc\nd")
|
||||
requests.post(f"{BASE_URL}/projects/{slug}/files/replace-lines", headers=_h(key),
|
||||
data={"path": "f.txt", "start": 2, "end": 3, "content": "X\nY"}, allow_redirects=False)
|
||||
assert _raw(key, slug, "f.txt")["content"] == "a\nX\nY\nd"
|
||||
requests.post(f"{BASE_URL}/projects/{slug}/files/insert-lines", headers=_h(key),
|
||||
data={"path": "f.txt", "at": 1, "content": "TOP"}, allow_redirects=False)
|
||||
assert _raw(key, slug, "f.txt")["content"] == "TOP\na\nX\nY\nd"
|
||||
requests.post(f"{BASE_URL}/projects/{slug}/files/delete-lines", headers=_h(key),
|
||||
data={"path": "f.txt", "start": 1, "end": 1}, allow_redirects=False)
|
||||
assert _raw(key, slug, "f.txt")["content"] == "a\nX\nY\nd"
|
||||
requests.post(f"{BASE_URL}/projects/{slug}/files/append", headers=_h(key),
|
||||
data={"path": "f.txt", "content": "END"}, allow_redirects=False)
|
||||
assert _raw(key, slug, "f.txt")["content"] == "a\nX\nY\nd\nEND"
|
||||
|
||||
|
||||
def test_http_lines_missing_file_404(app_server):
|
||||
_, key = _signup()
|
||||
proj = _create_project(key, "HTTP Lines 404")
|
||||
slug = proj["slug"] or proj["uid"]
|
||||
r = requests.get(f"{BASE_URL}/projects/{slug}/files/lines", headers=_h(key), params={"path": "nope.txt"})
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_http_replace_lines_non_owner_denied(app_server):
|
||||
_, owner_key = _signup()
|
||||
_, other_key = _signup()
|
||||
proj = _create_project(owner_key, "HTTP Owner Guard")
|
||||
slug = proj["slug"] or proj["uid"]
|
||||
_write(owner_key, slug, "f.txt", "a\nb")
|
||||
r = requests.post(f"{BASE_URL}/projects/{slug}/files/replace-lines", headers=_h(other_key),
|
||||
data={"path": "f.txt", "start": 1, "end": 1, "content": "x"}, allow_redirects=False)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
# ---------------- agent read-before-write guard ----------------
|
||||
|
||||
class _FakeClient:
|
||||
authenticated = True
|
||||
username = "u"
|
||||
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
async def call(self, method, path, params=None, data=None, file_field=None, headers=None):
|
||||
self.calls.append((method, path))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
def _make_dispatcher():
|
||||
import devplacepy.services.devii.actions.dispatcher as disp
|
||||
from devplacepy.services.devii.actions.catalog import PLATFORM_CATALOG
|
||||
d = disp.Dispatcher.__new__(disp.Dispatcher)
|
||||
d._actions = PLATFORM_CATALOG.by_name()
|
||||
d._client = _FakeClient()
|
||||
d._read_files = set()
|
||||
return disp, d
|
||||
|
||||
|
||||
def test_write_blocked_until_read(monkeypatch):
|
||||
disp, d = _make_dispatcher()
|
||||
monkeypatch.setattr(disp, "format_response", lambda r: "ok")
|
||||
monkeypatch.setattr(disp, "record_mutation", lambda name: None)
|
||||
monkeypatch.setattr(disp, "get_store", lambda: None)
|
||||
actions = d._actions
|
||||
write = actions["project_write_file"]
|
||||
read = actions["project_read_file"]
|
||||
args = {"project_slug": "p", "path": "src/app.py", "content": "x"}
|
||||
|
||||
from devplacepy.services.devii.errors import ToolInputError
|
||||
with pytest.raises(ToolInputError):
|
||||
asyncio.run(d._run_http(write, args))
|
||||
assert d._client.calls == []
|
||||
|
||||
asyncio.run(d._run_http(read, {"project_slug": "p", "path": "src/app.py"}))
|
||||
asyncio.run(d._run_http(write, args))
|
||||
assert ("POST", "/projects/p/files/write") in d._client.calls
|
||||
|
||||
|
||||
def test_guard_path_normalization_matches(monkeypatch):
|
||||
disp, d = _make_dispatcher()
|
||||
monkeypatch.setattr(disp, "format_response", lambda r: "ok")
|
||||
monkeypatch.setattr(disp, "record_mutation", lambda name: None)
|
||||
monkeypatch.setattr(disp, "get_store", lambda: None)
|
||||
actions = d._actions
|
||||
# read with a messy path, write with the clean path - normalization should unify them
|
||||
asyncio.run(d._run_http(actions["project_read_file"], {"project_slug": "p", "path": "/src//app.py"}))
|
||||
asyncio.run(d._run_http(actions["project_write_file"], {"project_slug": "p", "path": "src/app.py", "content": "x"}))
|
||||
assert ("POST", "/projects/p/files/write") in d._client.calls
|
||||
@@ -0,0 +1,260 @@
|
||||
import time
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table
|
||||
from devplacepy.utils import clear_user_cache
|
||||
from devplacepy import project_files
|
||||
from devplacepy.project_files import ProjectFileError
|
||||
from devplacepy.services.devii.actions.dispatcher import confirmation_error, _is_confirmed
|
||||
|
||||
_counter = [0]
|
||||
|
||||
|
||||
def _signup():
|
||||
_counter[0] += 1
|
||||
name = f"pv{int(time.time() * 1000)}{_counter[0]}"
|
||||
session = requests.Session()
|
||||
session.post(f"{BASE_URL}/auth/signup", data={
|
||||
"username": name, "email": f"{name}@t.dev",
|
||||
"password": "secret123", "confirm_password": "secret123",
|
||||
}, allow_redirects=True)
|
||||
row = get_table("users").find_one(username=name)
|
||||
return name, row["uid"], row["api_key"]
|
||||
|
||||
|
||||
def _make_admin():
|
||||
name, uid, key = _signup()
|
||||
get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
|
||||
clear_user_cache(uid)
|
||||
return name, uid, key
|
||||
|
||||
|
||||
def _h(key=None):
|
||||
headers = {"Accept": "application/json"}
|
||||
if key:
|
||||
headers["X-API-KEY"] = key
|
||||
return headers
|
||||
|
||||
|
||||
def _create_project(key, title, is_private=False):
|
||||
data = {"title": title, "description": "visibility test",
|
||||
"project_type": "software", "status": "In Development"}
|
||||
if is_private:
|
||||
data["is_private"] = "on"
|
||||
r = requests.post(f"{BASE_URL}/projects/create", headers=_h(key), data=data)
|
||||
assert r.status_code == 200, r.text
|
||||
return r.json()["data"]
|
||||
|
||||
|
||||
def _project_uid(slug):
|
||||
return get_table("projects").find_one(slug=slug)["uid"]
|
||||
|
||||
|
||||
def _write(key, slug, path, content):
|
||||
return requests.post(f"{BASE_URL}/projects/{slug}/files/write", headers=_h(key),
|
||||
data={"path": path, "content": content}, allow_redirects=False)
|
||||
|
||||
|
||||
def _set_private(key, slug, value):
|
||||
return requests.post(f"{BASE_URL}/projects/{slug}/private", headers=_h(key),
|
||||
data={"value": 1 if value else 0}, allow_redirects=False)
|
||||
|
||||
|
||||
def _set_readonly(key, slug, value):
|
||||
return requests.post(f"{BASE_URL}/projects/{slug}/readonly", headers=_h(key),
|
||||
data={"value": 1 if value else 0}, allow_redirects=False)
|
||||
|
||||
|
||||
def _list_slugs(key=None, user_uid=None):
|
||||
params = {"user_uid": user_uid} if user_uid else None
|
||||
r = requests.get(f"{BASE_URL}/projects", headers=_h(key), params=params)
|
||||
return [p["slug"] for p in r.json()["projects"]]
|
||||
|
||||
|
||||
# ---------- privacy: listing ----------
|
||||
|
||||
def test_private_project_hidden_from_guest_listing(app_server):
|
||||
_, owner_uid, key = _signup()
|
||||
slug = _create_project(key, "Private Listing", is_private=True)["slug"]
|
||||
assert slug not in _list_slugs(key=None, user_uid=owner_uid)
|
||||
assert slug in _list_slugs(key=key, user_uid=owner_uid)
|
||||
|
||||
|
||||
def test_private_project_hidden_from_other_member(app_server):
|
||||
_, owner_uid, owner_key = _signup()
|
||||
_, _, other_key = _signup()
|
||||
slug = _create_project(owner_key, "Private FromOther", is_private=True)["slug"]
|
||||
assert slug not in _list_slugs(key=other_key, user_uid=owner_uid)
|
||||
|
||||
|
||||
def test_private_project_visible_to_admin(app_server):
|
||||
_, owner_uid, owner_key = _signup()
|
||||
_, _, admin_key = _make_admin()
|
||||
slug = _create_project(owner_key, "Private AdminSees", is_private=True)["slug"]
|
||||
assert slug in _list_slugs(key=admin_key, user_uid=owner_uid)
|
||||
|
||||
|
||||
# ---------- privacy: detail + files ----------
|
||||
|
||||
def test_private_detail_404_for_guest_200_for_owner(app_server):
|
||||
_, _, key = _signup()
|
||||
slug = _create_project(key, "Private Detail", is_private=True)["slug"]
|
||||
assert requests.get(f"{BASE_URL}/projects/{slug}", headers=_h()).status_code == 404
|
||||
assert requests.get(f"{BASE_URL}/projects/{slug}", headers=_h(key)).status_code == 200
|
||||
|
||||
|
||||
def test_private_files_hidden_from_guest(app_server):
|
||||
_, _, key = _signup()
|
||||
slug = _create_project(key, "Private Files", is_private=True)["slug"]
|
||||
_write(key, slug, "secret.txt", "classified")
|
||||
assert requests.get(f"{BASE_URL}/projects/{slug}/files", headers=_h()).status_code == 404
|
||||
assert requests.get(f"{BASE_URL}/projects/{slug}/files/raw", params={"path": "secret.txt"},
|
||||
headers=_h()).status_code == 404
|
||||
assert requests.get(f"{BASE_URL}/projects/{slug}/files", headers=_h(key)).status_code == 200
|
||||
|
||||
|
||||
def test_private_detail_visible_to_admin(app_server):
|
||||
_, _, owner_key = _signup()
|
||||
_, _, admin_key = _make_admin()
|
||||
slug = _create_project(owner_key, "Private AdminDetail", is_private=True)["slug"]
|
||||
assert requests.get(f"{BASE_URL}/projects/{slug}", headers=_h(admin_key)).status_code == 200
|
||||
|
||||
|
||||
# ---------- privacy: sitemap ----------
|
||||
|
||||
def test_private_project_excluded_from_sitemap(app_server):
|
||||
_, _, key = _signup()
|
||||
public_slug = _create_project(key, "Sitemap Public")["slug"]
|
||||
private_slug = _create_project(key, "Sitemap Private", is_private=True)["slug"]
|
||||
xml = requests.get(f"{BASE_URL}/sitemap.xml").text
|
||||
assert f"/projects/{public_slug}" in xml
|
||||
assert f"/projects/{private_slug}" not in xml
|
||||
|
||||
|
||||
# ---------- privacy: toggle ----------
|
||||
|
||||
def test_toggle_private_then_public(app_server):
|
||||
_, owner_uid, key = _signup()
|
||||
slug = _create_project(key, "Toggle Privacy")["slug"]
|
||||
assert slug in _list_slugs(key=None, user_uid=owner_uid)
|
||||
assert _set_private(key, slug, True).status_code == 200
|
||||
assert slug not in _list_slugs(key=None, user_uid=owner_uid)
|
||||
assert _set_private(key, slug, False).status_code == 200
|
||||
assert slug in _list_slugs(key=None, user_uid=owner_uid)
|
||||
|
||||
|
||||
def test_non_owner_cannot_toggle_flags(app_server):
|
||||
_, _, owner_key = _signup()
|
||||
_, _, other_key = _signup()
|
||||
slug = _create_project(owner_key, "Toggle Guarded")["slug"]
|
||||
assert _set_private(other_key, slug, True).status_code == 403
|
||||
assert _set_readonly(other_key, slug, True).status_code == 403
|
||||
|
||||
|
||||
# ---------- read-only: HTTP enforcement ----------
|
||||
|
||||
def test_readonly_blocks_every_mutation(app_server):
|
||||
_, _, key = _signup()
|
||||
slug = _create_project(key, "Readonly Block")["slug"]
|
||||
_write(key, slug, "main.py", "print(1)\n")
|
||||
assert _set_readonly(key, slug, True).status_code == 200
|
||||
|
||||
assert _write(key, slug, "main.py", "print(2)\n").status_code == 400
|
||||
assert _write(key, slug, "new.py", "print(3)\n").status_code == 400
|
||||
assert requests.post(f"{BASE_URL}/projects/{slug}/files/mkdir", headers=_h(key),
|
||||
data={"path": "docs"}, allow_redirects=False).status_code == 400
|
||||
assert requests.post(f"{BASE_URL}/projects/{slug}/files/append", headers=_h(key),
|
||||
data={"path": "main.py", "content": "x"}, allow_redirects=False).status_code == 400
|
||||
assert requests.post(f"{BASE_URL}/projects/{slug}/files/replace-lines", headers=_h(key),
|
||||
data={"path": "main.py", "start": 1, "end": 1, "content": "z"},
|
||||
allow_redirects=False).status_code == 400
|
||||
assert requests.post(f"{BASE_URL}/projects/{slug}/files/move", headers=_h(key),
|
||||
data={"from_path": "main.py", "to_path": "renamed.py"},
|
||||
allow_redirects=False).status_code == 400
|
||||
assert requests.post(f"{BASE_URL}/projects/{slug}/files/delete", headers=_h(key),
|
||||
data={"path": "main.py"}, allow_redirects=False).status_code == 400
|
||||
assert requests.post(f"{BASE_URL}/projects/{slug}/files/upload", headers=_h(key),
|
||||
files={"file": ("u.py", b"x=1\n")}, data={"path": ""}).status_code == 400
|
||||
|
||||
|
||||
def test_readonly_unchanged_content(app_server):
|
||||
_, _, key = _signup()
|
||||
slug = _create_project(key, "Readonly Content")["slug"]
|
||||
_write(key, slug, "a.txt", "original")
|
||||
_set_readonly(key, slug, True)
|
||||
_write(key, slug, "a.txt", "tampered")
|
||||
body = requests.get(f"{BASE_URL}/projects/{slug}/files/raw", params={"path": "a.txt"},
|
||||
headers=_h(key)).json()
|
||||
assert body["content"] == "original"
|
||||
|
||||
|
||||
def test_toggle_readonly_off_restores_writes(app_server):
|
||||
_, _, key = _signup()
|
||||
slug = _create_project(key, "Readonly Restore")["slug"]
|
||||
_write(key, slug, "a.txt", "one")
|
||||
_set_readonly(key, slug, True)
|
||||
assert _write(key, slug, "a.txt", "two").status_code == 400
|
||||
assert _set_readonly(key, slug, False).status_code == 200
|
||||
assert _write(key, slug, "a.txt", "two").status_code == 200
|
||||
|
||||
|
||||
def test_delete_project_works_when_readonly(app_server):
|
||||
_, _, key = _signup()
|
||||
project = _create_project(key, "Readonly Delete")
|
||||
slug = project["slug"]
|
||||
_write(key, slug, "a.txt", "one")
|
||||
_set_readonly(key, slug, True)
|
||||
r = requests.post(f"{BASE_URL}/projects/delete/{slug}", headers=_h(key), allow_redirects=False)
|
||||
assert r.status_code == 200 and r.json()["ok"] is True
|
||||
assert requests.get(f"{BASE_URL}/projects/{slug}", headers=_h(key)).status_code == 404
|
||||
|
||||
|
||||
# ---------- read-only: service layer enforcement ----------
|
||||
|
||||
def test_readonly_guards_service_layer(app_server):
|
||||
_, owner_uid, key = _signup()
|
||||
slug = _create_project(key, "Readonly Service")["slug"]
|
||||
project_uid = _project_uid(slug)
|
||||
user = get_table("users").find_one(uid=owner_uid)
|
||||
project_files.write_text_file(project_uid, user, "seed.txt", "seed")
|
||||
_set_readonly(key, slug, True)
|
||||
|
||||
assert project_files.is_readonly(project_uid) is True
|
||||
with pytest.raises(ProjectFileError):
|
||||
project_files.write_text_file(project_uid, user, "seed.txt", "blocked")
|
||||
with pytest.raises(ProjectFileError):
|
||||
project_files.make_dir(project_uid, user, "docs")
|
||||
|
||||
|
||||
def test_readonly_blocks_import_from_dir(app_server, tmp_path):
|
||||
_, owner_uid, key = _signup()
|
||||
slug = _create_project(key, "Readonly Import")["slug"]
|
||||
project_uid = _project_uid(slug)
|
||||
user = get_table("users").find_one(uid=owner_uid)
|
||||
(tmp_path / "imported.txt").write_text("data")
|
||||
_set_readonly(key, slug, True)
|
||||
with pytest.raises(ProjectFileError):
|
||||
project_files.import_from_dir(project_uid, str(tmp_path), user)
|
||||
|
||||
|
||||
# ---------- devii confirmation gate ----------
|
||||
|
||||
def test_readonly_action_requires_confirmation():
|
||||
assert confirmation_error("project_set_readonly", {}) is not None
|
||||
assert confirmation_error("project_set_readonly", {"confirm": "false"}) is not None
|
||||
assert confirmation_error("project_set_readonly", {"confirm": "true"}) is None
|
||||
|
||||
|
||||
def test_private_action_needs_no_confirmation():
|
||||
assert confirmation_error("project_set_private", {"value": "true"}) is None
|
||||
assert confirmation_error("vote", {}) is None
|
||||
|
||||
|
||||
def test_is_confirmed_accepts_common_truthy():
|
||||
assert _is_confirmed({"confirm": True})
|
||||
assert _is_confirmed({"confirm": "yes"})
|
||||
assert not _is_confirmed({"confirm": ""})
|
||||
assert not _is_confirmed({})
|
||||
Reference in New Issue
Block a user