import copy import tempfile import unittest from datetime import datetime, timedelta, timezone from pathlib import Path from fastapi.testclient import TestClient from netplan_v4.store import PlannerStore from netplan_v4.service import create_app from netplan_v4 import controlled_trial as t AT=datetime(2026,10,2,12,tzinfo=timezone.utc) PLANT='00000000-0000-4000-8000-000000000001' SESSION='00000000-0000-4000-8000-000000000002' PLAN='00000000-0000-4000-8000-000000000003' TOKEN='synthetic-only-control-test-token' def fixture(): context={'batteries':{'b':{'capacityKwh':10.}},'limits':{'importW':10000.,'exportW':None,'managerMonthLimitsW':{}}} plan={'planId':PLAN,'installationId':PLANT,'sourceFamily':'3','configRevision':1, 'runMode':'shadow','liveEnabled':False,'executable':True,'validUntil':(AT+timedelta(hours=1)).isoformat(), 'controlContext':context,'inputQuality':{'loadBasis':'base_load','accountingEvidenceId':'synthetic-meter-boundary-v1'}, 'peakCostIsEstimate':True} view={'installationId':PLANT,'liveEnabled':False,'fresh':True,'plan':plan,'settings':{'family':'3','revision':1}} request={'sessionId':SESSION,'expectedPlanId':PLAN,'expectedRevision':1,'durationSeconds':300, 'maxChargeW':5000.,'maxDischargeW':5000.,'assetId':'b','managerId':11111,'batteryInstanceId':22222, 'acceptEstimatedPeak':True,'actuatorWatchdogEvidenceId':'synthetic-hardware-watchdog', 'confirmation':'ARM_BOUNDED_CONTROL_TRIAL'} return view,request class ControlledTrialTest(unittest.TestCase): def setUp(self): self.s=PlannerStore(':memory:');self.v,self.r=fixture() def tearDown(self): self.s.close() def arm(self,**kw): return t.arm(self.s,PLANT,kw.get('request',self.r),kw.get('view',self.v),kw.get('now',AT),kw.get('allowed',{PLANT})) def test_default_has_no_authority(self): self.assertIsNone(t.authority(self.s,PLANT,self.v,AT,set())) with self.assertRaises(ValueError):self.arm(allowed=set()) def test_shadow_plan_never_becomes_live(self): before=copy.deepcopy(self.v);self.arm() a=t.authority(self.s,PLANT,self.v,AT,{PLANT}) self.assertEqual(self.v,before);self.assertEqual(a['kind'],'controlled_trial_authority') self.assertTrue(a['sourcePlanRemainsShadow']);self.assertFalse(self.v['plan']['liveEnabled']) def test_short_authority_lease(self): self.arm();a=t.authority(self.s,PLANT,self.v,AT,{PLANT}) self.assertEqual(t.timestamp(a['validUntil'])-AT,timedelta(seconds=90)) def test_session_expiry(self): self.arm();self.assertIsNone(t.authority(self.s,PLANT,self.v,AT+timedelta(seconds=300),{PLANT})) def test_revoke(self): self.arm();t.revoke(self.s,PLANT,SESSION,AT);self.assertIsNone(t.authority(self.s,PLANT,self.v,AT,{PLANT})) def test_expired_or_revoked_session_not_resurrected(self): self.arm();t.revoke(self.s,PLANT,SESSION,AT) with self.assertRaises(ValueError):self.arm() def test_active_session_cannot_be_extended(self): self.arm() with self.assertRaises(ValueError):self.arm(now=AT+timedelta(seconds=30)) def test_other_plant(self): self.arm();self.assertIsNone(t.authority(self.s,SESSION,self.v,AT,{SESSION})) def test_revocation_mismatched_id_leaves_real_session(self): self.arm();t.revoke(self.s,PLANT,PLAN,AT);self.assertIsNotNone(t.authority(self.s,PLANT,self.v,AT,{PLANT})) def test_stale_source_plan_cannot_arm(self): self.v['fresh']=False with self.assertRaises(ValueError):self.arm() def test_stale_source_removes_authority(self): self.arm();self.v['fresh']=False;self.assertIsNone(t.authority(self.s,PLANT,self.v,AT,{PLANT})) def test_revision_bool_not_an_integer(self): self.r['expectedRevision']=True with self.assertRaises(ValueError):self.arm() def test_wrong_revision(self): self.r['expectedRevision']=2 with self.assertRaises(ValueError):self.arm() def test_revision_change_invalidates_authority(self): self.arm();self.v['settings']['revision']=2;self.v['plan']['configRevision']=2 self.assertIsNone(t.authority(self.s,PLANT,self.v,AT,{PLANT})) def test_policy_change_invalidates_authority(self): self.arm();self.v['plan']['controlContext']['limits']['importW']=12000. self.assertIsNone(t.authority(self.s,PLANT,self.v,AT,{PLANT})) def test_newer_plan_same_policy_supported(self): self.arm();self.v['plan']['planId']=SESSION self.assertEqual(t.authority(self.s,PLANT,self.v,AT,{PLANT})['sourceShadowPlanId'],SESSION) def test_auto_family_not_enabled_for_first_trial(self): self.v['settings']['family']='auto' with self.assertRaises(ValueError):self.arm() def test_aggregate_house_refused(self): self.v['plan']['inputQuality']['loadBasis']='house_total' with self.assertRaises(ValueError):self.arm() def test_no_accounting_evidence_no_trial(self): del self.v['plan']['inputQuality']['accountingEvidenceId'] with self.assertRaises(ValueError):self.arm() def test_no_watchdog_evidence_no_trial(self): self.r['actuatorWatchdogEvidenceId']='' with self.assertRaises(ValueError):self.arm() def test_no_implicit_estimate_acceptance(self): self.r['acceptEstimatedPeak']=False with self.assertRaises(ValueError):self.arm() def test_power_cap_and_nonfinite(self): for v in (5001, float('nan'),float('inf'),True,0,'5000'): with self.subTest(v=v),self.assertRaises(ValueError):self.arm(request={**self.r,'maxChargeW':v}) def test_duration_bound(self): for v in (0,1801,True,30.5): with self.subTest(v=v),self.assertRaises(ValueError):self.arm(request={**self.r,'durationSeconds':v}) def test_cannot_replace_asset(self): self.r['assetId']='other' with self.assertRaises(ValueError):self.arm() def test_two_assets_not_supported(self): self.v['plan']['controlContext']['batteries']['other']={} with self.assertRaises(ValueError):self.arm() def test_operator_endpoint_requires_auth(self): with tempfile.TemporaryDirectory() as td: app=create_app(Path(td)/'test.sqlite',TOKEN,[PLANT],start_worker=False) with TestClient(app) as c: self.assertEqual(c.post(f'/internal/v2/prognosis/{PLANT}/planner/trial/arm',json=self.r).status_code,401) r=c.post(f'/internal/v2/prognosis/{PLANT}/planner/trial/arm',headers={'X-Enelix-Service-Token':TOKEN},json=self.r) self.assertEqual(r.status_code,409);self.assertIn('allowlist',r.text) self.assertIsNone(c.get(f'/internal/v2/prognosis/{PLANT}/planner',headers={'X-Enelix-Service-Token':TOKEN}).json()['controlledTrial']) def test_trial_allowlist_cannot_expand_installation_access(self): with tempfile.TemporaryDirectory() as td,self.assertRaises(ValueError): create_app(Path(td)/'test.sqlite',TOKEN,[PLANT],start_worker=False,controlled_trial_plants=[SESSION]) if __name__=='__main__':unittest.main()