import unittest,sys,tempfile,json,time,threading from pathlib import Path ROOT=Path(__file__).resolve().parents[1] sys.path[:0]=[str(ROOT/'coordinator'),str(ROOT/'robobrain')] class RoboBrainTests(unittest.TestCase): def service(self,raw): from robot_robobrain.service import BrainService from robot_robobrain.backends import FixtureBackend self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) return BrainService(FixtureBackend(raw),self.tmp.name) def goal(self): return dict(task_id='t',task_revision=1,planning_generation=1,instruction='two waters then one doll',known_info={'items':[{'target_name':'water','quantity':2,'source_location':'shelf_A'},{'target_name':'doll','quantity':1,'source_location':'shelf_A'}],'destination':'tote_A'},context={},constraints={'route':'OBJECT_TABLE','schema_version':2},timeout=2) def test_plan_raw_archive_and_semantic_mismatch(self): from test_plan_v2 import sample p=sample();s=self.service(json.dumps(p));r=s.plan(self.goal());self.assertEqual(r['status'],'PLAN_READY');self.assertTrue(Path(r['record_ref']).is_file()) g=self.goal();g['known_info']['items'][0]['quantity']=1 self.assertEqual(s.plan(g)['error_code'],'SEMANTIC_MISMATCH') def test_invalid_json_and_missing_information(self): self.assertEqual(self.service('```json {} ```').plan(self.goal())['error_code'],'PARSE_ERROR') r=self.service('{"missing_information":["which destination?"]}').plan(self.goal()) self.assertEqual(r['status'],'NEEDS_CLARIFICATION') def test_shelf_ambiguity_freshness_and_geometry_untrusted(self): from robot_robobrain.observations import Observation media=tempfile.TemporaryDirectory();self.addCleanup(media.cleanup);path=Path(media.name)/'image.jpg';path.write_bytes(b'fixture-image') o=Observation('obs',100,'camera',str(path),'observe_A',1,'shelf_A') s=self.service('{"status":"SUCCEEDED","shelf_id":"shelf_A","side_id":"FRONT","column_id":"1","tier_id":"2","confidence":0.95}') g=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) self.assertEqual(s.shelf(g,o,now_ns=110,max_age_ns=50)['status'],'SUCCEEDED') self.assertEqual(s.shelf(g,o,now_ns=200,max_age_ns=50)['error_code'],'OBSERVATION_INVALID') self.assertEqual(self.service('{"status":"AMBIGUOUS"}').shelf(g,o,now_ns=110,max_age_ns=50)['status'],'AMBIGUOUS') r=self.service('{"point":[1,2,3]}').localize(g,o,now_ns=110,max_age_ns=50) self.assertFalse(r['geometry_valid']);self.assertEqual(r['target_point']['point'],[1,2,3]) def test_progress_never_completes_and_rejects_stale(self): from robot_robobrain.progress import ProgressMonitor m=ProgressMonitor('run','pick',capture_after=1,max_age=2,stall_seconds=3) for n in range(1,8):r=m.update(dict(run_id='run',subtask_id='pick',sequence=n,stamp=float(n),progress=.8,hop={'raw':1}),now=float(n)) self.assertEqual(r['state'],'STALLED');self.assertFalse(r['completion_authority']) self.assertEqual(m.update(dict(run_id='old',subtask_id='pick',sequence=9,stamp=9.,progress=1.,hop=1),now=9.)['state'],'UNKNOWN') self.assertEqual(m.snapshot(20.)['state'],'UNKNOWN') class WorkerAndMonitorTests(unittest.TestCase): def test_worker_roundtrip_timeout_and_cancel(self): from robot_robobrain.backends import ProcessBackend,InferenceError command=[sys.executable,'-u','-c','import sys,json,time\nfor l in sys.stdin:\n q=json.loads(l);time.sleep(q.get("delay",0));print(json.dumps({"raw":"ok"}),flush=True)'] b=ProcessBackend(command,'test-worker');self.addCleanup(b.close) self.assertEqual(b.infer({},1),'ok');self.assertEqual(b.infer({},1),'ok') with self.assertRaises(InferenceError) as ctx:b.infer({'delay':2},.1) self.assertEqual(ctx.exception.code,'TIMEOUT');self.assertIsNone(b.process) cancel=threading.Event();timer=threading.Timer(.05,cancel.set);timer.start() with self.assertRaises(InferenceError) as ctx:b.infer({'delay':2},1,cancel) timer.join();self.assertEqual(ctx.exception.code,'CANCELED');self.assertIsNone(b.process) def test_worker_crash_and_unbounded_output(self): from robot_robobrain.backends import ProcessBackend,InferenceError for source in ['import sys;sys.exit(1)','import sys;sys.stdin.readline();print("x"*300000,flush=True)']: b=ProcessBackend([sys.executable,'-u','-c',source],'test');self.addCleanup(b.close) with self.assertRaises(InferenceError):b.infer({},1) def test_monitor_full_progress_does_not_complete_regression_and_nan(self): from robot_robobrain.progress import ProgressMonitor m=ProgressMonitor('r','s',1) for n,p in enumerate([1,1,1,.1,.1,.1,.1,.1],1):r=m.update(dict(run_id='r',subtask_id='s',sequence=n,stamp=n,progress=p,hop=1),n) self.assertEqual(r['state'],'REGRESSED');self.assertFalse(r['completion_authority']) for p in [float('nan'),True,-.1,1.01]:self.assertEqual(m.update(dict(run_id='r',subtask_id='s',sequence=9,stamp=9,progress=p),9)['state'],'UNKNOWN') def test_dense_window_service_and_future_window(self): from robot_robobrain.backends import FixtureBackend from robot_robobrain.dopamine import DenseFeedbackService with tempfile.TemporaryDirectory() as tmp: frame=Path(tmp)/'frame.jpg';frame.write_bytes(b'fixture-image') s=DenseFeedbackService(FixtureBackend('{"progress":1,"hop":3}'),tmp) g=dict(task_id='t',run_id='r',subtask_id='pick',task_description='pick water',capture_after=1,sequence=1,timeout=1) r=s.evaluate(g,[{'stamp':2.,'views':{'front':str(frame)}}],2.) self.assertEqual(r['state'],'RUNNING');self.assertFalse(r['completion_authority']) self.assertEqual(s.evaluate(g,[{'stamp':9.,'views':{'front':str(frame)}}],2.)['state'],'UNKNOWN') def test_observation_path_cannot_escape_media_root(self): from robot_robobrain.observations import Observation,ObservationCache with tempfile.TemporaryDirectory() as tmp: root=Path(tmp)/'media';root.mkdir();outside=Path(tmp)/'private';outside.write_text('x');(root/'link').symlink_to(outside) cache=ObservationCache(root) with self.assertRaises(ValueError):cache.put(Observation('o',1,'camera',str(root/'link'),'s',1,'shelf')) class IntentTests(unittest.TestCase): def test_explicit_quantity_and_order_are_checked_without_known_info(self): from robot_robobrain.intent import extract from robot_robobrain.service import BrainService from robot_robobrain.backends import FixtureBackend from robot_bt_coordinator.plan_v2 import make_plan site=json.loads((ROOT/'config/sim_site_object_table.json').read_text());instruction='把货架A的两瓶水和一个玩偶放到周转箱A' slots=extract(instruction,site);self.assertEqual([x['quantity'] for x in slots['items']],[2,1]) bad=json.loads(json.dumps(slots));bad['items'][0]['quantity']=1 with tempfile.TemporaryDirectory() as tmp: s=BrainService(FixtureBackend(json.dumps(make_plan(instruction,bad,'OBJECT_TABLE'))),tmp) g=dict(task_id='t',task_revision=1,planning_generation=1,instruction=instruction,known_info={},context=site,constraints={'route':'OBJECT_TABLE'},timeout=1) self.assertEqual(s.plan(g)['error_code'],'SEMANTIC_MISMATCH') g['instruction']='拿一些水' self.assertEqual(s.plan(g)['status'],'NEEDS_CLARIFICATION') def test_frame_window_is_bounded_and_reset_on_clock_reversal(self): from robot_robobrain.windows import FrameWindow w=FrameWindow(3) for i in range(10):w.add(i,'front',str(i)) self.assertEqual(len(w.since(0,9)),3);w.add(1,'front','new');self.assertEqual(len(w.since(0,1)),1)