from argparse import Namespace

import pytest
from laplace_spark.app.smartstore_src_to_log0 import SparkAppSmartstoreSrcToLog0
from laplace_spark.constants import DATE_ID_COLUMN_NAME
from laplace_spark.modules.provider import Provider
from laplace_spark.modules.utils.laplace_utils import LaplaceUtils
from tests.utils import recursive_delete_s3_key


@pytest.fixture()
def spark_app_smartstore_src_to_log0(spark_session):
    yield SparkAppSmartstoreSrcToLog0(spark_session=spark_session)


@pytest.fixture()
def data_category():
    yield "order"


@pytest.fixture()
def mall_id():
    yield "dummy_mall_id"


@pytest.fixture()
def login_type():
    yield "NAVER"


@pytest.fixture()
def mall_name():
    yield "dummy_mall_name"


@pytest.fixture()
def args(data_category, mall_id, login_type, mall_name):
    yield [
        "--data-category",
        data_category,
        "--mall-id",
        mall_id,
        "--login-type",
        login_type,
        "--mall-name",
        mall_name,
    ]


@pytest.fixture()
def args_namespace(data_category, mall_id, login_type, mall_name):
    yield Namespace(
        data_category=data_category,
        mall_id=mall_id,
        login_type=login_type,
        mall_name=mall_name,
    )


@pytest.fixture()
def data_set_key(mall_id, login_type, mall_name):
    laplace_utils = LaplaceUtils(provider=Provider.SMARTSTORE.value)
    yield laplace_utils.hash_creator(
        {
            "mall_id": mall_id,
            "login_type": login_type,
            "mall_name": mall_name,
        }
    )


@pytest.fixture()
def smartstore_sourcing_delta_path(
    laplace_dashboard_bucket_name,
    df_smartstore_sourcing,
    s3,
):
    key = "dummy"
    path = f"s3a://{laplace_dashboard_bucket_name}/{key}"
    df_smartstore_sourcing.write.format("delta").save(path)
    yield path

    recursive_delete_s3_key(s3, laplace_dashboard_bucket_name, key)


@pytest.fixture()
def smartstore_log0_different_schema_table_path(
    laplace_dashboard_bucket_name,
    df_smartstore_sourcing,
    s3,
):
    df_no_date_id = df_smartstore_sourcing.drop(DATE_ID_COLUMN_NAME)
    key = "dummy"
    path = f"s3a://{laplace_dashboard_bucket_name}/{key}"
    df_no_date_id.write.format("delta").save(path)
    yield path

    recursive_delete_s3_key(s3, laplace_dashboard_bucket_name, key)


class TestClassSparkAppSmartstoreSrcToLog0:
    def test_get_arg_parser_success(
        self,
        spark_app_smartstore_src_to_log0,
        args,
        data_category,
        mall_id,
        login_type,
        mall_name,
    ):
        arg_parser = spark_app_smartstore_src_to_log0.get_arg_parser()
        parsed = arg_parser.parse_args(args)
        assert parsed.data_category == data_category
        assert parsed.mall_id == mall_id
        assert parsed.login_type == login_type
        assert parsed.mall_name == mall_name

    def test_get_path_prefix_success(
        self,
        spark_app_smartstore_src_to_log0,
        mall_id,
        login_type,
        mall_name,
        data_category,
        data_set_key,
    ):
        path_prefix = spark_app_smartstore_src_to_log0.get_path_prefix(
            mall_id=mall_id,
            login_type=login_type,
            mall_name=mall_name,
            data_category=data_category,
        )

        assert path_prefix == (
            "s3a://laplace-dashboard"
            f"/{Provider.SMARTSTORE.value}/{data_set_key}/{data_category}"
        )

    def test_get_src_path_success(
        self, spark_app_smartstore_src_to_log0, args_namespace, data_set_key
    ):
        src_path = spark_app_smartstore_src_to_log0.get_src_path(args_namespace)
        assert src_path == (
            "s3a://laplace-dashboard"
            f"/{Provider.SMARTSTORE.value}/{data_set_key}/{args_namespace.data_category}/sourcing"
        )

    def test_get_dest_path_success(
        self,
        spark_app_smartstore_src_to_log0,
        args_namespace,
        data_set_key,
    ):
        dest_path = spark_app_smartstore_src_to_log0.get_dest_path(args_namespace)
        assert dest_path == (
            "s3a://laplace-dashboard"
            f"/{Provider.SMARTSTORE.value}/{data_set_key}/{args_namespace.data_category}/log0"
        )

    def test_read_success(
        self,
        spark_app_smartstore_src_to_log0,
        smartstore_sourcing_delta_path,
        df_smartstore_sourcing,
    ):
        df = spark_app_smartstore_src_to_log0.read(smartstore_sourcing_delta_path)
        assert df.schema == df_smartstore_sourcing.schema
        assert df.collect() == df_smartstore_sourcing.collect()

    def test_write_success(
        self,
        spark_app_smartstore_src_to_log0,
        df_smartstore_sourcing,
        laplace_dashboard_bucket_name,
        spark_session,
    ):
        path = f"s3a://{laplace_dashboard_bucket_name}/"
        spark_app_smartstore_src_to_log0.write(path, df_smartstore_sourcing)
        df = spark_session.read.format("delta").load(path)

        assert df.schema == df_smartstore_sourcing.schema
        collected_from_s3 = df.collect()
        collected_original = df_smartstore_sourcing.collect()
        assert all(row in collected_original for row in collected_from_s3)
        assert all(row in collected_from_s3 for row in collected_original)

    def test_write_success_with_different_schema_existing(
        self,
        spark_app_smartstore_src_to_log0,
        df_smartstore_sourcing,
        smartstore_log0_different_schema_table_path,
        spark_session,
    ):
        spark_app_smartstore_src_to_log0.write(
            smartstore_log0_different_schema_table_path,
            df_smartstore_sourcing,
        )
        df = spark_session.read.format("delta").load(
            smartstore_log0_different_schema_table_path,
        )
        assert df.schema == df_smartstore_sourcing.schema
        collected_from_s3 = df.collect()
        collected_original = df_smartstore_sourcing.collect()
        assert all(row in collected_original for row in collected_from_s3)
        assert all(row in collected_from_s3 for row in collected_original)
