wenhuadian/xas-test-library
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initialize XAS test console with native execution

Browse Source
This commit is contained in:
AdamWen95
2026-10-14 09:38:58 +08:00
commit 438844b01a
83 changed files with 9334 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_test_catalog.py
+103
View File
@@ -0,0 +1,103 @@
"""A test entry represents behavior, not every source file under a directory."""
from pathlib import Path
import tempfile
import unittest
from fastapi.testclient import TestClient
from console.app import create_app
from console.settings import Settings
from console.store import Store
class TestCatalogTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
root = Path(self.temp.name)
self.source = root / "scripts"
self.source.mkdir()
(self.source / "run-suite.sh").write_text("python3 helper.py\n")
(self.source / "helper.py").write_text("print('helper')\n")
(self.source / "tests").mkdir()
(self.source / "tests/test_internal.py").write_text("assert True\n")
self.settings = Settings(root / "runtime", [{"id": "suite", "path": str(self.source)}],
allowed_hosts=["testserver"], allowed_networks=["127.0.0.0/8"])
self.settings.test_definitions = [{"key": "suite", "source_group": "suite", "entry": "run-suite.sh",
"name": "模型综合测试", "description": "一轮完整验证", "kind": "suite",
"members": [{"source_group": "suite", "pattern": "**/*"}]}]
self.client = TestClient(create_app(self.settings), client=("127.0.0.1", 5555))
self.headers = {"X-XAS-Console": "1"}
self.client.post("/api/discover", headers=self.headers)
self.store = Store(self.settings)
def tearDown(self):
self.client.close()
self.temp.cleanup()
def test_suite_is_one_item_with_readonly_members_not_three_cards(self):
items = self.client.get("/api/scripts").json()["items"]
self.assertEqual(len(items), 1)
self.assertEqual(items[0]["name"], "模型综合测试")
self.assertEqual(items[0]["test_kind"], "suite")
self.assertEqual(items[0]["file_count"], 3)
files = self.client.get(f"/api/scripts/{items[0]['id']}/files").json()["items"]
self.assertEqual(len(files), 3)
self.assertEqual(sum(f["role"] == "entry" for f in files), 1)
summary = self.client.get("/api/summary").json()
self.assertEqual(summary["scripts"], 1)
self.assertEqual(summary["source_files"], 3)
def test_refresh_does_not_promote_new_support_file_or_clear_configuration(self):
entry = next(s for s in self.store.scripts() if s["source_path"].endswith("run-suite.sh"))
self.client.patch(f"/api/scripts/{entry['id']}", json={"name": "我的综合测试"}, headers=self.headers)
(self.source / "new_module.py").write_text("pass\n")
self.client.post("/api/discover", headers=self.headers)
items = self.client.get("/api/scripts").json()["items"]
self.assertEqual(len(items), 1)
self.assertEqual(items[0]["name"], "我的综合测试")
self.assertEqual(items[0]["file_count"], 4)
def test_member_cannot_be_enabled_or_run_by_direct_api(self):
helper = next(s for s in self.store.scripts() if s["source_path"].endswith("helper.py"))
self.assertFalse(self.client.get(f"/api/scripts/{helper['id']}").json()["is_test"])
enabled = self.client.patch(f"/api/scripts/{helper['id']}", json={"enabled": True}, headers=self.headers)
self.assertEqual(enabled.status_code, 400)
# Even a pre-existing enabled inventory record cannot bypass the entry boundary.
helper.update(enabled=True, status="ready")
self.store.save_script(helper)
run = self.client.post(f"/api/scripts/{helper['id']}/runs",
json={"request_id": "caa3943d-6872-472c-86c2-500000000001"}, headers=self.headers)
self.assertEqual(run.status_code, 400)
self.assertEqual(self.store.runs(), [])
def test_upload_remains_one_test_and_examples_are_opt_in(self):
uploaded = self.client.post("/api/scripts/upload", headers=self.headers,
files={"file": ("test.py", b"print('custom')\n")}, data={"name": "用户测试"}).json()
example = self.client.post("/api/scripts/upload", headers=self.headers,
files={"file": ("demo.py", b"print('demo')\n")}, data={"name": "演示"}).json()
item = self.store.script(example["id"])
item["synthetic"] = True
self.store.save_script(item)
regular = self.client.get("/api/scripts").json()["items"]
self.assertEqual({s["name"] for s in regular}, {"模型综合测试", "用户测试"})
all_items = self.client.get("/api/scripts?include_examples=true").json()["items"]
self.assertEqual(len(all_items), 3)
self.assertTrue(next(s for s in all_items if s["id"] == uploaded["id"])["is_test"])
def test_historical_runs_keep_ids_and_source_contents_unchanged(self):
entry = next(s for s in self.store.scripts() if s["source_path"].endswith("run-suite.sh"))
historical = self.store.queue(entry, "past-request", "127.0.0.1")
before = (self.source / "run-suite.sh").read_bytes()
self.client.get("/api/scripts")
self.client.post("/api/discover", headers=self.headers)
self.assertEqual(self.store.run(historical["id"])["script_name"], "run-suite.sh")
self.assertEqual(self.store.run(historical["id"])["script_id"], entry["id"])
self.assertEqual(before, (self.source / "run-suite.sh").read_bytes())
def test_new_run_uses_one_entry_and_the_test_name(self):
entry = self.client.get("/api/scripts").json()["items"][0]
response = self.client.patch(f"/api/scripts/{entry['id']}", json={"enabled": True}, headers=self.headers)
self.assertEqual(response.status_code, 200)
run = self.client.post(f"/api/scripts/{entry['id']}/runs", headers=self.headers,
json={"request_id": "caa3943d-6872-472c-86c2-500000000002"}).json()
self.assertEqual(run["script_name"], "模型综合测试")
self.assertEqual(run["source_path"], str(self.source / "run-suite.sh"))
self.assertEqual(len(self.store.runs()), 1)