Add âž•
@@ -11,10 +11,7 @@
|
Tags
- {% if user.is_authenticated and user.is_superuser and request.archivebox_config.OPENCODE_ENABLED %}
-
💬 AI
-
|
- {% endif %}
+ {% plugin_ui "opencode" "navigation" %}
Docs
|
API
diff --git a/archivebox/tests/test_opencode_agent.py b/archivebox/tests/test_opencode_agent.py
index af0ef60c..9e6ea78b 100644
--- a/archivebox/tests/test_opencode_agent.py
+++ b/archivebox/tests/test_opencode_agent.py
@@ -1,19 +1,18 @@
import asyncio
-import json
import os
-import signal
+import shutil
import socket
-import subprocess
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from types import SimpleNamespace
-from urllib.parse import quote
+from urllib.parse import parse_qs, quote, urlsplit
import pytest
import requests
from asgiref.testing import ApplicationCommunicator
from archivebox.tests.conftest import ADMIN_TEST_HOST, run_archivebox_cmd
+from archivebox.config.common import get_config
pytestmark = pytest.mark.django_db(transaction=True)
@@ -78,8 +77,8 @@ def opencode_archive_config(initialized_archive):
@pytest.fixture
-def live_opencode(opencode_archive_config):
- from archivebox.opencode import views
+def installed_opencode(opencode_archive_config):
+ from abx_plugins.plugins.opencode import runtime
install = run_archivebox_cmd(
["install", "opencode", "--binproviders=env,pnpm"],
@@ -90,56 +89,61 @@ def live_opencode(opencode_archive_config):
assert install.returncode == 0, install.stderr or install.stdout
_reset_runtime_config()
- config = views._machine_config()
- settings = views._settings(config)
+ config = get_config().model_dump(mode="json")
+ settings = runtime._settings(config, opencode_archive_config.data_dir)
settings["archivebox_base_url"] = "http://admin.archivebox.localhost:8000"
settings["archivebox_admin_url"] = "http://admin.archivebox.localhost:8000/admin"
settings["archivebox_api_url"] = "http://admin.archivebox.localhost:8000/api/"
- binary, _, binary_env = views._resolve_binary(settings["binary"], settings["config"])
+ binary, binary_env = runtime._resolve_binary(settings["binary"], settings["config"])
version = binary.exec(
cmd=("--version",),
env={**os.environ, **binary_env},
timeout=120,
)
assert version.returncode == 0, version.stderr or version.stdout
- ok, error = views._ensure_opencode(settings)
+ return SimpleNamespace(config=opencode_archive_config, settings=settings)
+
+
+@pytest.fixture
+def live_opencode(installed_opencode):
+ from abx_plugins.plugins.opencode import runtime
+
+ settings = installed_opencode.settings
+ ok, error = runtime._ensure_opencode(settings)
assert ok, error
- process = views._PROCESS
+ process = runtime._PROCESS
assert process is not None
try:
- yield SimpleNamespace(config=opencode_archive_config, settings=settings, process=process)
+ yield SimpleNamespace(config=installed_opencode.config, settings=settings, process=process)
finally:
- views._stop_owned_process()
+ runtime._stop_owned_process()
def test_opencode_disabled_route_does_not_start_server(client, initialized_archive):
from archivebox.machine.models import Machine
- from archivebox.opencode import views
+ from abx_plugins.plugins.opencode import runtime
os.chdir(initialized_archive)
Machine.from_json({"config": {"OPENCODE_ENABLED": False}})
_reset_runtime_config()
- assert views._machine_config()["OPENCODE_ENABLED"] is False
+ assert get_config().model_dump(mode="json")["OPENCODE_ENABLED"] is False
response = client.get("/admin/agent", HTTP_HOST=ADMIN_TEST_HOST)
assert response.status_code == 404
- assert views._PROCESS is None or views._PROCESS.poll() is not None
+ assert runtime._PROCESS is None or runtime._PROCESS.poll() is not None
-def test_stop_owned_process_falls_back_for_stopped_process_without_dedicated_group():
- from archivebox.opencode import views
+def test_opencode_disabled_via_cli_stays_disabled(admin_client, initialized_archive):
+ _set_archivebox_config(initialized_archive, "OPENCODE_ENABLED=False")
- process = subprocess.Popen(["sleep", "60"])
- try:
- process.send_signal(signal.SIGSTOP)
- views._stop_owned_process(process)
- assert process.returncode == -signal.SIGTERM
- finally:
- if process.poll() is None:
- process.kill()
- process.wait()
+ assert get_config().OPENCODE_ENABLED is False
+ assert admin_client.get("/admin/agent", HTTP_HOST=ADMIN_TEST_HOST).status_code == 404
+ for path in ("/add/", "/admin/core/snapshot/"):
+ response = admin_client.get(path, HTTP_HOST=ADMIN_TEST_HOST)
+ assert response.status_code == 200
+ assert b'href="/admin/agent"' not in response.content
def test_opencode_agent_requires_superuser_when_enabled(client, db, django_user_model, live_opencode):
@@ -147,6 +151,11 @@ def test_opencode_agent_requires_superuser_when_enabled(client, db, django_user_
assert response.status_code == 302
assert "/admin/login/" in response.headers["Location"]
+ next_path = "/admin/agent?x=1&next=https://example.com"
+ response = client.get(next_path, HTTP_HOST=ADMIN_TEST_HOST)
+ assert response.status_code == 302
+ assert parse_qs(urlsplit(response.headers["Location"]).query) == {"next": [next_path]}
+
user = django_user_model.objects.create_user(username="regular", password="testpassword")
client.force_login(user)
response = client.get("/admin/agent", HTTP_HOST=ADMIN_TEST_HOST)
@@ -178,18 +187,22 @@ def test_opencode_proxy_blocks_cross_site_fetch_metadata(admin_client, db, live_
def test_opencode_agent_superuser_gets_admin_wrapper(admin_client, live_opencode):
- from archivebox.opencode import views
+ from abx_plugins.plugins.opencode import runtime
response = admin_client.get("/admin/agent", HTTP_HOST=ADMIN_TEST_HOST)
recent_session_id = response.context["recent_session_id"]
- session_path = views._project_route(live_opencode.config.data_dir, recent_session_id)
+ session_path = runtime._project_route(live_opencode.config.data_dir, recent_session_id)
assert response.status_code == 200
assert recent_session_id
assert f'
Get the extension