Files
aios/tests/test_model_governance_admin.py
T

83 lines
2.6 KiB
Python

from __future__ import annotations
import json
from pathlib import Path
import sys
ROOT=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(ROOT/"active-runtime/app"))
from aios_model_governance_admin import apply_config
from aios_model_governance import ModelGovernanceStore
def config():
return {
"schema":"aios.model-governance.bootstrap.v1",
"providers":[{
"provider_id":"provider-a",
"credential_ref":"runtime-profile:provider-a",
"max_failures":2,
"cooldown_seconds":60,
}],
"models":[{
"model_id":"model-a",
"provider_id":"provider-a",
"version":"v1",
"capabilities":["text"],
"input_cost_micros_per_1k":1,
"output_cost_micros_per_1k":2,
"max_input_tokens":1000,
"max_output_tokens":100,
"evaluation_state":"APPROVED",
"evaluation_sha256":"a"*64,
"priority":10,
}],
"policies":[{
"scope_type":"legacy",
"tenant_id":None,
"project_id":None,
"allowed_models":["model-a"],
"required_capabilities":["text"],
"daily_token_budget":10000,
"daily_cost_budget_micros":1000,
"fallback_limit":0,
}],
}
def test_admin_apply_is_idempotent(tmp_path):
cfg=tmp_path/"config.json"
cfg.write_text(json.dumps(config()),encoding="utf-8")
db=tmp_path/"db.sqlite3"
first=apply_config(db,cfg)
second=apply_config(db,cfg)
assert first["changes"]["providers"]==1
assert first["changes"]["models"]==1
assert first["changes"]["policies"]==1
assert second["changes"]["providers"]==0
assert second["changes"]["models"]==0
assert second["changes"]["policies"]==0
assert second["changes"]["unchanged"]==3
store=ModelGovernanceStore(db)
assert store.verify_audit_chain() is True
store.close()
def test_admin_config_change_uses_cas_revision(tmp_path):
value=config()
cfg=tmp_path/"config.json"
cfg.write_text(json.dumps(value),encoding="utf-8")
db=tmp_path/"db.sqlite3"
apply_config(db,cfg)
value["providers"][0]["cooldown_seconds"]=120
cfg.write_text(json.dumps(value),encoding="utf-8")
result=apply_config(db,cfg)
assert result["changes"]["providers"]==1
store=ModelGovernanceStore(db)
row=store.conn.execute(
"select revision,cooldown_seconds from model_governance_providers where provider_id='provider-a'"
).fetchone()
assert tuple(row)==(2,120)
store.close()