"""Behavioral checks for TP02 contracts and edge cases. Run: python -m unittest -v."""
import contextlib
import io
import tempfile
import unittest
from pathlib import Path

import pandas as pd
from pandas.testing import assert_frame_equal
from solution import (BASE, COURSE, DEFAULT_INPUT, CSVReader, MissingValueStrategy,
    DropMissing, FillMean, FillMode, DataCleaner, NormalizeColumns,
    RemoveDuplicates, StandardizeText, TransformFactory, DataPipeline,
    CSVDataPipeline, log_action, log_time)


class CleaningTests(unittest.TestCase):
    def setUp(self):
        self.df = pd.read_csv(DEFAULT_INPUT)

    def test_drop_retains_only_complete_rows(self):
        self.assertEqual(DropMissing().handle(self.df)["id"].tolist(), [1, 5])

    def test_mean_values_and_input_immutability(self):
        before = self.df.copy(deep=True)
        result = DataCleaner(FillMean()).clean(self.df)
        self.assertEqual(result.loc[1, "age"], 30.25)
        self.assertEqual(result.loc[3, "height_cm"], 167)
        self.assertEqual(result.loc[1, "weight_kg"], 70)
        self.assertEqual(result.loc[2, "score"], 86)
        self.assertEqual(result.isna().sum().sum(), 0)
        assert_frame_equal(self.df, before)

    def test_mode_tie_rule(self):
        result = FillMode().handle(self.df)
        self.assertEqual(result.loc[1, "age"], 27)
        self.assertEqual(result.loc[3, "height_cm"], 160)
        self.assertEqual(result.loc[1, "weight_kg"], 55)
        self.assertEqual(result.loc[2, "score"], 75)

    def test_all_missing_column_warns_and_remains_missing(self):
        df = pd.DataFrame({"age": [float("nan")]})
        for strategy in (FillMean(), FillMode()):
            with self.assertLogs("tp02", level="WARNING"):
                self.assertTrue(strategy.handle(df)["age"].isna().all())

    def test_empty_frame_is_supported(self):
        df = self.df.iloc[:0]
        for strategy in (DropMissing(), FillMean(), FillMode()):
            self.assertTrue(strategy.handle(df).empty)

    def test_normalize_preserves_id_and_handles_constant(self):
        df = pd.DataFrame({"id": [4, 9], "x": [10, 20], "c": [7, 7]})
        result = NormalizeColumns().apply(df)
        self.assertEqual(result["id"].tolist(), [4, 9])
        self.assertEqual(result["x"].tolist(), [0.0, 1.0])
        self.assertEqual(result["c"].tolist(), [0.0, 0.0])

    def test_text_preserves_missing_and_deduplication_order(self):
        df = pd.DataFrame({"city": [" Paris ", "paris", None]})
        standardized = StandardizeText().apply(df)
        self.assertTrue(pd.isna(standardized.loc[2, "city"]))
        self.assertEqual(len(RemoveDuplicates().apply(standardized)), 2)
        self.assertEqual(df.loc[0, "city"], " Paris ")

    def test_factory_validation(self):
        self.assertIsInstance(TransformFactory.create("normalize"), NormalizeColumns)
        with self.assertRaisesRegex(ValueError, "Unknown transform"):
            TransformFactory.create("not_a_transform")

    def test_reader_lazy_preview_and_copy(self):
        reader = CSVReader(DEFAULT_INPUT)
        with contextlib.redirect_stdout(io.StringIO()):
            top = reader.preview(2)
        top.loc[0, "name"] = "changed"
        with contextlib.redirect_stdout(io.StringIO()):
            self.assertEqual(reader.preview(1).loc[0, "name"], "Alice")
        with self.assertRaises(ValueError):
            reader.preview(-1)

    def test_decorators_preserve_result_metadata_and_exception(self):
        @log_action
        @log_time
        def divide(a, b):
            return a / b
        self.assertEqual(divide.__name__, "divide")
        with self.assertLogs("tp02", level="INFO") as logged:
            self.assertEqual(divide(6, 2), 3)
        self.assertTrue(any("TIME" in line for line in logged.output))
        with self.assertLogs("tp02", level="INFO") as logged:
            with self.assertRaises(ZeroDivisionError):
                divide(1, 0)
        self.assertIn("FINISH", logged.output[-1])

    def test_template_call_order(self):
        calls = []
        class TracePipeline(DataPipeline):
            def load(self):
                calls.append("load"); return 1
            def clean(self, df):
                calls.append("clean"); return df + 1
            def transform(self, df):
                calls.append("transform"); return df * 2
            def save(self, df):
                calls.append("save"); self.result = df
        pipeline = TracePipeline()
        self.assertEqual(pipeline.run(), 4)
        self.assertEqual(calls, ["load", "clean", "transform", "save"])

    def test_pipeline_round_trip_and_write_boundary(self):
        with tempfile.TemporaryDirectory(dir=BASE) as folder:
            output = Path(folder) / "clean.csv"
            pipeline = CSVDataPipeline(DEFAULT_INPUT, output, FillMean(),
                [StandardizeText(), RemoveDuplicates(), NormalizeColumns()])
            result = pipeline.run()
            reread = pd.read_csv(output)
            self.assertEqual(reread.shape, (7, 7))
            self.assertEqual(reread.isna().sum().sum(), 0)
            self.assertEqual(result["id"].tolist(), list(range(1, 8)))
            self.assertEqual(reread.loc[0, "name"], "alice")
            self.assertAlmostEqual(reread.loc[1, "age"], 0.40625)
        with self.assertRaises(ValueError):
            CSVDataPipeline(DEFAULT_INPUT, COURSE / "unsafe.csv", FillMean(), [])

    def test_strategy_is_abstract(self):
        with self.assertRaises(TypeError):
            MissingValueStrategy()


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