55 lines
2.6 KiB
Python
55 lines
2.6 KiB
Python
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
from fastapi import FastAPI
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
from api import xingbang_workspace as workspace
|
||
|
|
|
||
|
|
class WorkflowTests(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
||
|
|
self.old = workspace.DB_PATH
|
||
|
|
workspace.DB_PATH = Path(self.tmp.name) / 'test.sqlite3'
|
||
|
|
app = FastAPI()
|
||
|
|
app.include_router(workspace.router)
|
||
|
|
self.client = TestClient(app)
|
||
|
|
self.base = '/api/xingbang/workspace/projects'
|
||
|
|
self.project = self.client.post(self.base, json={'name':'测试','client':'客户','amount':5000,'assets':10000,'liabilities':7500,'lease_value':5500}).json()
|
||
|
|
def tearDown(self):
|
||
|
|
self.client.close()
|
||
|
|
workspace.DB_PATH = self.old
|
||
|
|
self.tmp.cleanup()
|
||
|
|
def act(self, kind, **kwargs):
|
||
|
|
return self.client.post(self.base+'/'+self.project['id']+'/actions', json={'kind':kind,'actor':'测试','note':'人工核对依据',**kwargs})
|
||
|
|
def test_missing_and_evidence(self):
|
||
|
|
p = self.act('assess').json()
|
||
|
|
self.assertEqual(p['risks'][2]['category'], '信息缺失')
|
||
|
|
self.assertEqual(p['risks'][0]['value'], .75)
|
||
|
|
self.assertEqual(p['risks'][0]['evidence']['snapshot'], p['snapshots'][-1]['id'])
|
||
|
|
self.assertEqual(self.act('report').status_code,409)
|
||
|
|
def test_report_lifecycle_and_invalidation(self):
|
||
|
|
self.act('assess')
|
||
|
|
for target in ['debt','lease','insurance']:
|
||
|
|
self.assertEqual(self.act('confirm',target=target).status_code,200)
|
||
|
|
p = self.act('report').json()
|
||
|
|
report = p['reports'][-1]
|
||
|
|
p = self.act('finalize',target=report['id']).json()
|
||
|
|
self.assertEqual(p['cards'][0]['status'],'草稿')
|
||
|
|
self.assertEqual(self.act('assess').status_code,409)
|
||
|
|
p = self.act('change',amount=6000).json()
|
||
|
|
self.assertTrue(p['reports'][0]['stale'])
|
||
|
|
self.assertEqual(p['snapshots'][0]['data']['amount'],5000)
|
||
|
|
self.assertEqual(self.act('report').status_code,409)
|
||
|
|
def test_task_requires_submission_before_close(self):
|
||
|
|
self.act('assess')
|
||
|
|
p = self.act('task',target='debt').json()
|
||
|
|
task=p['tasks'][0]['id']
|
||
|
|
self.assertEqual(self.act('close_task',target=task).status_code,409)
|
||
|
|
self.assertEqual(self.act('submit_task',target=task).status_code,200)
|
||
|
|
self.assertEqual(self.act('close_task',target=task).json()['tasks'][0]['status'],'已解决')
|
||
|
|
def test_invalid_amount(self):
|
||
|
|
self.assertEqual(self.act('change',amount=-1).status_code,422)
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
unittest.main()
|