85 lines
2.9 KiB
Python
85 lines
2.9 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_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_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"])
|