Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions dslr/config.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from dataclasses import dataclass
from urllib.parse import urlparse
from urllib.parse import unquote, urlparse

from .console import console

Expand Down Expand Up @@ -28,8 +28,8 @@ def initialize(self, *, url: str, debug: bool):
self.db = DatabaseConnection(
host=parsed.hostname or "",
port=parsed.port or 5432,
username=parsed.username or "",
password=parsed.password or "",
username=unquote(parsed.username or ""),
password=unquote(parsed.password or ""),
name=parsed.path[1:],
)

Expand Down
37 changes: 37 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import os
from datetime import datetime
from itertools import product
from typing import Any, List, Tuple
from unittest import TestCase, mock
from urllib.parse import quote

from click.testing import CliRunner

Expand Down Expand Up @@ -305,3 +307,38 @@ def test_settings_preference_order(self, mock_cli_settings, mock_get_snapshots):
mock.call(debug=False, url="postgres://cli:pw@test:5432/my_db"),
],
)

@mock.patch("dslr.config.DatabaseConnection")
def test_encoded_username_password(
self, mock_database_connection, mock_get_snapshots
):
usernames = ["user", "user@example.com"]
passwords = ["pw", "f4A$#%&?!_@:asvb"]

for username, password in product(usernames, passwords):
with self.subTest(username=username, password=password):
mock_database_connection.reset_mock()

username_encoded = quote(username)
password_encoded = quote(password)

# DATABASE_URL environment variable is used
with mock.patch.dict(
os.environ,
{
"DATABASE_URL": (
f"postgres://{username_encoded}:{password_encoded}"
"@test:5432/my_db"
)
},
):
runner = CliRunner()
result = runner.invoke(cli.cli, ["list"])

self.assertEqual(result.exit_code, 0)

self.assertEqual(1, mock_database_connection.call_count)

_, call_kwargs = mock_database_connection.call_args
self.assertEqual(call_kwargs["username"], username)
self.assertEqual(call_kwargs["password"], password)