import json from pathlib import Path import random import tempfile import threading import time import subprocess import sys import unittest from tools.region_flow_shadow.core import FlowShadow, EntryCatalog, lineage, infomap_partition, region_metrics def event(n,u='a',v='b',action='click',**kw): return dict(run_id='r',event_seq=n,event_kind='TRANSITION',source_state=u,target_state=v, transition_identity='exec'+str(n),model_action_identity=action,source_activity='Activity', execution_confirmed=True,is_back=False,**kw) def group(id='g',ids=('exec1',),**kw): return dict(group_id=id,run_id='r',function=id,category='I',source_activity='Activity', oracle=True,transition_ids=list(ids),**kw) def simple(nodes,weights):return [sorted(nodes)], 0 class FlowTests(unittest.TestCase): def test_01_confirmed_once_and_identity_conflict(self): f=FlowShadow('r');e=event(1);self.assertTrue(f.consume(e));self.assertFalse(f.consume(e)) self.assertFalse(f.consume(event(2,transition_identity_override='ignored') | {'transition_identity':'exec1'})) self.assertEqual(sum(f.weights.values()),1) with self.assertRaises(ValueError):f.consume(event(2,'a','c')) def test_02_topology_and_unconfirmed_excluded(self): f=FlowShadow('r');f.consume(event(1)|{'execution_confirmed':False});f.consume(event(2)|{'event_kind':'TOPOLOGY'}) self.assertFalse(f.nodes);self.assertEqual(len(f.events),0) def test_03_multi_action_aggregation(self): f=FlowShadow('r');f.consume(event(1));f.consume(event(2,action='longclick')) self.assertEqual(f.weights['a','b'],2);self.assertEqual(len(f.action_weights),2) def test_04_self_loop(self): f=FlowShadow('r');f.consume(event(1,'a','a'));self.assertEqual(f.weights['a','a'],1) def test_05_back_kept(self): f=FlowShadow('r');f.consume(event(1)|{'action_type':'MODEL_BACK','is_back':True});self.assertEqual(f.weights['a','b'],1) def test_06_restart_reset_boundaries(self): f=FlowShadow('r');f.consume(event(1));f.consume(dict(run_id='r',event_seq=2,event_kind='RESET',is_restart=True)) f.consume(event(3,'b','a'));self.assertEqual(len(f.events),2);self.assertEqual(f.events[-1]['_segment'],1) self.assertEqual(sum(f.weights.values()),2) def test_07_birth(self): current=[dict(id='s1c',members=['a'])];_,_,_=lineage([],current,1);self.assertEqual(current[0]['lineage_status'],'birth') def test_08_expansion(self): old=[dict(id='c',members=['a','b'],lineage_id='L1')];cur=[dict(id='d',members=['a','b','c'])] links,changes,_=lineage(old,cur,2);self.assertEqual(cur[0]['lineage_id'],'L1');self.assertEqual(links[0]['relation'],'expansion') def test_09_merge(self): old=[dict(id='a',members=['1','2'],lineage_id='L1'),dict(id='b',members=['3','4'],lineage_id='L2')];cur=[dict(id='c',members=['1','2','3','4'])] _,changes,_=lineage(old,cur,3);self.assertIn('merge',[x['kind'] for x in changes]);self.assertEqual(cur[0]['lineage_id'],'L00003') def test_10_split(self): old=[dict(id='a',members=['1','2','3','4'],lineage_id='L1')];cur=[dict(id='b',members=['1','2']),dict(id='c',members=['3','4'])] _,changes,_=lineage(old,cur,2);self.assertIn('split',[x['kind'] for x in changes]);self.assertNotEqual(cur[0]['lineage_id'],cur[1]['lineage_id']) def test_11_one_entry_multiple_targets(self): f=FlowShadow('r',EntryCatalog([group(ids=('exec1','exec2'))],True),simple) f.consume(event(1));f.consume(event(2,'a','c'));s=f.snapshot();self.assertEqual(len(s['entry_occurrences']),2) self.assertEqual({o['target_state'] for o in f.occurrences},{'b','c'}) def test_12_multi_entries_same_community(self): f=FlowShadow('r',EntryCatalog([group('one'),group('two',('exec2',))],True),simple) f.consume(event(1));f.consume(event(2,'b','a'));s=f.snapshot();self.assertEqual(len(s['communities'][0]['functions']),2) self.assertEqual(sum(r['metrics']['source_steps'] for r in s['communities']),2) def test_13_one_entry_multiple_communities(self): f=FlowShadow('r',EntryCatalog([group(ids=('exec1','exec2'))],True),lambda ns,ws:([[s] for s in sorted(ns)],0)) f.consume(event(1));f.consume(event(2,'a','c'));s=f.snapshot();self.assertEqual(sum(bool(r['entry_associations']) for r in s['communities']),2) def test_14_oracle_unknown_activity_and_category(self): f=FlowShadow('r',EntryCatalog([group()],False));f.consume(event(1));self.assertEqual(f.occurrences,[]) for g in [group()|{'source_activity':'Other'},group()|{'category':'III'},group()|{'status':'unknown'}]: f=FlowShadow('r',EntryCatalog([g],True));f.consume(event(1));self.assertFalse(f.occurrences) def test_15_snapshot_delay_frozen_occurrence(self): f=FlowShadow('r',EntryCatalog([group('g',('exec2',))],True),simple);f.consume(event(1));f.snapshot();f.consume(event(2,'a','new')) self.assertIsNone(f.occurrences[-1]['community_at_execution']);f.snapshot();self.assertIsNone(f.occurrences[-1]['community_at_execution']) def test_16_prefix_future_absence_and_forward_windows(self): f=FlowShadow('r',EntryCatalog([group()],True),simple);f.consume(event(1));s=f.snapshot();self.assertFalse(s['trajectory_windows'][0]['events']) f.consume(event(2,'b','future'));t=f.snapshot();self.assertNotIn('future',s['communities'][0]['members']);self.assertIn('future',t['communities'][0]['members']) self.assertEqual(t['trajectory_windows'][0]['events'],[2]);self.assertEqual(s['max_event_seq'],1) def test_17_analysis_failure_does_not_erase_events(self): def fail(*_):raise RuntimeError('injected') f=FlowShadow('r',partitioner=fail);f.consume(event(1)) with self.assertRaises(RuntimeError):f.snapshot() f.consume(event(2));self.assertEqual(sum(f.weights.values()),2) def test_18_out_of_order_and_cross_run(self): f=FlowShadow('r');f.consume(event(2)) with self.assertRaises(ValueError):f.consume(event(1)) with self.assertRaises(ValueError):f.consume(event(3)|{'run_id':'other'}) def test_19_native_rebuild_not_execution(self): f=FlowShadow('r');f.consume(event(1)|{'history_rebuild':True});self.assertFalse(f.events) def test_20_dwell_censor_and_budget(self): f=FlowShadow('r') for e in [event(1,'a','b'),event(2,'b','b'),event(3,'b','a'),event(4,'a','b'),event(5,'b','b')]:f.consume(e) m=region_metrics(f.events,['b']);self.assertEqual(m['boundary_entries'],2);self.assertEqual(m['reentries'],1) self.assertEqual(m['complete_dwell_count'],1);self.assertEqual(m['dwell_max'],2);self.assertTrue(m['episodes'][-1]['right_censored']) def test_21_identity_requires_concrete(self): with self.assertRaises(ValueError):EntryCatalog([group(members=[dict(source_state='a',model_action_identity='click')])]) def test_22_nonoracle_known_time(self): g=group(ids=('exec1','exec2'))|{'oracle':False,'known_at_event_seq':2} f=FlowShadow('r',EntryCatalog([g]),simple);f.consume(event(1));f.consume(event(2));self.assertEqual([o['event_seq'] for o in f.occurrences],[2]) def test_23_stream_follow_same_reducer(self): with tempfile.TemporaryDirectory() as td: t=Path(td);src=t/'events.jsonl';src.write_text('');out=t/'out' p=subprocess.Popen([sys.executable,'-m','tools.region_flow_shadow','--input',str(src),'--output',str(out),'--run-id','r','--follow','--idle-timeout','1']) try: with src.open('a') as f: for i in range(1,8):f.write(json.dumps(event(i,'a','b' if i%2 else 'a'))+'\n');f.flush();time.sleep(.02) self.assertEqual(p.wait(15),0) s=json.loads((out/'S0001.json').read_text());self.assertEqual(s['total_edge_weight'],7);self.assertEqual(s['max_event_seq'],7) finally: if p.poll() is None:p.terminate();p.wait() def test_24_no_rng_calls(self): random.seed(42);before=random.getstate();f=FlowShadow('r',partitioner=simple);f.consume(event(1));f.snapshot();self.assertEqual(before,random.getstate()) def test_25_real_infomap_reproducibility(self): ns={'a','b','c'};w={('a','b'):10,('b','a'):10,('b','c'):1,('c','c'):30} self.assertEqual(infomap_partition(ns,w),infomap_partition(ns,w)) if __name__=='__main__':unittest.main()