feat: add takeover fields to Authorization model
Add takeover_enabled, takeover_mode, and takeover_delay_seconds columns to the Authorization model with a SQLite migration, and include them in to_dict(). Test: test_takeover_model.py.
This commit is contained in:
@@ -35,6 +35,9 @@ def init_db():
|
||||
("knowledge_docs", "chunk_count", "INTEGER DEFAULT 0"),
|
||||
("knowledge_docs", "vectorized_at", "TIMESTAMP"),
|
||||
("avatars", "owner_id", "VARCHAR DEFAULT ''"),
|
||||
("authorizations", "takeover_enabled", "BOOLEAN DEFAULT 0"),
|
||||
("authorizations", "takeover_mode", "VARCHAR DEFAULT 'immediate'"),
|
||||
("authorizations", "takeover_delay_seconds", "INTEGER DEFAULT 30"),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -51,6 +51,9 @@ class Authorization(Base):
|
||||
target_name = Column(String, default="")
|
||||
permissions = Column(JSON, default=list)
|
||||
status = Column(String, default="active") # active | inactive
|
||||
takeover_enabled = Column(Boolean, default=False) # 是否开启分身接管
|
||||
takeover_mode = Column(String, default="immediate") # immediate | delayed
|
||||
takeover_delay_seconds = Column(Integer, default=30) # 延迟秒数
|
||||
created_at = Column(DateTime, server_default=func.now())
|
||||
|
||||
def to_dict(self):
|
||||
@@ -62,6 +65,9 @@ class Authorization(Base):
|
||||
"targetName": self.target_name,
|
||||
"permissions": self.permissions or [],
|
||||
"status": self.status,
|
||||
"takeoverEnabled": self.takeover_enabled,
|
||||
"takeoverMode": self.takeover_mode,
|
||||
"takeoverDelaySeconds": self.takeover_delay_seconds,
|
||||
"createdAt": _iso(self.created_at),
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import pytest
|
||||
from database import init_db, SessionLocal
|
||||
from models import Authorization
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def setup_database():
|
||||
"""Initialize DB tables and seed an Authorization row."""
|
||||
init_db()
|
||||
db = SessionLocal()
|
||||
try:
|
||||
existing = db.query(Authorization).first()
|
||||
if existing is None:
|
||||
auth = Authorization(
|
||||
id="test-auth-1",
|
||||
avatar_id="test-avatar-1",
|
||||
target_type="user",
|
||||
target_id="test-user-1",
|
||||
target_name="Test User",
|
||||
permissions=["read", "write"],
|
||||
status="active",
|
||||
)
|
||||
db.add(auth)
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
@@ -0,0 +1,30 @@
|
||||
from database import SessionLocal
|
||||
from models import Authorization
|
||||
|
||||
|
||||
def test_authorization_takeover_fields():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
auth = db.query(Authorization).first()
|
||||
assert auth is not None
|
||||
# Check new fields exist and have default values
|
||||
assert hasattr(auth, 'takeover_enabled')
|
||||
assert hasattr(auth, 'takeover_mode')
|
||||
assert hasattr(auth, 'takeover_delay_seconds')
|
||||
assert auth.takeover_enabled == False
|
||||
assert auth.takeover_mode == 'immediate'
|
||||
assert auth.takeover_delay_seconds == 30
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_authorization_to_dict_includes_takeover():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
auth = db.query(Authorization).first()
|
||||
d = auth.to_dict()
|
||||
assert 'takeoverEnabled' in d
|
||||
assert 'takeoverMode' in d
|
||||
assert 'takeoverDelaySeconds' in d
|
||||
finally:
|
||||
db.close()
|
||||
Reference in New Issue
Block a user