database: detect ProxySQL DB user at connect time

[ideaship]

Problem

osism status database and osism manage loadbalancer open a superuser connection to the cluster's MariaDB, and both chose the DB user from enable_proxysql read off a raw environments/kolla/configuration.yml load.

That value is not in the operator config — it is a release-gated Jinja default in osism/defaults ({{ 'no' if openstack_version in ['2024.1','2024.2','2025.1'] else 'yes' }}). A raw file read never sees it, so config.get("enable_proxysql", False) always returned False and the commands connected as the plain user.

On a ProxySQL cluster (the 2025.2 default) the proxied superuser is the sharded name <user>_shard_0, so ProxySQL denies the plain user with a 1045 and the check fails against a perfectly healthy database:

ERROR Failed to connect to MariaDB: (1045,
  "ProxySQL Error: Access denied for user 'root'@'…' (using password: YES)")

The bug was latent before 2025.2: enable_proxysql defaulted to no on every earlier release, so the wrong read (False → plain user) happened to match reality.

Fix

Reading the value correctly would mean resolving it the way kolla-ansible does (Ansible templating with openstack_version in scope); a raw read — and even ansible-inventory --host, which returns the unrendered Jinja — cannot see it.

Instead of resolving it, this sidesteps the question. New osism/utils/mariadb.connect() tries <user>_shard_0 (ProxySQL) first and falls back to <user> (plain HAProxy) only on an access-denied error, so the connection itself selects the topology with no dependency on defaults resolution. Genuine connectivity failures (non-1045) surface immediately without a retry. Both call sites now use the helper and drop their duplicated enable_proxysql/db_user logic.

Tests

• New tests/unit/utils/test_mariadb.py: shard-user-first success, fallback on 1045, no retry on a non-auth error (2003), and last-error propagation when both users are denied.

[sourcery-ai]

Hey - I've found 2 issues

Prompt for AI Agents

Please address the comments from this code review:

## Individual Comments

### Comment 1
<location path="tests/unit/utils/test_mariadb.py" line_range="15-24" />
<code_context>
+    return pymysql.err.OperationalError(1045, f"Access denied for user '{user}'")
+
+
+def test_connect_prefers_proxysql_shard_user():
+    """A ProxySQL cluster: the sharded user connects on the first try."""
+    sentinel = object()
+    with mock.patch("pymysql.connect", return_value=sentinel) as connect:
+        result = mariadb.connect("vip", "root", "pw", port=3306)
+
+    assert result is sentinel
+    connect.assert_called_once_with(
+        host="vip", user="root_shard_0", password="pw", port=3306
+    )
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test to verify that extra connection kwargs are passed through unchanged to `pymysql.connect`

The helper is also used in `status` and `loadbalancer` with extra kwargs like `database`, `cursorclass`, and `connect_timeout`, but the current tests only cover `port`. Please add a test (e.g. `test_connect_passes_through_extra_kwargs`) that calls `mariadb.connect` with several extra kwargs and asserts that `pymysql.connect` receives them unchanged, so future refactors don’t accidentally drop or modify these parameters.

```suggestion
def test_connect_prefers_proxysql_shard_user():
    """A ProxySQL cluster: the sharded user connects on the first try."""
    sentinel = object()
    with mock.patch("pymysql.connect", return_value=sentinel) as connect:
        result = mariadb.connect("vip", "root", "pw", port=3306)

    assert result is sentinel
    connect.assert_called_once_with(
        host="vip", user="root_shard_0", password="pw", port=3306
    )


def test_connect_passes_through_extra_kwargs():
    """Extra connection kwargs are passed through unchanged to pymysql.connect."""
    sentinel = object()
    cursor_cls = object()

    with mock.patch("pymysql.connect", return_value=sentinel) as connect:
        result = mariadb.connect(
            "vip",
            "root",
            "pw",
            port=3306,
            database="mydb",
            cursorclass=cursor_cls,
            connect_timeout=5,
        )

    assert result is sentinel
    connect.assert_called_once_with(
        host="vip",
        user="root_shard_0",
        password="pw",
        port=3306,
        database="mydb",
        cursorclass=cursor_cls,
        connect_timeout=5,
    )
```
</issue_to_address>

### Comment 2
<location path="tests/unit/utils/test_mariadb.py" line_range="53-60" />
<code_context>
+    connect.assert_called_once()
+
+
+def test_connect_raises_last_error_when_both_users_denied():
+    """Neither user authenticates: the last access-denied error propagates."""
+    errors = [_access_denied("root_shard_0"), _access_denied("root")]
+    with mock.patch("pymysql.connect", side_effect=errors):
+        with pytest.raises(pymysql.err.OperationalError) as excinfo:
+            mariadb.connect("vip", "root", "pw")
+
+    assert "root'" in excinfo.value.args[1]
</code_context>
<issue_to_address>
**nitpick (testing):** Tighten the assertion to clearly prove that the last (plain user) error is propagated

This test should clearly verify that the *last* access-denied error (for the plain user) is raised. The current assertion only checks for "root'" in the message, which would also succeed if the wrong error were raised. Please assert both the error code (1045) and that the message refers to the plain user, e.g. `assert excinfo.value.args[0] == 1045` and `assert "Access denied for user 'root'" in excinfo.value.args[1]`, so the "last error propagates" behavior is explicitly covered.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

suggestion (testing): Add a test to verify that extra connection kwargs are passed through unchanged to pymysql.connect

The helper is also used in status and loadbalancer with extra kwargs like database, cursorclass, and connect_timeout, but the current tests only cover port. Please add a test (e.g. test_connect_passes_through_extra_kwargs) that calls mariadb.connect with several extra kwargs and asserts that pymysql.connect receives them unchanged, so future refactors don’t accidentally drop or modify these parameters.

Suggested change

	def test_connect_prefers_proxysql_shard_user():
	"""A ProxySQL cluster: the sharded user connects on the first try."""
	sentinel = object()
	with mock.patch("pymysql.connect", return_value=sentinel) as connect:
	result = mariadb.connect("vip", "root", "pw", port=3306)
	assert result is sentinel
	connect.assert_called_once_with(
	host="vip", user="root_shard_0", password="pw", port=3306
	)
	def test_connect_prefers_proxysql_shard_user():
	"""A ProxySQL cluster: the sharded user connects on the first try."""
	sentinel = object()
	with mock.patch("pymysql.connect", return_value=sentinel) as connect:
	result = mariadb.connect("vip", "root", "pw", port=3306)
	assert result is sentinel
	connect.assert_called_once_with(
	host="vip", user="root_shard_0", password="pw", port=3306
	)
	def test_connect_passes_through_extra_kwargs():
	"""Extra connection kwargs are passed through unchanged to pymysql.connect."""
	sentinel = object()
	cursor_cls = object()
	with mock.patch("pymysql.connect", return_value=sentinel) as connect:
	result = mariadb.connect(
	"vip",
	"root",
	"pw",
	port=3306,
	database="mydb",
	cursorclass=cursor_cls,
	connect_timeout=5,
	)
	assert result is sentinel
	connect.assert_called_once_with(
	host="vip",
	user="root_shard_0",
	password="pw",
	port=3306,
	database="mydb",
	cursorclass=cursor_cls,
	connect_timeout=5,
	)

[sourcery-ai]

nitpick (testing): Tighten the assertion to clearly prove that the last (plain user) error is propagated

This test should clearly verify that the last access-denied error (for the plain user) is raised. The current assertion only checks for "root'" in the message, which would also succeed if the wrong error were raised. Please assert both the error code (1045) and that the message refers to the plain user, e.g. assert excinfo.value.args[0] == 1045 and assert "Access denied for user 'root'" in excinfo.value.args[1], so the "last error propagates" behavior is explicitly covered.