154 lines
4.9 KiB
Python
154 lines
4.9 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
MODULE_PATH = Path(__file__).resolve().parents[1] / "core" / "identification.py"
|
|
|
|
|
|
def load_identification_module():
|
|
get_v = types.ModuleType("get_V")
|
|
get_v.measure_volume = lambda *args, **kwargs: None
|
|
ind_collector = types.ModuleType("ind_collector")
|
|
ind_collector.collect_data_with_prbs = lambda *args, **kwargs: {}
|
|
api = types.ModuleType("api")
|
|
api.base_url = "https://cloud.example"
|
|
api.data_record_url = "https://cloud.example/data_record"
|
|
api.the_folder = "customer-a"
|
|
requests = types.ModuleType("requests")
|
|
|
|
spec = importlib.util.spec_from_file_location(
|
|
"identification_under_test", MODULE_PATH
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
with patch.dict(sys.modules, {
|
|
"get_V": get_v,
|
|
"ind_collector": ind_collector,
|
|
"api": api,
|
|
"requests": requests,
|
|
}):
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
IDENTIFICATION = load_identification_module()
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self):
|
|
self.now = 0.0
|
|
|
|
def monotonic(self):
|
|
self.now += 0.001
|
|
return self.now
|
|
|
|
def sleep(self, duration):
|
|
self.now += max(0.0, duration)
|
|
|
|
|
|
class FakeConnectionManager:
|
|
def __init__(self):
|
|
self.distance = 0
|
|
|
|
def set_motor_position(self, distance):
|
|
self.distance = int(distance)
|
|
return True
|
|
|
|
def read_pressure(self):
|
|
return self.distance / 100.0
|
|
|
|
|
|
class InitialTravelScanTests(unittest.TestCase):
|
|
def test_uploads_distance_and_pressure_json_without_time_fields(self):
|
|
manager = IDENTIFICATION.IdentificationManager()
|
|
manager._identifying = True
|
|
captured = {}
|
|
|
|
def capture_upload(body, filename, folder):
|
|
captured.update(body=body, filename=filename, folder=folder)
|
|
return True
|
|
|
|
manager._upload_to_server = capture_upload
|
|
clock = FakeClock()
|
|
with patch.object(IDENTIFICATION.time, "monotonic", clock.monotonic), \
|
|
patch.object(IDENTIFICATION.time, "sleep", clock.sleep):
|
|
payload = manager._run_initial_travel_scan(
|
|
FakeConnectionManager()
|
|
)
|
|
|
|
records = payload["stable_pressures"]
|
|
expected_distances = list(range(1000, -1, -100))
|
|
self.assertEqual(
|
|
[record["distance"] for record in records], expected_distances
|
|
)
|
|
self.assertEqual(
|
|
[record["pressure"] for record in records],
|
|
[distance / 100.0 for distance in expected_distances],
|
|
)
|
|
self.assertTrue(captured["filename"].endswith(".json"))
|
|
self.assertEqual(captured["folder"], "customer-a/ind_data")
|
|
self.assertEqual(json.loads(captured["body"]), payload)
|
|
self.assertTrue(all(
|
|
set(record) == {"distance", "pressure"} for record in records
|
|
))
|
|
|
|
def test_identification_uploads_collector_csv_and_notifies_filename(self):
|
|
manager = IDENTIFICATION.IdentificationManager()
|
|
manager._run_initial_travel_scan = lambda conn_mgr: {}
|
|
uploaded = {}
|
|
callbacks = []
|
|
csv_data = b"t,u,p,q_in,V\n0.0,10.0,20.0,50.0,5.0\n"
|
|
csv_filename = "identification_data_test.csv"
|
|
|
|
manager._upload_to_server = lambda content, filename, folder: (
|
|
uploaded.update(
|
|
content=content, filename=filename, folder=folder
|
|
) or True
|
|
)
|
|
manager.set_identification_upload_callback(
|
|
lambda success, filename, error:
|
|
callbacks.append((success, filename, error))
|
|
)
|
|
|
|
class ConnectedManager:
|
|
def is_connected(self):
|
|
return True
|
|
|
|
collector_result = {
|
|
"success": True,
|
|
"csv_data": csv_data,
|
|
"filename": csv_filename,
|
|
}
|
|
with patch.object(
|
|
IDENTIFICATION, "collect_data_with_prbs",
|
|
return_value=collector_result):
|
|
started = manager.start_identification(
|
|
conn_mgr=ConnectedManager(),
|
|
running_flag_check=lambda: False,
|
|
q_in_val=50.0,
|
|
dt=0.1,
|
|
n_order=6,
|
|
t_c=2.5,
|
|
levels=[10, 20, 30, 40, 50, 60, 70, 80],
|
|
dead_area=240.0,
|
|
xa_full=1000.0,
|
|
V_val=5.0,
|
|
repeat=2,
|
|
)
|
|
manager._task_thread.join(timeout=2)
|
|
|
|
self.assertTrue(started)
|
|
self.assertFalse(manager._task_thread.is_alive())
|
|
self.assertEqual(uploaded["content"], csv_data)
|
|
self.assertEqual(uploaded["filename"], csv_filename)
|
|
self.assertEqual(uploaded["folder"], "customer-a/ind_data")
|
|
self.assertEqual(callbacks, [(True, csv_filename, None)])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|