ws50529
364 lines
12 KiB
Python
364 lines
12 KiB
Python
import asyncio
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.responses import JSONResponse
|
|
|
|
import dea.csv_processing as csv_processing
|
|
from dea.csv_processing import PRESET_FILE_ID
|
|
from dea_api import (
|
|
BodeCascadeParams,
|
|
BodeCompareParams,
|
|
BodeParams,
|
|
DesignParams,
|
|
MCUWriteParams,
|
|
add_security_headers,
|
|
calculate_bode,
|
|
calculate_bode_compare,
|
|
calculate_bode_compare_cascade,
|
|
design_filter,
|
|
filter_csv,
|
|
filter_csv_download,
|
|
preset_csv,
|
|
upload_csv,
|
|
write_mcu_command,
|
|
)
|
|
|
|
|
|
class InMemoryUpload:
|
|
def __init__(self, content, filename):
|
|
self.content = content
|
|
self.filename = filename
|
|
|
|
async def read(self, _size=None):
|
|
return self.content
|
|
|
|
|
|
class DeaApiTest(unittest.TestCase):
|
|
async def upload_and_filter_csv(self, upload, b="1", a="1", col_idx=0, **kwargs):
|
|
upload_body = await upload_csv(upload)
|
|
self.assertIn("file_id", upload_body)
|
|
params = {
|
|
"b_int": None,
|
|
"a_int": None,
|
|
"shift_in": 14,
|
|
"shift_out": 14,
|
|
"shift_b": 14,
|
|
"shift_a": 14,
|
|
"use_round": False,
|
|
}
|
|
params.update(kwargs)
|
|
return await filter_csv(
|
|
file_id=upload_body["file_id"],
|
|
b=b,
|
|
a=a,
|
|
col_idx=col_idx,
|
|
**params,
|
|
)
|
|
|
|
def test_design_returns_normalized_coefficients_for_default_lowpass(self):
|
|
body = design_filter(
|
|
DesignParams(
|
|
filter_type="Lowpass (低通)",
|
|
fs=100000,
|
|
lp_fc=1000,
|
|
lp_order=1,
|
|
)
|
|
)
|
|
|
|
self.assertGreater(len(body["b"]), 0)
|
|
self.assertGreater(len(body["a"]), 0)
|
|
self.assertAlmostEqual(body["a"][0], 1.0)
|
|
|
|
def test_design_rejects_cutoff_at_or_above_nyquist(self):
|
|
try:
|
|
design_filter(
|
|
DesignParams(
|
|
filter_type="Lowpass (低通)",
|
|
fs=1000,
|
|
lp_fc=500,
|
|
lp_order=1,
|
|
)
|
|
)
|
|
except HTTPException as exc:
|
|
self.assertEqual(exc.status_code, 400)
|
|
self.assertIn("Nyquist", exc.detail)
|
|
else:
|
|
raise AssertionError("Expected Nyquist validation to fail")
|
|
|
|
def test_design_supports_2p1z_filter(self):
|
|
body = design_filter(
|
|
DesignParams(
|
|
filter_type="2P1Z (二極一零)",
|
|
fs=100000,
|
|
tp1z_fz=200,
|
|
tp1z_fp1=10,
|
|
tp1z_fp2=5000,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(len(body["b"]), 3)
|
|
self.assertEqual(len(body["a"]), 3)
|
|
self.assertAlmostEqual(body["a"][0], 1.0)
|
|
|
|
|
|
def test_bode_rejects_zero_a0(self):
|
|
try:
|
|
calculate_bode(BodeParams(fs=1000, b=[1.0], a=[0.0, 1.0]))
|
|
except HTTPException as exc:
|
|
self.assertEqual(exc.status_code, 400)
|
|
self.assertIn("a[0]", exc.detail)
|
|
else:
|
|
raise AssertionError("Expected a[0] validation to fail")
|
|
|
|
|
|
def test_bode_compare_reuses_one_frequency_axis(self):
|
|
body = calculate_bode_compare(
|
|
BodeCompareParams(
|
|
ideal=BodeParams(fs=1000, b=[1.0], a=[1.0]),
|
|
fixed=BodeParams(fs=1000, b=[0.5, 0.5], a=[1.0]),
|
|
)
|
|
)
|
|
|
|
self.assertEqual(len(body["freq"]), len(body["ideal"]["mag"]))
|
|
self.assertEqual(len(body["freq"]), len(body["fixed"]["mag"]))
|
|
self.assertEqual(len(body["ideal"]["phase"]), len(body["fixed"]["phase"]))
|
|
|
|
def test_bode_compare_cascade_combines_active_stages(self):
|
|
body = calculate_bode_compare_cascade(
|
|
BodeCascadeParams(
|
|
fs=1000,
|
|
stages=[
|
|
{
|
|
"b": [1.0],
|
|
"a": [1.0],
|
|
"b_fixed": [1.0],
|
|
"a_fixed": [1.0],
|
|
"isActive": True,
|
|
},
|
|
{
|
|
"b": [0.5, 0.5],
|
|
"a": [1.0],
|
|
"b_fixed": [0.5, 0.5],
|
|
"a_fixed": [1.0],
|
|
"isActive": True,
|
|
},
|
|
],
|
|
)
|
|
)
|
|
|
|
self.assertEqual(len(body["freq"]), len(body["ideal"]["mag"]))
|
|
self.assertEqual(len(body["freq"]), len(body["fixed"]["mag"]))
|
|
self.assertEqual(len(body["ideal"]["phase"]), len(body["fixed"]["phase"]))
|
|
|
|
def test_mcu_write_rejects_invalid_command_format(self):
|
|
try:
|
|
write_mcu_command(MCUWriteParams(command="hello"))
|
|
except HTTPException as exc:
|
|
self.assertEqual(exc.status_code, 400)
|
|
self.assertIn("MCU", exc.detail)
|
|
else:
|
|
raise AssertionError("Expected invalid MCU command validation to fail")
|
|
|
|
|
|
|
|
def test_preset_csv_endpoint_returns_local_ignored_file_metadata(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
preset_path = Path(tmp) / "preset_signals.csv"
|
|
preset_path.write_text("time,value\n0,1\n1,2\n", encoding="utf-8")
|
|
|
|
with patch.object(csv_processing, "PRESET_CSV_PATH", str(preset_path)):
|
|
body = preset_csv()
|
|
|
|
self.assertTrue(body["available"])
|
|
self.assertEqual(body["file_id"], PRESET_FILE_ID)
|
|
self.assertEqual(body["columns"], ["time", "value"])
|
|
self.assertEqual(body["default_col_idx"], 1)
|
|
self.assertEqual(body["rows"], 2)
|
|
|
|
def test_filter_can_use_preset_file_id_and_skip_blank_signal_cells(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
preset_path = Path(tmp) / "preset_signals.csv"
|
|
preset_path.write_text("time,value\n0,1\n1,\n2,3\n", encoding="utf-8")
|
|
|
|
with patch.object(csv_processing, "PRESET_CSV_PATH", str(preset_path)):
|
|
body = asyncio.run(
|
|
filter_csv(
|
|
file_id=PRESET_FILE_ID,
|
|
b="1",
|
|
a="1",
|
|
col_idx=1,
|
|
b_int=None,
|
|
a_int=None,
|
|
shift_in=14,
|
|
shift_out=14,
|
|
shift_b=14,
|
|
shift_a=14,
|
|
use_round=False,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(body["total_points"], 2)
|
|
self.assertEqual(body["index"], [0, 2])
|
|
self.assertEqual(body["original"], [1.0, 3.0])
|
|
|
|
def test_filter_downsamples_plot_response_for_large_csv(self):
|
|
rows = ["value"] + [str(i) for i in range(6001)]
|
|
upload = InMemoryUpload(("\n".join(rows) + "\n").encode("utf-8"), "input.csv")
|
|
|
|
body = asyncio.run(self.upload_and_filter_csv(upload, b="1", a="1", col_idx=0))
|
|
|
|
self.assertEqual(body["total_points"], 6001)
|
|
self.assertLessEqual(body["plot_points"], 5000)
|
|
self.assertGreater(body["downsample_step"], 1)
|
|
|
|
|
|
def test_filter_rejects_non_numeric_signal_column(self):
|
|
upload = InMemoryUpload(b"value\n1\nbad\n3\n", "input.csv")
|
|
|
|
try:
|
|
asyncio.run(self.upload_and_filter_csv(upload, b="1", a="1", col_idx=0))
|
|
except HTTPException as exc:
|
|
self.assertEqual(exc.status_code, 400)
|
|
self.assertIn("非數值", exc.detail)
|
|
else:
|
|
raise AssertionError("Expected non-numeric column validation to fail")
|
|
|
|
def test_filter_accepts_quoted_csv_fields_and_filters_selected_column(self):
|
|
upload = InMemoryUpload('label,value\n"a,1",1\n"a,2",2\n'.encode("utf-8"), "input.csv")
|
|
|
|
body = asyncio.run(self.upload_and_filter_csv(upload, b="1", a="1", col_idx=1))
|
|
|
|
self.assertEqual(body["col_name"], "value")
|
|
self.assertEqual(body["original"], [1.0, 2.0])
|
|
self.assertEqual(body["filtered"], [1.0, 2.0])
|
|
|
|
def test_filter_preview_accepts_index_window_for_lod_zoom(self):
|
|
upload = InMemoryUpload(b"value\n0\n1\n2\n3\n4\n", "input.csv")
|
|
|
|
body = asyncio.run(
|
|
self.upload_and_filter_csv(
|
|
upload,
|
|
b="1",
|
|
a="1",
|
|
col_idx=0,
|
|
start_idx=1,
|
|
end_idx=4,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(body["total_points"], 5)
|
|
self.assertEqual(body["index"], [1, 2, 3])
|
|
self.assertEqual(body["original"], [1.0, 2.0, 3.0])
|
|
|
|
def test_filter_accepts_cascade_stages_payload(self):
|
|
upload = InMemoryUpload(b"value\n1\n2\n3\n", "input.csv")
|
|
stages = [
|
|
{
|
|
"b_str": "1",
|
|
"a_str": "1",
|
|
"b_int_str": "16384",
|
|
"a_int_str": "16384",
|
|
"shift_in": 14,
|
|
"shift_out": 14,
|
|
"shift_b": 14,
|
|
"shift_a": 14,
|
|
"use_round": False,
|
|
"isActive": True,
|
|
},
|
|
{
|
|
"b_str": "1",
|
|
"a_str": "1",
|
|
"b_int_str": "16384",
|
|
"a_int_str": "16384",
|
|
"shift_in": 14,
|
|
"shift_out": 14,
|
|
"shift_b": 14,
|
|
"shift_a": 14,
|
|
"use_round": False,
|
|
"isActive": True,
|
|
},
|
|
]
|
|
|
|
body = asyncio.run(
|
|
self.upload_and_filter_csv(
|
|
upload,
|
|
b="1",
|
|
a="1",
|
|
col_idx=0,
|
|
stages=json.dumps(stages),
|
|
)
|
|
)
|
|
|
|
self.assertEqual(body["original"], [1.0, 2.0, 3.0])
|
|
self.assertEqual(body["filtered"], [1.0, 2.0, 3.0])
|
|
self.assertEqual(body["filtered_fixed"], [1.0, 2.0, 3.0])
|
|
|
|
def test_filter_download_can_export_compact_four_column_csv(self):
|
|
async def run_case():
|
|
upload = InMemoryUpload(b"value\n1\n2\n", "input.csv")
|
|
upload_body = await upload_csv(upload)
|
|
response = await filter_csv_download(
|
|
file_id=upload_body["file_id"],
|
|
b="1",
|
|
a="1",
|
|
col_idx=0,
|
|
fs=1000.0,
|
|
compact=True,
|
|
)
|
|
chunks = []
|
|
async for chunk in response.body_iterator:
|
|
chunks.append(chunk)
|
|
return b"".join(chunk if isinstance(chunk, bytes) else chunk.encode() for chunk in chunks).decode()
|
|
|
|
csv_text = asyncio.run(run_case())
|
|
|
|
self.assertTrue(csv_text.startswith("Time (s),value,value_filtered_ideal,value_filtered_fixed\n"))
|
|
self.assertIn("0.0,1.0,1.0,1.0", csv_text)
|
|
self.assertIn("0.001,2.0,2.0,2.0", csv_text)
|
|
|
|
def test_filter_rejects_non_csv_filename(self):
|
|
upload = InMemoryUpload(b"value\n1\n", "input.txt")
|
|
|
|
try:
|
|
asyncio.run(upload_csv(upload))
|
|
except HTTPException as exc:
|
|
self.assertEqual(exc.status_code, 400)
|
|
self.assertIn("CSV", exc.detail)
|
|
else:
|
|
raise AssertionError("Expected non-CSV upload validation to fail")
|
|
|
|
def test_filter_rejects_empty_csv_upload(self):
|
|
upload = InMemoryUpload(b" \n", "input.csv")
|
|
|
|
try:
|
|
asyncio.run(upload_csv(upload))
|
|
except HTTPException as exc:
|
|
self.assertEqual(exc.status_code, 400)
|
|
self.assertIn("CSV", exc.detail)
|
|
else:
|
|
raise AssertionError("Expected empty CSV validation to fail")
|
|
|
|
def test_filter_rejects_infinite_signal_values(self):
|
|
upload = InMemoryUpload(b"value\n1\ninf\n3\n", "input.csv")
|
|
|
|
try:
|
|
asyncio.run(self.upload_and_filter_csv(upload, b="1", a="1", col_idx=0))
|
|
except HTTPException as exc:
|
|
self.assertEqual(exc.status_code, 400)
|
|
self.assertIn("有限", exc.detail)
|
|
else:
|
|
raise AssertionError("Expected infinite signal validation to fail")
|
|
|
|
def test_security_headers_are_applied(self):
|
|
response = add_security_headers(JSONResponse({"ok": True}))
|
|
|
|
self.assertEqual(response.headers["X-Content-Type-Options"], "nosniff")
|
|
self.assertEqual(response.headers["X-Frame-Options"], "DENY")
|
|
self.assertIn("frame-ancestors 'none'", response.headers["Content-Security-Policy"])
|
|
self.assertNotIn("unsafe-eval", response.headers["Content-Security-Policy"])
|