|
@@ -0,0 +1,566 @@
|
|
|
|
|
+"""月度本地批次:40天双系统取数、身份边界、PDF接口与跨会话档案复用。"""
|
|
|
|
|
+from contextlib import asynccontextmanager, contextmanager
|
|
|
|
|
+import asyncio
|
|
|
|
|
+from datetime import timedelta
|
|
|
|
|
+import json
|
|
|
|
|
+from types import SimpleNamespace
|
|
|
|
|
+
|
|
|
|
|
+import pytest
|
|
|
|
|
+
|
|
|
|
|
+import config
|
|
|
|
|
+from agents.workspace import session_root
|
|
|
|
|
+from db import business_store, chat_store, main_fan_store
|
|
|
|
|
+from extensions.main_fan import exploration, gateway_queries, history, report_archive
|
|
|
|
|
+from extensions.main_fan.batch import BatchRunner
|
|
|
|
|
+from extensions.main_fan.batch_context import BatchScope, current_scope
|
|
|
|
|
+from extensions.main_fan.batch_models import BatchRequest, HistoryPoint, MinePlan, ShaftPlan
|
|
|
|
|
+from extensions.main_fan.batch_tools import checked_plan
|
|
|
|
|
+from extensions.main_fan.report_api import ReportClient, mock_report_client
|
|
|
|
|
+from extensions.registry import load_extensions
|
|
|
|
|
+from extensions.shaanxi.client import GatewayError
|
|
|
|
|
+from runtime.python_sandbox import capability, digest, run as run_python, workspace_file
|
|
|
|
|
+from test_main_fan_documents import native_pdf
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.fixture
|
|
|
|
|
+def batch_workspace(monkeypatch, tmp_path):
|
|
|
|
|
+ for directory in ('skills', 'db'):
|
|
|
|
|
+ (tmp_path / directory).symlink_to(config.ROOT / directory, target_is_directory=True)
|
|
|
|
|
+ monkeypatch.setattr(config, 'ROOT', tmp_path)
|
|
|
|
|
+ for key, value in {'MAIN_FAN_TARGETS_FILE': '', 'MAIN_FAN_RULES_FILE': '',
|
|
|
|
|
+ 'MAIN_FAN_SERIES_MAX_HOURS': '24', 'MAIN_FAN_HISTORY_CONCURRENCY': '2',
|
|
|
|
|
+ 'RESULT_PLATFORM_ENABLED': '1', 'MAIN_FAN_AUTO_PUBLISH': '1'}.items():
|
|
|
|
|
+ monkeypatch.setenv(key, value)
|
|
|
|
|
+ load_extensions()
|
|
|
|
|
+ source = tmp_path / 'original.pdf'
|
|
|
|
|
+ native_pdf(source)
|
|
|
|
|
+ return SimpleNamespace(root=tmp_path, pdf=source,
|
|
|
|
|
+ mine={'dept_id': '1219', 'mine_id': '00001219', 'mine_name': '测试甲矿'})
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@contextmanager
|
|
|
|
|
+def bound(workspace, *, client=None, shaft=None, days=40, mine=None):
|
|
|
|
|
+ session = chat_store.create_session('batch-test', 'full')
|
|
|
|
|
+ mine = mine or workspace.mine
|
|
|
|
|
+ shaft = shaft or ShaftPlan(shaft_id='north', shaft_name='北风井', identity_status='confirmed', identity_reason='测试明确井号',
|
|
|
|
|
+ points=[HistoryPoint(system='safety', node_id='state1', role='fan_state'),
|
|
|
|
|
+ HistoryPoint(system='safety', node_id='state2', role='fan_state'),
|
|
|
|
|
+ HistoryPoint(system='equipment', node_id='flow1', role='flow'),
|
|
|
|
|
+ HistoryPoint(system='equipment', node_id='flow2', role='flow')])
|
|
|
|
|
+ scope = BatchScope('batch-1', mine, 'batch-test', session['id'], workspace.root / 'out',
|
|
|
|
|
+ workspace.root / 'archives', '2026-10-20T15:00:00+08:00', days, client, shaft)
|
|
|
|
|
+ token = current_scope.set(scope)
|
|
|
|
|
+ prepared = exploration.prepare({'mine_code': mine['mine_id'], 'mine_name': mine['mine_name'],
|
|
|
|
|
+ 'dept_id': mine['dept_id'], 'use_gateway': True, 'measurements': {'batch': {
|
|
|
|
|
+ 'batch_id': scope.batch_id, 'cutoff': scope.cutoff, 'lookback_days': scope.lookback_days,
|
|
|
|
|
+ 'shaft_plan': shaft.model_dump(mode='json')}}}, scope.user, scope.session, job_id='local-test-no-push')
|
|
|
|
|
+ scope.active_workflow_id = prepared['workflow_id']
|
|
|
|
|
+ try:
|
|
|
|
|
+ yield scope, prepared['workflow_id']
|
|
|
|
|
+ finally:
|
|
|
|
|
+ current_scope.reset(token)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class HistoryGateway:
|
|
|
|
|
+ def __init__(self):
|
|
|
|
|
+ self.calls = []
|
|
|
|
|
+ self.fail = set()
|
|
|
|
|
+
|
|
|
|
|
+ async def proxy(self, agent, path, *, params):
|
|
|
|
|
+ self.calls.append((path, params))
|
|
|
|
|
+ node = params.get('nodeIds') or params['nodeId']
|
|
|
|
|
+ if (node, params['startTime']) in self.fail:
|
|
|
|
|
+ raise GatewayError('test', 503, retryable=False)
|
|
|
|
|
+ if 'nodeIds' in params:
|
|
|
|
|
+ return {'code': 200, 'data': [[{'NodeID': node, 'DataTime': params['startTime'],
|
|
|
|
|
+ 'RealData': 1, 'KgDesc': '开'}], [{'NodeID': node}]]}
|
|
|
|
|
+ return {'code': 200, 'data': {'nodeId': node, 'unit': 'm3/min',
|
|
|
|
|
+ 'samples': [{'time': params['startTime'], 'value': 3800}]}}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_daily_windows_are_contiguous_and_rolling():
|
|
|
|
|
+ windows = history.daily_windows('2026-10-20 15:00:00', 40)
|
|
|
|
|
+ assert len(windows) == 40
|
|
|
|
|
+ assert windows[0][1].isoformat() == '2026-10-20T15:00:00+08:00'
|
|
|
|
|
+ assert windows[-1][0].isoformat() == '2026-09-10T15:00:00+08:00'
|
|
|
|
|
+ assert all(end - start == timedelta(days=1) for start, end in windows)
|
|
|
|
|
+ assert all(windows[index][0] == windows[index + 1][1] for index in range(39))
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_four_points_160_daily_queries_one_operation_and_resumption(batch_workspace, monkeypatch):
|
|
|
|
|
+ gateway = HistoryGateway()
|
|
|
|
|
+ monkeypatch.setattr(history, 'get_service', lambda _: gateway)
|
|
|
|
|
+ with bound(batch_workspace) as (scope, identifier):
|
|
|
|
|
+ receipt = await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert receipt['coverage_complete'] and receipt['completed_segments'] == 160
|
|
|
|
|
+ assert receipt['source_policy']['fan_state'] == 'safety'
|
|
|
|
|
+ assert receipt['source_policy']['flow'] == 'equipment'
|
|
|
|
|
+ assert len(gateway.calls) == 160
|
|
|
|
|
+ assert sum('nodeIds' in params for _, params in gateway.calls) == 80
|
|
|
|
|
+ assert sum('nodeId' in params for _, params in gateway.calls) == 80
|
|
|
|
|
+ assert all(params['deptId'] == '1219' for _, params in gateway.calls)
|
|
|
|
|
+ workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert len(workflow['stages']) == 2, '按日传输不消耗160次Agent操作'
|
|
|
|
|
+ assert len(workflow['context']['history_manifest']['segments']) == 160
|
|
|
|
|
+ archive = workspace_file(scope.user, scope.session, receipt['evidence_archive_path'])
|
|
|
|
|
+ assert archive.is_file() and len(list(archive.parent.glob('history/**/*.json'))) == 0
|
|
|
|
|
+ compact = json.loads(workspace_file(scope.user, scope.session, receipt['agent_data_path']).read_text())
|
|
|
|
|
+ assert len(compact['points']) == 4 and len(compact['points'][0]['state_periods']) == 1
|
|
|
|
|
+ assert compact['source_policy'] == receipt['source_policy']
|
|
|
|
|
+ await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert len(gateway.calls) == 160, '恢复成功段不能重复发请求'
|
|
|
|
|
+ exploration.validate_inputs(main_fan_store.workflow(identifier, scope.user, scope.session), scope.user, scope.session)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_history_failure_keeps_partial_and_only_failed_interval_retries(batch_workspace, monkeypatch):
|
|
|
|
|
+ gateway = HistoryGateway()
|
|
|
|
|
+ gateway.fail.add(('flow1', '2026-10-19 15:00:00'))
|
|
|
|
|
+ monkeypatch.setattr(history, 'get_service', lambda _: gateway)
|
|
|
|
|
+ with bound(batch_workspace, days=2) as (scope, identifier):
|
|
|
|
|
+ first = await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert first['failed_segments'] == 1 and not first['coverage_complete']
|
|
|
|
|
+ assert len(gateway.calls) == 8
|
|
|
|
|
+ gateway.fail.clear()
|
|
|
|
|
+ second = await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert second['completed_segments'] == 8 and second['coverage_complete']
|
|
|
|
|
+ assert len(gateway.calls) == 9
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_single_query_rejects_another_mine_before_gateway_access(batch_workspace, monkeypatch):
|
|
|
|
|
+ gateway = HistoryGateway()
|
|
|
|
|
+ monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
|
|
|
|
|
+ with bound(batch_workspace) as (scope, identifier):
|
|
|
|
|
+ with pytest.raises(ValueError, match='不属于'):
|
|
|
|
|
+ await gateway_queries.fetch(identifier, scope.user, scope.session, '/ignored', {'deptId': '1338'})
|
|
|
|
|
+ assert not gateway.calls
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_mock_report_interface_returns_original_bytes_and_fixture_provenance(batch_workspace):
|
|
|
|
|
+ async with mock_report_client(batch_workspace.pdf) as http:
|
|
|
|
|
+ client = ReportClient(http)
|
|
|
|
|
+ files, _, response = await client.list('00001219', 'north')
|
|
|
|
|
+ assert response['code'] == 200 and files[0]['is_mock']
|
|
|
|
|
+ assert files[0]['binding'] == 'test_fixture_not_mine_report'
|
|
|
|
|
+ target = batch_workspace.root / 'download.pdf'
|
|
|
|
|
+ assert await client.download(files[0], target) == digest(batch_workspace.pdf)
|
|
|
|
|
+ assert target.read_bytes() == batch_workspace.pdf.read_bytes()
|
|
|
|
|
+ assert (await http.get('/api/main-fan/report-files/not-found.pdf')).status_code == 404
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_report_archive_reuses_across_sessions_and_corruption_reparses(batch_workspace):
|
|
|
|
|
+ if not capability()['available']:
|
|
|
|
|
+ pytest.skip('需要真实Python沙箱')
|
|
|
|
|
+ async with mock_report_client(batch_workspace.pdf) as http:
|
|
|
|
|
+ client = ReportClient(http)
|
|
|
|
|
+ with bound(batch_workspace, client=client) as (scope, identifier):
|
|
|
|
|
+ first = await report_archive.fetch(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert first['reports'][0]['mode'] == 'analyze'
|
|
|
|
|
+ facts = {'source_sha256': digest(batch_workspace.pdf), 'page_count': 1, 'read_pages': [1],
|
|
|
|
|
+ 'page_sources': {'1': '/workspace/page_1.txt'},
|
|
|
|
|
+ 'reports': [{'mine_name': 'Synthetic Native Test Mine', 'report_id': 'NATIVE-TEST-001', 'page': 1}],
|
|
|
|
|
+ 'analysis': {'is_test': True}, 'limitations': ['不同矿的模拟原件']}
|
|
|
|
|
+ script = workspace_file(scope.user, scope.session, '/workspace/extract.py')
|
|
|
|
|
+ script.write_text('from pathlib import Path\nimport json\nPath("page_1.txt").write_text("Actual test page text")\nPath("facts.json").write_text(' + repr(json.dumps(facts)) + ')\n')
|
|
|
|
|
+ execution = await run_python(scope.user, scope.session, '/workspace/extract.py',
|
|
|
|
|
+ output_paths=['/workspace/facts.json', '/workspace/page_1.txt'])
|
|
|
|
|
+ assert execution['exit_code'] == 0, execution['stderr']
|
|
|
|
|
+ exploration.record_python(identifier, scope.user, scope.session, execution)
|
|
|
|
|
+ with pytest.raises(ValueError, match='没有可核验'):
|
|
|
|
|
+ report_archive.save(identifier, scope.user, scope.session, '/workspace/facts.json', [execution['run_id']])
|
|
|
|
|
+ assert not list(scope.cache_root.glob('*.json'))
|
|
|
|
|
+ from extensions.main_fan.middleware import record_page_read
|
|
|
|
|
+ from langchain_core.messages import ToolMessage
|
|
|
|
|
+ record_page_read(SimpleNamespace(tool_call={'name': 'read_file', 'args': {'file_path': '/workspace/page_1.txt'}}),
|
|
|
|
|
+ ToolMessage(content='Actual test page text', tool_call_id='page-read'))
|
|
|
|
|
+ receipt = report_archive.save(identifier, scope.user, scope.session, '/workspace/facts.json', [execution['run_id']])
|
|
|
|
|
+ assert receipt['page_count'] == 1
|
|
|
|
|
+ with bound(batch_workspace, client=client) as (scope, identifier):
|
|
|
|
|
+ reused = await report_archive.fetch(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert reused['reused_count'] == 1 and reused['reports'][0]['mode'] == 'reuse'
|
|
|
|
|
+ assert 'archive_python_path' in reused['reports'][0]
|
|
|
|
|
+ archive = next((batch_workspace.root / 'archives').glob('*.json'))
|
|
|
|
|
+ archive.write_text('{broken')
|
|
|
|
|
+ with bound(batch_workspace, client=client) as (scope, identifier):
|
|
|
|
|
+ assert (await report_archive.fetch(identifier, scope.user, scope.session))['reports'][0]['mode'] == 'analyze'
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_report_content_change_invalidates_archive(batch_workspace):
|
|
|
|
|
+ async with mock_report_client(batch_workspace.pdf) as http:
|
|
|
|
|
+ with bound(batch_workspace, client=ReportClient(http)) as (scope, identifier):
|
|
|
|
|
+ first = await report_archive.fetch(identifier, scope.user, scope.session)
|
|
|
|
|
+ sha = first['reports'][0]['sha256']
|
|
|
|
|
+ assert not report_archive.load_cached(scope, sha)
|
|
|
|
|
+ original = batch_workspace.pdf.read_bytes()
|
|
|
|
|
+ batch_workspace.pdf.write_bytes(original + b'\n% updated fixture\n')
|
|
|
|
|
+ async with mock_report_client(batch_workspace.pdf) as http:
|
|
|
|
|
+ with bound(batch_workspace, client=ReportClient(http)) as (scope, identifier):
|
|
|
|
|
+ second = await report_archive.fetch(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert second['reports'][0]['sha256'] != sha and second['reused_count'] == 0
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_claiming_26_pages_after_reading_only_10_cannot_create_cache(batch_workspace):
|
|
|
|
|
+ if not capability()['available']:
|
|
|
|
|
+ pytest.skip('需要真实沙箱')
|
|
|
|
|
+ import pypdfium2
|
|
|
|
|
+ from extensions.main_fan.middleware import record_page_read
|
|
|
|
|
+ from langchain_core.messages import ToolMessage
|
|
|
|
|
+ original = batch_workspace.pdf.read_bytes()
|
|
|
|
|
+ with pypdfium2.PdfDocument(original) as source, pypdfium2.PdfDocument.new() as combined:
|
|
|
|
|
+ for _ in range(26):
|
|
|
|
|
+ combined.import_pages(source)
|
|
|
|
|
+ combined.save(batch_workspace.pdf)
|
|
|
|
|
+ async with mock_report_client(batch_workspace.pdf) as http:
|
|
|
|
|
+ with bound(batch_workspace, client=ReportClient(http)) as (scope, identifier):
|
|
|
|
|
+ await report_archive.fetch(identifier, scope.user, scope.session)
|
|
|
|
|
+ facts = {'source_sha256': digest(batch_workspace.pdf), 'page_count': 26,
|
|
|
|
|
+ 'read_pages': list(range(1, 27)), 'page_sources': {str(page): f'/workspace/page_{page}.txt' for page in range(1, 27)},
|
|
|
|
|
+ 'reports': [{'report_id': 'synthetic'}]}
|
|
|
|
|
+ script = workspace_file(scope.user, scope.session, '/workspace/incomplete.py')
|
|
|
|
|
+ script.write_text('from pathlib import Path\nfor page in range(1,27): Path(f"page_{page}.txt").write_text(f"actual page {page}")\nPath("incomplete.json").write_text(' + repr(json.dumps(facts)) + ')\n')
|
|
|
|
|
+ receipt = await run_python(scope.user, scope.session, '/workspace/incomplete.py', output_paths=['/workspace/incomplete.json'])
|
|
|
|
|
+ assert receipt['exit_code'] == 0
|
|
|
|
|
+ exploration.record_python(identifier, scope.user, scope.session, receipt)
|
|
|
|
|
+ for page in range(1, 11):
|
|
|
|
|
+ record_page_read(SimpleNamespace(tool_call={'name': 'read_file', 'args': {'file_path': f'/workspace/page_{page}.txt'}}),
|
|
|
|
|
+ ToolMessage(content=f'actual page {page}', tool_call_id=str(page)))
|
|
|
|
|
+ with pytest.raises(ValueError, match='第11页'):
|
|
|
|
|
+ report_archive.save(identifier, scope.user, scope.session, '/workspace/incomplete.json', [receipt['run_id']])
|
|
|
|
|
+ assert not list(scope.cache_root.glob('*.json'))
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_plan_requires_real_points_and_complete_pagination(batch_workspace):
|
|
|
|
|
+ with bound(batch_workspace) as (scope, identifier):
|
|
|
|
|
+ workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
|
|
|
|
|
+ plan = MinePlan(shafts=[scope.shaft.model_copy(update={'shaft_id': 'unresolved-main',
|
|
|
|
|
+ 'shaft_name': '主备系统(待确认)', 'identity_status': 'candidate'})])
|
|
|
|
|
+ with pytest.raises(ValueError, match='两个实时列表'):
|
|
|
|
|
+ checked_plan(workflow, scope.user, scope.session, plan)
|
|
|
|
|
+ context = workflow['context']
|
|
|
|
|
+ for system, collection in [('safety', 'sensor_queries'), ('equipment', 'equipment_queries')]:
|
|
|
|
|
+ virtual = '/workspace/' + system + '.json'
|
|
|
|
|
+ local = workspace_file(scope.user, scope.session, virtual)
|
|
|
|
|
+ exploration._write(local, {'payload': {'rows': []}})
|
|
|
|
|
+ context[collection] = [{'data_path': virtual, 'page_num': 1, 'next_page': None}]
|
|
|
|
|
+ with pytest.raises(ValueError, match='未从本矿'):
|
|
|
|
|
+ checked_plan(workflow, scope.user, scope.session, plan)
|
|
|
|
|
+ context['sensor_queries'][0]['next_page'] = 2
|
|
|
|
|
+ with pytest.raises(ValueError, match='未获取分页'):
|
|
|
|
|
+ checked_plan(workflow, scope.user, scope.session, plan)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_runner_selected_mines_local_files_no_publish_even_global_auto_enabled(batch_workspace):
|
|
|
|
|
+ calls = []
|
|
|
|
|
+ mine = batch_workspace.mine
|
|
|
|
|
+ snapshot = {'snapshot_id': 'a' * 32, 'mines': [mine,
|
|
|
|
|
+ {'dept_id': '1338', 'mine_id': '00001338', 'mine_name': '不应执行的矿'}]}
|
|
|
|
|
+ plan = MinePlan(shafts=[ShaftPlan(shaft_id=value, shaft_name=value, identity_reason='测试数据明确井号')
|
|
|
|
|
+ for value in ('north', 'south')])
|
|
|
|
|
+ async def executor(**kwargs):
|
|
|
|
|
+ data = kwargs['graph_config']['configurable']['business_task_input']
|
|
|
|
|
+ phase = data['measurements']['batch']['phase']
|
|
|
|
|
+ calls.append((data['dept_id'], kwargs['session_id'], phase))
|
|
|
|
|
+ workspace = session_root(kwargs['user_name'], kwargs['session_id']) / 'workspace'
|
|
|
|
|
+ workspace.mkdir(exist_ok=True)
|
|
|
|
|
+ outside = batch_workspace.root / 'not-an-evidence.txt'
|
|
|
|
|
+ outside.write_text('do not export')
|
|
|
|
|
+ link = workspace / 'external-link.txt'
|
|
|
|
|
+ if not link.is_symlink():
|
|
|
|
|
+ link.symlink_to(outside)
|
|
|
|
|
+ if phase == 'discovery':
|
|
|
|
|
+ result = {'assessment': {'plan': plan.model_dump(mode='json')}}
|
|
|
|
|
+ else:
|
|
|
|
|
+ result = {'result': '# 本地测试报告', 'analysis_id': 'analysis-test', 'workflow_id': 'workflow-test',
|
|
|
|
|
+ 'platform_payload': {'judgment': {'resultCode': 'UNDETERMINED'}},
|
|
|
|
|
+ 'assessment': {'history_coverage': {'coverage_complete': True}, 'report_archives': []}}
|
|
|
|
|
+ return SimpleNamespace(status='success', run_id='run-test', error='', output='', usage={}, structured_output=result)
|
|
|
|
|
+ output = batch_workspace.root / 'output'
|
|
|
|
|
+ runner = BatchRunner(output, batch_workspace.pdf, executor=executor)
|
|
|
|
|
+ request = BatchRequest(dept_ids=['1219'], cutoff='2026-10-20T15:00:00+08:00')
|
|
|
|
|
+ result = await runner.run(request, snapshot=snapshot)
|
|
|
|
|
+ assert result['status'] == 'completed' and result['publish'] is False
|
|
|
|
|
+ assert len(calls) == 3 and {row[0] for row in calls} == {'1219'}
|
|
|
|
|
+ assert len({row[1] for row in calls}) == 1, '同矿多井复用矿级Agent会话'
|
|
|
|
|
+ assert (output / '1219/north/报告.md').is_file() and (output / '1219/south/结果.json').is_file()
|
|
|
|
|
+ assert not (output / '1338').exists()
|
|
|
|
|
+ assert not (output / '1219/workspace/external-link.txt').exists()
|
|
|
|
|
+ assert not business_store.list_jobs('main_fan', 'batch-test', 100)
|
|
|
|
|
+ await runner.run(request)
|
|
|
|
|
+ assert len(calls) == 3, '恢复已完成批次不能重新调用Agent'
|
|
|
|
|
+ await runner.run(request.model_copy(update={'reassess': True}))
|
|
|
|
|
+ assert len(calls) == 5, '重新分析只执行原矿的两井,不重复发现和遍历其他矿'
|
|
|
|
|
+ assert (output / '1219/north/revisions/analysis-test/报告.md').is_file()
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_plan_rejects_vibration_as_fan_state_and_fan_number_as_shaft(batch_workspace):
|
|
|
|
|
+ with bound(batch_workspace) as (scope, identifier):
|
|
|
|
|
+ workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
|
|
|
|
|
+ for collection in ('sensor_queries', 'equipment_queries'):
|
|
|
|
|
+ virtual = '/workspace/' + collection + '.json'
|
|
|
|
|
+ exploration._write(workspace_file(scope.user, scope.session, virtual), {'payload': {'rows': [
|
|
|
|
|
+ {'nodeId': 'vibration', 'nodeTypeName': '水平振动', 'sensorUnit': 'mm/s'}]}})
|
|
|
|
|
+ workflow['context'][collection] = [{'data_path': virtual, 'page_num': 1, 'next_page': None}]
|
|
|
|
|
+ shaft = ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统(待确认)', identity_reason='没有井号',
|
|
|
|
|
+ points=[HistoryPoint(system='equipment', node_id='vibration', role='fan_state')])
|
|
|
|
|
+ with pytest.raises(ValueError, match='实际类型'):
|
|
|
|
|
+ checked_plan(workflow, scope.user, scope.session, MinePlan(shafts=[shaft]))
|
|
|
|
|
+ shaft.points = []
|
|
|
|
|
+ shaft.shaft_id, shaft.shaft_name = 'shaft-1', '1号通风井'
|
|
|
|
|
+ with pytest.raises(ValueError, match='风机号不能'):
|
|
|
|
|
+ checked_plan(workflow, scope.user, scope.session, MinePlan(shafts=[shaft]))
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.parametrize('system,role,row', [
|
|
|
|
|
+ ('equipment', 'fan_state', {'nodeId': 'wrong-source', 'nodeTypeName': '风机运行状态'}),
|
|
|
|
|
+ ('safety', 'flow', {'nodeId': 'wrong-source', 'sensorName': '风量'}),
|
|
|
|
|
+])
|
|
|
|
|
+def test_plan_rejects_valid_measurement_from_wrong_authoritative_system(batch_workspace, system, role, row):
|
|
|
|
|
+ with bound(batch_workspace) as (scope, identifier):
|
|
|
|
|
+ workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
|
|
|
|
|
+ for name, collection in [('safety', 'sensor_queries'), ('equipment', 'equipment_queries')]:
|
|
|
|
|
+ virtual = '/workspace/' + collection + '.json'
|
|
|
|
|
+ exploration._write(workspace_file(scope.user, scope.session, virtual),
|
|
|
|
|
+ {'payload': {'rows': [row] if name == system else []}})
|
|
|
|
|
+ workflow['context'][collection] = [{'data_path': virtual, 'page_num': 1, 'next_page': None}]
|
|
|
|
|
+ shaft = ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统(待确认)', identity_reason='没有井号',
|
|
|
|
|
+ points=[HistoryPoint(system=system, node_id='wrong-source', role=role)])
|
|
|
|
|
+ with pytest.raises(ValueError, match='开停以安全监控为准'):
|
|
|
|
|
+ checked_plan(workflow, scope.user, scope.session, MinePlan(shafts=[shaft]))
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_history_rejects_old_equipment_state_plan_before_gateway_access(batch_workspace, monkeypatch):
|
|
|
|
|
+ gateway = HistoryGateway()
|
|
|
|
|
+ monkeypatch.setattr(history, 'get_service', lambda _: gateway)
|
|
|
|
|
+ shaft = ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统(待确认)', identity_reason='旧计划',
|
|
|
|
|
+ points=[HistoryPoint(system='equipment', node_id='state1', role='fan_state')])
|
|
|
|
|
+ with bound(batch_workspace, shaft=shaft) as (scope, identifier):
|
|
|
|
|
+ with pytest.raises(ValueError, match='开停以安全监控为准'):
|
|
|
|
|
+ await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert gateway.calls == []
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_history_reuses_old_archive_but_excludes_equipment_states_from_analysis(batch_workspace, monkeypatch):
|
|
|
|
|
+ gateway = HistoryGateway()
|
|
|
|
|
+ monkeypatch.setattr(history, 'get_service', lambda _: gateway)
|
|
|
|
|
+ with bound(batch_workspace, days=2) as (scope, identifier):
|
|
|
|
|
+ scope.shaft.points.append(HistoryPoint(system='equipment', node_id='old-state', role='fan_state'))
|
|
|
|
|
+ # 先按旧版本行为生成真实格式的合成归档,再应用当前来源选择,检验历史兼容。
|
|
|
|
|
+ with monkeypatch.context() as legacy:
|
|
|
|
|
+ legacy.setattr(history, 'select_history_points', lambda points: list(points))
|
|
|
|
|
+ old = await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert old['completed_segments'] == 10 and len(gateway.calls) == 10
|
|
|
|
|
+ archive = workspace_file(scope.user, scope.session, old['evidence_archive_path'])
|
|
|
|
|
+ archived_bytes = archive.read_bytes()
|
|
|
|
|
+ current = await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert current['coverage_complete'] and current['completed_segments'] == 8
|
|
|
|
|
+ assert current['point_count'] == 4 and len(gateway.calls) == 10
|
|
|
|
|
+ assert archive.read_bytes() == archived_bytes, '来源调整不应重拉或删除旧原件'
|
|
|
|
|
+ workflow = main_fan_store.workflow(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert len(workflow['context']['history_index']) == 10
|
|
|
|
|
+ manifest = workflow['context']['history_manifest']
|
|
|
|
|
+ assert len(manifest['segments']) == 8
|
|
|
|
|
+ assert manifest['ignored_plan_points'][0]['node_id'] == 'old-state'
|
|
|
|
|
+ compact = json.loads(workspace_file(scope.user, scope.session, current['agent_data_path']).read_text())
|
|
|
|
|
+ assert all(p['node_id'] != 'old-state' for p in compact['points'])
|
|
|
|
|
+ exploration.validate_inputs(workflow, scope.user, scope.session)
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_actual_agent_graph_two_phases_pdf_history_sandbox_and_local_delivery(batch_workspace, monkeypatch, pricing_ready):
|
|
|
|
|
+ if not capability()['available']:
|
|
|
|
|
+ pytest.skip('需要实际沙箱')
|
|
|
|
|
+ from agents import agent_factory, agent_registry
|
|
|
|
|
+ from runtime.environment import RuntimeEnvironment
|
|
|
|
|
+ from sandbox_agent_helpers import tool_call
|
|
|
|
|
+ from test_main_fan_timeseries import ToolModel
|
|
|
|
|
+ monkeypatch.setenv('RESULT_PLATFORM_ENABLED', '0')
|
|
|
|
|
+ points = [HistoryPoint(system='safety', node_id='state1', role='fan_state'),
|
|
|
|
|
+ HistoryPoint(system='equipment', node_id='flow1', role='flow', unit='m3/min')]
|
|
|
|
|
+ plan = MinePlan(shafts=[ShaftPlan(shaft_id='unresolved-main', shaft_name='主备系统(待确认)',
|
|
|
|
|
+ identity_reason='实际列表未给井号', points=points)])
|
|
|
|
|
+ gateway = HistoryGateway()
|
|
|
|
|
+ original_proxy = gateway.proxy
|
|
|
|
|
+ async def proxy(agent, path, *, params):
|
|
|
|
|
+ if path.endswith('getNodeRealtimeDetails'):
|
|
|
|
|
+ return {'code': 200, 'total': 1, 'rows': [{'nodeId': 'state1', 'sensorTypeId': '1010', 'nodePlace': '主通1一级'}]}
|
|
|
|
|
+ if path.endswith('selectSbNodeList'):
|
|
|
|
|
+ return {'code': 200, 'total': 1, 'rows': [{'nodeId': 'flow1', 'nodeTypeName': '风量', 'sensorUnit': 'm3/min'}]}
|
|
|
|
|
+ return await original_proxy(agent, path, params=params)
|
|
|
|
|
+ gateway.proxy = proxy
|
|
|
|
|
+ monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1')
|
|
|
|
|
+ monkeypatch.setenv('SHANXI_GATEWAY_URL', 'https://gateway.invalid')
|
|
|
|
|
+ monkeypatch.setenv('SHANXI_GATEWAY_TOKEN', 'batch-test-not-a-real-token')
|
|
|
|
|
+ @asynccontextmanager
|
|
|
|
|
+ async def gateway_service(dependencies):
|
|
|
|
|
+ yield gateway
|
|
|
|
|
+ monkeypatch.setattr('extensions.shaanxi.extension.gateway_service', gateway_service)
|
|
|
|
|
+ monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
|
|
|
|
|
+ monkeypatch.setattr(history, 'get_service', lambda _: gateway)
|
|
|
|
|
+ def latest():
|
|
|
|
|
+ from db import db_conn
|
|
|
|
|
+ row = db_conn.query_one("SELECT id,user_name,session_id FROM main_fan_workflows WHERE status='running' ORDER BY created_at DESC LIMIT 1")
|
|
|
|
|
+ return main_fan_store.workflow(row['id'], row['user_name'], row['session_id'])
|
|
|
|
|
+ def messages():
|
|
|
|
|
+ yield tool_call('prepare_main_fan_task', {'target': {}})
|
|
|
|
|
+ identifier = latest()['id']
|
|
|
|
|
+ yield tool_call('query_fan_sensor_realtime', {'workflow_id': identifier, 'dept_id': '1219', 'sensor_type_list': ['1010']})
|
|
|
|
|
+ yield tool_call('query_fan_equipment_realtime', {'workflow_id': identifier, 'dept_id': '1219'})
|
|
|
|
|
+ yield tool_call('submit_main_fan_mine_plan', {'workflow_id': identifier, 'plan': plan.model_dump(mode='json')})
|
|
|
|
|
+ if current_scope.get().shaft is None:
|
|
|
|
|
+ snapshot = agent_registry.get_agent('main_fan').get_state({'configurable': {'thread_id': current_scope.get().session}})
|
|
|
|
|
+ raise AssertionError(str([(getattr(message, 'name', ''), message.content) for message in snapshot.values['messages'][-4:]]))
|
|
|
|
|
+ yield tool_call('prepare_main_fan_task', {'target': {}})
|
|
|
|
|
+ workflow = latest(); identifier = workflow['id']
|
|
|
|
|
+ yield tool_call('fetch_main_fan_reports', {'workflow_id': identifier})
|
|
|
|
|
+ yield tool_call('query_main_fan_history', {'workflow_id': identifier})
|
|
|
|
|
+ workflow = latest(); pdf = workflow['context']['report_archives'][0]
|
|
|
|
|
+ prefix = 'exploration/' + identifier
|
|
|
|
|
+ code = f'''from pathlib import Path
|
|
|
|
|
+import json, hashlib, pypdfium2 as pdfium
|
|
|
|
|
+source=Path({pdf['python_path']!r})
|
|
|
|
|
+with pdfium.PdfDocument(source) as document:
|
|
|
|
|
+ texts=[document[index].get_textpage().get_text_range() for index in range(len(document))]
|
|
|
|
|
+facts={{'source_sha256':hashlib.sha256(source.read_bytes()).hexdigest(),'page_count':len(texts),
|
|
|
|
|
+ 'read_pages':list(range(1,len(texts)+1)),'page_sources':{{'1':'/workspace/page_1.txt'}},'reports':[{{'raw_text':texts[0],'page':1}}],
|
|
|
|
|
+ 'analysis':{{}},'limitations':['模拟报告,不移植矿井身份']}}
|
|
|
|
|
+manifest=json.loads(Path({(prefix + '/history-manifest.json')!r}).read_text())
|
|
|
|
|
+compact=json.loads(Path({(prefix + '/history-agent.json')!r}).read_text())
|
|
|
|
|
+counts={{row['node_id']:row['quality']['raw_sample_count'] for row in compact['points']}}
|
|
|
|
|
+Path('facts.json').write_text(json.dumps(facts,ensure_ascii=False))
|
|
|
|
|
+Path('page_1.txt').write_text(texts[0])
|
|
|
|
|
+Path('report.md').write_text('# 本地判识\\n\\n完整查询并保留归属限制。')
|
|
|
|
|
+assessment={{'result_code':'UNDETERMINED','summary':'已查询原件与时序,通风井及整机归属待确认。',
|
|
|
|
|
+ 'details':{{'segments':manifest['expected_segments'],'total_records':sum(counts.values()),'switch_analysis':{{'status':'insufficient_data','method':'单级状态不能代替整机','events':[],'limitations':['测试仅有单级状态']}},'flow_analysis':{{'status':'unrecognized_structure','raw_sample_count':None,'limitations':['保留未知格式,不当空数据']}}}},
|
|
|
|
|
+ 'evidence':[{{'source':'/workspace/{prefix}/history-manifest.json'}}]}}
|
|
|
|
|
+Path('assessment.json').write_text(json.dumps(assessment,ensure_ascii=False))
|
|
|
|
|
+'''
|
|
|
|
|
+ yield tool_call('write_file', {'file_path': '/workspace/analyze.py', 'content': code})
|
|
|
|
|
+ yield tool_call('run_python', {'workflow_id': identifier, 'script_path': '/workspace/analyze.py',
|
|
|
|
|
+ 'output_paths': ['/workspace/facts.json', '/workspace/report.md', '/workspace/assessment.json', '/workspace/page_1.txt']})
|
|
|
|
|
+ receipt = latest()['context']['python_runs'][-1]
|
|
|
|
|
+ assert receipt['exit_code'] == 0
|
|
|
|
|
+ yield tool_call('read_file', {'file_path': '/workspace/page_1.txt'})
|
|
|
|
|
+ yield tool_call('archive_main_fan_report', {'workflow_id': identifier, 'facts_path': '/workspace/facts.json',
|
|
|
|
|
+ 'python_run_ids': [receipt['run_id']]})
|
|
|
|
|
+ # 即使宿主全局自动推送开关为真,本地批次的task_id也必须阻止入队。
|
|
|
|
|
+ monkeypatch.setenv('RESULT_PLATFORM_ENABLED', '1')
|
|
|
|
|
+ yield tool_call('submit_main_fan_report', {'workflow_id': identifier, 'report_path': '/workspace/report.md',
|
|
|
|
|
+ 'calculations_path': '/workspace/assessment.json', 'python_run_ids': [receipt['run_id']]})
|
|
|
|
|
+ model = ToolModel(messages=messages())
|
|
|
|
|
+ monkeypatch.setattr(agent_factory.model_factory, 'get_model', lambda: model)
|
|
|
|
|
+ agent_registry.invalidate()
|
|
|
|
|
+ output = batch_workspace.root / 'graph-output'
|
|
|
|
|
+ snapshot = {'snapshot_id': 'b' * 32, 'mines': [batch_workspace.mine]}
|
|
|
|
|
+ async with RuntimeEnvironment(prewarm=False, billing_retry=False, recover_interrupted=False):
|
|
|
|
|
+ result = await BatchRunner(output, batch_workspace.pdf).run(
|
|
|
|
|
+ {'dept_ids': ['1219'], 'cutoff': '2026-10-20 15:00:00', 'lookback_days': 2}, snapshot=snapshot)
|
|
|
|
|
+ assert result['status'] == 'completed', result
|
|
|
|
|
+ from db import db_conn
|
|
|
|
|
+ assert db_conn.query_one('SELECT COUNT(*) AS n FROM business_jobs')['n'] == 0
|
|
|
|
|
+ delivered = json.loads((output / '1219/unresolved-main/结果.json').read_text())
|
|
|
|
|
+ assert delivered['assessment']['agent_result']['details']['segments'] == 4
|
|
|
|
|
+ assert delivered['assessment']['history_coverage']['coverage_complete']
|
|
|
|
|
+ assert delivered['assessment']['report_archives'][0]['archived']
|
|
|
|
|
+ assert (output / 'index.html').is_file()
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_reassessment_reuses_same_batch_history_but_other_batch_refetches(batch_workspace, monkeypatch):
|
|
|
|
|
+ gateway = HistoryGateway()
|
|
|
|
|
+ monkeypatch.setattr(history, 'get_service', lambda _: gateway)
|
|
|
|
|
+ with bound(batch_workspace, days=2) as (scope, identifier):
|
|
|
|
|
+ await history.query(identifier, scope.user, scope.session)
|
|
|
|
|
+ assert len(gateway.calls) == 8
|
|
|
|
|
+ old = main_fan_store.workflow(identifier, scope.user, scope.session)
|
|
|
|
|
+ main_fan_store.update_workflow(identifier, status='completed')
|
|
|
|
|
+ data = old['context']['task_input']
|
|
|
|
|
+ second = exploration.prepare(data, scope.user, scope.session, job_id='local-review')
|
|
|
|
|
+ scope.active_workflow_id = second['workflow_id']
|
|
|
|
|
+ receipt = await history.query(second['workflow_id'], scope.user, scope.session)
|
|
|
|
|
+ assert receipt['coverage_complete'] and len(gateway.calls) == 8
|
|
|
|
|
+ scope.batch_id = 'different-batch'
|
|
|
|
|
+ data['measurements']['batch']['batch_id'] = scope.batch_id
|
|
|
|
|
+ third = exploration.prepare(data, scope.user, scope.session, job_id='other-month')
|
|
|
|
|
+ scope.active_workflow_id = third['workflow_id']
|
|
|
|
|
+ await history.query(third['workflow_id'], scope.user, scope.session)
|
|
|
|
|
+ assert len(gateway.calls) == 16, '不同月度批次仍重新取得当期原件'
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_equipment_sbnodecurve_zero_samples_are_data_not_empty():
|
|
|
|
|
+ from extensions.main_fan.equipment import _curve_data
|
|
|
|
|
+ body = {'code': 200, 'data': [{'nodeId': 'flow1', 'nodePlace': '1号风机风量', 'sbNodeCurve': [
|
|
|
|
|
+ {'dataTime': '2026-10-19 16:23:20', 'realData': '0'},
|
|
|
|
|
+ {'dataTime': '2026-10-19 17:13:21', 'realData': '0'}]}]}
|
|
|
|
|
+ data, summary = _curve_data(body, 'flow1')
|
|
|
|
|
+ assert summary['sample_count'] == 2 and summary['normalization'] == 'sbNodeCurve'
|
|
|
|
|
+ assert data['payload'] is body and data['series'][0]['samples'][0]['value'] == '0'
|
|
|
|
|
+ assert data['series'][0]['samples'][0]['time'] == '2026-10-19T16:23:20+08:00'
|
|
|
|
|
+ unknown, metadata = _curve_data({'code': 200, 'data': [{'otherCurve': [1, 2]}]})
|
|
|
|
|
+ assert unknown['series'] is None and metadata['sample_count'] is None
|
|
|
|
|
+ with pytest.raises(ValueError, match='测点与本次请求'):
|
|
|
|
|
+ _curve_data(body, 'another-point')
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+def test_independent_verification_excludes_conflicting_states_and_keeps_zero_flow(tmp_path):
|
|
|
|
|
+ from scripts.verify_main_fan_batch import point_statistics
|
|
|
|
|
+ path = tmp_path / 'workspace' / 'samples.json'
|
|
|
|
|
+ path.parent.mkdir()
|
|
|
|
|
+ segment = {'status': 'completed', 'node_id': 'flow', 'data_path': '/workspace/samples.json',
|
|
|
|
|
+ 'start_time': '2026-10-01T00:00:00+08:00', 'end_time': '2026-10-02T00:00:00+08:00'}
|
|
|
|
|
+ path.write_text(json.dumps({'samples': [
|
|
|
|
|
+ {'time': '2026-10-01T01:00:00+08:00', 'value_label': '开', 'status_label': '正常'},
|
|
|
|
|
+ {'time': '2026-10-01T02:00:00+08:00', 'value_label': '开', 'status_label': '正常'},
|
|
|
|
|
+ {'time': '2026-10-01T02:00:00+08:00', 'value_label': '停', 'status_label': '正常'},
|
|
|
|
|
+ {'time': '2026-10-01T03:00:00+08:00', 'value_label': '停', 'status_label': '故障'},
|
|
|
|
|
+ {'time': '2026-09-30T03:00:00+08:00', 'value_label': '停', 'status_label': '正常'},
|
|
|
|
|
+ ]}))
|
|
|
|
|
+ safety = point_statistics([segment], tmp_path, 'safety')
|
|
|
|
|
+ assert safety['raw_sample_count'] == 5 and safety['unique_usable_sample_count'] == 1
|
|
|
|
|
+ assert safety['conflicting_timestamp_count'] == 1 and safety['invalid_status_count'] == 1
|
|
|
|
|
+ assert safety['out_of_window_count'] == 1 and safety['transitions'] == []
|
|
|
|
|
+ path.write_text(json.dumps({'payload': {'data': [{'nodeId': 'flow', 'sbNodeCurve': [
|
|
|
|
|
+ {'dataTime': '2026-10-01 01:00:00', 'realData': '0'},
|
|
|
|
|
+ {'dataTime': '2026-10-01 01:00:00', 'realData': '0'},
|
|
|
|
|
+ {'dataTime': '2026-10-01 02:00:00', 'realData': 'NaN'},
|
|
|
|
|
+ ]}]}}))
|
|
|
|
|
+ flow = point_statistics([segment], tmp_path, 'equipment')
|
|
|
|
|
+ assert flow['raw_sample_count'] == 3 and flow['zero_count'] == 1
|
|
|
|
|
+ assert flow['repeated_timestamp_count'] == 1 and flow['unrecognized_value_count'] == 1
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@pytest.mark.asyncio
|
|
|
|
|
+async def test_monthly_local_queue_idempotence_and_shutdown_before_start(batch_workspace, monkeypatch):
|
|
|
|
|
+ from extensions.main_fan import batch_service
|
|
|
|
|
+ monkeypatch.setenv('MAIN_FAN_MOCK_REPORT_PDF', str(batch_workspace.pdf))
|
|
|
|
|
+ monkeypatch.setenv('MAIN_FAN_BATCH_OUTPUT_DIR', str(batch_workspace.root / 'service-output'))
|
|
|
|
|
+ monkeypatch.setenv('MAIN_FAN_BATCH_DEPT_IDS', '1219')
|
|
|
|
|
+ calls = []
|
|
|
|
|
+ class Runner:
|
|
|
|
|
+ def __init__(self, folder, pdf, **kwargs):
|
|
|
|
|
+ self.folder = folder
|
|
|
|
|
+ async def run(self, request):
|
|
|
|
|
+ calls.append(request.dept_ids)
|
|
|
|
|
+ exploration._write(self.folder / 'batch.json', {'status': 'completed'})
|
|
|
|
|
+ return {'status': 'completed'}
|
|
|
|
|
+ monkeypatch.setattr(batch_service, 'BatchRunner', Runner)
|
|
|
|
|
+ controller = batch_service.LocalBatchController()
|
|
|
|
|
+ monkeypatch.setattr('extensions.get_service', lambda _: controller)
|
|
|
|
|
+ first = await batch_service.monthly_callback()
|
|
|
|
|
+ second = await batch_service.monthly_callback()
|
|
|
|
|
+ assert first['batch_id'] == second['batch_id'] and len(controller.tasks) == 1
|
|
|
|
|
+ await asyncio.gather(*list(controller.tasks.values()))
|
|
|
|
|
+ assert (await batch_service.monthly_callback())['status'] == 'completed'
|
|
|
|
|
+ assert calls == [['1219']]
|
|
|
|
|
+ queued = controller.submit({'dept_ids': ['1219']})
|
|
|
|
|
+ await controller.close()
|
|
|
|
|
+ assert controller.get(queued['batch_id'])['status'] == 'interrupted'
|
|
|
|
|
+ assert not controller.tasks
|