115 lines
3.9 KiB
Python
115 lines
3.9 KiB
Python
import ast
|
|
import logging
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import schedule
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_web_bootstrap_does_not_eagerly_import_scheduler_runtime():
|
|
source = (ROOT / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
imports = []
|
|
for node in tree.body:
|
|
if isinstance(node, ast.Import):
|
|
imports.extend(alias.name for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
imports.append(node.module)
|
|
|
|
assert "scheduler" not in imports
|
|
assert "schedule" not in imports
|
|
assert "services.web_scheduler_compat" in source
|
|
|
|
|
|
def test_openclaw_blueprint_registration_has_no_worker_side_effect():
|
|
source = (ROOT / "routes/openclaw_bot_routes.py").read_text(encoding="utf-8")
|
|
registration = source[source.index("@openclaw_bot_bp.record_once") :]
|
|
|
|
assert "start_scheduler()" not in registration
|
|
assert "timed jobs delegated to momo-scheduler" in registration
|
|
|
|
|
|
def test_openclaw_schedules_are_centralized_and_lazy(monkeypatch):
|
|
from services.openclaw_bot import scheduled_jobs
|
|
|
|
schedule.clear(scheduled_jobs.JOB_TAG)
|
|
try:
|
|
jobs = scheduled_jobs.register_openclaw_schedules(schedule, logging.getLogger("test"))
|
|
assert len(jobs) == 10
|
|
assert len(schedule.get_jobs(scheduled_jobs.JOB_TAG)) == 10
|
|
assert "routes.openclaw_bot_routes" not in scheduled_jobs.__dict__
|
|
|
|
called = []
|
|
fake_routes = SimpleNamespace(send_morning_report=lambda: called.append("morning"))
|
|
monkeypatch.setattr(
|
|
scheduled_jobs.importlib,
|
|
"import_module",
|
|
lambda name: fake_routes,
|
|
)
|
|
result = scheduled_jobs.run_openclaw_job("send_morning_report")
|
|
assert result["status"] == "completed"
|
|
assert called == ["morning"]
|
|
finally:
|
|
schedule.clear(scheduled_jobs.JOB_TAG)
|
|
|
|
|
|
def test_embedding_worker_requires_explicit_idempotent_start(monkeypatch):
|
|
from services.openclaw_bot import embedding_worker_runtime as runtime
|
|
|
|
created = []
|
|
|
|
class FakeThread:
|
|
def __init__(self, *, target, daemon, name):
|
|
self.target = target
|
|
self.daemon = daemon
|
|
self.name = name
|
|
self.started = False
|
|
created.append(self)
|
|
|
|
def start(self):
|
|
self.started = True
|
|
|
|
def is_alive(self):
|
|
return self.started
|
|
|
|
monkeypatch.setattr(runtime.threading, "Thread", FakeThread)
|
|
monkeypatch.setattr(runtime, "_worker_thread", None)
|
|
|
|
first = runtime.start_embedding_worker()
|
|
second = runtime.start_embedding_worker()
|
|
|
|
assert first is second
|
|
assert len(created) == 1
|
|
assert runtime.embedding_worker_runtime_status() == {
|
|
"runtime_owner": "momo-scheduler",
|
|
"thread_name": "openclaw-embedding-worker",
|
|
"running": True,
|
|
}
|
|
|
|
|
|
def test_versioned_static_assets_use_local_compression_and_immutable_policy():
|
|
app_source = (ROOT / "app.py").read_text(encoding="utf-8")
|
|
requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8")
|
|
|
|
assert "Flask-Compress>=1.24,<2" in requirements
|
|
assert "COMPRESS_ALGORITHM=['br', 'gzip']" in app_source
|
|
assert "public, max-age=31536000, immutable" in app_source
|
|
static_branch = app_source[app_source.index("if request.endpoint == 'static':") :]
|
|
assert static_branch.index("return response") < static_branch.index("session.get('logged_in')")
|
|
|
|
|
|
def test_login_first_viewport_has_no_third_party_or_inline_styles():
|
|
template = (ROOT / "templates/login.html").read_text(encoding="utf-8")
|
|
stylesheet = (ROOT / "web/static/css/page-login.css").read_text(encoding="utf-8")
|
|
|
|
assert "https://" not in template
|
|
assert "bootstrap" not in template.lower()
|
|
assert "font-awesome" not in template.lower()
|
|
assert "<style>" not in template
|
|
assert "css/page-login.css" in template
|
|
assert "gradient(" not in stylesheet
|
|
assert "font-size: clamp" not in stylesheet
|