159 lines
9.3 KiB
Python
159 lines
9.3 KiB
Python
import copy
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(ROOT/'coordinator'), str(ROOT/'robobrain'), str(ROOT/'ros2/robobrain_services')]
|
|
from robot_robobrain.backends import FixtureBackend
|
|
from robot_robobrain.service import BrainService
|
|
from robot_robobrain.observations import Observation
|
|
from robot_bt_coordinator.plan_v2 import make_plan
|
|
|
|
class ConsistencyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.site = json.loads((ROOT/'config/sim_site_object_table.json').read_text(encoding='utf-8'))
|
|
self.slots = {'items': [{'target_name':'water', 'quantity':2, 'source_location':'shelf_A'}, {'target_name':'doll', 'quantity':1, 'source_location':'shelf_A'}], 'destination':'tote_A'}
|
|
self.instruction = '把货架A的两瓶水和一个玩偶放到周转箱A'
|
|
|
|
def run_plan(self, slots, revision=1, confirmed=False, instruction=None):
|
|
text = instruction or self.instruction
|
|
service = BrainService(FixtureBackend(json.dumps(make_plan(text, slots, 'OBJECT_TABLE'))), self.tmp.name)
|
|
return service.plan(dict(task_id='t', task_revision=revision, planning_generation=1,
|
|
instruction=text, known_info=slots, context=self.site,
|
|
constraints={'route':'OBJECT_TABLE', 'clarification_confirmed':confirmed}, timeout=1))
|
|
|
|
def test_initial_explicit_quantity_order_item_destination_conflicts_fail(self):
|
|
variants=[]
|
|
p=copy.deepcopy(self.slots);p['items'][0]['quantity']=1;variants.append(p)
|
|
p=copy.deepcopy(self.slots);p['items'].reverse();variants.append(p)
|
|
p=copy.deepcopy(self.slots);p['items'][0]['target_name']='doll';variants.append(p)
|
|
p=copy.deepcopy(self.slots);p['destination']='tote_B';variants.append(p)
|
|
for slots in variants:
|
|
with self.subTest(slots=slots):
|
|
result=self.run_plan(slots)
|
|
self.assertEqual(result['status'], 'FAILED')
|
|
self.assertEqual(result['error_code'], 'SEMANTIC_MISMATCH')
|
|
self.assertTrue(Path(result['record_ref']).is_file())
|
|
|
|
def test_revision_alone_and_initial_confirmation_marker_do_not_override(self):
|
|
slots=copy.deepcopy(self.slots);slots['items'][0]['quantity']=1
|
|
for revision,confirmed in [(2,False),(1,True),(2,'true')]:
|
|
with self.subTest(revision=revision,confirmed=confirmed):
|
|
self.assertEqual(self.run_plan(slots,revision,confirmed)['status'],'FAILED')
|
|
|
|
def test_confirmed_clarification_can_override_original(self):
|
|
slots=copy.deepcopy(self.slots);slots['items'][0]['quantity']=1
|
|
self.assertEqual(self.run_plan(slots,2,True)['status'],'PLAN_READY')
|
|
|
|
def test_matching_and_unrecognized_language_are_not_false_proofs(self):
|
|
self.assertEqual(self.run_plan(self.slots)['status'],'PLAN_READY')
|
|
self.assertEqual(self.run_plan(self.slots,instruction='carry out the supplied item list')['status'],'PLAN_READY')
|
|
|
|
def test_shelf_non_success_retains_observation_identity(self):
|
|
path=Path(self.tmp.name)/'image.png';path.write_bytes(b'fixture')
|
|
obs=Observation('observation-7',100,'camera',str(path),'observe_A',1,'shelf_A')
|
|
goal=dict(task_id='t',subtask_id='s',target_ref='water',source_region_ref='shelf_A',observation_station_id='observe_A',station_registry_version=1,capture_after=90,timeout=1)
|
|
for status in ['NOT_FOUND','AMBIGUOUS']:
|
|
result=BrainService(FixtureBackend(json.dumps({'status':status})),self.tmp.name).shelf(goal,obs,110)
|
|
self.assertEqual(result.get('observation_id'),'observation-7')
|
|
self.assertEqual(result.get('observed_at'),100)
|
|
self.assertNotIn('column_id',result)
|
|
|
|
def test_unexpected_backend_failure_is_archived(self):
|
|
def broken(_): raise RuntimeError('model driver crashed')
|
|
service=BrainService(FixtureBackend(broken),self.tmp.name)
|
|
goal=dict(task_id='t',task_revision=1,planning_generation=1,instruction=self.instruction,known_info=self.slots,context=self.site,constraints={'route':'OBJECT_TABLE'},timeout=1)
|
|
try:
|
|
result=service.plan(goal)
|
|
except RuntimeError:
|
|
self.fail('unexpected backend exception escaped without diagnostic result')
|
|
self.assertEqual(result['status'],'FAILED')
|
|
self.assertEqual(result['error_code'],'INFERENCE_FAILED')
|
|
self.assertIn('model driver crashed',json.loads(Path(result['record_ref']).read_text())['result']['message'])
|
|
|
|
class AdapterConfigTests(unittest.TestCase):
|
|
def test_missing_loader_configuration_has_actionable_error(self):
|
|
from robot_robobrain.model_adapter import create
|
|
for config in [{}, {'loader':'missing-separator','model':{},'task_mapping':{}}]:
|
|
try:
|
|
create(config)
|
|
except Exception as ex:
|
|
self.assertIsInstance(ex,ValueError)
|
|
self.assertIn('loader',str(ex))
|
|
else:self.fail('invalid loader configuration accepted')
|
|
|
|
def test_single_image_adapter_rejects_dense_frames_before_loader_import(self):
|
|
from robot_robobrain.model_adapter import create
|
|
config={'loader':'deployment_models:create_robobrain','model':{'checkpoint':'/not/present'},'task_mapping':{'dense_feedback':'general'}}
|
|
try:create(config)
|
|
except Exception as ex:
|
|
self.assertIsInstance(ex,ValueError)
|
|
self.assertIn('dense',str(ex))
|
|
else:self.fail('single-image adapter accepted dense task mapping')
|
|
|
|
class ServiceBoundaryTests(unittest.TestCase):
|
|
def test_malformed_ros_request_is_archived_without_parsing_it(self):
|
|
from robobrain_services import nodes
|
|
self.assertTrue(callable(getattr(nodes,'guarded_work',None)), 'ROS worker needs a failure-archiving boundary')
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
service=BrainService(FixtureBackend('{}'),tmp)
|
|
request=types.SimpleNamespace(task_id='t',instruction='original text',known_info_json='{malformed',timeout=types.SimpleNamespace(sec=1,nanosec=0))
|
|
def bad_request(): raise ValueError('invalid JSON input')
|
|
result=nodes.guarded_work(service,'plan',request,bad_request)
|
|
self.assertEqual(result['error_code'],'SERVICE_ERROR')
|
|
record=json.loads(Path(result['record_ref']).read_text())
|
|
self.assertEqual(record['input']['known_info_json'],'{malformed')
|
|
self.assertEqual(record['input']['timeout']['sec'],1)
|
|
self.assertIn('invalid JSON input',record['result']['message'])
|
|
|
|
def test_archive_failure_still_returns_terminal_failure(self):
|
|
from robobrain_services import nodes
|
|
self.assertTrue(callable(getattr(nodes,'guarded_work',None)), 'ROS worker needs a failure-archiving boundary')
|
|
class UnavailableArchive:
|
|
def _record(self,*args):raise OSError('disk full')
|
|
def failure():raise RuntimeError('service crashed')
|
|
result=nodes.guarded_work(UnavailableArchive(),'shelf',{'task_id':'t'},failure)
|
|
self.assertEqual(result['status'],'FAILED')
|
|
self.assertEqual(result.get('record_ref'),'')
|
|
self.assertIn('archive',result['message'])
|
|
self.assertIn('disk full',result['message'])
|
|
|
|
def test_success_result_is_not_rewritten(self):
|
|
from robobrain_services import nodes
|
|
self.assertTrue(callable(getattr(nodes,'guarded_work',None)), 'ROS worker needs a failure-archiving boundary')
|
|
expected={'status':'PLAN_READY','record_ref':'existing'}
|
|
self.assertIs(nodes.guarded_work(None,'plan',{},lambda:expected),expected)
|
|
|
|
class LocalizeDiagnosticTests(unittest.TestCase):
|
|
def test_non_success_keeps_observation_without_inventing_point(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path=Path(tmp)/'image.png';path.write_bytes(b'fixture')
|
|
obs=Observation('obs3d',100,'camera',str(path),'observe_A',1,'shelf_A','cal-1',7)
|
|
goal=dict(task_id='t',subtask_id='s',target_ref='water',capture_after=90,timeout=1)
|
|
for status in ('NOT_FOUND','AMBIGUOUS'):
|
|
result=BrainService(FixtureBackend(json.dumps({'status':status})),tmp).localize(goal,obs,110)
|
|
self.assertEqual(result.get('observation_id'),'obs3d')
|
|
self.assertEqual(result.get('observed_at'),100)
|
|
self.assertEqual(result.get('calibration_id'),'cal-1')
|
|
self.assertEqual(result.get('geometry_epoch'),7)
|
|
self.assertFalse(result['geometry_valid'])
|
|
self.assertNotIn('target_point',result)
|
|
|
|
class DemoClarificationTests(unittest.TestCase):
|
|
def test_demo_propagates_trusted_clarification_marker(self):
|
|
from robot_robobrain.demo_backend import BrainDemoBackend
|
|
backend=BrainDemoBackend.__new__(BrainDemoBackend)
|
|
backend.plans=[];backend.site={'execution_route':'OBJECT_TABLE'}
|
|
goals=[]
|
|
backend.brain=types.SimpleNamespace(plan=lambda goal:(goals.append(goal) or {'status':'FAILED','record_ref':'r'}))
|
|
backend.emit=lambda event:None
|
|
task={'task_id':'t','task_revision':2,'planning_generation':3,'clarification_confirmed':True,'request':{'instruction':'original','known_info':{}}}
|
|
backend.start_planning(task)
|
|
self.assertIs(goals[0]['constraints'].get('clarification_confirmed'),True)
|