46 lines
2.7 KiB
Python
46 lines
2.7 KiB
Python
import copy
|
|
from datetime import timedelta
|
|
import tempfile
|
|
from pathlib import Path
|
|
import unittest
|
|
from netplan_v4.receiver_contract import control_context, provenance
|
|
from netplan_v4.store import PlannerStore
|
|
from netplan_v4.service import ingest, run_once, status
|
|
from test_v4 import inputs, AT, AID
|
|
|
|
class ReceiverContractTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.s=PlannerStore(str(Path(self.tmp.name)/'qa.sqlite'))
|
|
op,fc,tar=inputs()
|
|
for kind,value in zip(('operation','forecast','tariffs'),(op,fc,tar)):ingest(self.s,AID,kind,value,AT)
|
|
def tearDown(self):self.s.close();self.tmp.cleanup()
|
|
def test_published_plan_and_response_bound_to_same_plant(self):
|
|
p=run_once(self.s,AT);r=status(self.s,AID,AT)
|
|
self.assertEqual(p['installationId'],AID);self.assertEqual(r['installationId'],AID)
|
|
self.assertEqual(r['receiverProtocolVersion'],1);self.assertTrue(r['fresh'])
|
|
self.assertFalse(r['liveEnabled']);self.assertFalse(p['liveEnabled'])
|
|
self.assertEqual(p['inputRefs'],{'operation':'op1','forecast':'fc1','tariffs':'tar1'})
|
|
def test_control_context_tracks_stable_policy_not_volatile_soc(self):
|
|
op=inputs()[0];a=control_context(op);op['batteries'][0]['socPercent']=80.;op['gridW']=1000.
|
|
self.assertEqual(a,control_context(op));op['batteries'][0]['minSocPercent']=15.
|
|
self.assertNotEqual(a,control_context(op))
|
|
def test_context_is_snapshot_not_reference_to_future_changes(self):
|
|
op,fc,tar=inputs();p=provenance({'operation':op,'forecast':fc,'tariffs':tar})
|
|
op['limits']['importW']=100.;self.assertEqual(p['controlContext']['limits']['importW'],10000.)
|
|
def test_no_plan_for_another_installation(self):
|
|
run_once(self.s,AT);r=status(self.s,'00000000-0000-4000-8000-000000000002',AT)
|
|
self.assertIsNone(r['plan']);self.assertFalse(r['fresh'])
|
|
def test_shadow_ack_is_not_execution(self):
|
|
p=run_once(self.s,AT)
|
|
self.s.acknowledge(AID,p['planId'],p['configRevision'],AT,step=p['points'][0]['time'],status='shadow_seen')
|
|
self.assertEqual(status(self.s,AID,AT)['acknowledgement']['status'],'shadow_seen')
|
|
with self.assertRaises(ValueError):self.s.acknowledge(AID,p['planId'],p['configRevision'],AT,status='applied')
|
|
def test_new_configuration_invalidates_fresh_flag(self):
|
|
run_once(self.s,AT);self.s.save_settings(AID,{'family':'23'},0,AT+timedelta(seconds=1))
|
|
self.assertFalse(status(self.s,AID,AT+timedelta(seconds=2))['fresh'])
|
|
def test_rearm_default_matches_php_contract(self):
|
|
op=inputs()[0];op['batteries'][0]['rearmSocPercent']=None
|
|
self.assertEqual(control_context(op)['batteries']['b']['rearmSocPercent'],10.)
|
|
|
|
if __name__=='__main__':unittest.main()
|