Skip to content

Commit f4cbbc9

Browse files
authored
Cover missing test cases for internal users (#357)
* test_password_rotation | added coverage for missing system users: `monitoring`, `backup`, `rewind` * Add test_db_connection_with_empty_password | checking that user can't connect with empty password * removed the excess self-hosted test mark
1 parent 41fd24d commit f4cbbc9

File tree

1 file changed

+47
-1
lines changed

1 file changed

+47
-1
lines changed

tests/integration/test_password_rotation.py

Lines changed: 47 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import json
55
import time
66

7+
import psycopg2
78
import pytest
89
from pytest_operator.plugin import OpsTest
910

@@ -12,8 +13,11 @@
1213
CHARM_SERIES,
1314
METADATA,
1415
check_patroni,
16+
db_connect,
1517
get_leader_unit,
1618
get_password,
19+
get_primary,
20+
get_unit_address,
1721
restart_patroni,
1822
set_password,
1923
)
@@ -46,6 +50,9 @@ async def test_password_rotation(ops_test: OpsTest):
4650
any_unit_name = ops_test.model.applications[APP_NAME].units[0].name
4751
superuser_password = await get_password(ops_test, any_unit_name)
4852
replication_password = await get_password(ops_test, any_unit_name, "replication")
53+
monitoring_password = await get_password(ops_test, any_unit_name, "monitoring")
54+
backup_password = await get_password(ops_test, any_unit_name, "backup")
55+
rewind_password = await get_password(ops_test, any_unit_name, "rewind")
4956

5057
# Get the leader unit name (because passwords can only be set through it).
5158
leader = None
@@ -67,11 +74,40 @@ async def test_password_rotation(ops_test: OpsTest):
6774
assert "password" in result.keys()
6875
await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
6976

70-
new_superuser_password = await get_password(ops_test, any_unit_name)
77+
# For monitoring, generate a specific password and pass it to the action.
78+
new_monitoring_password = "test-password"
79+
result = await set_password(
80+
ops_test, unit_name=leader, username="monitoring", password=new_monitoring_password
81+
)
82+
assert "password" in result.keys()
83+
await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
84+
85+
# For backup, generate a specific password and pass it to the action.
86+
new_backup_password = "test-password"
87+
result = await set_password(
88+
ops_test, unit_name=leader, username="backup", password=new_backup_password
89+
)
90+
assert "password" in result.keys()
91+
await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
7192

93+
# For rewind, generate a specific password and pass it to the action.
94+
new_rewind_password = "test-password"
95+
result = await set_password(
96+
ops_test, unit_name=leader, username="rewind", password=new_rewind_password
97+
)
98+
assert "password" in result.keys()
99+
await ops_test.model.wait_for_idle(apps=[APP_NAME], status="active", timeout=1000)
100+
101+
new_superuser_password = await get_password(ops_test, any_unit_name)
72102
assert superuser_password != new_superuser_password
73103
assert new_replication_password == await get_password(ops_test, any_unit_name, "replication")
74104
assert replication_password != new_replication_password
105+
assert new_monitoring_password == await get_password(ops_test, any_unit_name, "monitoring")
106+
assert monitoring_password != new_monitoring_password
107+
assert new_backup_password == await get_password(ops_test, any_unit_name, "backup")
108+
assert backup_password != new_backup_password
109+
assert new_rewind_password == await get_password(ops_test, any_unit_name, "rewind")
110+
assert rewind_password != new_rewind_password
75111

76112
# Restart Patroni on any non-leader unit and check that
77113
# Patroni and PostgreSQL continue to work.
@@ -122,6 +158,16 @@ async def test_empty_password(ops_test: OpsTest) -> None:
122158
assert password == "None"
123159

124160

161+
@pytest.mark.group(1)
162+
async def test_db_connection_with_empty_password(ops_test: OpsTest):
163+
"""Test that user can't connect with empty password."""
164+
primary = await get_primary(ops_test, f"{APP_NAME}/0")
165+
address = get_unit_address(ops_test, primary)
166+
with pytest.raises(psycopg2.Error):
167+
with db_connect(address, "") as connection:
168+
connection.close()
169+
170+
125171
@pytest.mark.group(1)
126172
async def test_no_password_change_on_invalid_password(ops_test: OpsTest) -> None:
127173
"""Test that in general, there is no change when password validation fails."""

0 commit comments

Comments
 (0)