tegwick
52dcc08f23
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
101 lines
5.7 KiB
Python
101 lines
5.7 KiB
Python
from datetime import datetime,timezone,timedelta
|
|
import pytest
|
|
from activity_core.release_sampler import BASELINE, Sampler, invariant_matches
|
|
|
|
NOW=datetime(2026,10,11,20,tzinfo=timezone.utc)
|
|
|
|
|
|
def test_gap_sensitive_actual_sample_interval_survives_restart_without_backfill(tmp_path):
|
|
database=tmp_path/'health.sqlite'
|
|
sampler=Sampler(database,lambda:(BASELINE,True))
|
|
assert sampler.sample(NOW)['continuous_24h_ready'] is False
|
|
later=NOW+timedelta(days=1)
|
|
assert sampler.sample(later)['healthy_since']==later.isoformat()
|
|
for minute in range(1,1441):sampler.sample(later+timedelta(minutes=minute))
|
|
restarted=Sampler(database,lambda:(BASELINE,True))
|
|
result=restarted.sample(later+timedelta(days=1,seconds=60))
|
|
assert result['continuous_24h_ready'] and result['attestation']['continuous']
|
|
|
|
|
|
def test_unavailable_probe_records_failure_and_resets_interval(tmp_path):
|
|
sampler=Sampler(tmp_path/'health.sqlite',lambda:(BASELINE,True))
|
|
sampler.sample(NOW)
|
|
def unavailable():raise RuntimeError('PRIVATE-TOKEN-CANARY')
|
|
sampler.probe=unavailable
|
|
result=sampler.sample(NOW+timedelta(minutes=1))
|
|
assert result['healthy'] is False and result['healthy_since'] is None
|
|
assert 'CANARY' not in str(result)
|
|
sampler.probe=lambda:(BASELINE,True)
|
|
result=sampler.sample(NOW+timedelta(minutes=2))
|
|
assert result['healthy_since']==(NOW+timedelta(minutes=2)).isoformat()
|
|
|
|
|
|
def test_revision_and_clock_change_refuse_old_observation(tmp_path):
|
|
sampler=Sampler(tmp_path/'health.sqlite',lambda:(BASELINE,True));sampler.sample(NOW)
|
|
with pytest.raises(ValueError):sampler.sample(NOW)
|
|
sampler.probe=lambda:('b'*40,True)
|
|
assert sampler.sample(NOW+timedelta(minutes=1))['healthy_since']==(NOW+timedelta(minutes=1)).isoformat()
|
|
|
|
|
|
@pytest.mark.parametrize('delta',[{'healthy':False},{'schedules_match':False},{'definition_count':2},
|
|
{'observed_at':(NOW-timedelta(seconds=91)).isoformat()},{'observed_at':(NOW+timedelta(seconds=1)).isoformat()},
|
|
{'error':'private canary'},{'schema_version':'other'}])
|
|
def test_invariant_response_cannot_manufacture_health(delta):
|
|
value={'schema_version':'activity-core-release-health/v1','healthy':True,'schedules_match':True,
|
|
'definitions_and_report_sinks_match':True,'definition_count':3,'schedule_count':3,'observed_at':NOW.isoformat()}
|
|
assert not invariant_matches(value|delta,NOW)
|
|
|
|
|
|
@pytest.fixture
|
|
def native_probe():
|
|
import json
|
|
from activity_core.release_sampler import Probe,CLUSTER_UID,RESOURCE_SET
|
|
from activity_core.release_transport import ACTIVITY_URL
|
|
app={'apiVersion':'argoproj.io/v1alpha1','kind':'Application',
|
|
'metadata':{'name':'activity-core','namespace':'argocd'},
|
|
'spec':{'project':'activity-core','source':{'repoURL':ACTIVITY_URL,'path':'k8s/gitops','targetRevision':BASELINE},
|
|
'destination':{'namespace':'activity-core','server':'https://kubernetes.default.svc'},
|
|
'syncPolicy':{'syncOptions':['CreateNamespace=false','ApplyOutOfSyncOnly=true','PruneLast=true','FailOnSharedResource=true']}},
|
|
'status':{'sync':{'status':'Synced','revision':BASELINE},'health':{'status':'Healthy'},
|
|
'resources':[{'kind':kind,'name':name,'namespace':'activity-core','status':'Synced','group':'apps' if kind=='Deployment' else ''} for kind,name in RESOURCE_SET]}}
|
|
deployment={'metadata':{'generation':1},'spec':{'replicas':1},'status':{'observedGeneration':1,'readyReplicas':1,'updatedReplicas':1,'availableReplicas':1}}
|
|
objects={'nodes':{'items':[{'status':{'addresses':[{'address':'92.205.62.239'}]}}]},
|
|
'namespace':{'metadata':{'uid':CLUSTER_UID}},'application':app,'deployment':deployment,
|
|
'service':{'spec':{'clusterIP':'10.43.1.2','ports':[{'port':8010}]}}}
|
|
calls=[]
|
|
def runner(argv,**kwargs):
|
|
calls.append(argv)
|
|
assert 'get' in argv and not any(verb in argv for verb in ['patch','apply','create','secrets'])
|
|
return json.dumps(objects[argv[argv.index('get')+1]])
|
|
def reader(url):
|
|
assert url=='http://10.43.1.2:8010/release/health'
|
|
return {'schema_version':'activity-core-release-health/v1','healthy':True,'schedules_match':True,
|
|
'definitions_and_report_sinks_match':True,'definition_count':3,'schedule_count':3,
|
|
'observed_at':datetime.now(timezone.utc).isoformat()}
|
|
return Probe(runner,reader),objects,calls
|
|
|
|
|
|
def test_native_probe_fixed_reads_and_invariants(native_probe):
|
|
probe,_,calls=native_probe
|
|
assert probe()==(BASELINE,True)
|
|
assert len(calls)==7
|
|
|
|
|
|
@pytest.mark.parametrize('change',['estate','uid','automated','missing_resource','extra_resource','wrong_group','wrong_namespace','stale_generation','unready','revision','operation'])
|
|
def test_native_probe_drift_fails_closed(native_probe,change):
|
|
probe,objects,_=native_probe
|
|
app=objects['application']
|
|
if change=='estate':objects['nodes']['items'][0]['status']['addresses'][0]['address']='192.0.2.1'
|
|
elif change=='uid':objects['namespace']['metadata']['uid']='foreign-cluster'
|
|
elif change=='automated':app['spec']['syncPolicy']['automated']={}
|
|
elif change=='missing_resource':app['status']['resources'].pop()
|
|
elif change=='extra_resource':app['status']['resources'].append(dict(app['status']['resources'][0]))
|
|
elif change=='wrong_group':app['status']['resources'][0]['group']='foreign'
|
|
elif change=='wrong_namespace':app['status']['resources'][0]['namespace']='foreign'
|
|
elif change=='stale_generation':objects['deployment']['status']['observedGeneration']=0
|
|
elif change=='unready':objects['deployment']['status']['readyReplicas']=0
|
|
elif change=='revision':app['status']['sync']['revision']='b'*40
|
|
elif change=='operation':app['operation']={'sync':{}}
|
|
try:_,healthy=probe()
|
|
except (ValueError,KeyError):return
|
|
assert healthy is False
|