186 lines
10 KiB
Python
186 lines
10 KiB
Python
"""Neutral fixtures; only model/RAG boundaries mocked, SQLite is real."""
|
|
import importlib.util
|
|
import pytest
|
|
from types import SimpleNamespace
|
|
from sqlmodel import SQLModel, Session, create_engine
|
|
from fastapi import HTTPException
|
|
from unittest.mock import patch
|
|
|
|
|
|
def test_persistent_workflow_empty_evidence_and_permission():
|
|
from app import training_map as tm
|
|
assert hasattr(tm, 'create_draft'), 'persistent API missing'
|
|
from app.models import AppUser, KnowledgeBase
|
|
engine = create_engine('sqlite://')
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as s:
|
|
user = AppUser(username='neutral', role='editor', id=91)
|
|
request = SimpleNamespace(state=SimpleNamespace(user=user))
|
|
kb = KnowledgeBase(name='时间管理资料', owner_user_id=91, visibility='private')
|
|
s.add(kb); s.commit(); s.refresh(kb)
|
|
row = tm.create_draft(tm.DraftInput(topic='时间管理训练营', kb_ids=[kb.id]), request, s)
|
|
with pytest.raises(HTTPException):
|
|
tm.generate_day(row.id, 1, tm.VersionInput(version=row.version), request, s)
|
|
with patch.object(tm, 'search_kbs_remote', return_value=[]), patch.object(tm, 'chat_json') as ai:
|
|
with pytest.raises(HTTPException) as exc:
|
|
tm.assess(row.id, tm.AssessmentInput(mode='quick', version=row.version), request, s)
|
|
assert exc.value.status_code == 422
|
|
ai.assert_not_called()
|
|
assert s.get(tm.TrainingMap, row.id).topic == '时间管理训练营'
|
|
stranger = SimpleNamespace(state=SimpleNamespace(user=AppUser(id=92, role='editor')))
|
|
with pytest.raises(HTTPException):
|
|
tm.get_draft(row.id, stranger, s)
|
|
with pytest.raises(HTTPException):
|
|
tm.create_draft(tm.DraftInput(topic='中性测试', kb_ids=[kb.id]), stranger, s)
|
|
|
|
|
|
def test_schema_rejects_empty_leaves_and_wrong_day_count():
|
|
from app import training_map as tm
|
|
assert hasattr(tm, 'DayMap'), 'strict schema missing'
|
|
from pydantic import ValidationError
|
|
with pytest.raises(ValidationError):
|
|
tm.DayMap.model_validate({'day': 1, 'title': '时间管理', 'children': [{'title': '练习', 'children': [{'title': '空标题'}]}]})
|
|
with pytest.raises(ValidationError):
|
|
tm.Plan.model_validate({'recommended_days': 2, 'rationale': '只有一日', 'insufficiencies': [], 'days': [{'day': 1, 'title': '时间记录', 'source_ids': ['E1'], 'support': '足够'}]})
|
|
with pytest.raises(ValidationError, match='duplicate day titles'):
|
|
tm.Plan.model_validate({'recommended_days': 2, 'rationale': '两天内容安排', 'insufficiencies': [], 'days': [
|
|
{'day': 1, 'title': '时间记录', 'source_ids': ['E1'], 'support': '第一天记录方法'},
|
|
{'day': 2, 'title': '时间 记录', 'source_ids': ['E1'], 'support': '第二天不应重复标题'},
|
|
]})
|
|
|
|
|
|
def test_missing_key_blocks_before_network():
|
|
from app import training_map as tm
|
|
from app.models import ModelProvider
|
|
engine = create_engine('sqlite://')
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as s:
|
|
p = ModelProvider(name='neutral-no-key', base_url='https://invalid.example/v1', api_key='', default_model='neutral', enabled=True)
|
|
s.add(p); s.commit(); s.refresh(p)
|
|
with patch.object(tm, 'chat_json') as ai:
|
|
with pytest.raises(HTTPException) as exc:
|
|
tm.model_json(s, tm.TrainingMap(), tm.Plan, 'neutral', [])
|
|
assert exc.value.status_code == 503
|
|
ai.assert_not_called()
|
|
|
|
|
|
def test_full_confirm_retry_version_and_source_workflow():
|
|
with patch('app.secret_store.resolve_secret', return_value='unit-test-only'):
|
|
_full_workflow()
|
|
|
|
|
|
def _full_workflow():
|
|
from app import training_map as tm
|
|
from app.models import AppUser, KnowledgeBase, ModelProvider, TaskQueueItem
|
|
from sqlmodel import select
|
|
engine = create_engine('sqlite://')
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as s:
|
|
req = SimpleNamespace(state=SimpleNamespace(user=AppUser(id=91, role='admin')))
|
|
kb = KnowledgeBase(name='专注力中性资料', owner_user_id=91)
|
|
provider = ModelProvider(name='unit-mock-only', base_url='http://127.0.0.1:9/v1', api_key='unit-test-not-a-real-key', default_model='neutral', enabled=True)
|
|
s.add_all([kb, provider]); s.commit(); s.refresh(kb)
|
|
row = tm.create_draft(tm.DraftInput(topic='专注力训练营', kb_ids=[kb.id]), req, s)
|
|
hit = {'kb_id': kb.id, 'text': '先连续记录三天时间开销，每半小时记录任务、干扰与完成量，再识别主要干扰来源。', 'meta': {'file': 'neutral.txt'}}
|
|
plan = {'recommended_days': 1, 'rationale': '资料可以支持时间记录与干扰识别练习。', 'insufficiencies': ['没有高级干预资料'], 'days': [{'day': 1, 'title': '记录时间开销', 'source_ids': ['E1'], 'support': '三天记录方法有具体步骤'}]}
|
|
tree = {'day': 1, 'title': '记录时间开销', 'children': [{'title': '时间观察', 'children': [{'title': '记录练习', 'children': [{'title': '半小时记录法', 'content': hit['text'], 'source_ids': ['E1']}]}]}]}
|
|
with patch.object(tm, 'search_kbs_remote', return_value=[hit, hit]), patch.object(tm, 'chat_json', return_value={'status': 'ok', 'json': plan}) as ai:
|
|
row = tm.assess(row.id, tm.AssessmentInput(mode='quick', version=row.version), req, s)
|
|
assert len(row.evidence) == 1 and ai.call_count == 1
|
|
v = row.version
|
|
with pytest.raises(HTTPException): tm.confirm(row.id, tm.VersionInput(version=v-1), req, s)
|
|
row = tm.confirm(row.id, tm.VersionInput(version=v), req, s)
|
|
with patch.object(tm, 'chat_json', return_value={'status': 'ok', 'json': {'day': 1}}):
|
|
with pytest.raises(HTTPException): tm.generate_day(row.id, 1, tm.VersionInput(version=v), req, s)
|
|
assert tm.get_draft(row.id, req, s).maps == {}
|
|
with patch.object(tm, 'chat_json', return_value={'status': 'ok', 'json': tree}) as ai:
|
|
row = tm.generate_day(row.id, 1, tm.VersionInput(version=v), req, s)
|
|
row = tm.generate_day(row.id, 1, tm.VersionInput(version=v), req, s)
|
|
assert ai.call_count == 1 and row.status == 'done'
|
|
assert len(s.exec(select(TaskQueueItem)).all()) == 1
|
|
exported = tm.export_draft(row.id, req, 'markdown', s)['content']
|
|
assert '半小时记录法' in exported
|
|
assert '钩子：' in exported and 'data:image/svg+xml' in exported
|
|
with pytest.raises(ValueError): tm.validate_sources({'source_ids': ['invented']}, row.evidence)
|
|
row = tm.update_draft(row.id, tm.UpdateInput(topic=row.topic, kb_ids=row.kb_ids, requested_days=2, version=v), req, s)
|
|
assert row.confirmed_version is None and row.plan == {} and row.version > v
|
|
with pytest.raises(HTTPException): tm.generate_day(row.id, 1, tm.VersionInput(version=v), req, s)
|
|
ident = row.id
|
|
tm.delete_draft(ident, req, s)
|
|
assert s.get(tm.TrainingMap, ident) is None and s.exec(select(TaskQueueItem)).all() == []
|
|
|
|
|
|
def test_training_map_has_confirmed_generation_gate():
|
|
assert importlib.util.find_spec('app.training_map') is not None, 'training-map backend missing'
|
|
from app.training_map import TrainingMap, require_confirmed
|
|
from fastapi import HTTPException
|
|
import pytest
|
|
with pytest.raises(HTTPException) as exc:
|
|
require_confirmed(TrainingMap(topic='时间管理训练营'), 1)
|
|
assert exc.value.status_code == 409
|
|
|
|
|
|
def test_generated_map_rejects_repeated_content_and_adds_hooks():
|
|
from app.training_map import validate_map_logic
|
|
|
|
tree = {
|
|
'day': 1,
|
|
'title': '专注力训练',
|
|
'children': [{
|
|
'title': '认识干扰',
|
|
'children': [
|
|
{'title': '记录干扰', 'content': '连续三天记录每次干扰发生的时间、来源和当时正在做的事情。', 'source_ids': ['E1']},
|
|
{'title': '复盘记录', 'content': '对三天记录进行归类，找出出现频率最高的干扰来源并确定优先处理项。', 'source_ids': ['E1']},
|
|
],
|
|
}],
|
|
}
|
|
result = validate_map_logic(tree)
|
|
assert result['children'][0]['hook']
|
|
assert result['children'][0]['children'][0]['hook'].startswith('承接下一步')
|
|
assert result['children'][0]['children'][1]['hook']
|
|
|
|
duplicate = {
|
|
'day': 1,
|
|
'title': '专注力训练',
|
|
'children': [{
|
|
'title': '认识干扰',
|
|
'children': [
|
|
{'title': '记录干扰', 'content': '连续三天记录每次干扰发生的时间、来源和当时正在做的事情。', 'source_ids': ['E1']},
|
|
{'title': '再次记录', 'content': '连续三天记录每次干扰发生的时间、来源和当时正在做的事情。', 'source_ids': ['E1']},
|
|
],
|
|
}],
|
|
}
|
|
with pytest.raises(ValueError, match='重复内容'):
|
|
validate_map_logic(duplicate)
|
|
|
|
next_day = {
|
|
'day': 2,
|
|
'title': '持续专注',
|
|
'children': [{
|
|
'title': '执行计划',
|
|
'children': [{'title': '复用段落', 'content': '连续三天记录每次干扰发生的时间、来源和当时正在做的事情。', 'source_ids': ['E1']}],
|
|
}],
|
|
}
|
|
with pytest.raises(ValueError, match='重复内容'):
|
|
validate_map_logic(next_day, {'1': tree})
|
|
|
|
|
|
def test_daily_generation_cannot_skip_previous_day():
|
|
from app import training_map as tm
|
|
from app.models import AppUser, KnowledgeBase
|
|
engine = create_engine('sqlite://')
|
|
SQLModel.metadata.create_all(engine)
|
|
with Session(engine) as s:
|
|
req = SimpleNamespace(state=SimpleNamespace(user=AppUser(id=91, role='admin')))
|
|
kb = KnowledgeBase(name='递进课程资料', owner_user_id=91)
|
|
s.add(kb); s.commit(); s.refresh(kb)
|
|
row = tm.TrainingMap(owner_user_id=91, organization_id=1, topic='递进课程', kb_ids=[kb.id],
|
|
version=1, confirmed_version=1,
|
|
evidence=[{'id': 'E1', 'kb_id': kb.id, 'text': '证据'}],
|
|
plan={'recommended_days': 2, 'rationale': '递进', 'insufficiencies': [],
|
|
'days': [{'day': 1, 'title': '基础', 'source_ids': ['E1'], 'support': '基础'},
|
|
{'day': 2, 'title': '应用', 'source_ids': ['E1'], 'support': '应用'}]})
|
|
s.add(row); s.commit(); s.refresh(row)
|
|
with pytest.raises(HTTPException, match='先生成并查看第1天'):
|
|
tm.generate_day(row.id, 2, tm.VersionInput(version=1), req, s)
|