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()