866 lines
37 KiB
Python
866 lines
37 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_run_events",
|
|
"agent_runs",
|
|
"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 _seed_approved_member_without_project(self) -> dict[str, Any]:
|
|
now = self.db_module.utc_now()
|
|
admin_id = "acct_admin"
|
|
member_id = "acct_member_noproject"
|
|
model_id = "model_default"
|
|
admin_token = "token_admin"
|
|
member_token = "token_member_noproject"
|
|
|
|
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_noproject", "Member No Project", "operator", admin_id, now, model_id, 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,
|
|
"admin_headers": {"Authorization": f"Bearer {admin_token}"},
|
|
"member_headers": {"Authorization": f"Bearer {member_token}"},
|
|
}
|
|
|
|
def test_agent_run_creation_snapshots_governance_and_needs_confirmation(self) -> None:
|
|
response = self.client.post(
|
|
"/v2/oneliner/runs",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"source_screen": "dashboard",
|
|
"source_action_key": "homepage-primary-action",
|
|
"title": "跟进重点账号",
|
|
"summary": "先由主 Agent 评估优先级",
|
|
"intent_key": "track_account",
|
|
"platform": "douyin",
|
|
"platform_scope": "single_platform",
|
|
"plan_request": {
|
|
"goal": "跟进重点账号",
|
|
"steps": ["读取当前项目上下文", "检查重点账号变化", "决定下一步"],
|
|
},
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
payload = response.json()
|
|
self.assertEqual(payload["run_status"], "needs_confirmation")
|
|
self.assertEqual(payload["source_screen"], "dashboard")
|
|
self.assertEqual(payload["platform"], "douyin")
|
|
self.assertEqual(payload["platform_scope"], "single_platform")
|
|
self.assertEqual(payload["session_id"][:5], "oline")
|
|
self.assertEqual(payload["plan"]["goal"], "跟进重点账号")
|
|
self.assertEqual(payload["recommended_preview_action"]["action"], "goto-discovery")
|
|
self.assertEqual(payload["recommended_preview_action"]["screen"], "discovery")
|
|
self.assertEqual(payload["governance"]["project_id"], self.ctx["project_id"])
|
|
self.assertIn("layers", payload["governance"])
|
|
self.assertEqual(payload["events"][0]["event_type"], "run.created")
|
|
|
|
def test_agent_run_confirm_transitions_to_queue_or_running_and_logs_events(self) -> None:
|
|
create = self.client.post(
|
|
"/v2/oneliner/runs",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"source_screen": "strategy",
|
|
"source_action_key": "handoff-to-main-agent",
|
|
"title": "调整当前平台策略",
|
|
"summary": "让主 Agent 先给执行计划",
|
|
"intent_key": "custom",
|
|
"platform": "douyin",
|
|
"platform_scope": "single_platform",
|
|
"plan_request": {
|
|
"goal": "调整当前平台策略",
|
|
"steps": ["读取当前平台策略", "生成调整建议"],
|
|
},
|
|
},
|
|
)
|
|
self.assertEqual(create.status_code, 200, create.text)
|
|
run_id = create.json()["id"]
|
|
|
|
confirm = self.client.post(
|
|
f"/v2/oneliner/runs/{run_id}/confirm",
|
|
headers=self.ctx["member_headers"],
|
|
json={"reason": "user confirmed"},
|
|
)
|
|
self.assertEqual(confirm.status_code, 200, confirm.text)
|
|
payload = confirm.json()
|
|
self.assertIn(payload["run_status"], {"queued", "running"})
|
|
self.assertEqual(payload["recommended_preview_action"]["action"], "goto-strategy")
|
|
self.assertEqual(payload["recommended_preview_action"]["screen"], "strategy")
|
|
event_types = [item["event_type"] for item in payload["events"]]
|
|
self.assertIn("run.created", event_types)
|
|
self.assertIn("run.confirmed", event_types)
|
|
self.assertTrue("run.queued" in event_types or "run.started" in event_types)
|
|
|
|
def test_running_agent_run_detail_advances_to_progress_and_done(self) -> None:
|
|
create = self.client.post(
|
|
"/v2/oneliner/runs",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"source_screen": "dashboard",
|
|
"source_action_key": "homepage-primary-action",
|
|
"title": "安排今日动作",
|
|
"summary": "让主 Agent 给出执行收口",
|
|
"intent_key": "custom",
|
|
"platform": "douyin",
|
|
"platform_scope": "single_platform",
|
|
"plan_request": {
|
|
"goal": "安排今日动作",
|
|
"steps": ["读取当前项目上下文", "给出执行建议", "输出下一步"],
|
|
},
|
|
},
|
|
)
|
|
self.assertEqual(create.status_code, 200, create.text)
|
|
run_id = create.json()["id"]
|
|
|
|
confirm = self.client.post(
|
|
f"/v2/oneliner/runs/{run_id}/confirm",
|
|
headers=self.ctx["member_headers"],
|
|
json={"reason": "user confirmed"},
|
|
)
|
|
self.assertEqual(confirm.status_code, 200, confirm.text)
|
|
|
|
detail = self.client.get(
|
|
f"/v2/oneliner/runs/{run_id}",
|
|
headers=self.ctx["member_headers"],
|
|
)
|
|
self.assertEqual(detail.status_code, 200, detail.text)
|
|
payload = detail.json()
|
|
self.assertEqual(payload["run_status"], "done")
|
|
self.assertTrue(payload["finished_at"])
|
|
self.assertEqual(payload["result"]["result_kind"], "main_agent_plan")
|
|
self.assertEqual(payload["result"]["recommended_action"]["action"], "goto-discovery")
|
|
self.assertEqual(payload["result"]["recommended_action"]["screen"], "discovery")
|
|
event_types = [item["event_type"] for item in payload["events"]]
|
|
self.assertIn("run.progress", event_types)
|
|
self.assertIn("run.done", event_types)
|
|
|
|
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"])
|
|
self.assertEqual(payload["active_admin_override_notice"]["title"], "Global safety override")
|
|
|
|
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_scope_read_endpoints_keep_current_version_on_active_release_not_future_schedule(self) -> None:
|
|
user_first = self.client.put(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "User global baseline",
|
|
"policy": {"tone": {"style": "baseline"}},
|
|
"reason": "seed baseline",
|
|
},
|
|
)
|
|
self.assertEqual(user_first.status_code, 200, user_first.text)
|
|
|
|
user_future = self.client.put(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "User global future",
|
|
"policy": {"tone": {"style": "future"}},
|
|
"effect_mode": "scheduled",
|
|
"starts_at": "2099-01-01T00:00:00Z",
|
|
"reason": "future rollout",
|
|
},
|
|
)
|
|
self.assertEqual(user_future.status_code, 200, user_future.text)
|
|
|
|
user_read = self.client.get(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"]},
|
|
)
|
|
self.assertEqual(user_read.status_code, 200, user_read.text)
|
|
self.assertEqual(user_read.json()["current_version"]["title"], "User global baseline")
|
|
|
|
system_first = self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"title": "System baseline",
|
|
"policy": {"homepage": {"focus": "stable"}},
|
|
"reason": "stable baseline",
|
|
},
|
|
)
|
|
self.assertEqual(system_first.status_code, 200, system_first.text)
|
|
|
|
system_future = self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"title": "System future",
|
|
"policy": {"homepage": {"focus": "future"}},
|
|
"effect_mode": "scheduled",
|
|
"starts_at": "2099-01-01T00:00:00Z",
|
|
"reason": "future rollout",
|
|
},
|
|
)
|
|
self.assertEqual(system_future.status_code, 200, system_future.text)
|
|
|
|
system_read = self.client.get(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
)
|
|
self.assertEqual(system_read.status_code, 200, system_read.text)
|
|
self.assertEqual(system_read.json()["current_version"]["title"], "System baseline")
|
|
|
|
def test_governance_read_endpoints_do_not_create_default_project_when_project_is_missing(self) -> None:
|
|
self._clear_tables()
|
|
ctx = self._seed_approved_member_without_project()
|
|
before_count = self.core.db.fetch_one("SELECT COUNT(*) AS count FROM projects WHERE user_id = ?", (ctx["member_id"],))
|
|
self.assertEqual(int((before_count or {}).get("count") or 0), 0)
|
|
|
|
effective_response = self.client.get(
|
|
"/v2/oneliner/governance/effective",
|
|
headers=ctx["member_headers"],
|
|
params={"platform": "douyin"},
|
|
)
|
|
self.assertEqual(effective_response.status_code, 200, effective_response.text)
|
|
effective_payload = effective_response.json()
|
|
self.assertEqual(effective_payload["project_id"], "")
|
|
self.assertEqual(effective_payload["layers"], [])
|
|
|
|
global_response = self.client.get(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=ctx["member_headers"],
|
|
)
|
|
self.assertEqual(global_response.status_code, 200, global_response.text)
|
|
self.assertEqual(global_response.json()["scope"]["subject_project_id"], "")
|
|
self.assertIsNone(global_response.json()["current_version"])
|
|
|
|
after_count = self.core.db.fetch_one("SELECT COUNT(*) AS count FROM projects WHERE user_id = ?", (ctx["member_id"],))
|
|
self.assertEqual(int((after_count or {}).get("count") or 0), 0)
|
|
|
|
def test_admin_governance_directory_lists_accounts_and_projects(self) -> None:
|
|
response = self.client.get(
|
|
"/v2/admin/oneliner/governance/directory",
|
|
headers=self.ctx["admin_headers"],
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
payload = response.json()
|
|
self.assertGreaterEqual(payload["count"], 2)
|
|
member = next((item for item in payload["items"] if item["id"] == self.ctx["member_id"]), None)
|
|
self.assertIsNotNone(member)
|
|
assert member is not None
|
|
self.assertEqual(member["project_count"], 1)
|
|
self.assertEqual(member["projects"][0]["id"], self.ctx["project_id"])
|
|
|
|
def test_admin_override_versions_support_rollback(self) -> None:
|
|
first_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": "Override v1",
|
|
"summary": "first override",
|
|
"policy": {"actions": {"max_cards": 2}},
|
|
"reason": "first override",
|
|
},
|
|
)
|
|
self.assertEqual(first_response.status_code, 200, first_response.text)
|
|
first_version_id = first_response.json()["current_version"]["id"]
|
|
|
|
second_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": "Override v2",
|
|
"summary": "second override",
|
|
"policy": {"actions": {"max_cards": 5}},
|
|
"reason": "second override",
|
|
},
|
|
)
|
|
self.assertEqual(second_response.status_code, 200, second_response.text)
|
|
|
|
versions_before = self.client.get(
|
|
"/v2/admin/oneliner/governance/overrides/versions",
|
|
headers=self.ctx["admin_headers"],
|
|
params={
|
|
"target_user_id": self.ctx["member_id"],
|
|
"target_project_id": self.ctx["project_id"],
|
|
"platform": "douyin",
|
|
},
|
|
)
|
|
self.assertEqual(versions_before.status_code, 200, versions_before.text)
|
|
self.assertEqual(versions_before.json()["count"], 2)
|
|
|
|
rollback_response = self.client.post(
|
|
"/v2/admin/oneliner/governance/overrides/rollback",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"target_user_id": self.ctx["member_id"],
|
|
"target_project_id": self.ctx["project_id"],
|
|
"platform": "douyin",
|
|
"version_id": first_version_id,
|
|
"reason": "rollback to v1",
|
|
},
|
|
)
|
|
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"]["actions"]["max_cards"], 2)
|
|
|
|
versions_after = self.client.get(
|
|
"/v2/admin/oneliner/governance/overrides/versions",
|
|
headers=self.ctx["admin_headers"],
|
|
params={
|
|
"target_user_id": self.ctx["member_id"],
|
|
"target_project_id": self.ctx["project_id"],
|
|
"platform": "douyin",
|
|
},
|
|
)
|
|
self.assertEqual(versions_after.status_code, 200, versions_after.text)
|
|
self.assertEqual(versions_after.json()["count"], 3)
|
|
|
|
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_user_platform_versions_support_rollback_by_creating_new_version(self) -> None:
|
|
first_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 v1",
|
|
"policy": {"douyin": {"benchmark_mode": "strict"}},
|
|
"reason": "first platform 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/platforms/douyin",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "Douyin strategy v2",
|
|
"policy": {"douyin": {"benchmark_mode": "aggressive"}},
|
|
"reason": "push harder",
|
|
},
|
|
)
|
|
self.assertEqual(second_response.status_code, 200, second_response.text)
|
|
|
|
versions_before = self.client.get(
|
|
"/v2/oneliner/governance/user/platforms/douyin/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/platforms/douyin/rollback",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"version_id": first_version_id,
|
|
"reason": "restore previous platform strategy",
|
|
},
|
|
)
|
|
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"]["douyin"]["benchmark_mode"], "strict")
|
|
|
|
versions_after = self.client.get(
|
|
"/v2/oneliner/governance/user/platforms/douyin/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_user_policy_audits_include_personal_and_admin_layers_for_project(self) -> None:
|
|
self.client.put(
|
|
"/v2/oneliner/governance/user/global",
|
|
headers=self.ctx["member_headers"],
|
|
json={
|
|
"project_id": self.ctx["project_id"],
|
|
"title": "Global strategy",
|
|
"policy": {"tone": {"style": "analytical"}},
|
|
"reason": "personalize defaults",
|
|
},
|
|
)
|
|
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": "Admin override",
|
|
"policy": {"actions": {"max_cards": 4}},
|
|
"reason": "contain drift",
|
|
},
|
|
)
|
|
|
|
response = self.client.get(
|
|
"/v2/oneliner/governance/user/audits",
|
|
headers=self.ctx["member_headers"],
|
|
params={"project_id": self.ctx["project_id"], "platform": "douyin"},
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
payload = response.json()
|
|
self.assertGreaterEqual(payload["count"], 2)
|
|
scope_kinds = [item["scope_kind"] for item in payload["items"]]
|
|
self.assertIn("user_global", scope_kinds)
|
|
self.assertIn("admin_override", scope_kinds)
|
|
first_item = payload["items"][0]
|
|
self.assertIn("version", first_item)
|
|
self.assertIn("scope", first_item)
|
|
self.assertNotIn("actor_user_id", first_item)
|
|
self.assertNotIn("policy", first_item["version"])
|
|
self.assertNotIn("reason", first_item["version"])
|
|
self.assertNotIn("actor_user_id", first_item["version"])
|
|
|
|
def test_admin_policy_audits_include_target_and_system_layers(self) -> None:
|
|
self.client.put(
|
|
"/v2/admin/oneliner/governance/system/main-agent",
|
|
headers=self.ctx["admin_headers"],
|
|
json={
|
|
"title": "System main",
|
|
"policy": {"homepage": {"focus": "ops"}},
|
|
"reason": "seed system baseline",
|
|
},
|
|
)
|
|
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": "Admin override",
|
|
"policy": {"actions": {"max_cards": 5}},
|
|
"reason": "focus target account",
|
|
},
|
|
)
|
|
|
|
response = self.client.get(
|
|
"/v2/admin/oneliner/governance/audits",
|
|
headers=self.ctx["admin_headers"],
|
|
params={
|
|
"target_user_id": self.ctx["member_id"],
|
|
"target_project_id": self.ctx["project_id"],
|
|
"platform": "douyin",
|
|
"include_system": "1",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
payload = response.json()
|
|
self.assertGreaterEqual(payload["count"], 2)
|
|
scope_kinds = [item["scope_kind"] for item in payload["items"]]
|
|
self.assertIn("system_main", scope_kinds)
|
|
self.assertIn("admin_override", scope_kinds)
|
|
|
|
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)
|