dh
74a2d6c685
Tests / test (push) Successful in 1m1s
Approved by Daniel Haefliger for develop and beta. Author dh_Agent, authenticated account dh. Preserve published battery, charging and overall Energy Pie changes. No deployment or plant control authorization.
62 lines
2.8 KiB
Python
62 lines
2.8 KiB
Python
"""Cross-language contract for the ordinary Manager's generated measurement map."""
|
|
from hashlib import sha256
|
|
import json
|
|
from pathlib import Path
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import unittest
|
|
|
|
from netplan_v4 import measurement_pipeline as pipeline
|
|
|
|
|
|
@unittest.skipUnless(shutil.which('php'), 'Native PHP CLI required for capture contract')
|
|
class NativeMeasurementSetupTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
fixture = Path(__file__).resolve().parents[3] / 'tests' / 'V4Setup' / 'checks.php'
|
|
result = subprocess.run(['php', str(fixture), '--fixture'], check=True, capture_output=True, text=True)
|
|
if result.stderr:
|
|
raise AssertionError(result.stderr)
|
|
cls.fixture = json.loads(result.stdout)
|
|
|
|
def test_dataset_matches_backend_and_exact_php_capture_hash(self):
|
|
setup = self.fixture['setup']
|
|
config = pipeline.validate_config(setup['dataset'])
|
|
self.assertEqual(config['mappingSha256'], sha256(self.fixture['captureJson'].encode()).hexdigest())
|
|
self.assertEqual(config['mappingSha256'], self.fixture['record']['mappingSha256'])
|
|
self.assertEqual(config['inventorySha256'], self.fixture['record']['reportedInventorySha256'])
|
|
self.assertEqual(config['datasetId'], setup['datasetId'])
|
|
self.assertEqual(len(config['sources']), 3)
|
|
|
|
def test_capture_ingests_without_virtual_double_count_or_permissions(self):
|
|
setup = self.fixture['setup']
|
|
record = self.fixture['record']
|
|
config = setup['dataset']
|
|
plant = record['installationId']
|
|
now = pipeline.epoch(record['capturedAt'])
|
|
con = sqlite3.connect(':memory:')
|
|
try:
|
|
pipeline.schema(con)
|
|
registered = pipeline.register_dataset(con, plant, config, now)
|
|
self.assertFalse(registered['controlEnabled'])
|
|
receipt = pipeline.ingest_batch(con, plant, {'version': 1, 'datasetId': setup['datasetId'], 'records': [record]}, now)
|
|
self.assertEqual(receipt['stored'], 1)
|
|
self.assertFalse(receipt['controlEnabled'])
|
|
values = {key: source['value'] for key, source in record['raw'].items()}
|
|
self.assertEqual(pipeline.physical_value(values, config), 4000)
|
|
self.assertFalse(record['trainingEligible'])
|
|
self.assertFalse(record['measurementBoundaryVerified'])
|
|
changed = json.loads(json.dumps(config))
|
|
changed['sources'][0]['factorToW'] *= -1
|
|
with self.assertRaises(ValueError):
|
|
pipeline.register_dataset(con, plant, changed, now)
|
|
with self.assertRaises(ValueError):
|
|
pipeline.ingest_batch(con, 'other-installation', {'version': 1, 'datasetId': setup['datasetId'], 'records': [record]}, now)
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|