158 lines
9.2 KiB
Python
158 lines
9.2 KiB
Python
"""长期记忆人工确认、版本替代及管理边界的回归测试。"""
|
|
from types import SimpleNamespace
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
from app.access import canvas_allows_resource
|
|
from app.agents.snapshot import freeze_agent_snapshot
|
|
from app.main import create_pending_memory, decide_memory, list_memories, update_memory
|
|
from app.models import Agent, AppUser, Canvas, CanvasGrant, Memory, PendingMemory
|
|
|
|
@pytest.fixture
|
|
def scope():
|
|
engine = create_engine('sqlite://')
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
user = AppUser(username='memory-owner', role='admin', organization_id=1)
|
|
other = AppUser(username='memory-other', role='editor', organization_id=1)
|
|
foreign = AppUser(username='memory-foreign', role='admin', organization_id=2)
|
|
session.add_all([user, other, foreign]); session.commit()
|
|
agent = Agent(name='记忆测试', role='content', owner_user_id=user.id, organization_id=1)
|
|
session.add(agent); session.commit()
|
|
yield session, user, other, foreign, agent
|
|
engine.dispose()
|
|
|
|
def request(user):
|
|
return SimpleNamespace(state=SimpleNamespace(user=user))
|
|
|
|
def candidate(session, user, agent, content='偏好口语表达', **kwargs):
|
|
row = PendingMemory(target_agent_id=agent.id, content=content, owner_user_id=user.id,
|
|
organization_id=user.organization_id, **kwargs)
|
|
session.add(row); session.commit(); session.refresh(row)
|
|
return row
|
|
|
|
def test_metadata_edit_remains_recallable_and_get_does_not_count(scope):
|
|
session, user, _, _, agent = scope
|
|
row = Memory(agent_id=agent.id, title='旧标题', content='保持口语表达', owner_user_id=user.id)
|
|
session.add(row); session.commit(); session.refresh(row)
|
|
before = freeze_agent_snapshot(session, agent, user=user)
|
|
update_memory(row.id, {'title': '新标题', 'description': '口播场景'}, session, request(user))
|
|
after = freeze_agent_snapshot(session, agent, user=user)
|
|
assert before['memories'][0]['title'] == '旧标题'
|
|
assert after['memories'][0]['title'] == '新标题'
|
|
assert after['memories'][0]['id'] == row.id
|
|
listing = list_memories(q='口播场景', s=session, request=request(user))
|
|
assert listing[0]['index_status'] == 'indexed'
|
|
assert listing[0]['usage_count'] == 0 and listing[0]['last_used_at'] == ''
|
|
assert listing[0]['can_manage']
|
|
|
|
def test_confirm_refines_candidate_and_replaces_immutable_old_content(scope):
|
|
session, user, _, _, agent = scope
|
|
old = Memory(agent_id=agent.id, content='多举促销例子', owner_user_id=user.id)
|
|
session.add(old); session.commit(); session.refresh(old)
|
|
pending = candidate(session, user, agent, '这段不要总说促销')
|
|
result = decide_memory(pending.id, {'action': 'approve', 'content': '教学段优先用生活例子',
|
|
'type': 'L2', 'replace_memory_id': old.id}, session, request(user))
|
|
session.refresh(old); session.refresh(pending)
|
|
new = session.get(Memory, result['memory_id'])
|
|
assert old.status == 'expired' and old.content == '多举促销例子'
|
|
assert new.content == '教学段优先用生活例子' and new.layer == 'L2' and new.supersedes_id == old.id
|
|
assert pending.content == '这段不要总说促销' and pending.status == 'approved'
|
|
with pytest.raises(HTTPException) as exc:
|
|
update_memory(old.id, {'status': 'active'}, session, request(user))
|
|
assert exc.value.status_code == 409
|
|
|
|
def test_restoring_ancestor_cannot_conflict_with_active_later_revision(scope):
|
|
session, user, _, _, agent = scope
|
|
first = Memory(agent_id=agent.id, content='版本一', status='expired', owner_user_id=user.id)
|
|
session.add(first); session.commit(); session.refresh(first)
|
|
second = Memory(agent_id=agent.id, content='版本二', status='expired', owner_user_id=user.id, supersedes_id=first.id)
|
|
session.add(second); session.commit(); session.refresh(second)
|
|
third = Memory(agent_id=agent.id, content='版本三', owner_user_id=user.id, supersedes_id=second.id)
|
|
session.add(third); session.commit()
|
|
with pytest.raises(HTTPException) as exc:
|
|
update_memory(first.id, {'status': 'active'}, session, request(user))
|
|
assert exc.value.status_code == 409
|
|
session.rollback()
|
|
update_memory(third.id, {'status': 'expired'}, session, request(user))
|
|
update_memory(first.id, {'status': 'active'}, session, request(user))
|
|
assert first.status == 'active'
|
|
|
|
def test_unknown_empty_and_repeat_approval_are_rejected(scope):
|
|
session, user, _, _, agent = scope
|
|
row = candidate(session, user, agent)
|
|
for body in ({'action': 'unknown'}, {'action': 'approve', 'content': ' '},
|
|
{'action': 'approve', 'type': 'other'}, {'action': 'approve', 'tags': 'oops'}):
|
|
with pytest.raises(HTTPException) as exc:
|
|
decide_memory(row.id, body, session, request(user))
|
|
assert exc.value.status_code == 422
|
|
decide_memory(row.id, {'action': 'approve'}, session, request(user))
|
|
with pytest.raises(HTTPException) as exc:
|
|
decide_memory(row.id, {'action': 'approve'}, session, request(user))
|
|
assert exc.value.status_code == 409
|
|
assert len(session.exec(select(Memory)).all()) == 1
|
|
|
|
def test_dedup_does_not_merge_users_private_preferences(scope):
|
|
session, user, other, _, agent = scope
|
|
first = candidate(session, user, agent)
|
|
second = candidate(session, other, agent)
|
|
third = candidate(session, user, agent)
|
|
a = decide_memory(first.id, {'action': 'approve'}, session, request(user))
|
|
b = decide_memory(second.id, {'action': 'approve'}, session, request(user))
|
|
c = decide_memory(third.id, {'action': 'approve'}, session, request(user))
|
|
assert a['memory_id'] != b['memory_id'] and a['memory_id'] == c['memory_id']
|
|
|
|
def test_replacement_cannot_expire_another_users_memory(scope):
|
|
session, user, other, foreign, agent = scope
|
|
old = Memory(agent_id=agent.id, content='他人的偏好', owner_user_id=other.id)
|
|
session.add(old); session.commit(); session.refresh(old)
|
|
row = candidate(session, user, agent)
|
|
with pytest.raises(HTTPException) as exc:
|
|
decide_memory(row.id, {'action': 'approve', 'replace_memory_id': old.id}, session, request(user))
|
|
assert exc.value.status_code == 422
|
|
with pytest.raises(HTTPException) as exc:
|
|
decide_memory(row.id, {'action': 'approve'}, session, request(foreign))
|
|
assert exc.value.status_code == 404
|
|
assert old.status == 'active' and row.status == 'pending'
|
|
|
|
def test_manual_memory_is_private_pending_and_manage_permission_required(scope):
|
|
session, user, other, _, agent = scope
|
|
payload = {'agent_id': agent.id, 'content': '用具体例子教学', 'type': 'L1', 'visibility': 'global'}
|
|
created = create_pending_memory(payload, session, request(user))
|
|
row = session.get(PendingMemory, created['id'])
|
|
assert row.status == 'pending' and row.visibility == 'private' and row.owner_user_id == user.id
|
|
assert create_pending_memory(payload, session, request(user))['id'] == created['id']
|
|
assert session.exec(select(Memory)).all() == []
|
|
with pytest.raises(HTTPException) as exc:
|
|
create_pending_memory(payload, session, request(other))
|
|
assert exc.value.status_code == 404
|
|
|
|
def test_canvas_shares_configured_experience_without_other_users_private_memory(scope):
|
|
session, owner, viewer, foreign, agent = scope
|
|
canvas = Canvas(name='共享测试画布', owner_user_id=owner.id, visibility='shared', nodes=[{'agent_id': agent.id}])
|
|
session.add(canvas); session.commit(); session.refresh(canvas)
|
|
session.add(CanvasGrant(canvas_id=canvas.id, user_id=viewer.id, permission='use'))
|
|
configured = Memory(agent_id=agent.id, content='配置者经验', owner_user_id=owner.id)
|
|
private = Memory(agent_id=agent.id, content='另一使用者偏好', owner_user_id=999)
|
|
cross_tenant = Memory(agent_id=agent.id, content='其他组织', owner_user_id=owner.id, organization_id=2)
|
|
session.add_all([configured, private, cross_tenant]); session.commit()
|
|
assert canvas_allows_resource(session, configured, viewer, 'use')
|
|
assert not canvas_allows_resource(session, private, viewer, 'use')
|
|
assert not canvas_allows_resource(session, cross_tenant, viewer, 'use')
|
|
assert not canvas_allows_resource(session, configured, viewer, 'manage')
|
|
|
|
|
|
def test_replacement_preserves_explicit_grants_and_visibility(scope):
|
|
from app.models import MemoryGrant
|
|
session, user, other, _, agent = scope
|
|
old = Memory(agent_id=agent.id, content='旧共享经验', owner_user_id=user.id, visibility='shared', tags=['教学'])
|
|
session.add(old); session.commit(); session.refresh(old)
|
|
session.add(MemoryGrant(memory_id=old.id, user_id=other.id, permission='use'))
|
|
session.commit()
|
|
row = candidate(session, user, agent, '新共享经验', visibility='shared')
|
|
response = decide_memory(row.id, {'action': 'approve', 'replace_memory_id': old.id}, session, request(user))
|
|
new = session.get(Memory, response['memory_id'])
|
|
grant = session.exec(select(MemoryGrant).where(MemoryGrant.memory_id == new.id)).one()
|
|
assert grant.user_id == other.id and grant.permission == 'use' and grant.enabled
|
|
assert new.visibility == 'shared' and new.tags == ['教学']
|