import unittest

from pandas import Timestamp
from feldera import PipelineBuilder
from tests import TEST_CLIENT, unique_pipeline_name
from feldera.runtime_config import RuntimeConfig
from feldera.testutils import FELDERA_TEST_NUM_WORKERS, FELDERA_TEST_NUM_HOSTS


class TestIssue_4457(unittest.TestCase):
    def test_local(self):
        sql = """
        CREATE TABLE test_events (
            id VARCHAR NOT NULL PRIMARY KEY,
            a VARCHAR,
            t TIMESTAMP NOT NULL LATENESS INTERVAL 1 MINUTE
        );
        CREATE VIEW V AS SELECT * FROM test_events;
        """

        # Make sure the pipelines are unique if we have concurrent runs
        pipeline = PipelineBuilder(
            TEST_CLIENT,
            name=unique_pipeline_name("test_issue4457"),
            sql=sql,
            runtime_config=RuntimeConfig(
                workers=FELDERA_TEST_NUM_WORKERS,
                hosts=FELDERA_TEST_NUM_HOSTS,
            ),
        ).create_or_replace()

        pipeline.start_paused()
        # TODO: use .query() instead
        out = pipeline.listen("v")

        pipeline.resume()

        pipeline.input_json(
            "test_events",
            [{"id": "a", "a": "test4", "t": "2025-05-20 21:00:17.920"}],
        )
        pipeline.wait_for_idle()

        output = out.to_dict()
        assert output == [
            {
                "id": "a",
                "a": "test4",
                "t": Timestamp("2025-05-20 21:00:17.920"),
                "insert_delete": 1,
            }
        ]

        pipeline.input_json(
            "test_events",
            [{"id": "a", "a": "test5", "t": "2025-03-20 21:00:17.920"}],
        )
        pipeline.wait_for_idle()

        output = out.to_dict()
        assert output == []

        pipeline.stop(force=True)


if __name__ == "__main__":
    unittest.main()
