retoor/devplacepy
1
0
Fork 1
Code Issues 44 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

chore: reorganize test files into domain-specific subdirectories

Browse Source
This commit is contained in:
retoor
2026-06-13 14:32:33 +00:00
parent 8eac4cafba
commit eccec02336
268 changed files with 20634 additions and 9206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/api/projects/files/__init__.py
View File
tests/api/projects/files/lines.py
+113
View File
@@ -0,0 +1,113 @@
# retoor <retoor@molodetz.nl>
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_project_file_lines():
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
_counter_project_file_lines = [0]
def _signup_project_file_lines():
_counter_project_file_lines[0] += 1
name = f"pl{int(time.time() * 1000)}{_counter_project_file_lines[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_project_file_lines(key):
return {"X-API-KEY": key, "Accept": "application/json"}
def _create_project_project_file_lines(key, title):
r = requests.post(
f"{BASE_URL}/projects/create",
headers=_h_project_file_lines(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_project_file_lines(key, slug, path, content):
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
headers=_h_project_file_lines(key),
data={"path": path, "content": content},
allow_redirects=False,
)
assert r.status_code in (200, 302), r.text
def _raw_project_file_lines(key, slug, path):
return requests.get(
f"{BASE_URL}/projects/{slug}/files/raw", headers=_h_project_file_lines(key), params={"path": path}
).json()
class _FakeResponse:
def __init__(self, status_code=200):
self.status_code = status_code
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 _FakeResponse()
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_http_read_lines(app_server):
_, key = _signup_project_file_lines()
proj = _create_project_project_file_lines(key, "HTTP Read Lines")
slug = proj["slug"] or proj["uid"]
_write_project_file_lines(key, slug, "f.txt", "a\nb\nc\nd")
r = requests.get(
f"{BASE_URL}/projects/{slug}/files/lines",
headers=_h_project_file_lines(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_lines_missing_file_404(app_server):
_, key = _signup_project_file_lines()
proj = _create_project_project_file_lines(key, "HTTP Lines 404")
slug = proj["slug"] or proj["uid"]
r = requests.get(
f"{BASE_URL}/projects/{slug}/files/lines",
headers=_h_project_file_lines(key),
params={"path": "nope.txt"},
)
assert r.status_code == 404
tests/api/projects/files/raw.py
+116
View File
@@ -0,0 +1,116 @@
# retoor <retoor@molodetz.nl>
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_project_visibility = [0]
def _signup_project_visibility():
_counter_project_visibility[0] += 1
name = f"pv{int(time.time() * 1000)}{_counter_project_visibility[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_project_visibility():
name, uid, key = _signup_project_visibility()
get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
clear_user_cache(uid)
return name, uid, key
def _h_project_visibility(key=None):
headers = {"Accept": "application/json"}
if key:
headers["X-API-KEY"] = key
return headers
def _create_project_project_visibility(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_project_visibility(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_project_visibility(key, slug, path, content):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
headers=_h_project_visibility(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_project_visibility(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_project_visibility(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_project_visibility(key), params=params)
return [p["slug"] for p in r.json()["projects"]]
def test_private_files_hidden_from_guest(app_server):
_, _, key = _signup_project_visibility()
slug = _create_project_project_visibility(key, "Private Files", is_private=True)["slug"]
_write_project_visibility(key, slug, "secret.txt", "classified")
assert (
requests.get(f"{BASE_URL}/projects/{slug}/files", headers=_h_project_visibility()).status_code
== 404
)
assert (
requests.get(
f"{BASE_URL}/projects/{slug}/files/raw",
params={"path": "secret.txt"},
headers=_h_project_visibility(),
).status_code
== 404
)
assert (
requests.get(f"{BASE_URL}/projects/{slug}/files", headers=_h_project_visibility(key)).status_code
== 200
)
def test_readonly_unchanged_content(app_server):
_, _, key = _signup_project_visibility()
slug = _create_project_project_visibility(key, "Readonly Content")["slug"]
_write_project_visibility(key, slug, "a.txt", "original")
_set_readonly(key, slug, True)
_write_project_visibility(key, slug, "a.txt", "tampered")
body = requests.get(
f"{BASE_URL}/projects/{slug}/files/raw",
params={"path": "a.txt"},
headers=_h_project_visibility(key),
).json()
assert body["content"] == "original"
tests/api/projects/files/replacelines.py
+272
View File
@@ -0,0 +1,272 @@
# retoor <retoor@molodetz.nl>
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_project_file_lines():
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
_counter_project_file_lines = [0]
def _signup_project_file_lines():
_counter_project_file_lines[0] += 1
name = f"pl{int(time.time() * 1000)}{_counter_project_file_lines[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_project_file_lines(key):
return {"X-API-KEY": key, "Accept": "application/json"}
def _create_project_project_file_lines(key, title):
r = requests.post(
f"{BASE_URL}/projects/create",
headers=_h_project_file_lines(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_project_file_lines(key, slug, path, content):
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
headers=_h_project_file_lines(key),
data={"path": path, "content": content},
allow_redirects=False,
)
assert r.status_code in (200, 302), r.text
def _raw_project_file_lines(key, slug, path):
return requests.get(
f"{BASE_URL}/projects/{slug}/files/raw", headers=_h_project_file_lines(key), params={"path": path}
).json()
class _FakeResponse:
def __init__(self, status_code=200):
self.status_code = status_code
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 _FakeResponse()
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
from devplacepy.database import get_table
from devplacepy.utils import clear_user_cache
from devplacepy import project_files
from devplacepy.services.devii.actions.dispatcher import (
confirmation_error,
_is_confirmed,
)
_counter_project_visibility = [0]
def _signup_project_visibility():
_counter_project_visibility[0] += 1
name = f"pv{int(time.time() * 1000)}{_counter_project_visibility[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_project_visibility():
name, uid, key = _signup_project_visibility()
get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
clear_user_cache(uid)
return name, uid, key
def _h_project_visibility(key=None):
headers = {"Accept": "application/json"}
if key:
headers["X-API-KEY"] = key
return headers
def _create_project_project_visibility(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_project_visibility(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_project_visibility(key, slug, path, content):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
headers=_h_project_visibility(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_project_visibility(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_project_visibility(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_project_visibility(key), params=params)
return [p["slug"] for p in r.json()["projects"]]
def test_http_replace_insert_delete_append_roundtrip(app_server):
_, key = _signup_project_file_lines()
proj = _create_project_project_file_lines(key, "HTTP Edit Roundtrip")
slug = proj["slug"] or proj["uid"]
_write_project_file_lines(key, slug, "f.txt", "a\nb\nc\nd")
requests.post(
f"{BASE_URL}/projects/{slug}/files/replace-lines",
headers=_h_project_file_lines(key),
data={"path": "f.txt", "start": 2, "end": 3, "content": "X\nY"},
allow_redirects=False,
)
assert _raw_project_file_lines(key, slug, "f.txt")["content"] == "a\nX\nY\nd"
requests.post(
f"{BASE_URL}/projects/{slug}/files/insert-lines",
headers=_h_project_file_lines(key),
data={"path": "f.txt", "at": 1, "content": "TOP"},
allow_redirects=False,
)
assert _raw_project_file_lines(key, slug, "f.txt")["content"] == "TOP\na\nX\nY\nd"
requests.post(
f"{BASE_URL}/projects/{slug}/files/delete-lines",
headers=_h_project_file_lines(key),
data={"path": "f.txt", "start": 1, "end": 1},
allow_redirects=False,
)
assert _raw_project_file_lines(key, slug, "f.txt")["content"] == "a\nX\nY\nd"
requests.post(
f"{BASE_URL}/projects/{slug}/files/append",
headers=_h_project_file_lines(key),
data={"path": "f.txt", "content": "END"},
allow_redirects=False,
)
assert _raw_project_file_lines(key, slug, "f.txt")["content"] == "a\nX\nY\nd\nEND"
def test_http_replace_lines_non_owner_denied(app_server):
_, owner_key = _signup_project_file_lines()
_, other_key = _signup_project_file_lines()
proj = _create_project_project_file_lines(owner_key, "HTTP Owner Guard")
slug = proj["slug"] or proj["uid"]
_write_project_file_lines(owner_key, slug, "f.txt", "a\nb")
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/replace-lines",
headers=_h_project_file_lines(other_key),
data={"path": "f.txt", "start": 1, "end": 1, "content": "x"},
allow_redirects=False,
)
assert r.status_code == 403
def test_readonly_blocks_every_mutation(app_server):
_, _, key = _signup_project_visibility()
slug = _create_project_project_visibility(key, "Readonly Block")["slug"]
_write_project_visibility(key, slug, "main.py", "print(1)\n")
assert _set_readonly(key, slug, True).status_code == 200
assert _write_project_visibility(key, slug, "main.py", "print(2)\n").status_code == 400
assert _write_project_visibility(key, slug, "new.py", "print(3)\n").status_code == 400
assert (
requests.post(
f"{BASE_URL}/projects/{slug}/files/mkdir",
headers=_h_project_visibility(key),
data={"path": "docs"},
allow_redirects=False,
).status_code
== 400
)
assert (
requests.post(
f"{BASE_URL}/projects/{slug}/files/append",
headers=_h_project_visibility(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_project_visibility(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_project_visibility(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_project_visibility(key),
data={"path": "main.py"},
allow_redirects=False,
).status_code
== 400
)
assert (
requests.post(
f"{BASE_URL}/projects/{slug}/files/upload",
headers=_h_project_visibility(key),
files={"file": ("u.py", b"x=1\n")},
data={"path": ""},
).status_code
== 400
)
tests/api/projects/files/upload.py
+147
View File
@@ -0,0 +1,147 @@
# retoor <retoor@molodetz.nl>
import time
import requests
from playwright.sync_api import expect
from tests.conftest import BASE_URL
from devplacepy.database import get_table
_counter_project_files = [0]
def _signup_project_files():
_counter_project_files[0] += 1
name = f"pf{int(time.time() * 1000)}{_counter_project_files[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,
)
key = get_table("users").find_one(username=name)["api_key"]
return name, key
def _h_project_files(key):
return {"X-API-KEY": key, "Accept": "application/json"}
def _create_project_project_files(key, title):
r = requests.post(
f"{BASE_URL}/projects/create",
headers=_h_project_files(key),
data={
"title": title,
"description": "filesystem test",
"project_type": "software",
"status": "In Development",
},
)
assert r.status_code == 200, r.text
return r.json()["data"]
def _write_project_files(key, slug, path, content):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
headers=_h_project_files(key),
data={"path": path, "content": content},
allow_redirects=False,
)
def _mkdir(key, slug, path):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/mkdir",
headers=_h_project_files(key),
data={"path": path},
allow_redirects=False,
)
def _move(key, slug, from_path, to_path):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/move",
headers=_h_project_files(key),
data={"from_path": from_path, "to_path": to_path},
allow_redirects=False,
)
def _delete(key, slug, path):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/delete",
headers=_h_project_files(key),
data={"path": path},
allow_redirects=False,
)
def _list(slug, key=None):
return requests.get(
f"{BASE_URL}/projects/{slug}/files",
headers=_h_project_files(key) if key else {"Accept": "application/json"},
)
def _raw_project_files(slug, path, key=None):
return requests.get(
f"{BASE_URL}/projects/{slug}/files/raw",
params={"path": path},
headers=_h_project_files(key) if key else {"Accept": "application/json"},
)
def _paths(slug, key=None):
return sorted(f["path"] for f in _list(slug, key).json()["files"])
def _make_project_ui(page, title):
page.goto(f"{BASE_URL}/projects", wait_until="domcontentloaded")
page.locator("#create-project-btn").click()
page.fill("#title", title)
page.fill("#description", "Project for filesystem UI test")
page.click("button:has-text('Create Project')")
page.wait_for_url(f"{BASE_URL}/projects/*", wait_until="domcontentloaded")
return page.url
def _open_files(page, title):
proj_url = _make_project_ui(page, title)
slug = proj_url.rstrip("/").split("/")[-1]
page.goto(proj_url + "/files", wait_until="domcontentloaded")
return slug
def _dialog_fill(page, value):
page.locator(".dialog-overlay.visible .dialog-input").wait_for(state="visible")
page.fill(".dialog-overlay.visible .dialog-input", value)
page.click(".dialog-overlay.visible .dialog-confirm")
def _dialog_confirm(page):
page.locator(".dialog-overlay.visible .dialog-confirm").wait_for(state="visible")
page.click(".dialog-overlay.visible .dialog-confirm")
def _dialog_cancel(page):
page.locator(".dialog-overlay.visible .dialog-cancel").wait_for(state="visible")
page.click(".dialog-overlay.visible .dialog-cancel")
def _new_folder(page, name):
page.click("#pf-new-folder")
_dialog_fill(page, name)
page.wait_for_selector(f".pf-node-row:has-text('{name.split('/')[-1]}')")
def _new_file(page, name):
page.click("#pf-new-file")
_dialog_fill(page, name)
def _row(page, name):
return page.locator(f".pf-node-row:has-text('{name}')").first
def _alice_key():
return get_table("users").find_one(username="alice_test")["api_key"]
def test_upload_text_is_editable(app_server):
_, key = _signup_project_files()
slug = _create_project_project_files(key, "FS UploadText")["slug"]
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/upload",
headers=_h_project_files(key),
files={"file": ("util.py", b"x = 1\n")},
data={"path": "src"},
)
assert r.status_code == 200
node = _raw_project_files(slug, "src/util.py", key).json()
assert node["is_binary"] is False and node["content"] == "x = 1\n"
def test_upload_binary_is_served(app_server):
_, key = _signup_project_files()
slug = _create_project_project_files(key, "FS UploadBinary")["slug"]
blob = bytes(range(256))
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/upload",
headers=_h_project_files(key),
files={"file": ("logo.bin", blob)},
data={"path": "assets"},
)
assert r.status_code == 200
node = _raw_project_files(slug, "assets/logo.bin", key).json()
assert node["is_binary"] is True and node["url"].startswith(
"/static/uploads/project_files/"
)
served = requests.get(BASE_URL + node["url"])
assert served.status_code == 200 and served.content == blob
tests/api/projects/files/write.py
+126
View File
@@ -0,0 +1,126 @@
# retoor <retoor@molodetz.nl>
import time
import requests
from playwright.sync_api import expect
from tests.conftest import BASE_URL
from devplacepy.database import get_table
_counter_project_files = [0]
def _signup_project_files():
_counter_project_files[0] += 1
name = f"pf{int(time.time() * 1000)}{_counter_project_files[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,
)
key = get_table("users").find_one(username=name)["api_key"]
return name, key
def _h_project_files(key):
return {"X-API-KEY": key, "Accept": "application/json"}
def _create_project_project_files(key, title):
r = requests.post(
f"{BASE_URL}/projects/create",
headers=_h_project_files(key),
data={
"title": title,
"description": "filesystem test",
"project_type": "software",
"status": "In Development",
},
)
assert r.status_code == 200, r.text
return r.json()["data"]
def _write_project_files(key, slug, path, content):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
headers=_h_project_files(key),
data={"path": path, "content": content},
allow_redirects=False,
)
def _mkdir(key, slug, path):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/mkdir",
headers=_h_project_files(key),
data={"path": path},
allow_redirects=False,
)
def _move(key, slug, from_path, to_path):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/move",
headers=_h_project_files(key),
data={"from_path": from_path, "to_path": to_path},
allow_redirects=False,
)
def _delete(key, slug, path):
return requests.post(
f"{BASE_URL}/projects/{slug}/files/delete",
headers=_h_project_files(key),
data={"path": path},
allow_redirects=False,
)
def _list(slug, key=None):
return requests.get(
f"{BASE_URL}/projects/{slug}/files",
headers=_h_project_files(key) if key else {"Accept": "application/json"},
)
def _raw_project_files(slug, path, key=None):
return requests.get(
f"{BASE_URL}/projects/{slug}/files/raw",
params={"path": path},
headers=_h_project_files(key) if key else {"Accept": "application/json"},
)
def _paths(slug, key=None):
return sorted(f["path"] for f in _list(slug, key).json()["files"])
def _make_project_ui(page, title):
page.goto(f"{BASE_URL}/projects", wait_until="domcontentloaded")
page.locator("#create-project-btn").click()
page.fill("#title", title)
page.fill("#description", "Project for filesystem UI test")
page.click("button:has-text('Create Project')")
page.wait_for_url(f"{BASE_URL}/projects/*", wait_until="domcontentloaded")
return page.url
def _open_files(page, title):
proj_url = _make_project_ui(page, title)
slug = proj_url.rstrip("/").split("/")[-1]
page.goto(proj_url + "/files", wait_until="domcontentloaded")
return slug
def _dialog_fill(page, value):
page.locator(".dialog-overlay.visible .dialog-input").wait_for(state="visible")
page.fill(".dialog-overlay.visible .dialog-input", value)
page.click(".dialog-overlay.visible .dialog-confirm")
def _dialog_confirm(page):
page.locator(".dialog-overlay.visible .dialog-confirm").wait_for(state="visible")
page.click(".dialog-overlay.visible .dialog-confirm")
def _dialog_cancel(page):
page.locator(".dialog-overlay.visible .dialog-cancel").wait_for(state="visible")
page.click(".dialog-overlay.visible .dialog-cancel")
def _new_folder(page, name):
page.click("#pf-new-folder")
_dialog_fill(page, name)
page.wait_for_selector(f".pf-node-row:has-text('{name.split('/')[-1]}')")
def _new_file(page, name):
page.click("#pf-new-file")
_dialog_fill(page, name)
def _row(page, name):
return page.locator(f".pf-node-row:has-text('{name}')").first
def _alice_key():
return get_table("users").find_one(username="alice_test")["api_key"]
def test_guest_write_blocked(app_server):
_, key = _signup_project_files()
slug = _create_project_project_files(key, "FS Guest")["slug"]
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
data={"path": "x.py", "content": "1"},
allow_redirects=False,
)
assert r.status_code in (303, 401)
assert _paths(slug, key) == []
tests/api/projects/files/zip.py
+143
View File
@@ -0,0 +1,143 @@
# retoor <retoor@molodetz.nl>
import asyncio
import io
import json
import time
import zipfile
from pathlib import Path
import requests
from playwright.sync_api import expect
from tests.conftest import BASE_URL
from devplacepy.database import get_table, refresh_snapshot
from devplacepy.services.jobs import queue
from devplacepy.services.jobs.zip_service import ZipService
from tests.conftest import run_async
_counter_zip_download = [0]
def _signup_zip_download():
_counter_zip_download[0] += 1
name = f"zd{int(time.time() * 1000)}{_counter_zip_download[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,
)
refresh_snapshot()
key = get_table("users").find_one(username=name)["api_key"]
return name, key
def _h_zip_download(key):
return {"X-API-KEY": key, "Accept": "application/json"}
def _create_project_zip_download(key, title):
r = requests.post(
f"{BASE_URL}/projects/create",
headers=_h_zip_download(key),
data={
"title": title,
"description": "zip download test",
"project_type": "software",
"status": "In Development",
},
)
assert r.status_code == 200, r.text
return r.json()["data"]
def _write_zip_download(key, slug, path, content):
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/write",
headers=_h_zip_download(key),
data={"path": path, "content": content},
allow_redirects=False,
)
assert r.status_code in (200, 302), r.text
def _process_uid(uid):
async def drive():
svc = ZipService()
for _ in range(400):
await svc.run_once()
refresh_snapshot()
job = queue.get_job(uid)
if job and job["status"] in ("done", "failed"):
return
await asyncio.sleep(0.05)
run_async(drive())
def _drain_pending():
async def drive():
svc = ZipService()
appeared = False
for _ in range(600):
refresh_snapshot()
pending = [
r
for r in get_table("jobs").find(kind="zip")
if r["status"] in ("pending", "running")
]
if pending:
appeared = True
await svc.run_once()
refresh_snapshot()
pending = [
r
for r in get_table("jobs").find(kind="zip")
if r["status"] in ("pending", "running")
]
if appeared and not pending and not svc._inflight:
return
await asyncio.sleep(0.05)
run_async(drive())
def _cleanup_archives():
refresh_snapshot()
jobs = get_table("jobs")
for row in list(jobs.find(kind="zip")):
result = json.loads(row.get("result") or "{}")
local_path = result.get("local_path")
if local_path:
Path(local_path).unlink(missing_ok=True)
jobs.delete(uid=row["uid"])
def _login_zip_download(page, name):
page.goto(f"{BASE_URL}/auth/login", wait_until="domcontentloaded")
page.fill("#email", f"{name}@t.dev")
page.fill("#password", "secret123")
page.click("button:has-text('Sign in')")
page.wait_for_url("**/feed", timeout=10000, wait_until="domcontentloaded")
def test_files_zip_subtree_download(app_server):
try:
_, key = _signup_zip_download()
proj = _create_project_zip_download(key, "Subtree Flow")
slug = proj["slug"] or proj["uid"]
_write_zip_download(key, slug, "src/app.py", "print(1)\n")
_write_zip_download(key, slug, "docs/readme.md", "x")
uid = requests.post(
f"{BASE_URL}/projects/{slug}/files/zip",
headers=_h_zip_download(key),
params={"path": "src"},
).json()["uid"]
_process_uid(uid)
done = requests.get(
f"{BASE_URL}/zips/{uid}", headers={"Accept": "application/json"}
).json()
dl = requests.get(f"{BASE_URL}{done['download_url']}")
names = sorted(zipfile.ZipFile(io.BytesIO(dl.content)).namelist())
assert names == ["src/", "src/app.py"]
finally:
_cleanup_archives()
def test_files_zip_rejects_traversal(app_server):
_, key = _signup_zip_download()
proj = _create_project_zip_download(key, "Traversal Guard")
slug = proj["slug"] or proj["uid"]
r = requests.post(
f"{BASE_URL}/projects/{slug}/files/zip",
headers=_h_zip_download(key),
params={"path": "../../etc/passwd"},
)
assert r.status_code == 400, r.text