tcad-bodeplot/tests/test_dea_api.py

import asyncio
import json
import unittest

from fastapi import HTTPException
from fastapi.responses import JSONResponse

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,
    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_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"])