import concurrent.futures import http.client import json from pathlib import Path import tempfile import threading import unittest from unittest.mock import patch from agent_control.store import Store, Invalid, Conflict from agent_control.server import Server, EtaClock def definition(**changes): value = dict(id='choice',title='Choose',question='Which approach?', options=[dict(id='a',label='Approach A',description='Synthetic option')]) value.update(changes) return value def answer(**changes): value = dict(option_id='a',comment='Keep this exact comment',idempotency_key='key') value.update(changes) return value class DecisionTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = Path(self.tmp.name)/'test.sqlite3' self.store = Store(self.path) def tearDown(self): self.tmp.cleanup() def test_task_progress_independent_timestamp_and_retention(self): task = self.store.task(dict(id='t',name='Task',eta_basis='Estimate only')) self.assertEqual(task['current_progress'],'') self.assertIsNone(task['progress_updated']) with patch('agent_control.store.time.time',return_value=100): task = self.store.task(dict(current_progress='Validated parser'), 't') self.assertEqual(task['progress_updated'],100) with patch('agent_control.store.time.time',return_value=200): task = self.store.task(dict(status='blocked',decision='Choose next step'),'t') self.assertEqual(task['current_progress'],'Validated parser') self.assertEqual(task['progress_updated'],100) self.assertEqual(task['updated'],200) with patch('agent_control.store.time.time',return_value=300): task = self.store.task(dict(status='completed',phase='done'),'t') self.assertEqual(task['progress_updated'],100) self.assertEqual(task['current_progress'],'Validated parser') with patch('agent_control.store.time.time',return_value=400): task = self.store.task(dict(current_progress=''),'t') self.assertEqual(task['progress_updated'],400) self.assertEqual(task['current_progress'],'') def test_task_progress_validation_and_legacy_defaults(self): self.store.task(dict(id='t',name='Task')) for data in [dict(current_progress=None),dict(current_progress=1), dict(current_progress='x'*1001),dict(progress_updated=1)]: with self.assertRaises(Invalid): self.store.task(data,'t') with self.store.db() as db: db.execute('INSERT INTO tasks(id,data) VALUES(?,?)',('legacy',json.dumps(dict(id='legacy',name='Legacy')))) task = next(t for t in self.store.state()['tasks'] if t['id']=='legacy') self.assertEqual(task['current_progress'],'') self.assertIsNone(task['progress_updated']) def eta_task(self, **changes): data = dict(id='eta',name='Synthetic ETA',status='running', eta_min_minutes=5,eta_max_minutes=10,eta_basis='Synthetic remaining checks') data.update(changes) with patch('agent_control.store.time.time',return_value=1000): return self.store.task(data) def eta_notice(self, now=1300): self.store.eta_tick(now) return self.store.eta_notifications()['notifications'][0] def test_eta_checkpoint_clock_and_persistent_dedup(self): task = self.eta_task() self.assertEqual(task['eta_revision'],1) self.assertEqual(task['eta_estimated_at'],1000) self.assertEqual(task['eta_change_reason'],task['eta_basis']) self.assertEqual(self.store.eta_tick(1299)['created'],0) self.assertEqual(self.store.eta_tick(1300)['created'],1) first = self.store.eta_notifications()['notifications'][0] self.assertEqual(first['checkpoint'],'half') self.assertEqual(first['target'],'control') self.assertEqual(first['revision'],1) self.assertEqual(Store(self.path).eta_tick(1300)['created'],0) self.assertEqual(self.store.eta_tick(1479)['created'],0) self.assertEqual(self.store.eta_tick(1480)['created'],1) self.assertEqual(self.store.eta_tick(1600)['created'],1) self.assertEqual(self.store.state()['tasks'][0]['eta_review_status'],'overdue') self.assertEqual(self.store.eta_notifications()['pending_count'],3) def test_eta_heartbeat_keep_and_ack_never_move_estimate(self): self.eta_task() nid = self.eta_notice()['notification_id'] with patch('agent_control.store.time.time',return_value=1310): task = self.store.task(dict(eta_min_minutes=5,eta_max_minutes=10,current_progress='Running checks'),'eta') self.assertEqual(task['eta_estimated_at'],1000) self.assertEqual(task['eta_revision'],1) self.assertEqual(self.store.eta_notifications()['notifications'][0]['task']['current_progress'],'Running checks') ack = dict(notification_id=nid) self.assertFalse(self.store.eta_ack(ack)['duplicate']) self.assertTrue(self.store.eta_ack(ack)['duplicate']) self.assertEqual(self.store.eta_notifications()['pending_count'],0) body = dict(result='keep',reason='Half the checks passed',current_progress='Verified outputs') with patch('agent_control.store.time.time',return_value=1320): result = self.store.eta_complete(nid,body) self.assertEqual(result['eta_estimated_at'],1000) self.assertEqual(result['eta_revision'],1) self.assertEqual(result['eta_last_reviewed_at'],1320) self.assertEqual(result['eta_review_reason'],body['reason']) self.assertEqual(result['progress_updated'],1320) self.assertEqual(Store(self.path).eta_complete(nid,body),result) with self.assertRaises(Conflict): self.store.eta_complete(nid,dict(body,reason='Changed answer')) def test_eta_revise_changes_version_and_cancels_old_notifications(self): self.eta_task() self.store.eta_tick(1480) notices = self.store.eta_notifications()['notifications'] body = dict(result='revise',reason='New failing test found',eta_min_minutes=5,eta_max_minutes=20) with patch('agent_control.store.time.time',return_value=1481): task = self.store.eta_complete(notices[0]['notification_id'],body) self.assertEqual((task['eta_revision'],task['eta_estimated_at']),(2,1481)) self.assertEqual(task['eta_extension_count'],1) self.assertEqual(task['eta_change_reason'],body['reason']) self.assertEqual(self.store.eta_notifications()['pending_count'],0) with self.assertRaises(Conflict): self.store.eta_complete(notices[1]['notification_id'],dict(result='keep',reason='Late response')) self.assertEqual(self.store.eta_tick(2081)['created'],1) self.assertEqual(self.store.eta_notifications()['notifications'][0]['revision'],2) with self.store.db() as db: self.assertEqual(db.execute('SELECT count(*) FROM eta_reviews WHERE cancelled_at IS NOT NULL').fetchone()[0],2) def test_eta_unknown_does_not_extend(self): self.eta_task() nid = self.eta_notice()['notification_id'] with patch('agent_control.store.time.time',return_value=1310): task = self.store.eta_complete(nid,dict(result='unknown',reason='Waiting on real evidence')) self.assertEqual(task['eta_review_status'],'unknown') self.assertEqual(task['eta_estimated_at'],1000) self.assertEqual(task['eta_revision'],1) self.assertEqual(task['eta_max_minutes'],10) def test_eta_extension_limit_persists_and_cannot_be_bypassed_by_task_patch(self): task = self.eta_task() for index in range(3): checkpoint = task['eta_estimated_at'] + task['eta_max_minutes']*30 nid = self.eta_notice(checkpoint)['notification_id'] with patch('agent_control.store.time.time',return_value=checkpoint+1): task = self.store.eta_complete(nid,dict(result='revise',reason='Measured additional work', eta_min_minutes=5,eta_max_minutes=20+index*10)) self.assertEqual(task['eta_extension_count'],3) old = dict(task) checkpoint = task['eta_estimated_at'] + task['eta_max_minutes']*30 nid = self.eta_notice(checkpoint)['notification_id'] with patch('agent_control.store.time.time',return_value=checkpoint+1): with self.assertRaises(Conflict): self.store.eta_complete(nid,dict(result='revise',reason='Fourth extension', eta_min_minutes=5,eta_max_minutes=50)) with self.assertRaises(Conflict): self.store.task(dict(eta_max_minutes=50,eta_change_reason='Try task patch'),'eta') task = Store(self.path).state()['tasks'][0] self.assertEqual(task['eta_review_status'],'needs_decision') self.assertEqual(task['eta_revision'],old['eta_revision']) self.assertEqual(task['eta_estimated_at'],old['eta_estimated_at']) self.assertEqual(task['eta_max_minutes'],old['eta_max_minutes']) self.assertEqual(task['eta_extension_count'],3) self.store.eta_tick(checkpoint+10000) self.assertEqual(self.store.state()['tasks'][0]['eta_review_status'],'needs_decision') def test_eta_same_range_revise_and_unknown_cannot_clear_decision_limit(self): self.eta_task() nid = self.eta_notice()['notification_id'] with patch('agent_control.store.time.time',return_value=1400): task = self.store.eta_complete(nid,dict(result='revise',reason='Same measured range', eta_min_minutes=5,eta_max_minutes=10)) self.assertEqual(task['eta_revision'],1) self.assertEqual(task['eta_estimated_at'],1000) self.assertEqual(task['eta_review_status'],'reviewed') self.store.eta_tick(1480) nid = self.store.eta_notifications()['notifications'][0]['notification_id'] with self.store.db() as db: task['eta_review_status']='needs_decision' self.store._put(db,'tasks',task) task = self.store.eta_complete(nid,dict(result='unknown',reason='Still awaiting evidence')) self.assertEqual(task['eta_review_status'],'needs_decision') def test_eta_invalid_changes_zero_and_managed_fields(self): with self.assertRaises(Invalid): self.eta_task(eta_min_minutes=0,eta_max_minutes=0) self.eta_task() for data in [dict(eta_max_minutes=11),dict(eta_max_minutes=11,eta_change_reason=''), dict(eta_revision=9),dict(eta_estimated_at=1),dict(eta_extension_count=0), dict(eta_max_minutes=True),dict(eta_max_minutes=float('inf'))]: with self.assertRaises(Invalid): self.store.task(data,'eta') self.assertEqual(self.store.state()['tasks'][0]['eta_revision'],1) task = self.store.task(dict(status='completed',phase='done',eta_min_minutes=0, eta_max_minutes=0,eta_change_reason='Checks finished'),'eta') self.assertEqual(task['eta_max_minutes'],0) self.assertEqual(self.store.eta_tick(100000)['created'],0) def test_eta_legacy_and_no_estimate_do_not_invent_start(self): self.store.task(dict(id='none',name='No estimate')) with self.store.db() as db: db.execute('INSERT INTO tasks(id,data) VALUES(?,?)',('legacy',json.dumps( dict(id='legacy',name='Legacy',status='running',eta_min_minutes=5,eta_max_minutes=10, eta_basis='Legacy explanation')))) self.assertEqual(self.store.eta_tick(100000)['created'],0) task = next(t for t in self.store.state()['tasks'] if t['id']=='legacy') self.assertIsNone(task['eta_estimated_at']) self.assertEqual(task['eta_revision'],0) with patch('agent_control.store.time.time',return_value=2000): task = self.store.task(dict(current_progress='Reading baseline'),'legacy') self.assertIsNone(task['eta_estimated_at']) with patch('agent_control.store.time.time',return_value=2001): task = self.store.task(dict(eta_min_minutes=5,eta_max_minutes=10),'legacy') self.assertEqual(task['eta_revision'],1) self.assertEqual(task['eta_estimated_at'],2001) def test_eta_finished_and_revised_work_does_not_notify_old_reviews(self): for index,status in enumerate(['completed','cancelled','failed']): tid = 'finished-'+str(index) self.eta_task(id=tid) notices = self.store.eta_tick(1300) notice = next(n for n in self.store.eta_notifications()['notifications'] if n['task']['id']==tid) self.store.task(dict(status=status,phase='done'),'finished-'+str(index)) self.assertFalse(any(n['task']['id']==tid for n in self.store.eta_notifications()['notifications'])) with self.assertRaises(Conflict): self.store.eta_complete(notice['notification_id'],dict(result='keep',reason='Late result')) self.assertEqual(self.store.eta_tick(100000)['created'],0) def test_eta_invalid_completion_does_not_mutate(self): self.eta_task() nid = self.eta_notice()['notification_id'] for body in [dict(result='keep',reason=''),dict(result='bad',reason='Reason'), dict(result='keep',reason='Reason',eta_max_minutes=10), dict(result='revise',reason='Reason',eta_max_minutes=20), dict(result='revise',reason='Reason',eta_min_minutes=0,eta_max_minutes=0), dict(result='revise',reason='Reason',eta_min_minutes=None,eta_max_minutes=None), dict(result='unknown',reason='Reason',current_progress=None), dict(result='keep',reason='Reason',authorization=True)]: with self.subTest(body=body): with self.assertRaises(Invalid): self.store.eta_complete(nid,body) self.assertEqual(self.store.state()['tasks'][0]['eta_revision'],1) self.assertEqual(self.store.eta_notifications()['pending_count'],1) def test_eta_concurrent_ticks_and_completions(self): self.eta_task() with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: list(pool.map(lambda _:self.store.eta_tick(1300),range(12))) self.assertEqual(self.store.eta_notifications()['pending_count'],1) nid = self.store.eta_notifications()['notifications'][0]['notification_id'] body = dict(result='keep',reason='Measured progress') with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda _:self.store.eta_complete(nid,body),range(12))) self.assertTrue(all(r==results[0] for r in results)) self.assertEqual(self.store.eta_notifications()['pending_count'],0) def test_eta_background_clock_without_state_poll_and_stop(self): self.eta_task() seen = threading.Event() original = self.store.eta_tick def tick(): original(1300) seen.set() with patch.object(self.store,'eta_tick',side_effect=tick): clock = EtaClock(self.store,interval=0.01) try: self.assertTrue(seen.wait(2)) self.assertEqual(clock.interval,0.01) self.assertEqual(self.store.eta_notifications()['pending_count'],1) finally: clock.close() self.assertFalse(clock.thread.is_alive()) def test_empty_and_legacy_task(self): self.assertEqual(self.store.state()['decisions'],[]) self.assertEqual(self.store.decision_notifications(),dict(notifications=[],pending_count=0)) self.store.session(dict(id='s',name='Synthetic')) self.store.task(dict(id='t',name='Task',session_id='s',decision='Legacy question')) self.store.decision(definition(task_id='t')) self.assertEqual(self.store.state()['tasks'][0]['decision'],'Legacy question') def test_normalized_creation_and_immutable_definition(self): first = self.store.decision(definition()) self.assertEqual(first['status'],'open') self.assertIsNone(first['response']) explicit = definition(impact='',priority='normal',details='',recommendation='', deadline=None,requires_authorization=False,task_id=None) self.assertEqual(first,self.store.decision(explicit)) with self.assertRaises(Conflict): self.store.decision(definition(question='Changed')) self.assertEqual(self.store.decision_notifications()['pending_count'],0) def test_persistence_response_dedup_and_ack(self): self.store.decision(definition()) with patch('subprocess.run',side_effect=AssertionError('must not execute')): first = self.store.decision_respond('choice',answer()) reopened = Store(self.path) self.assertEqual(first,reopened.decision_respond('choice',answer())) self.assertEqual(first,reopened.decision(definition())) self.assertEqual(first['status'],'decided') self.assertFalse(first['response']['authorization']) self.assertEqual(first['response']['comment'],answer()['comment']) for changes in [dict(comment='changed'),dict(idempotency_key='other'),dict(option_id=None)]: with self.assertRaises(Conflict): reopened.decision_respond('choice',answer(**changes)) queue = reopened.decision_notifications() self.assertEqual(queue['pending_count'],1) notice = queue['notifications'][0] self.assertEqual(set(notice),{'notification_id','target','decision'}) self.assertEqual(notice['target'],'control') self.assertEqual(notice['decision'],first) ack = dict(notification_id=notice['notification_id']) self.assertFalse(reopened.decision_ack(ack)['duplicate']) self.assertTrue(Store(self.path).decision_ack(ack)['duplicate']) self.assertEqual(reopened.decision_notifications()['pending_count'],0) reopened.decision_respond('choice',answer()) self.assertEqual(reopened.decision_notifications()['pending_count'],0) def test_authorization_is_only_pending_verification(self): self.store.decision(definition(requires_authorization=True,options=[])) value = self.store.decision_respond('choice',answer(option_id=None,comment='Approve sudo')) self.assertEqual(value['status'],'pending_verification') self.assertIs(value['response']['authorization'],False) def test_invalid_definitions(self): for changes in [dict(shell='x'),dict(title=''),dict(question=None),dict(impact=None), dict(priority=[]),dict(priority='critical'),dict(requires_authorization=1), dict(options={}),dict(options=[dict(id='a',label='a')]), dict(options=[dict(id='a',label='a',description='')]*2), dict(deadline='2026-02-30T01:00:00Z'),dict(deadline='tomorrow'), dict(task_id='missing'),dict(details='x'*8001)]: with self.subTest(changes=changes): with self.assertRaises(Invalid): self.store.decision(definition(**changes)) self.assertEqual(self.store.state()['decisions'],[]) self.store.decision(definition(deadline='2026-10-19T13:00:00+08:00')) def test_invalid_answers_and_ack(self): self.store.decision(definition()) for changes in [dict(option_id=None),dict(option_id='bad'),dict(comment=None), dict(comment='x'*8001),dict(idempotency_key=''),dict(authorization=True)]: with self.subTest(changes=changes): with self.assertRaises(Invalid): self.store.decision_respond('choice',answer(**changes)) with self.assertRaises(Invalid): self.store.decision_respond('missing',answer()) with self.assertRaises(Invalid): self.store.decision_ack(dict(notification_id='missing')) with self.assertRaises(Invalid): self.store.decision_ack(dict(notification_id='missing',target='human')) self.assertEqual(self.store.state()['decisions'][0]['status'],'open') self.assertEqual(self.store.decision_notifications()['pending_count'],0) self.store.decision(definition(id='free',options=[])) for option,comment in [(None,' '),('a','text')]: with self.assertRaises(Invalid): self.store.decision_respond('free',answer(option_id=option,comment=comment)) def test_concurrent_same_key_has_one_response_and_notification(self): self.store.decision(definition()) with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: values = list(pool.map(lambda _:self.store.decision_respond('choice',answer()),range(16))) self.assertTrue(all(v == values[0] for v in values)) self.assertEqual(self.store.decision_notifications()['pending_count'],1) def test_concurrent_distinct_answers_have_one_winner(self): self.store.decision(definition()) def respond(index): try: return self.store.decision_respond('choice',answer(idempotency_key=str(index))) except Conflict: return None with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: values = list(pool.map(respond,range(16))) self.assertEqual(sum(v is not None for v in values),1) self.assertEqual(self.store.decision_notifications()['pending_count'],1) def test_atomic_rollback_on_outbox_failure(self): self.store.decision(definition()) with self.store.db() as db: db.execute("CREATE TRIGGER fail_notice BEFORE INSERT ON decision_notifications BEGIN SELECT RAISE(ABORT,'synthetic'); END") with self.assertRaises(Exception): self.store.decision_respond('choice',answer()) self.assertEqual(self.store.state()['decisions'][0]['status'],'open') self.assertEqual(self.store.decision_notifications()['pending_count'],0) def test_pending_order_and_limit(self): for index in range(103): did = 'choice-'+str(index) self.store.decision(definition(id=did)) self.store.decision_respond(did,answer()) queue = self.store.decision_notifications() self.assertEqual(queue['pending_count'],103) self.assertEqual(len(queue['notifications']),100) self.assertEqual(queue['notifications'][0]['decision']['id'],'choice-0') self.assertEqual(queue['notifications'][-1]['decision']['id'],'choice-99') class DecisionHTTPTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() root = Path(self.tmp.name) root.chmod(0o700) token = root / 'token' self.secret = 'h'*48 token.write_text(self.secret) token.chmod(0o600) self.metrics = patch('agent_control.server.MetricsCache') mock = self.metrics.start() mock.return_value.snapshot.return_value = {'status':'synthetic'} self.server = Server(dict(bind='127.0.0.1',port=0,runtime_root=str(root),token_file=str(token))) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join() self.metrics.stop() self.tmp.cleanup() def request(self,path,method='GET',payload=None,headers=None): conn = http.client.HTTPConnection(*self.server.server_address,timeout=3) merged = {'Authorization':'Bearer '+self.secret,'X-CSRF-Token':'1','Content-Type':'application/json'} merged.update(headers or {}) conn.request(method,path,json.dumps(payload) if payload is not None else None,merged) response = conn.getresponse() status, body = response.status,json.loads(response.read()) conn.close() return status, body def test_routes_and_security(self): create='/api/decisions' respond='/api/decisions/choice/respond' notifications='/api/decision-notifications' ack=notifications+'/ack' for path,body in [(create,definition()),(respond,answer()),(ack,dict(notification_id='missing'))]: for headers,status in [({'Authorization':''},401),({'X-CSRF-Token':''},403), ({'Origin':'https://untrusted.example'},403)]: self.assertEqual(self.request(path,'POST',body,headers)[0],status) self.assertEqual(self.request(notifications,headers={'Authorization':''})[0],401) self.assertEqual(self.request('/api/state')[1]['decisions'],[]) self.assertEqual(self.request(create,'POST',definition())[0],200) self.assertEqual(self.request(create,'POST',definition(question='different'))[0],409) self.assertEqual(self.request(respond,'POST',answer(option_id='bad'))[0],400) code,value=self.request(respond,'POST',answer()) self.assertEqual(code,200) self.assertEqual(self.request(respond,'POST',answer())[1],value) self.assertEqual(self.request(respond,'POST',answer(idempotency_key='new'))[0],409) notice=self.request(notifications)[1]['notifications'][0] self.assertEqual(self.request(ack,'POST',dict(notification_id=notice['notification_id']))[0],200) self.assertEqual(self.request(notifications)[1]['pending_count'],0) self.assertEqual(self.request(notifications+'?target=human')[0],404) self.assertEqual(self.request(respond+'?x=1','POST',answer())[0],404) self.assertEqual(self.request('/api/decisions/choice','PATCH',definition())[0],404) self.assertEqual(self.request('/api/state')[1]['decisions'][0],value) def test_cross_site_only_public_top_level_navigation(self): web = Path(self.tmp.name)/'web' web.mkdir() (web/'index.html').write_text('Public login') self.server.web_root = web navigation = {'Sec-Fetch-Site':'cross-site','Sec-Fetch-Mode':'navigate','Sec-Fetch-Dest':'document'} conn = http.client.HTTPConnection(*self.server.server_address,timeout=3) conn.request('GET','/',headers=navigation) response = conn.getresponse() self.assertEqual(response.status,200) self.assertIn(b'Public login',response.read()) conn.close() self.assertEqual(self.request('/api/state',headers=navigation)[0],403) self.assertEqual(self.request('/api/decisions','POST',definition(),navigation)[0],403) self.assertEqual(self.request('/','POST',definition(),navigation)[0],403) for overrides in [{'Sec-Fetch-Mode':'no-cors','Sec-Fetch-Dest':'script'}, {'Sec-Fetch-Dest':'iframe'},{'Sec-Fetch-Mode':'cors'}, {'Origin':'https://untrusted.example'},{'Host':'untrusted.example'}]: self.assertEqual(self.request('/',headers=dict(navigation,**overrides))[0],403) def test_eta_http_security_ack_and_complete(self): self.server.eta_clock.close() with patch('agent_control.store.time.time',return_value=1000): self.server.store.task(dict(id='http-eta',name='ETA',status='running', eta_min_minutes=5,eta_max_minutes=10,eta_basis='Measured')) self.server.store.eta_tick(1300) self.assertEqual(self.request('/api/eta-notifications',headers={'Authorization':''})[0],401) self.assertEqual(self.request('/api/eta-notifications',headers={'Sec-Fetch-Site':'cross-site'})[0],403) self.assertEqual(self.request('/api/eta-notifications?x=1')[0],404) notice=self.request('/api/eta-notifications')[1]['notifications'][0] nid=notice['notification_id'] ack='/api/eta-notifications/ack' complete='/api/eta-reviews/'+nid+'/complete' body=dict(result='keep',reason='Evidence received') self.assertEqual(self.request(ack,'POST',dict(notification_id=nid),{'X-CSRF-Token':''})[0],403) self.assertEqual(self.request(complete,'POST',body,{'Origin':'https://untrusted.example'})[0],403) self.assertEqual(self.request(ack,'POST',dict(notification_id=nid))[0],200) self.assertEqual(self.request(complete,'POST',body)[0],200) self.assertEqual(self.request(complete,'POST',dict(body,reason='Changed'))[0],409) self.assertEqual(self.request('/api/eta-notifications')[1]['pending_count'],0) self.assertEqual(self.server.eta_clock.interval,15) self.assertFalse(self.server.eta_clock.thread.is_alive()) def test_server_starts_clock_without_http_polling(self): self.assertIsInstance(self.server.eta_clock,EtaClock) self.assertTrue(self.server.eta_clock.thread.is_alive()) self.assertTrue(self.server.eta_clock.thread.name=='eta-clock')