381 lines
16 KiB
Python
381 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_ROOT = ROOT / "collector-service"
|
|
if str(APP_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(APP_ROOT))
|
|
|
|
|
|
class MainAgentGovernanceTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.tempdir = tempfile.TemporaryDirectory()
|
|
temp_root = Path(cls.tempdir.name)
|
|
os.environ["DATA_DIR"] = str(temp_root / "data")
|
|
os.environ["DATABASE_PATH"] = str(temp_root / "data" / "storyforge.db")
|
|
os.environ["DOWNLOADS_DIR"] = str(temp_root / "downloads")
|
|
os.environ["JOBS_DIR"] = str(temp_root / "jobs")
|
|
os.environ["MODELS_DIR"] = str(temp_root / "models")
|
|
os.environ["ORCHESTRATOR_SHARED_SECRET"] = "test-secret"
|
|
os.environ["WEB_AUTOLOGIN_ENABLED"] = "0"
|
|
os.environ.setdefault("BOOTSTRAP_SUPERADMIN_USERNAME", "")
|
|
os.environ.setdefault("BOOTSTRAP_SUPERADMIN_PASSWORD", "")
|
|
|
|
cls.db_module = importlib.reload(importlib.import_module("app.database"))
|
|
cls.core = importlib.reload(importlib.import_module("app.core_main"))
|
|
cls.app_main = importlib.reload(importlib.import_module("app.main"))
|
|
cls.core.db.init_schema()
|
|
cls.client = TestClient(cls.app_main.app)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls) -> None:
|
|
cls.client.close()
|
|
cls.tempdir.cleanup()
|
|
|
|
def setUp(self) -> None:
|
|
self._clear_tables()
|
|
self.ctx = self._seed_accounts()
|
|
|
|
def _clear_tables(self) -> None:
|
|
tables = [
|
|
"agent_policy_audit_logs",
|
|
"agent_policy_effectivity",
|
|
"agent_policy_versions",
|
|
"agent_policy_scopes",
|
|
"agent_skill_versions",
|
|
"agent_skills",
|
|
"agent_memories",
|
|
"platform_agent_profiles",
|
|
"oneliner_messages",
|
|
"oneliner_sessions",
|
|
"oneliner_profiles",
|
|
"auth_tokens",
|
|
"projects",
|
|
"accounts",
|
|
"model_profiles",
|
|
]
|
|
for table in tables:
|
|
try:
|
|
self.core.db.execute(f"DELETE FROM {table}")
|
|
except Exception:
|
|
continue
|
|
|
|
def _seed_accounts(self) -> dict[str, Any]:
|
|
now = self.db_module.utc_now()
|
|
admin_id = "acct_admin"
|
|
member_id = "acct_member"
|
|
project_id = "proj_member"
|
|
model_id = "model_default"
|
|
admin_token = "token_admin"
|
|
member_token = "token_member"
|
|
|
|
self.core.db.execute(
|
|
"""
|
|
INSERT INTO accounts (
|
|
id, username, password_hash, password_salt, display_name, role, approval_status,
|
|
approved_by, approved_at, preferred_analysis_model_id, created_at, updated_at
|
|
) VALUES (?, ?, 'hash', 'salt', ?, ?, 'approved', ?, ?, ?, ?, ?)
|
|
""",
|
|
(admin_id, "admin", "Admin", "super_admin", admin_id, now, model_id, now, now),
|
|
)
|
|
self.core.db.execute(
|
|
"""
|
|
INSERT INTO accounts (
|
|
id, username, password_hash, password_salt, display_name, role, approval_status,
|
|
approved_by, approved_at, preferred_analysis_model_id, created_at, updated_at
|
|
) VALUES (?, ?, 'hash', 'salt', ?, ?, 'approved', ?, ?, ?, ?, ?)
|
|
""",
|
|
(member_id, "member", "Member", "operator", admin_id, now, model_id, now, now),
|
|
)
|
|
self.core.db.execute(
|
|
"""
|
|
INSERT INTO projects (id, user_id, name, description, created_at, updated_at)
|
|
VALUES (?, ?, ?, '', ?, ?)
|
|
""",
|
|
(project_id, member_id, "Member Project", now, now),
|
|
)
|
|
self.core.db.execute(
|
|
"""
|
|
INSERT INTO model_profiles (
|
|
id, owner_account_id, name, provider, base_url, api_key, model_name,
|
|
is_system, is_default, created_at, updated_at
|
|
) VALUES (?, NULL, 'Default Model', 'openai_compat', 'http://127.0.0.1:8317/v1', '', 'GLM-5', 1, 1, ?, ?)
|
|
""",
|
|
(model_id, now, now),
|
|
)
|
|
self.core.db.execute(
|
|
"INSERT INTO auth_tokens (token, account_id, created_at) VALUES (?, ?, ?)",
|
|
(admin_token, admin_id, now),
|
|
)
|
|
self.core.db.execute(
|
|
"INSERT INTO auth_tokens (token, account_id, created_at) VALUES (?, ?, ?)",
|
|
(member_token, member_id, now),
|
|
)
|
|
return {
|
|
"admin_id": admin_id,
|
|
"member_id": member_id,
|
|
"project_id": project_id,
|
|
"admin_headers": {"Authorization": f"Bearer {admin_token}"},
|
|
"member_headers": {"Authorization": f"Bearer {member_token}"},
|
|
}
|
|
|
|
def test_effective_policy_merges_system_user_global_and_platform_layers(self) -> None:
|
|
system_response = self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"title": "System main agent",
|
|
"summary": "Default baseline",
|
|
"policy": {
|
|
"tone": {"style": "default"},
|
|
"homepage": {"focus": "ops"},
|
|
"actions": {"max_cards": 3},
|
|
},
|
|
"reason": "seed system baseline",
|
|
},
|
|
)
|
|
self.assertEqual(system_response.status_code, 200, system_response.text)
|
|
|
|
global_response = self.client.put(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "Member global strategy",
|
|
"summary": "Personal operating style",
|
|
"policy": {
|
|
"tone": {"style": "analytical"},
|
|
"memory": {"default_window": "30d"},
|
|
"actions": {"max_cards": 2},
|
|
},
|
|
"reason": "personalize global defaults",
|
|
},
|
|
)
|
|
self.assertEqual(global_response.status_code, 200, global_response.text)
|
|
|
|
platform_response = self.client.put(
|
|
"/v2/oneliner/governance/user/platforms/douyin",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "Douyin strategy",
|
|
"summary": "Tighter benchmark workflow",
|
|
"policy": {
|
|
"actions": {"max_cards": 1},
|
|
"douyin": {"benchmark_mode": "strict"},
|
|
},
|
|
"reason": "tighten douyin execution",
|
|
},
|
|
)
|
|
self.assertEqual(platform_response.status_code, 200, platform_response.text)
|
|
|
|
effective_response = self.client.get(
|
|
"/v2/oneliner/governance/effective",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"], "platform": "douyin"},
|
|
)
|
|
self.assertEqual(effective_response.status_code, 200, effective_response.text)
|
|
payload = effective_response.json()
|
|
self.assertEqual(
|
|
[item["scope_kind"] for item in payload["layers"]],
|
|
["system_main", "user_global", "user_platform"],
|
|
)
|
|
self.assertEqual(payload["effective_policy"]["tone"]["style"], "analytical")
|
|
self.assertEqual(payload["effective_policy"]["homepage"]["focus"], "ops")
|
|
self.assertEqual(payload["effective_policy"]["actions"]["max_cards"], 1)
|
|
self.assertEqual(payload["effective_policy"]["douyin"]["benchmark_mode"], "strict")
|
|
|
|
def test_admin_override_takes_precedence_in_effective_policy(self) -> None:
|
|
self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"title": "System main agent",
|
|
"policy": {"actions": {"max_cards": 3}},
|
|
"reason": "seed baseline",
|
|
},
|
|
)
|
|
self.client.put(
|
|
"/v2/oneliner/governance/user/platforms/douyin",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "Douyin strategy",
|
|
"policy": {"actions": {"max_cards": 1}},
|
|
"reason": "tighten douyin execution",
|
|
},
|
|
)
|
|
|
|
override_response = self.client.post(
|
|
"/v2/admin/oneliner/governance/overrides",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"target_user_id": self.ctx["member_id"],
|
|
"target_project_id": self.ctx["project_id"],
|
|
"platform": "douyin",
|
|
"title": "Safety override",
|
|
"summary": "Require review after recent drift",
|
|
"policy": {
|
|
"actions": {"max_cards": 5},
|
|
"guardrails": {"require_admin_review": True},
|
|
},
|
|
"reason": "contain unexpected drift",
|
|
},
|
|
)
|
|
self.assertEqual(override_response.status_code, 200, override_response.text)
|
|
|
|
effective_response = self.client.get(
|
|
"/v2/oneliner/governance/effective",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"], "platform": "douyin"},
|
|
)
|
|
self.assertEqual(effective_response.status_code, 200, effective_response.text)
|
|
payload = effective_response.json()
|
|
self.assertEqual(payload["layers"][-1]["scope_kind"], "admin_override")
|
|
self.assertEqual(payload["effective_policy"]["actions"]["max_cards"], 5)
|
|
self.assertTrue(payload["effective_policy"]["guardrails"]["require_admin_review"])
|
|
|
|
def test_admin_override_without_target_project_applies_to_member_projects(self) -> None:
|
|
override_response = self.client.post(
|
|
"/v2/admin/oneliner/governance/overrides",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"target_user_id": self.ctx["member_id"],
|
|
"title": "Global safety override",
|
|
"summary": "Apply guardrails across every project",
|
|
"policy": {
|
|
"guardrails": {"require_admin_review": True},
|
|
"actions": {"max_cards": 4},
|
|
},
|
|
"reason": "global containment",
|
|
},
|
|
)
|
|
self.assertEqual(override_response.status_code, 200, override_response.text)
|
|
|
|
effective_response = self.client.get(
|
|
"/v2/oneliner/governance/effective",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"], "platform": "douyin"},
|
|
)
|
|
self.assertEqual(effective_response.status_code, 200, effective_response.text)
|
|
payload = effective_response.json()
|
|
self.assertEqual(payload["layers"][-1]["scope_kind"], "admin_override")
|
|
self.assertEqual(payload["effective_policy"]["actions"]["max_cards"], 4)
|
|
self.assertTrue(payload["effective_policy"]["guardrails"]["require_admin_review"])
|
|
|
|
def test_effective_policy_skips_future_scheduled_versions_until_window_opens(self) -> None:
|
|
first_response = self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"title": "Current system baseline",
|
|
"summary": "Active now",
|
|
"policy": {"tone": {"style": "default"}},
|
|
"reason": "baseline",
|
|
},
|
|
)
|
|
self.assertEqual(first_response.status_code, 200, first_response.text)
|
|
|
|
second_response = self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"title": "Future strategy",
|
|
"summary": "Should not be active yet",
|
|
"policy": {"tone": {"style": "future"}},
|
|
"effect_mode": "scheduled",
|
|
"starts_at": "2099-01-01T00:00:00Z",
|
|
"reason": "future rollout",
|
|
},
|
|
)
|
|
self.assertEqual(second_response.status_code, 200, second_response.text)
|
|
|
|
effective_response = self.client.get(
|
|
"/v2/oneliner/governance/effective",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"], "platform": "douyin"},
|
|
)
|
|
self.assertEqual(effective_response.status_code, 200, effective_response.text)
|
|
payload = effective_response.json()
|
|
self.assertEqual(payload["effective_policy"]["tone"]["style"], "default")
|
|
self.assertEqual(payload["layers"][0]["current_version"]["title"], "Current system baseline")
|
|
|
|
def test_user_global_versions_support_rollback_by_creating_new_version(self) -> None:
|
|
first_response = self.client.put(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "Global strategy v1",
|
|
"policy": {"tone": {"style": "analytical"}},
|
|
"reason": "first pass",
|
|
},
|
|
)
|
|
self.assertEqual(first_response.status_code, 200, first_response.text)
|
|
first_version_id = first_response.json()["current_version"]["id"]
|
|
|
|
second_response = self.client.put(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "Global strategy v2",
|
|
"policy": {"tone": {"style": "decisive"}},
|
|
"reason": "refine tone",
|
|
},
|
|
)
|
|
self.assertEqual(second_response.status_code, 200, second_response.text)
|
|
|
|
versions_before = self.client.get(
|
|
"/v2/oneliner/governance/user/global/versions",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"]},
|
|
)
|
|
self.assertEqual(versions_before.status_code, 200, versions_before.text)
|
|
self.assertEqual(versions_before.json()["count"], 2)
|
|
|
|
rollback_response = self.client.post(
|
|
"/v2/oneliner/governance/user/global/rollback",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"version_id": first_version_id,
|
|
"reason": "restore best baseline",
|
|
},
|
|
)
|
|
self.assertEqual(rollback_response.status_code, 200, rollback_response.text)
|
|
rollback_payload = rollback_response.json()
|
|
self.assertEqual(rollback_payload["current_version"]["rollback_from_version_id"], first_version_id)
|
|
self.assertEqual(rollback_payload["effective_policy"]["tone"]["style"], "analytical")
|
|
|
|
versions_after = self.client.get(
|
|
"/v2/oneliner/governance/user/global/versions",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"]},
|
|
)
|
|
self.assertEqual(versions_after.status_code, 200, versions_after.text)
|
|
self.assertEqual(versions_after.json()["count"], 3)
|
|
|
|
def test_non_admin_cannot_change_system_defaults(self) -> None:
|
|
response = self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"title": "Not allowed",
|
|
"policy": {"tone": {"style": "rogue"}},
|
|
"reason": "should be blocked",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 403, response.text)
|