/*
/* Copyright 2018-2026 contributors to the OpenLineage project
/* SPDX-License-Identifier: Apache-2.0
*/
package io.openlineage.spark.api.naming;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import io.openlineage.spark.api.OpenLineageContext;
import java.util.Optional;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

class AwsGlueApplicationJobNameProviderTest {
  private AwsGlueApplicationJobNameProvider provider;

  private OpenLineageContext noSparkContextOlContext;
  private OpenLineageContext noGluePropertiesOlContext;
  private OpenLineageContext autogeneratedGlueJobNameOlContext1;
  private OpenLineageContext autogeneratedGlueJobNameOlContext2;
  private OpenLineageContext userProvidedJobNameOlContext1;
  private OpenLineageContext userProvidedJobNameOlContext2;

  @BeforeEach
  void setUp() {
    provider = new AwsGlueApplicationJobNameProvider();

    // Case 1: OpenLineage has no SparkContext set
    noSparkContextOlContext = mock(OpenLineageContext.class);
    when(noSparkContextOlContext.getSparkContext()).thenReturn(Optional.empty());

    // Case 2: No Glue variables in the context
    noGluePropertiesOlContext = buildOpenLineageContextWithEmptyConf();

    // Case 3: With Glue variables and autogenerated job name
    autogeneratedGlueJobNameOlContext1 =
        buildOpenLineageContextWithGlueVariables("jr_11111111111111111111", "Some job 1");
    autogeneratedGlueJobNameOlContext2 =
        buildOpenLineageContextWithGlueVariables(
            "jr_22222222222222222222", "Some job 2 cross-check");

    // Case 4: With Glue variables and user provided job name
    userProvidedJobNameOlContext1 =
        buildOpenLineageContextWithGlueVariables(
            "jr_33333333333333333333", "Some job 3", "User provided name");
    userProvidedJobNameOlContext2 =
        buildOpenLineageContextWithGlueVariables(
            "jr_44444444444444444444", "Some job 4", "User provided name cross-check");
  }

  @Test
  void isDefinedAt() {
    assertThat(provider.isDefinedAt(noSparkContextOlContext)).isFalse();
    assertThat(provider.isDefinedAt(noGluePropertiesOlContext)).isFalse();
    assertThat(provider.isDefinedAt(autogeneratedGlueJobNameOlContext1)).isTrue();
    assertThat(provider.isDefinedAt(autogeneratedGlueJobNameOlContext2)).isTrue();
    assertThat(provider.isDefinedAt(userProvidedJobNameOlContext1)).isTrue();
    assertThat(provider.isDefinedAt(userProvidedJobNameOlContext2)).isTrue();
  }

  @Test
  void getJobName() {
    assertThat(provider.getJobName(autogeneratedGlueJobNameOlContext1)).isEqualTo("Some job 1");
    assertThat(provider.getJobName(autogeneratedGlueJobNameOlContext2))
        .isEqualTo("Some job 2 cross-check");
    assertThat(provider.getJobName(userProvidedJobNameOlContext1)).isEqualTo("User provided name");
    assertThat(provider.getJobName(userProvidedJobNameOlContext2))
        .isEqualTo("User provided name cross-check");
  }

  static OpenLineageContext buildOpenLineageContextWithEmptyConf() {
    OpenLineageContext openLineageContext = mock(OpenLineageContext.class);
    SparkContext sparkContext = mock(SparkContext.class);
    SparkConf conf = new SparkConf();
    when(sparkContext.getConf()).thenReturn(conf);
    when(openLineageContext.getSparkContext()).thenReturn(Optional.of(sparkContext));
    return openLineageContext;
  }

  static OpenLineageContext buildOpenLineageContextWithGlueVariables(
      String runId, String jobName, String sparkAppNameProperty) {
    OpenLineageContext openLineageContext = mock(OpenLineageContext.class);
    SparkContext sparkContext = mock(SparkContext.class);
    SparkConf conf = new SparkConf();
    conf.set("spark.glue.GLUE_VERSION", "4.0");
    conf.set("spark.glue.JOB_NAME", jobName);
    conf.set("spark.glue.JOB_RUN_ID", runId);
    when(sparkContext.getConf()).thenReturn(conf);
    when(sparkContext.appName()).thenReturn(sparkAppNameProperty);
    when(openLineageContext.getSparkContext()).thenReturn(Optional.of(sparkContext));
    return openLineageContext;
  }

  static OpenLineageContext buildOpenLineageContextWithGlueVariables(String runId, String jobName) {
    return buildOpenLineageContextWithGlueVariables(
        runId, jobName, String.format("nativespark-%s-%s", jobName, runId));
  }
}
