Repository: apache/flink
File: flink-metrics/flink-metrics-prometheus/src/test/java/org/apache/flink/metrics/prometheus/PrometheusPushGatewayReporterTest.java (Java)

Code before:
        MetricConfig metricConfig = new MetricConfig();
        metricConfig.setProperty(HOST_URL.key(), "http://localhost:9091");
        metricConfig.setProperty(PASSWORD.key(), "flink-password");
        // Only password configured - should create reporter successfully without Basic Auth
        PrometheusPushGatewayReporter reporter = factory.createMetricReporter(metricConfig);
        assertThat(reporter).isNotNull();
        // Verify warning log is emitted
        assertThat(loggerExtension.getMessages())
                .anyMatch(
                        msg ->
                                msg.contains("Both username and password must be configured")
                                        && msg.contains("Currently only password is configured"));
    }

    @Test
    void testBasicAuthEnabledWithBothCredentials() {
        PrometheusPushGatewayReporterFactory factory = new PrometheusPushGatewayReporterFactory();
        MetricConfig metricConfig = new MetricConfig();
        metricConfig.setProperty(HOST_URL.key(), "http://localhost:9091");
        metricConfig.setProperty(USERNAME.key(), "flink-user");
        metricConfig.setProperty(PASSWORD.key(), "flink-password");
        // Both username and password configured - Basic Auth should be enabled
        PrometheusPushGatewayReporter reporter = factory.createMetricReporter(metricConfig);
        assertThat(reporter).isNotNull();
        assertThat(reporter.hostUrl.toString()).isEqualTo("http://localhost:9091");
    }
}


Review comment (security):
I think we need to add another assertion to this test to verify that authentication _was_ actually enabled. We may want to consider following a pattern similar to `hostUrl` to expose it for testing (e.g. @VisibleForTesting package-private boolean field), something like `basicAuthEnabled` and then we could simply check that property here:

```
assertThat(reporter.basicAuthEnabled).isTrue();
```

Either that or rely on a reflection based approach to inspect the underlying connection factory.

Code after:
    void testBasicAuthNotEnabledWithOnlyPassword() {
        PrometheusPushGatewayReporterFactory factory = new PrometheusPushGatewayReporterFactory();
        MetricConfig metricConfig = new MetricConfig();
        metricConfig.setProperty(HOST_URL.key(), "http://localhost:9091");
        metricConfig.setProperty(PASSWORD.key(), "flink-password");
        // Only password configured - should create reporter successfully without Basic Auth
        PrometheusPushGatewayReporter reporter = factory.createMetricReporter(metricConfig);
        assertThat(reporter).isNotNull();
        assertThat(reporter.basicAuthEnabled).isFalse();
        // Verify warning log is emitted
        assertThat(loggerExtension.getMessages())
                .anyMatch(
                        msg ->
                                msg.contains("Both username and password must be configured")
                                        && msg.contains("Currently only password is configured"));
    }

    @Test
    void testBasicAuthEnabledWithBothCredentials() {
        PrometheusPushGatewayReporterFactory factory = new PrometheusPushGatewayReporterFactory();
        MetricConfig metricConfig = new MetricConfig();
        metricConfig.setProperty(HOST_URL.key(), "http://localhost:9091");
        metricConfig.setProperty(USERNAME.key(), "flink-user");
        metricConfig.setProperty(PASSWORD.key(), "flink-password");
        // Both username and password configured - Basic Auth should be enabled
        PrometheusPushGatewayReporter reporter = factory.createMetricReporter(metricConfig);
        assertThat(reporter).isNotNull();
        assertThat(reporter.hostUrl.toString()).isEqualTo("http://localhost:9091");
        assertThat(reporter.basicAuthEnabled).isTrue();
    }
}