Repository: apache/flink
File: flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/catalog/CatalogTableITCase.scala (Scala)

Code before:
    val tableDDL =
      """
        |create table t1(
        |  a bigint,
        |  b varchar,
        |  c int
        |) with (
        |  'connector' = 'values',
        |  'password' = 'Secret1',
        |  'secret' = 'Secret2',
        |  'fs.azure.account.key' = 'Secret3',
        |  'apikey' =  'Secret4',
        |  'api-key' = 'Secret5',
        |  'auth-params' = 'Secret6',
        |  'service-key' = 'Secret7',
        |  'token' = 'Secret8',
        |  'basic-auth' = 'Secret9',
        |  'jaas.config' = 'Secret10',
        |  'http-headers' = '{"Authorization": "Secret11"}'
        |)
      """.stripMargin

    tableEnv.executeSql(tableDDL)

    // Test source table validation
    try {
      tableEnv.executeSql("select * from t1")
      fail("Expected ValidationException for source table")
    } catch {
      case e: ValidationException =>
        val errorMsg = e.getMessage
        // Verify source table sensitive info is masked
        assert(!errorMsg.contains("Secret"))

        assert(errorMsg.contains("'password'='******'"))
        assert(errorMsg.contains("'secret'='******'"))
        assert(errorMsg.contains("'fs.azure.account.key'='******'"))
        assert(errorMsg.contains("'apikey'='******'"))
        assert(errorMsg.contains("'api-key'='******'"))
        assert(errorMsg.contains("'auth-params'='******'"))
        assert(errorMsg.contains("'service-key'='******'"))
        assert(errorMsg.contains("'token'='******'"))
        assert(errorMsg.contains("'basic-auth'='******'"))
        assert(errorMsg.contains("'jaas.config'='******'"))
        assert(errorMsg.contains("'http-headers'='******'"))
    }

    // Test sink table validation
    try {
      tableEnv.executeSql("insert into t1 select 1, 'test', 1")
      fail("Expected ValidationException for sink table")

Review comment (refactor):
Seems like the diff between the 2 try blocks the difference is the SQL statement and source/sink word. If that so we can just extract it into a function with 2 params, right?

Code after:
    val tableDDL =
      """
        |create table t1(
        |  a bigint,
        |  b varchar,
        |  c int
        |) with (
        |  'connector' = 'values',
        |  'password' = 'Secret1',
        |  'secret' = 'Secret2',
        |  'fs.azure.account.key' = 'Secret3',
        |  'apikey' =  'Secret4',
        |  'api-key' = 'Secret5',
        |  'auth-params' = 'Secret6',
        |  'service-key' = 'Secret7',
        |  'token' = 'Secret8',
        |  'basic-auth' = 'Secret9',
        |  'jaas.config' = 'Secret10',
        |  'http-headers' = '{"Authorization": "Secret11"}'
        |)
    """.stripMargin

    tableEnv.executeSql(tableDDL)

    // Extract a generic validation function
    def validateSensitiveInfoMasking(sql: String, tableType: String): Unit = {
      try {
        tableEnv.executeSql(sql)
        fail(s"Expected ValidationException for $tableType table")
      } catch {
        case e: ValidationException =>
          val errorMsg = e.getMessage
          // Verify sensitive information is masked
          assert(!errorMsg.contains("Secret"))

          // Verify all sensitive fields are replaced with '******'
          assert(errorMsg.contains("'password'='******'"))
          assert(errorMsg.contains("'secret'='******'"))
          assert(errorMsg.contains("'fs.azure.account.key'='******'"))
          assert(errorMsg.contains("'apikey'='******'"))
          assert(errorMsg.contains("'api-key'='******'"))
          assert(errorMsg.contains("'auth-params'='******'"))
          assert(errorMsg.contains("'service-key'='******'"))
          assert(errorMsg.contains("'token'='******'"))
          assert(errorMsg.contains("'basic-auth'='******'"))
          assert(errorMsg.contains("'jaas.config'='******'"))
          assert(errorMsg.contains("'http-headers'='******'"))
      }
    }

    // Test source table (SELECT)