import asyncio import unittest from fastapi import HTTPException from fastapi.responses import JSONResponse from dea_api import ( BodeCompareParams, BodeParams, DesignParams, MCUWriteParams, add_security_headers, calculate_bode, calculate_bode_compare, design_filter, filter_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_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_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_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"])