Initial commit: one_os_compositor Odoo 20 addon — core OWL library + backend models + JWT auth
Includes:
- compositor.app, compositor.session, compositor.session.app, compositor.audit.log models
- /web/compositor/{token,apps,launch,audit} HTTP controllers with JWT (HS256)
- CompositorService OWL service (reusable via useService('compositor'))
- CompositorCanvas OWL component (reusable canvas wrapper)
- Security: compositor_group_user, compositor_group_manager groups + ir.access.csv rules
- 5 demo apps pre-loaded (gnome-calculator, xterm, gtk4-demo, glxgears, featherpad)
- Audit log capturing JWT issuance + app launch events
- Stub mode: launching apps returns a stub SurfaceHandle when proxy isn't connected
- Tested against Odoo 20 + PostgreSQL 17
Author: 1bspace.com
Parent company: 1B Global
License: Proprietary (1B Space internal use)
This commit is contained in:
commit
93a5a0a221
32 files changed
+2497
No files matched your search
@@ -0,0 +1,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for the 1B Space Compositor Core addon."""
|
||||
|
||||
from . import test_compositor_app
|
||||
from . import test_compositor_session
|
||||
from . import test_compositor_controller
|
||||
@@ -0,0 +1,59 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for compositor.app model."""
|
||||
|
||||
from odoo.exceptions import ValidationError, AccessError
|
||||
from odoo.tests import common, tagged
|
||||
|
||||
|
||||
@tagged("post_install", "-at_install")
|
||||
class TestCompositorApp(common.TransactionCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.App = self.env["compositor.app"]
|
||||
self.group_user = self.env.ref("base.group_user")
|
||||
self.group_manager = self.env.ref("one_os_compositor.compositor_group_manager")
|
||||
|
||||
def test_create_app_minimal(self):
|
||||
app = self.App.create({
|
||||
"name": "Test App",
|
||||
"app_id": "test-app",
|
||||
"executable": "/bin/true",
|
||||
})
|
||||
self.assertEqual(app.memory_limit_mb, 1024)
|
||||
self.assertEqual(app.cpu_quota_percent, 200)
|
||||
self.assertEqual(app.tasks_max, 256)
|
||||
self.assertTrue(app.sandbox_gpu)
|
||||
self.assertFalse(app.sandbox_network)
|
||||
|
||||
def test_app_id_unique(self):
|
||||
self.App.create({"name": "App 1", "app_id": "duplicate", "executable": "/bin/true"})
|
||||
with self.assertRaises(ValidationError):
|
||||
self.App.create({"name": "App 2", "app_id": "duplicate", "executable": "/bin/true"})
|
||||
|
||||
def test_memory_limit_too_low(self):
|
||||
with self.assertRaises(ValidationError):
|
||||
self.App.create({
|
||||
"name": "Bad App", "app_id": "bad-app",
|
||||
"executable": "/bin/true", "memory_limit_mb": 32,
|
||||
})
|
||||
|
||||
def test_get_launch_config(self):
|
||||
app = self.App.create({
|
||||
"name": "Test", "app_id": "test",
|
||||
"executable": "/usr/bin/test",
|
||||
"args": "--foo --bar",
|
||||
"env_vars": "GDK_SCALE=1\n# comment\nGTK_DEBUG=all",
|
||||
"memory_limit_mb": 512,
|
||||
"cpu_quota_percent": 100,
|
||||
"tasks_max": 64,
|
||||
"sandbox_network": True,
|
||||
"sandbox_gpu": False,
|
||||
})
|
||||
config = app.get_launch_config()
|
||||
self.assertEqual(config["app_id"], "test")
|
||||
self.assertEqual(config["executable"], "/usr/bin/test")
|
||||
self.assertEqual(config["args"], "--foo --bar")
|
||||
self.assertEqual(config["env"], {"GDK_SCALE": "1", "GTK_DEBUG": "all"})
|
||||
self.assertEqual(config["memory_limit_mb"], 512)
|
||||
self.assertTrue(config["sandbox_network"])
|
||||
self.assertFalse(config["sandbox_gpu"])
|
||||
@@ -0,0 +1,69 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for compositor HTTP controllers."""
|
||||
|
||||
from odoo.tests import HttpCase, tagged
|
||||
|
||||
|
||||
@tagged("post_install", "-at_install")
|
||||
class TestCompositorController(HttpCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
# Make sure a JWT secret is set
|
||||
self.env["ir.config_parameter"].set_str(
|
||||
"compositor.jwt_secret", "test-secret-1bspace"
|
||||
)
|
||||
|
||||
def test_get_token(self):
|
||||
"""Test that /web/compositor/token returns a valid JWT."""
|
||||
self.authenticate("admin", "admin")
|
||||
response = self.url_open(
|
||||
"/web/compositor/token",
|
||||
data="{}",
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=30,
|
||||
)
|
||||
self.assertEqual(response.status_code, 200, response.text)
|
||||
result = response.json()
|
||||
self.assertIn("jwt", result["result"])
|
||||
self.assertIn("proxy_url", result["result"])
|
||||
self.assertIn("expires_at", result["result"])
|
||||
self.assertIn("compositor_session_id", result["result"])
|
||||
|
||||
def test_list_apps(self):
|
||||
"""Test that /web/compositor/apps returns the pre-installed apps."""
|
||||
self.authenticate("admin", "admin")
|
||||
response = self.url_open(
|
||||
"/web/compositor/apps",
|
||||
data="{}",
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=30,
|
||||
)
|
||||
self.assertEqual(response.status_code, 200, response.text)
|
||||
result = response.json()["result"]
|
||||
self.assertIn("apps", result)
|
||||
app_ids = [a["app_id"] for a in result["apps"]]
|
||||
self.assertIn("gnome-calculator", app_ids)
|
||||
self.assertIn("xterm", app_ids)
|
||||
|
||||
def test_launch_app_stub(self):
|
||||
"""Test that launching an app returns a stub response when proxy isn't connected."""
|
||||
self.authenticate("admin", "admin")
|
||||
# First get a token
|
||||
token_resp = self.url_open(
|
||||
"/web/compositor/token",
|
||||
data="{}",
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=30,
|
||||
).json()["result"]
|
||||
# Then try to launch
|
||||
response = self.url_open(
|
||||
"/web/compositor/launch",
|
||||
data='{"app_id": "gnome-calculator"}',
|
||||
headers={"Content-Type": "application/json"},
|
||||
timeout=30,
|
||||
)
|
||||
self.assertEqual(response.status_code, 200, response.text)
|
||||
result = response.json()["result"]
|
||||
# Since proxy isn't connected, we should get a stub
|
||||
self.assertTrue(result.get("stub", False))
|
||||
self.assertIn("compositor_surface_id", result)
|
||||
@@ -0,0 +1,45 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""Tests for compositor.session model."""
|
||||
|
||||
from odoo.tests import common, tagged
|
||||
|
||||
|
||||
@tagged("post_install", "-at_install")
|
||||
class TestCompositorSession(common.TransactionCase):
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.Session = self.env["compositor.session"]
|
||||
self.App = self.env["compositor.app"]
|
||||
|
||||
def test_create_session_defaults(self):
|
||||
session = self.Session.create({})
|
||||
self.assertEqual(session.state, "pending")
|
||||
self.assertTrue(session.compositor_session_id)
|
||||
self.assertEqual(session.user_id, self.env.user)
|
||||
self.assertEqual(session.app_count, 0)
|
||||
self.assertEqual(session.running_app_count, 0)
|
||||
|
||||
def test_session_app_lifecycle(self):
|
||||
session = self.Session.create({})
|
||||
app = self.App.create({
|
||||
"name": "Test", "app_id": "test-lifecycle", "executable": "/bin/true",
|
||||
})
|
||||
inst = self.env["compositor.session.app"].create({
|
||||
"session_id": session.id, "app_id": app.id,
|
||||
})
|
||||
self.assertEqual(inst.state, "launching")
|
||||
inst.write({"state": "running"})
|
||||
self.assertEqual(session.running_app_count, 1)
|
||||
inst.action_terminate()
|
||||
self.assertEqual(inst.state, "terminated")
|
||||
self.assertEqual(session.running_app_count, 0)
|
||||
|
||||
def test_action_close(self):
|
||||
session = self.Session.create({})
|
||||
app = self.App.create({"name": "X", "app_id": "x", "executable": "/bin/true"})
|
||||
inst = self.env["compositor.session.app"].create({
|
||||
"session_id": session.id, "app_id": app.id, "state": "running",
|
||||
})
|
||||
session.action_close()
|
||||
self.assertEqual(session.state, "closed")
|
||||
self.assertEqual(inst.state, "terminated")
|
||||
Reference in new issue
Block a user