return {"username": credentials.username, "password": credentials.password}


client = TestClient(app)


def test_security_http_basic():
    response = client.get("/users/me", auth=("john", "secret"))
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "john", "password": "secret"}


def test_security_http_basic_no_credentials():

app = FastAPI()

security = HTTPBasic()


@app.get("/users/me")
def read_current_user(credentials: HTTPBasicCredentials = Depends(security)):

	opts.Port = 14223
	opts.Username = "testminio"
	opts.Password = "miniotest"
	s := natsserver.RunServer(&opts)
	defer s.Shutdown()

	clientConfig := &NATSArgs{
		Enable: true,
		Address: xnet.Host{
			Name:      "localhost",
			Port:      (xnet.Port(opts.Port)),
			IsPortSet: true,
		},
		Subject:  "test",
		Username: opts.Username,
		Password: opts.Password,
	}

	con, err := clientConfig.connectNats()

	Port               string    `json:"port"`     // default: 5432
	Username           string    `json:"username"` // default: user running minio
	Password           string    `json:"password"` // default: no password
	Database           string    `json:"database"` // default: same as user
	QueueDir           string    `json:"queueDir"`
	QueueLimit         uint64    `json:"queueLimit"`

    current_password_bytes = credentials.password.encode("utf8")
    correct_password_bytes = b"swordfish"
    is_correct_password = secrets.compare_digest(
        current_password_bytes, correct_password_bytes
    )
    if not (is_correct_username and is_correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",

ARGS:
broker*              (uri)       MQTT server endpoint e.g. `tcp://localhost:1883`
topic*               (string)    name of the MQTT topic to publish
username             (string)    MQTT username
password             (string)    MQTT password
qos                  (number)    set the quality of service priority, defaults to '0'
keep_alive_interval  (duration)  keep-alive interval for MQTT connections in s,m,h,d

    current_password_bytes = credentials.password.encode("utf8")
    correct_password_bytes = b"swordfish"
    is_correct_password = secrets.compare_digest(
        current_password_bytes, correct_password_bytes
    )
    if not (is_correct_username and is_correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",

        assertEquals("\\2a\\29\\28uid=\\2a\\29\\29\\28|\\28uid=\\2a", LdapUtil.escapeValue("*)(uid=*))(|(uid=*"));

        // Test complex injection: admin)(&(password=*))
        assertEquals("admin\\29\\28&\\28password=\\2a\\29\\29", LdapUtil.escapeValue("admin)(&(password=*))"));
    }

    @Test
    public void test_escapeValue_allSpecialCharsInSequence() {
        assertEquals("\\5c\\2a\\28\\29\\00", LdapUtil.escapeValue("\\*()\0"));

        assertEquals("********", SystemUtil.maskSensitiveValue("password", "secret"));
        assertEquals("********", SystemUtil.maskSensitiveValue("PASSWORD", "secret"));
        assertEquals("********", SystemUtil.maskSensitiveValue("Password", "secret"));
        assertEquals("********", SystemUtil.maskSensitiveValue("PaSsWoRd", "secret"));
    }

    @Test
    public void test_maskSensitiveValue_emptyValue() {

        TermQueryBuilder builder = regTermQ("password", password);
        if (opLambda != null) {
            opLambda.callback(builder);
        }
    }

    public void setPassword_NotEqual(String password) {
        setPassword_NotTerm(password, null);
    }

    public void setPassword_NotTerm(String password) {
        setPassword_NotTerm(password, null);