v0.3: SQLite persistence + ProjectService + dependency resolution
- New: services/sensu_db.py (SQLite, 5 tables) - New: services/project_service.py (project registry, port allocation, dep resolution) - Integrated into main.py (step 11.6) - Tests: 23/23 passing (15 original + 8 new)
This commit is contained in:
@@ -0,0 +1,75 @@
|
||||
"""v0.3 tests: SQLite DB, ProjectService, Dependency Resolution"""
|
||||
import pytest, os, sys, asyncio
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
||||
|
||||
class TestSenSuDB:
|
||||
@pytest.fixture
|
||||
def db(self):
|
||||
from services.sensu_db import SenSuDB
|
||||
import tempfile
|
||||
d = SenSuDB(":memory:")
|
||||
yield d
|
||||
d.close()
|
||||
|
||||
def test_save_and_get_plugin(self, db):
|
||||
db.save_plugin("test_plugin", version="1.0", status="running")
|
||||
p = db.get_plugin("test_plugin")
|
||||
assert p["name"] == "test_plugin"
|
||||
assert p["version"] == "1.0"
|
||||
assert p["status"] == "running"
|
||||
|
||||
def test_list_plugins(self, db):
|
||||
db.save_plugin("p1", version="1.0")
|
||||
db.save_plugin("p2", version="2.0")
|
||||
lst = db.list_plugins()
|
||||
assert len(lst) >= 2
|
||||
|
||||
def test_permissions(self, db):
|
||||
db.grant_permission("p1", "read")
|
||||
assert db.check_permission("p1", "read") is True
|
||||
assert db.check_permission("p1", "write") is False
|
||||
|
||||
def test_config_kv(self, db):
|
||||
db.set_config("theme", "dark")
|
||||
assert db.get_config("theme") == "dark"
|
||||
assert db.get_config("nonexistent", "default") == "default"
|
||||
|
||||
class TestProjectService:
|
||||
def test_register_project(self):
|
||||
from services.project_service import ProjectService
|
||||
ps = ProjectService(None)
|
||||
ok = ps.register_project("test_plugin", {"name": "TestProject", "port": 5000, "path": "/tmp/test"})
|
||||
assert ok is True
|
||||
p = ps.get_project("TestProject")
|
||||
assert p["port"] == 5000
|
||||
assert p["status"] == "registered"
|
||||
|
||||
def test_port_conflict(self):
|
||||
from services.project_service import ProjectService
|
||||
ps = ProjectService(None)
|
||||
ps.register_project("p1", {"name": "A", "port": 5000})
|
||||
ps.register_project("p2", {"name": "B", "port": 5000})
|
||||
b = ps.get_project("B")
|
||||
assert b["port"] != 5000 # should get a different port
|
||||
|
||||
def test_dependency_resolution(self):
|
||||
from services.project_service import ProjectService
|
||||
ps = ProjectService(None)
|
||||
plugins = {
|
||||
"base": {"depends_on": []},
|
||||
"middle": {"depends_on": ["base"]},
|
||||
"top": {"depends_on": ["middle"]},
|
||||
}
|
||||
order = ps.resolve_dependencies(plugins)
|
||||
assert order.index("base") < order.index("middle")
|
||||
assert order.index("middle") < order.index("top")
|
||||
|
||||
def test_circular_dependency(self):
|
||||
from services.project_service import ProjectService
|
||||
ps = ProjectService(None)
|
||||
plugins = {
|
||||
"a": {"depends_on": ["b"]},
|
||||
"b": {"depends_on": ["a"]},
|
||||
}
|
||||
order = ps.resolve_dependencies(plugins)
|
||||
assert len(order) == 2 # should handle gracefully
|
||||
Reference in New Issue
Block a user