AskaEth
d48ef65b35
- 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)
76 lines
2.7 KiB
Python
76 lines
2.7 KiB
Python
"""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
|