a513d8d373
joy pipeline list/get/tools/summary/doctor 读 NAS cluster-pipeline-registry.yaml；dashboard 并入 registry 与 扫盘对账；流水线 tab 展示 tools/登记；web 白名单只读。
98 lines
2.8 KiB
Python
98 lines
2.8 KiB
Python
"""pipeline registry 单元测试（stdlib unittest，无 pytest 依赖）"""
|
||
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
import yaml
|
||
|
||
SRC = Path(__file__).parent.parent / "src"
|
||
sys.path.insert(0, str(SRC))
|
||
|
||
from joy.core import pipeline_registry as pr # noqa: E402
|
||
|
||
|
||
SAMPLE = {
|
||
"version": "1.0",
|
||
"pipelines": [
|
||
{
|
||
"id": "studio/project/image-cover",
|
||
"vm": "studio",
|
||
"kind": "project",
|
||
"name": "image-cover",
|
||
"status": "active",
|
||
"source": "declared",
|
||
"tags": ["image"],
|
||
"readiness": {"invocable": 2},
|
||
},
|
||
{
|
||
"id": "studio/project/orphan-scan",
|
||
"vm": "studio",
|
||
"kind": "project",
|
||
"name": "ghost",
|
||
"status": "active",
|
||
"source": "scanned",
|
||
},
|
||
{
|
||
"id": "wenpai/workspace/wp-release",
|
||
"vm": "wenpai",
|
||
"kind": "workspace",
|
||
"name": "wp-release",
|
||
"status": "planned",
|
||
"source": "declared",
|
||
},
|
||
],
|
||
}
|
||
|
||
|
||
class TestPipelineRegistry(unittest.TestCase):
|
||
def setUp(self):
|
||
self._td = tempfile.TemporaryDirectory()
|
||
self.f = Path(self._td.name) / "cluster-pipeline-registry.yaml"
|
||
self.f.write_text(yaml.dump(SAMPLE, allow_unicode=True), encoding="utf-8")
|
||
self._patch = mock.patch.object(pr, "registry_path", return_value=self.f)
|
||
self._patch.start()
|
||
|
||
def tearDown(self):
|
||
self._patch.stop()
|
||
self._td.cleanup()
|
||
|
||
def test_list_all(self):
|
||
r = pr.list_pipelines()
|
||
self.assertEqual(r["count"], 3)
|
||
self.assertTrue(r["exists"])
|
||
|
||
def test_tools_only(self):
|
||
r = pr.list_pipelines(tools_only=True)
|
||
ids = [x["id"] for x in r["items"]]
|
||
self.assertEqual(ids, ["studio/project/image-cover"])
|
||
|
||
def test_get(self):
|
||
p = pr.get_pipeline("studio/project/image-cover")
|
||
self.assertEqual(p["name"], "image-cover")
|
||
p2 = pr.get_pipeline("image-cover")
|
||
self.assertEqual(p2["id"], "studio/project/image-cover")
|
||
|
||
def test_filter_vm(self):
|
||
r = pr.list_pipelines(vm="wenpai")
|
||
self.assertEqual(r["count"], 1)
|
||
|
||
def test_doctor(self):
|
||
scanned = [
|
||
{"vm": "studio", "name": "image-cover", "kind": "project"},
|
||
{"vm": "studio", "name": "only-on-disk", "kind": "project"},
|
||
]
|
||
d = pr.doctor_against_scan(scanned)
|
||
self.assertGreaterEqual(d["both_n"], 1)
|
||
self.assertIn("only-on-disk", {x["name"] for x in d["orphan_scanned"]})
|
||
|
||
def test_summary(self):
|
||
s = pr.registry_summary()
|
||
self.assertEqual(s["total"], 3)
|
||
self.assertEqual(s["active_tools"], 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|
