ws50529
145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
import asyncio
|
|
import io
|
|
import unittest
|
|
|
|
from fastapi import HTTPException, UploadFile
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from dea_api import BodeParams, DesignParams, add_security_headers, calculate_bode, design_filter, filter_csv
|
|
|
|
|
|
class DeaApiTest(unittest.TestCase):
|
|
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_filter_downsamples_plot_response_for_large_csv(self):
|
|
rows = ["value"] + [str(i) for i in range(6001)]
|
|
upload = UploadFile(
|
|
io.BytesIO(("\n".join(rows) + "\n").encode("utf-8")),
|
|
filename="input.csv",
|
|
)
|
|
|
|
body = asyncio.run(filter_csv(file=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 = UploadFile(io.BytesIO(b"value\n1\nbad\n3\n"), filename="input.csv")
|
|
|
|
try:
|
|
asyncio.run(filter_csv(file=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 = UploadFile(
|
|
io.BytesIO('label,value\n"a,1",1\n"a,2",2\n'.encode("utf-8")),
|
|
filename="input.csv",
|
|
)
|
|
|
|
body = asyncio.run(filter_csv(file=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_rejects_non_csv_filename(self):
|
|
upload = UploadFile(io.BytesIO(b"value\n1\n"), filename="input.txt")
|
|
|
|
try:
|
|
asyncio.run(filter_csv(file=upload, b="1", a="1", col_idx=0))
|
|
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 = UploadFile(io.BytesIO(b" \n"), filename="input.csv")
|
|
|
|
try:
|
|
asyncio.run(filter_csv(file=upload, b="1", a="1", col_idx=0))
|
|
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 = UploadFile(io.BytesIO(b"value\n1\ninf\n3\n"), filename="input.csv")
|
|
|
|
try:
|
|
asyncio.run(filter_csv(file=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"])
|