41 lines
2.5 KiB
Python
41 lines
2.5 KiB
Python
"""Root coordinator refusal/rearm tests; native integration tests real credentials."""
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
|
|
import c01_supervisor as s
|
|
import c01_children as children
|
|
|
|
class ClockRefreshTests(unittest.TestCase):
|
|
def test_collector_rearm_requires_successful_complete_retirement(self):
|
|
for active,reserved,result,stopping in [(True,False,0,False),(False,True,0,False),(False,False,1,False),(False,False,None,False),(False,False,0,True)]:
|
|
c=object.__new__(children.Children);c.children={'collector':object()} if active else {};c.reserved={'collector':object()} if reserved else {};c.results={} if result is None else {'collector':result};c.stopping=stopping
|
|
with self.assertRaises(ValueError):c.rearm_collector()
|
|
c=object.__new__(children.Children);c.children={};c.reserved={};c.results={'collector':0};c.stopping=False
|
|
c.rearm_collector();self.assertNotIn('collector',c.results)
|
|
with self.assertRaises(ValueError):c.rearm_collector()
|
|
|
|
def test_replay_unappointed_and_power_overlap_refuse_before_control_write(self):
|
|
request={'status':'CLOCK_REFRESH','sequence':1,'nonce':[1]*32}
|
|
for enabled,power,sequence in [(False,None,0),(True,object(),0),(True,None,1)]:
|
|
owner=SimpleNamespace(power_state=power,clock_sequence=sequence)
|
|
prepared=SimpleNamespace(epoch={'delegation':{'allow_refresh':enabled}})
|
|
with patch.object(s,'packet') as packet,self.assertRaises(ValueError):
|
|
s.clock_refresh(owner,prepared,None,None,None,None,None,None,request)
|
|
packet.assert_not_called()
|
|
|
|
def test_mismatched_retirement_nonce_consumes_request_without_minting(self):
|
|
owner=SimpleNamespace(power_state=None,clock_sequence=0,
|
|
receive=lambda *args:json.dumps({'status':'CLOCK_RETIRED','sequence':1,'nonce':[2]*32}).encode())
|
|
prepared=SimpleNamespace(epoch={'delegation':{'allow_refresh':True}})
|
|
authority=SimpleNamespace(channel=object())
|
|
with patch.object(s,'packet') as packet,patch.object(s,'start_collector') as collect,self.assertRaises(ValueError):
|
|
s.clock_refresh(owner,prepared,None,None,authority,None,{'uid':1,'gid':1},None,
|
|
{'status':'CLOCK_REFRESH','sequence':1,'nonce':[1]*32})
|
|
self.assertEqual(owner.clock_sequence,1);self.assertEqual(packet.call_count,1);collect.assert_not_called()
|
|
|
|
if __name__=='__main__':unittest.main()
|