83 lines
6.1 KiB
Python
83 lines
6.1 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from datetime import datetime,timedelta,timezone
|
|
import unittest
|
|
ROOT=Path(__file__).resolve().parents[1]
|
|
def module(name,path):
|
|
spec=importlib.util.spec_from_file_location(name,path);value=importlib.util.module_from_spec(spec);spec.loader.exec_module(value);return value
|
|
publisher=module('publisher',ROOT/'integrations/netplan_v4_publisher.py')
|
|
hooks=module('hooks',ROOT/'install_hooks.py')
|
|
AT=datetime(2026,10,1,12,0,tzinfo=timezone.utc)
|
|
|
|
class PublisherTest(unittest.TestCase):
|
|
def test_tariff_null_preserved_and_mode_explicit(self):
|
|
cfg={'tarif_bezug_modus':'statisch','tarif_bezug':'custom','tarif_bezug_fest':0.,'tarif_einspeisung_modus':'dynamisch','tarif_einspeisung':'dynamic','tarif_peak_fest':0.}
|
|
p=publisher.tariff_payload(cfg,AT);self.assertEqual(p['import']['staticChfKwh'],0.);self.assertEqual(p['peakChfKwMonth'],0.)
|
|
del cfg['tarif_bezug_modus']
|
|
with self.assertRaises(ValueError):publisher.tariff_payload(cfg,AT)
|
|
def test_native_ckw_unit_zero_and_delivery_range(self):
|
|
rows=[{'start_timestamp':AT.isoformat(),'end_timestamp':(AT+timedelta(minutes=15)).isoformat(),'integrated':[{'unit':'CHF_kWh','value':0.}]}]
|
|
p=publisher.ckw_payload('CKW Dynamisch Home',rows,AT-timedelta(hours=1),AT)
|
|
self.assertEqual(p['periods'][0]['value'],0.);self.assertEqual(p['periods'][0]['unit'],'CHF/kWh');self.assertEqual(p['periods'][0]['sourceKind'],'published_interval')
|
|
del rows[0]['end_timestamp']
|
|
with self.assertRaises(ValueError):publisher.ckw_payload('CKW',rows,at=AT)
|
|
def test_unknown_unit_and_future_publication_rejected(self):
|
|
rows=[{'start_timestamp':AT.isoformat(),'end_timestamp':(AT+timedelta(minutes=15)).isoformat(),'integrated':[{'unit':'unknown','value':100.}]}]
|
|
with self.assertRaises(ValueError):publisher.ckw_payload('CKW',rows,at=AT)
|
|
rows[0]['integrated'][0]['unit']='CHF_kWh'
|
|
with self.assertRaises(ValueError):publisher.ckw_payload('CKW',rows,AT+timedelta(minutes=1),AT)
|
|
def test_forecast_not_relabelled_as_base_load(self):
|
|
p=publisher.forecast_payload([(23,{AT:5000.},{AT:2000.})],AT)
|
|
self.assertEqual(p['families']['23']['loadBasis'],'house_total');self.assertIsNone(p['families']['23']['trainedUntil'])
|
|
|
|
class InstallerTest(unittest.TestCase):
|
|
def fixtures(self,root):
|
|
files={'prognosis-manager-enelix2/api/main.py':'def f():\n _portal_store_configuration(anlagen_id, configuration)\n',
|
|
'prognosis-manager-enelix2/forecast_engine/main.py':'def f():\n for a in []:\n for key, default in []:\n d[key] = float(d.get(key) or default)\n for c in []:\n p_3 = v3.predict(data_obj, p_1, p_2)\n',
|
|
'prognosis-manager-enelix2/tariff_importer/main.py':'def f():\n for a in []:\n for b in []:\n for c in []:\n if True:\n return candidate_points\n',
|
|
'license/server.mjs':'const server = createServer(async (req, res) => {\n if (url.pathname === "/healthz" && req.method === "GET") return json(res, 200, {status:"ok"});\n});',
|
|
'license/public/app.js':'// Parallel GUI work. Must remain byte-identical.'}
|
|
for name,data in files.items():p=root/name;p.parent.mkdir(parents=True,exist_ok=True);p.write_text(data)
|
|
return files
|
|
def test_install_idempotent_and_verified_rollback(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root=Path(tmp);originals=self.fixtures(root);receipt=hooks.apply(hooks.plan(root),root)
|
|
self.assertFalse(hooks.plan(root));self.assertEqual((root/'license/public/app.js').read_text(),originals['license/public/app.js'])
|
|
hooks.rollback(receipt)
|
|
for name,data in originals.items():self.assertEqual((root/name).read_text(),data)
|
|
self.assertFalse((root/'license/public/netplan-v4.html').exists())
|
|
def test_concurrent_changes_never_overwritten(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root=Path(tmp);self.fixtures(root);entries=hooks.plan(root);p=root/'license/server.mjs';p.write_text(p.read_text()+'// Concurrent')
|
|
with self.assertRaises(ValueError):hooks.apply(entries,root)
|
|
self.assertTrue(p.read_text().endswith('// Concurrent'))
|
|
def test_later_changes_block_rollback(self):
|
|
with TemporaryDirectory() as tmp:
|
|
root=Path(tmp);self.fixtures(root);receipt=hooks.apply(hooks.plan(root),root);p=root/'license/server.mjs';p.write_text(p.read_text()+'// Later')
|
|
with self.assertRaises(ValueError):hooks.rollback(receipt)
|
|
self.assertTrue(p.read_text().endswith('// Later'))
|
|
def test_changed_or_ambiguous_anchor_fails_closed(self):
|
|
with self.assertRaises(ValueError):hooks.once('x x','x','y')
|
|
with self.assertRaises(ValueError):hooks.patch_api('def f():pass')
|
|
|
|
class HorizonValidationTest(unittest.TestCase):
|
|
def test_synthetic_48h_energy_soc_limits(self):
|
|
import math
|
|
from netplan_v4.domain import Battery,Step,Price,Limits
|
|
from netplan_v4.optimizer import optimize
|
|
steps=[]
|
|
for i in range(576):
|
|
solar=max(0.,math.sin(math.pi*((i%288)/12-6)/12))*10000.
|
|
steps.append(Step(AT+timedelta(minutes=5*i),3000.,solar,Price(.15 if i%288<72 else .35),Price(.1)))
|
|
b=Battery('synthetic',40.,20.,10.,90.,10000.,8000.,AT,grid_charging=True)
|
|
p=optimize(steps,[b],at=AT,limits=Limits(import_w=12000.,export_w=10000.),observed_peaks={'2026-10':8.},peak_prices={'2026-10':5.})
|
|
self.assertTrue(p['executable']);self.assertEqual(len(p['points']),576)
|
|
for s,x in zip(steps,p['points']):
|
|
self.assertAlmostEqual(x['gridTargetW'],s.residual_w+x['batteryTargetW']+x['pvCurtailmentW'],places=3)
|
|
self.assertLessEqual(x['gridTargetW'],12000.001);self.assertGreaterEqual(x['gridTargetW'],-10000.001)
|
|
self.assertTrue(9.999<=x['socEndPercent']['synthetic']<=90.001)
|
|
self.assertGreaterEqual(p['points'][-1]['socEndPercent']['synthetic'],19.999)
|
|
|
|
if __name__=='__main__':unittest.main()
|