import asyncio
import io
import unittest

from fastapi import HTTPException, UploadFile
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,
    write_mcu_command,
)


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