Solved database credentials bug

This commit is contained in:
Kevin Veen-Birkenbach 2025-07-17 06:32:53 +02:00
parent 84322f81ef
commit 725fea1169
No known key found for this signature in database
GPG Key ID: 44D8F11FD62F878E
5 changed files with 45 additions and 99 deletions

View File

@ -1,50 +0,0 @@
# filter_plugins/get_cymais_path.py
"""
This plugin provides filters to extract the CyMaIS directory and file identifiers
from a given role name. It assumes the role name is structured as 'dir_file'.
If the structure is invalid (e.g., missing or too many underscores), it raises an error.
These filters are used to support internal processing within CyMaIS.
"""
from ansible.errors import AnsibleFilterError
class CymaisPathExtractor:
"""Extracts directory and file parts from role names in the format 'dir_file'."""
def __init__(self, value):
self.value = value
self._parts = self._split_value()
def _split_value(self):
parts = self.value.split("_")
if len(parts) != 2:
raise AnsibleFilterError(
f"Invalid format: '{self.value}' must contain exactly one underscore (_)"
)
return parts
def get_dir(self):
return self._parts[0]
def get_file(self):
return self._parts[1]
def get_cymais_dir(value):
return CymaisPathExtractor(value).get_dir()
def get_cymais_file(value):
return CymaisPathExtractor(value).get_file()
class FilterModule(object):
"""Ansible filter plugin for CyMaIS path parsing."""
def filters(self):
return {
"get_cymais_dir": get_cymais_dir,
"get_cymais_file": get_cymais_file,
}

View File

@ -0,0 +1,17 @@
class FilterModule(object):
def filters(self):
return {
'get_public_id': self.get_public_id
}
def get_public_id(self, value):
"""
Extract the substring after the last hyphen in the input string.
Example:
'service-user-abc123' => 'abc123'
"""
if not isinstance(value, str):
raise ValueError("Expected a string")
if '-' not in value:
raise ValueError("No hyphen found in input string")
return value.rsplit('-', 1)[-1]

View File

@ -1,13 +1,12 @@
# Helper variables # Helper variables
_database_id: "svc-db-{{ database_type }}" _database_id: "svc-db-{{ database_type }}"
_database_central_name: "{{ applications | get_app_conf( _database_id, 'docker.services.' ~ database_type ~ '.name') }}" _database_central_name: "{{ applications | get_app_conf( _database_id, 'docker.services.' ~ database_type ~ '.name') }}"
_database_central_user: "{{ database_type }}"
# Definition # Definition
database_name: "{{ applications | get_app_conf( database_application_id, 'database.name', false, _database_central_name ) }}" # The overwritte configuration is needed by bigbluebutton database_name: "{{ applications | get_app_conf( database_application_id, 'database.name', false, database_application_id | get_public_id ) }}" # The overwritte configuration is needed by bigbluebutton
database_instance: "{{ _database_central_name if applications | get_app_conf(database_application_id, 'features.central_database', False) else database_name }}" # This could lead to bugs at dedicated database @todo cleanup database_instance: "{{ _database_central_name if applications | get_app_conf(database_application_id, 'features.central_database', False) else database_name }}" # This could lead to bugs at dedicated database @todo cleanup
database_host: "{{ _database_central_name if applications | get_app_conf(database_application_id, 'features.central_database', False) else 'database' }}" # This could lead to bugs at dedicated database @todo cleanup database_host: "{{ _database_central_name if applications | get_app_conf(database_application_id, 'features.central_database', False) else 'database' }}" # This could lead to bugs at dedicated database @todo cleanup
database_username: "{{ applications | get_app_conf(database_application_id, 'database.username', false, _database_central_user )}}" # The overwritte configuration is needed by bigbluebutton database_username: "{{ applications | get_app_conf(database_application_id, 'database.username', false, database_application_id | get_public_id)}}" # The overwritte configuration is needed by bigbluebutton
database_password: "{{ applications | get_app_conf(database_application_id, 'credentials.database_password', true) }}" database_password: "{{ applications | get_app_conf(database_application_id, 'credentials.database_password', true) }}"
database_port: "{{ ports.localhost.database[ _database_id ] }}" database_port: "{{ ports.localhost.database[ _database_id ] }}"
database_env: "{{docker_compose.directories.env}}{{database_type}}.env" database_env: "{{docker_compose.directories.env}}{{database_type}}.env"

View File

@ -1,46 +0,0 @@
# tests/unit/filter_plugins/test_get_cymais_path.py
import unittest
import sys
import os
# Ensure the filter_plugins directory is in the import path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../../filter_plugins')))
from get_cymais_path import get_cymais_dir, get_cymais_file
from ansible.errors import AnsibleFilterError
class TestGetCymaisPath(unittest.TestCase):
def test_valid_input(self):
"""Test valid input with exactly one underscore"""
self.assertEqual(get_cymais_dir("web_app"), "web")
self.assertEqual(get_cymais_file("web_app"), "app")
self.assertEqual(get_cymais_dir("sys_timer"), "sys")
self.assertEqual(get_cymais_file("sys_timer"), "timer")
def test_invalid_no_underscore(self):
"""Test input with no underscore raises error"""
with self.assertRaises(AnsibleFilterError):
get_cymais_dir("invalid")
with self.assertRaises(AnsibleFilterError):
get_cymais_file("invalid")
def test_invalid_multiple_underscores(self):
"""Test input with more than one underscore raises error"""
with self.assertRaises(AnsibleFilterError):
get_cymais_dir("too_many_parts_here")
with self.assertRaises(AnsibleFilterError):
get_cymais_file("too_many_parts_here")
def test_empty_string(self):
"""Test empty string input raises error"""
with self.assertRaises(AnsibleFilterError):
get_cymais_dir("")
with self.assertRaises(AnsibleFilterError):
get_cymais_file("")
if __name__ == '__main__':
unittest.main()

View File

@ -0,0 +1,26 @@
import unittest
from filter_plugins.get_public_id import FilterModule
class TestGetPublicId(unittest.TestCase):
def setUp(self):
self.filter = FilterModule().filters()['get_public_id']
def test_extract_public_id(self):
self.assertEqual(self.filter("svc-user-abc123"), "abc123")
self.assertEqual(self.filter("something-simple-xyz"), "xyz")
self.assertEqual(self.filter("a-b-c-d-e"), "e")
def test_no_hyphen(self):
with self.assertRaises(ValueError):
self.filter("nohyphenhere")
def test_non_string_input(self):
with self.assertRaises(ValueError):
self.filter(12345)
def test_empty_string(self):
with self.assertRaises(ValueError):
self.filter("")
if __name__ == '__main__':
unittest.main()