from __future__ import annotations

from conftest import write_basic_snapshot, write_text

from errors import ManifestError
from metadata import Column, Table, TableName
from snapshot.schema import SnapshotTable, parse_create_table_sql
from snapshot.tables import load_snapshot_tables
from storage.object_store import LocalObjectStore
from tests.pytest_compat import pytest


def test_discovers_snapshot_tables_from_dumpling_schema_sql(workspace):
    write_basic_snapshot(workspace)
    store = LocalObjectStore(str(workspace))

    snapshot_tables = load_snapshot_tables(store)

    assert [table.name for table in snapshot_tables.tables] == [TableName("test", "worker")]
    assert [table.quoted for table in snapshot_tables.tables] == ["`test`.`worker`"]


def test_discovers_snapshot_tables_under_custom_generation_prefix(workspace):
    # A snapshot generation lives under snapshot/generations/<gen>/ — discovery must honour
    # the configured snapshot_prefix and not the hard-coded default.
    gen_prefix = "snapshot/generations/gen_abc"
    write_text(
        workspace / f"{gen_prefix}/test.worker-schema.sql",
        "CREATE TABLE `worker` (\n  `id` bigint NOT NULL,\n  PRIMARY KEY (`id`)\n) ENGINE=InnoDB;\n",
    )
    store = LocalObjectStore(str(workspace))

    snapshot_tables = load_snapshot_tables(store, snapshot_prefix=gen_prefix)
    assert [t.source_table for t in snapshot_tables.tables] == ["worker"]

    # The default "snapshot" prefix has no files here, so discovery must fail there.
    with pytest.raises(ManifestError):
        load_snapshot_tables(store)


def test_snapshot_schema_parser_handles_clustered_primary_key_comment():
    table = parse_create_table_sql(
        """/*!40014 SET FOREIGN_KEY_CHECKS=0*/;
/*!40101 SET NAMES binary*/;
CREATE TABLE `bank0` (
  `id` bigint NOT NULL,
  `LAST_UDT_TMS` datetime(3) DEFAULT CURRENT_TIMESTAMP(3),
  `RSRV_GLS_ID` varchar(30) DEFAULT 'Z',
  PRIMARY KEY (`id`) /*T![clustered_index] CLUSTERED */
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
""",
        source_schema="test",
        source_table="bank0",
        snapshot_file_stem="test.bank0",
    )

    assert table.primary_key == ("id",)
    assert [(column.name, column.source_type, column.nullable) for column in table.columns] == [
        ("id", "BIGINT", False),
        ("LAST_UDT_TMS", "DATETIME(3)", True),
        ("RSRV_GLS_ID", "VARCHAR(30)", True),
    ]


def test_snapshot_schema_parser_applies_decimal_defaults():
    table = parse_create_table_sql(
        """CREATE TABLE `payment` (
  `id` bigint NOT NULL,
  `default_amount` decimal DEFAULT NULL,
  `whole_amount` decimal(20) DEFAULT NULL,
  `scaled_amount` decimal(20,6) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB;
""",
        source_schema="test",
        source_table="payment",
        snapshot_file_stem="test.payment",
    )

    assert [(column.name, column.precision, column.scale) for column in table.columns] == [
        ("id", None, None),
        ("default_amount", 10, 0),
        ("whole_amount", 20, 0),
        ("scaled_amount", 20, 6),
    ]


def test_snapshot_schema_parser_keeps_only_mapping_relevant_type_params():
    table = parse_create_table_sql(
        """CREATE TABLE `payment` (
  `flag` tinyint(1) DEFAULT NULL,
  `bitmap` bit(8) DEFAULT NULL,
  `amount` decimal(20,6) DEFAULT NULL,
  `name` varchar(64) DEFAULT NULL,
  `created_at` datetime(3) DEFAULT NULL
) ENGINE=InnoDB;
""",
        source_schema="test",
        source_table="payment",
        snapshot_file_stem="test.payment",
    )

    assert [(column.name, column.precision, column.scale) for column in table.columns] == [
        ("flag", 1, None),
        ("bitmap", 8, None),
        ("amount", 20, 6),
        ("name", None, None),
        ("created_at", None, None),
    ]


def test_snapshot_table_path_key_quotes_object_path_segments():
    table = SnapshotTable(
        table=Table(
            name=TableName("test/tenant", ".."),
            primary_key=("id",),
            columns=(Column("id", "BIGINT", nullable=False),),
        ),
        fivetran_table="test_tenant",
        snapshot_file_stem="test%2Ftenant.%2E%2E",
    )

    assert table.path_key == "test%2Ftenant/%2E%2E"