142 lines
9.5 KiB
Python
142 lines
9.5 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'coordinator'))
|
|
from robot_bt_coordinator.service import Coordinator
|
|
from robot_bt_coordinator.backends import ManualBackend, demo_plan
|
|
from robot_bt_coordinator.errors import ApiError
|
|
from robot_bt_coordinator.plan import validate_plan
|
|
|
|
REQ = dict(client_request_id='r1', robot_id='robot_01', instruction='把水放入箱A', known_info=dict(target_name='water', quantity=1, source_location='shelf_A', destination='tote_A'))
|
|
|
|
class CoordinatorTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db = str(Path(self.tmp.name)/'state.db')
|
|
self.backend = ManualBackend()
|
|
self.c = Coordinator(self.db, self.backend, {'robot_01'})
|
|
def tearDown(self):
|
|
self.c.close(); self.tmp.cleanup()
|
|
def plan(self, tid):
|
|
self.c.tick()
|
|
t=self.c.get(tid)
|
|
self.backend.emit(dict(type='plan',task_id=tid,task_revision=t['task_revision'],planning_generation=t['planning_generation'],status='PLAN_READY',plan=demo_plan(REQ['known_info'])))
|
|
self.c.tick()
|
|
def result(self,tid,**changes):
|
|
t=self.c.get(tid)
|
|
e=dict(type='execution_result',task_id=tid,run_id=t['run_id'],status='SUCCEEDED',stop_confirmed=True,completed_quantity=1,evidence=dict(evidence_id='ev1',target_ref='water',destination_ref='tote_A',passed=True,empty_hand=True,in_destination=True,valid=True,safe_to_release=True))
|
|
e.update(changes); self.backend.emit(e);self.c.tick()
|
|
def test_same_request_is_idempotent_and_conflict_rejected(self):
|
|
a=self.c.submit(REQ); b=self.c.submit(REQ)
|
|
self.assertEqual(a['task_id'],b['task_id']);self.assertTrue(b['deduplicated'])
|
|
with self.assertRaises(ApiError) as e:self.c.submit(dict(REQ,instruction='different'))
|
|
self.assertEqual(e.exception.status,409)
|
|
def test_fifo_and_queue_cancel(self):
|
|
a=self.c.submit(REQ)['task_id'];b=self.c.submit(dict(REQ,client_request_id='r2'))['task_id']
|
|
self.plan(a);self.assertEqual(self.c.get(b)['status'],'QUEUED')
|
|
self.c.control(b,'cancel');self.assertEqual(self.c.get(b)['status'],'CANCELED')
|
|
def test_cancel_during_planning_ignores_late_plan(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.c.tick(); t=self.c.get(tid)
|
|
self.c.control(tid,'cancel')
|
|
self.backend.emit(dict(type='plan',task_id=tid,task_revision=t['task_revision'],planning_generation=t['planning_generation'],status='PLAN_READY',plan=demo_plan(REQ['known_info'])))
|
|
self.c.tick();self.assertEqual(self.c.get(tid)['status'],'CANCELED')
|
|
self.assertEqual(len(self.backend.executions),0)
|
|
def test_cancel_ack_does_not_release_execution(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid);self.c.control(tid,'cancel')
|
|
self.backend.emit(dict(type='cancel_ack',task_id=tid,run_id=self.c.get(tid)['run_id']))
|
|
self.c.tick();self.assertEqual(self.c.get(tid)['status'],'CANCELING')
|
|
def test_unknown_stop_quarantines_and_blocks_next_task(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid)
|
|
other=self.c.submit(dict(REQ,client_request_id='r2'))['task_id']
|
|
self.result(tid,stop_confirmed=False)
|
|
self.assertEqual(self.c.get(tid)['status'],'INTERVENTION_REQUIRED')
|
|
self.c.tick();self.assertEqual(self.c.get(other)['status'],'QUEUED')
|
|
def test_wrong_container_not_counted(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid)
|
|
self.result(tid,evidence=dict(evidence_id='ev',target_ref='water',destination_ref='tote_B',passed=True,empty_hand=True,in_destination=True,valid=True,safe_to_release=True))
|
|
self.assertEqual(self.c.get(tid)['completed_quantity'],0)
|
|
self.assertEqual(self.c.get(tid)['status'],'INTERVENTION_REQUIRED')
|
|
def test_delivery_transaction_is_idempotent_after_duplicate_and_restart(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid);self.result(tid);self.result(tid)
|
|
self.assertEqual(self.c.get(tid)['completed_quantity'],1)
|
|
self.c.close();self.c=Coordinator(self.db,ManualBackend(),{'robot_01'})
|
|
self.assertEqual(self.c.get(tid)['completed_quantity'],1)
|
|
self.assertEqual(self.c.get(tid)['status'],'SUCCEEDED')
|
|
def test_restart_quarantines_unfinished_motion(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid);self.c.close()
|
|
self.c=Coordinator(self.db,ManualBackend(),{'robot_01'})
|
|
self.assertEqual(self.c.get(tid)['status'],'INTERVENTION_REQUIRED')
|
|
self.c.tick();self.assertEqual(len(self.c.backend.executions),0)
|
|
def test_stale_clarification_rejected(self):
|
|
tid=self.c.submit(dict(REQ,known_info={}))['task_id'];self.c.tick();t=self.c.get(tid)
|
|
self.backend.emit(dict(type='plan',task_id=tid,task_revision=t['task_revision'],planning_generation=t['planning_generation'],status='NEEDS_CLARIFICATION',questions=['destination']))
|
|
self.c.tick();t=self.c.get(tid)
|
|
with self.assertRaises(ApiError): self.c.clarify(tid,dict(question_id='old',task_revision=t['task_revision'],known_info={'destination':'tote_A'}))
|
|
def test_input_unknown_robot_and_quantity_rejected(self):
|
|
for req in [dict(REQ,robot_id='other'),dict(REQ,instruction=' '*5),dict(REQ,known_info=dict(REQ['known_info'],quantity=2))]:
|
|
with self.assertRaises(ApiError):self.c.submit(req)
|
|
def test_cancel_with_unknown_hand_does_not_admit_next_task(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid);self.c.control(tid,'cancel')
|
|
self.result(tid,status='CANCELED',completed_quantity=0,evidence={})
|
|
self.assertEqual(self.c.get(tid)['status'],'INTERVENTION_REQUIRED')
|
|
def test_single_process_owns_scheduler_database(self):
|
|
with self.assertRaises(RuntimeError):Coordinator(self.db,ManualBackend(),{'robot_01'})
|
|
def test_planning_budget_expires_without_reply(self):
|
|
elapsed=[0.]
|
|
self.c.steady=lambda:elapsed[0]
|
|
tid=self.c.submit(REQ)['task_id'];self.c.tick()
|
|
elapsed[0]+=100
|
|
self.c.tick()
|
|
self.assertEqual(self.c.get(tid)['planning_attempts'],2)
|
|
elapsed[0]+=100
|
|
self.c.tick();self.assertEqual(self.c.get(tid)['status'],'FAILED')
|
|
def test_planner_cannot_change_user_known_slots(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.c.tick();t=self.c.get(tid)
|
|
changed=dict(REQ['known_info'],target_name='other')
|
|
self.backend.emit(dict(type='plan',task_id=tid,task_revision=t['task_revision'],planning_generation=t['planning_generation'],status='PLAN_READY',plan=demo_plan(changed)))
|
|
self.c.tick();self.assertEqual(len(self.backend.executions),0)
|
|
def test_ask_user_plan_enters_clarification_without_execution(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.c.tick();t=self.c.get(tid)
|
|
p=demo_plan(REQ['known_info']);p['subtasks']=[dict(id='Q1',skill='ASK_USER',arguments={'question':'请确认目标箱编号'},depends_on=[])]
|
|
self.backend.emit(dict(type='plan',task_id=tid,task_revision=t['task_revision'],planning_generation=t['planning_generation'],status='PLAN_READY',plan=p))
|
|
self.c.tick();self.assertEqual(self.c.get(tid)['status'],'NEEDS_CLARIFICATION')
|
|
self.assertEqual(len(self.backend.executions),0)
|
|
def test_late_plan_after_budget_cannot_dispatch(self):
|
|
elapsed=[0.];self.c.steady=lambda:elapsed[0]
|
|
tid=self.c.submit(REQ)['task_id'];self.c.tick();t=self.c.get(tid)
|
|
elapsed[0]=31.
|
|
self.backend.emit(dict(type='plan',task_id=tid,task_revision=t['task_revision'],planning_generation=t['planning_generation'],status='PLAN_READY',plan=demo_plan(REQ['known_info'])))
|
|
self.c.tick();self.assertEqual(len(self.backend.executions),0)
|
|
def test_delivered_item_retained_if_cleanup_requires_intervention(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid)
|
|
ev=dict(evidence_id='ev1',target_ref='water',destination_ref='tote_A',passed=True,empty_hand=True,in_destination=True,valid=True,safe_to_release=False)
|
|
self.result(tid,status='INTERVENTION_REQUIRED',evidence=ev)
|
|
self.assertEqual(self.c.get(tid)['completed_quantity'],1)
|
|
self.assertEqual(self.c.get(tid)['status'],'INTERVENTION_REQUIRED')
|
|
other=self.c.submit(dict(REQ,client_request_id='cleanup-next'))['task_id'];self.c.tick()
|
|
self.assertEqual(self.c.get(other)['status'],'QUEUED')
|
|
def test_event_cursor_monotonic(self):
|
|
tid=self.c.submit(REQ)['task_id'];self.plan(tid);es=self.c.events(tid)
|
|
self.assertTrue(len(es)>2); self.assertEqual(self.c.events(tid,es[-1]['event_id']),[])
|
|
|
|
class PlanTest(unittest.TestCase):
|
|
def test_valid_fixed_plan(self):
|
|
self.assertEqual(validate_plan(demo_plan(REQ['known_info']))['task_type'],'pick_transport_place')
|
|
def test_unknown_skill_cycle_missing_id_xml_injection_rejected(self):
|
|
base=demo_plan(REQ['known_info'])
|
|
plans=[]
|
|
a=json.loads(json.dumps(base));a['subtasks'][2]['skill']='SHELL';plans.append(a)
|
|
a=json.loads(json.dumps(base));a['subtasks'][0]['depends_on']=['S6'];plans.append(a)
|
|
a=json.loads(json.dumps(base));a['subtasks'][1]['depends_on']=['NO'];plans.append(a)
|
|
a=json.loads(json.dumps(base));a['xml']='<Script/>';plans.append(a)
|
|
for p in plans:
|
|
with self.subTest(plan=p),self.assertRaises(ApiError):validate_plan(p)
|
|
def test_nonfinite_and_bool_quantity_rejected(self):
|
|
for q in [True,float('nan'),2]:
|
|
p=demo_plan(REQ['known_info']);p['slots']['quantity']=q
|
|
with self.assertRaises(ApiError):validate_plan(p)
|
|
|
|
if __name__=='__main__':unittest.main()
|