45 lines
2.8 KiB
Python
45 lines
2.8 KiB
Python
import copy
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from fastapi.testclient import TestClient
|
|
from netplan_v4.service import create_app, ingest, run_once
|
|
from test_v4 import inputs, AT
|
|
from test_controlled_trial import fixture, PLANT, SESSION, TOKEN
|
|
|
|
class ControlledTrialPipelineTest(unittest.TestCase):
|
|
def test_real_planner_to_internal_trial_api_and_revoke(self):
|
|
with tempfile.TemporaryDirectory() as td:
|
|
app=create_app(Path(td)/'trial.sqlite',TOKEN,[PLANT],start_worker=False,controlled_trial_plants=[PLANT])
|
|
store=app.state.store_factory()
|
|
op,fc,tar=inputs()
|
|
for f in fc['families'].values():f['accountingEvidenceId']='synthetic-accounting-review'
|
|
for k,v in [('operation',op),('forecast',fc),('tariffs',tar)]:ingest(store,PLANT,k,v,AT)
|
|
plan=run_once(store,AT);self.assertEqual(plan['status'],'optimal');store.close()
|
|
headers={'X-Enelix-Service-Token':TOKEN};prefix=f'/internal/v2/prognosis/{PLANT}/planner'
|
|
with patch('netplan_v4.service.datetime',wraps=datetime) as clock,TestClient(app) as c:
|
|
clock.now.return_value=AT
|
|
view=c.get(prefix,headers=headers).json()
|
|
self.assertIsNone(view['controlledTrial']);self.assertTrue(view['fresh'])
|
|
_,request=fixture();request.update(expectedPlanId=plan['planId'],expectedRevision=0,acceptEstimatedPeak=False)
|
|
grant=c.post(prefix+'/trial/arm',headers=headers,json=request)
|
|
self.assertEqual(grant.status_code,200,grant.text)
|
|
view=c.get(prefix,headers=headers).json()
|
|
self.assertEqual(view['controlledTrial']['sourceShadowPlanId'],plan['planId'])
|
|
self.assertFalse(view['liveEnabled']);self.assertFalse(view['plan']['liveEnabled'])
|
|
self.assertEqual(view['plan']['runMode'],'shadow')
|
|
self.assertEqual(c.post(prefix+'/ack',headers=headers,json={'planId':plan['planId'],'revision':0,'status':'applied'}).status_code,400)
|
|
self.assertEqual(c.post(prefix+'/trial/revoke',headers=headers,json={'sessionId':SESSION}).status_code,200)
|
|
self.assertIsNone(c.get(prefix,headers=headers).json()['controlledTrial'])
|
|
self.assertEqual(c.post(prefix+'/trial/arm',headers=headers,json=request).status_code,409)
|
|
def test_accounting_marker_cannot_label_aggregate_load_verified(self):
|
|
from netplan_v4.store import PlannerStore
|
|
s=PlannerStore(':memory:');op,fc,tar=inputs()
|
|
f=fc['families']['3'];f['loadBasis']='house_total';f['accountingEvidenceId']='synthetic-incorrect-label'
|
|
with self.assertRaises(ValueError):ingest(s,PLANT,'forecast',fc,AT)
|
|
s.close()
|
|
|
|
if __name__=='__main__':unittest.main()
|