|
|
|
@@ -1,5 +1,6 @@ |
|
|
|
import os |
|
|
|
|
|
|
|
import pytest |
|
|
|
from flask import Flask |
|
|
|
from packaging.version import Version |
|
|
|
from yarl import URL |
|
|
|
@@ -137,3 +138,61 @@ def test_db_extras_options_merging(monkeypatch): |
|
|
|
options = engine_options["connect_args"]["options"] |
|
|
|
assert "search_path=myschema" in options |
|
|
|
assert "timezone=UTC" in options |
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize( |
|
|
|
("broker_url", "expected_host", "expected_port", "expected_username", "expected_password", "expected_db"), |
|
|
|
[ |
|
|
|
("redis://localhost:6379/1", "localhost", 6379, None, None, "1"), |
|
|
|
("redis://:password@localhost:6379/1", "localhost", 6379, None, "password", "1"), |
|
|
|
("redis://:mypass%23123@localhost:6379/1", "localhost", 6379, None, "mypass#123", "1"), |
|
|
|
("redis://user:pass%40word@redis-host:6380/2", "redis-host", 6380, "user", "pass@word", "2"), |
|
|
|
("redis://admin:complex%23pass%40word@127.0.0.1:6379/0", "127.0.0.1", 6379, "admin", "complex#pass@word", "0"), |
|
|
|
( |
|
|
|
"redis://user%40domain:secret%23123@redis.example.com:6380/3", |
|
|
|
"redis.example.com", |
|
|
|
6380, |
|
|
|
"user@domain", |
|
|
|
"secret#123", |
|
|
|
"3", |
|
|
|
), |
|
|
|
# Password containing %23 substring (double encoding scenario) |
|
|
|
("redis://:mypass%2523@localhost:6379/1", "localhost", 6379, None, "mypass%23", "1"), |
|
|
|
# Username and password both containing encoded characters |
|
|
|
("redis://user%2525%40:pass%2523@localhost:6379/1", "localhost", 6379, "user%25@", "pass%23", "1"), |
|
|
|
], |
|
|
|
) |
|
|
|
def test_celery_broker_url_with_special_chars_password( |
|
|
|
monkeypatch, broker_url, expected_host, expected_port, expected_username, expected_password, expected_db |
|
|
|
): |
|
|
|
"""Test that CELERY_BROKER_URL with various formats are handled correctly.""" |
|
|
|
from kombu.utils.url import parse_url |
|
|
|
|
|
|
|
# clear system environment variables |
|
|
|
os.environ.clear() |
|
|
|
|
|
|
|
# Set up basic required environment variables (following existing pattern) |
|
|
|
monkeypatch.setenv("CONSOLE_API_URL", "https://example.com") |
|
|
|
monkeypatch.setenv("CONSOLE_WEB_URL", "https://example.com") |
|
|
|
monkeypatch.setenv("DB_USERNAME", "postgres") |
|
|
|
monkeypatch.setenv("DB_PASSWORD", "postgres") |
|
|
|
monkeypatch.setenv("DB_HOST", "localhost") |
|
|
|
monkeypatch.setenv("DB_PORT", "5432") |
|
|
|
monkeypatch.setenv("DB_DATABASE", "dify") |
|
|
|
|
|
|
|
# Set the CELERY_BROKER_URL to test |
|
|
|
monkeypatch.setenv("CELERY_BROKER_URL", broker_url) |
|
|
|
|
|
|
|
# Create config and verify the URL is stored correctly |
|
|
|
config = DifyConfig() |
|
|
|
assert broker_url == config.CELERY_BROKER_URL |
|
|
|
|
|
|
|
# Test actual parsing behavior using kombu's parse_url (same as production) |
|
|
|
redis_config = parse_url(config.CELERY_BROKER_URL) |
|
|
|
|
|
|
|
# Verify the parsing results match expectations (using kombu's field names) |
|
|
|
assert redis_config["hostname"] == expected_host |
|
|
|
assert redis_config["port"] == expected_port |
|
|
|
assert redis_config["userid"] == expected_username # kombu uses 'userid' not 'username' |
|
|
|
assert redis_config["password"] == expected_password |
|
|
|
assert redis_config["virtual_host"] == expected_db # kombu uses 'virtual_host' not 'db' |