Files
archy/scripts/tests/test_c01_characterized_recovery.py

64 lines
3.7 KiB
Python

"""Characterized issuer mechanics, synthetic model/clock; never a profile."""
import copy
from dataclasses import replace
from pathlib import Path
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
import c01_nts as nts
import c01_recovery as c
from scripts.tests.test_c01_epoch_recovery import observation,challenge,qualification,binding
def model():
return dict(identity=dict(kernel=[1]*32,clocksource=[2]*32,hardware=[3]*32),evidence_sha256=[4]*32,
additive=c.duration(80_000_000),rate_numerator=1,rate_denominator=1_000_000,
max_raw_holdover=c.duration(64*10**9),max_reference_age=c.duration(8*10**9),
max_adjustment=c.duration(1_000_000),association_multiplier=2)
def raw(before,after):return dict(before=c.duration(before),after=c.duration(after),identity=model()['identity'])
class CharacterizedIssuerTests(unittest.TestCase):
def inputs(self):
o=replace(observation(),raw=raw(100*10**9,100*10**9+100))
ctx=challenge();ctx.update(characterized=model(),barrier_raw=raw(99*10**9,99*10**9+100),not_before=c.duration(o.boottime_ns-10**9))
q=replace(qualification(),characterized_model=model())
return o,ctx,q
def packet(self,o,ctx,q,refresh=False):
return c.epoch_packet(o,c.binding_bytes(binding()),q,ctx,c.EpochIssuerDelegation(bytes([3])*32,True,refresh))
def test_exact_raw_anchor_model_and_additive_term_are_transported(self):
o,ctx,q=self.inputs();p=self.packet(o,ctx,q)
self.assertEqual(p['raw_anchor'],o.raw)
self.assertEqual(p['challenge']['characterized']['additive'],c.duration(80_000_000))
self.assertEqual(p['recovery']['wall'],c.duration(o.utc_ns))
self.assertEqual(p['recovery']['uncertainty'],c.duration(o.observation_error_ns))
self.assertEqual(p['expires_at'],c.duration(o.boottime_ns+q.max_awake_ns))
def test_missing_raw_identity_change_and_cumulative_adjustment_refuse(self):
for mutation in range(4):
o,ctx,q=self.inputs()
if mutation==0:o=replace(o,raw=None)
elif mutation==1:o.raw['identity']['hardware']=[9]*32
elif mutation==2:o=replace(o,raw=raw(101*10**9,101*10**9+100))
else:ctx['characterized']['rate_denominator']=0
with self.subTest(mutation=mutation),self.assertRaises(nts.Unavailable):self.packet(o,ctx,q)
def test_refresh_is_fresh_explicit_service_scope_not_manual_approval(self):
o,ctx,q=self.inputs();ctx['cause']='Refresh'
with self.assertRaises(nts.Unavailable):self.packet(o,ctx,q)
self.assertEqual(self.packet(o,ctx,q,True)['challenge']['cause'],'Refresh')
def test_model_reference_age_additive_and_zero_denominator(self):
m=model();o=observation()
age=c.model_reference_age(m,o,o.utc_ns)
self.assertGreaterEqual(age,80_000_000)
e=c.model_error(m,raw(10**9,10**9),0,raw(10**9,10**9),0,0,0,0)
self.assertEqual(e,80_000_000)
with self.assertRaises(nts.Unavailable):c.model_reference_age(m,o,o.utc_ns-9*10**9)
m['rate_denominator']=0
with self.assertRaises(nts.Unavailable):c.validate_characterized_model(m)
def test_profile_identity_and_model_shape_refuse_before_collection(self):
o,ctx,q=self.inputs()
with patch.object(c.PlatformQualification,'validate'),patch.object(nts,'platform_identity',return_value=dict(kernel=[9]*32,clocksource=[2]*32,hardware=[3]*32)),self.assertRaises(nts.Unavailable):
q.validate_epoch(q.max_awake_ns)
bad=copy.deepcopy(model());bad['unreviewed_bound']=0
with self.assertRaises(nts.Unavailable):c.validate_characterized_model(bad)
if __name__=='__main__':unittest.main()