mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2025-08-30 07:18:09 +02:00
Enhance timeout_start_sec_for_domains filter to accept dict, list, or str
- Updated filter to handle dict (domain map), list (flattened domains), or single str inputs. - Prevents duplicate 'www.' prefixes by checking prefix before adding. - Adjusted unit tests: * Replaced old non-dict test with invalid type tests (int, None). * Added explicit tests for list and string input types. See conversation: https://chatgpt.com/share/68b1ae9a-1ac0-800f-b49d-2915386a1a23
This commit is contained in:
@@ -1,12 +1,7 @@
|
|||||||
|
# filter_plugins/timeout_start_sec_for_domains.py (nur Kern geändert)
|
||||||
from ansible.errors import AnsibleFilterError
|
from ansible.errors import AnsibleFilterError
|
||||||
|
|
||||||
class FilterModule(object):
|
class FilterModule(object):
|
||||||
"""
|
|
||||||
Compute a max TimeoutStartSec for systemd services that iterate over many domains.
|
|
||||||
The timeout scales with the number of unique domains (optionally including www.* clones)
|
|
||||||
and is clamped between configurable min/max bounds.
|
|
||||||
"""
|
|
||||||
|
|
||||||
def filters(self):
|
def filters(self):
|
||||||
return {
|
return {
|
||||||
"timeout_start_sec_for_domains": self.timeout_start_sec_for_domains,
|
"timeout_start_sec_for_domains": self.timeout_start_sec_for_domains,
|
||||||
@@ -23,28 +18,17 @@ class FilterModule(object):
|
|||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Args:
|
Args:
|
||||||
domains_dict (dict): Same structure you pass to generate_all_domains
|
domains_dict (dict | list[str] | str): Either the domain mapping dict
|
||||||
(values can be str | list[str] | dict[str,str]).
|
(values can be str | list[str] | dict[str,str]) or an already
|
||||||
include_www (bool): If true, also count "www.<domain>" variants.
|
flattened list of domains, or a single domain string.
|
||||||
per_domain_seconds (int): Budget per domain (default 25s).
|
include_www (bool): If true, add 'www.<domain>' for non-www entries.
|
||||||
overhead_seconds (int): Fixed overhead on top (default 30s).
|
...
|
||||||
min_seconds (int): Lower clamp (default 120s).
|
|
||||||
max_seconds (int): Upper clamp (default 3600s).
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
int: TimeoutStartSec in seconds (integer).
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
AnsibleFilterError: On invalid input types or unexpected failures.
|
|
||||||
"""
|
"""
|
||||||
try:
|
try:
|
||||||
if not isinstance(domains_dict, dict):
|
# Local flattener for dict inputs (like your generate_all_domains source)
|
||||||
raise AnsibleFilterError("Expected 'domains_dict' to be a dict.")
|
def _flatten_from_dict(domains_map):
|
||||||
|
|
||||||
# Local flatten similar to your generate_all_domains
|
|
||||||
def _flatten(domains):
|
|
||||||
flat = []
|
flat = []
|
||||||
for v in (domains or {}).values():
|
for v in (domains_map or {}).values():
|
||||||
if isinstance(v, str):
|
if isinstance(v, str):
|
||||||
flat.append(v)
|
flat.append(v)
|
||||||
elif isinstance(v, list):
|
elif isinstance(v, list):
|
||||||
@@ -53,18 +37,26 @@ class FilterModule(object):
|
|||||||
flat.extend(v.values())
|
flat.extend(v.values())
|
||||||
return flat
|
return flat
|
||||||
|
|
||||||
flat = _flatten(domains_dict)
|
# Accept dict | list | str
|
||||||
|
if isinstance(domains_dict, dict):
|
||||||
|
flat = _flatten_from_dict(domains_dict)
|
||||||
|
elif isinstance(domains_dict, list):
|
||||||
|
flat = list(domains_dict)
|
||||||
|
elif isinstance(domains_dict, str):
|
||||||
|
flat = [domains_dict]
|
||||||
|
else:
|
||||||
|
raise AnsibleFilterError(
|
||||||
|
"Expected 'domains_dict' to be dict | list | str."
|
||||||
|
)
|
||||||
|
|
||||||
if include_www:
|
if include_www:
|
||||||
# dedupe first so we don't generate duplicate www-variants
|
|
||||||
base_unique = sorted(set(flat))
|
base_unique = sorted(set(flat))
|
||||||
www_variants = [f"www.{d}" for d in base_unique if not str(d).startswith("www.")]
|
www_variants = [f"www.{d}" for d in base_unique if not str(d).lower().startswith("www.")]
|
||||||
flat.extend(www_variants)
|
flat.extend(www_variants)
|
||||||
|
|
||||||
unique_domains = sorted(set(flat))
|
unique_domains = sorted(set(flat))
|
||||||
count = len(unique_domains)
|
count = len(unique_domains)
|
||||||
|
|
||||||
# Compute and clamp
|
|
||||||
raw = overhead_seconds + per_domain_seconds * count
|
raw = overhead_seconds + per_domain_seconds * count
|
||||||
clamped = max(min_seconds, min(max_seconds, int(raw)))
|
clamped = max(min_seconds, min(max_seconds, int(raw)))
|
||||||
return clamped
|
return clamped
|
||||||
|
@@ -78,10 +78,28 @@ class TestTimeoutStartSecForDomains(unittest.TestCase):
|
|||||||
# raw = 30 + 25*4 = 130
|
# raw = 30 + 25*4 = 130
|
||||||
self.assertEqual(result, 130)
|
self.assertEqual(result, 130)
|
||||||
|
|
||||||
def test_raises_on_non_dict_input(self):
|
def test_raises_on_invalid_type_int(self):
|
||||||
with self.assertRaises(AnsibleFilterError):
|
with self.assertRaises(AnsibleFilterError):
|
||||||
_f()(["not-a-dict"])
|
_f()(123)
|
||||||
|
|
||||||
|
def test_raises_on_invalid_type_none(self):
|
||||||
|
with self.assertRaises(AnsibleFilterError):
|
||||||
|
_f()(None)
|
||||||
|
|
||||||
|
def test_accepts_list_input(self):
|
||||||
|
domains_list = ["a.com", "www.a.com", "b.com"]
|
||||||
|
result = _f()(domains_list, include_www=True,
|
||||||
|
per_domain_seconds=25, overhead_seconds=30,
|
||||||
|
min_seconds=1, max_seconds=10000)
|
||||||
|
# unique + www for b.com -> {"a.com","www.a.com","b.com","www.b.com"} = 4
|
||||||
|
self.assertEqual(result, 30 + 25*4)
|
||||||
|
|
||||||
|
def test_accepts_str_input(self):
|
||||||
|
result = _f()("a.com", include_www=True,
|
||||||
|
per_domain_seconds=25, overhead_seconds=30,
|
||||||
|
min_seconds=1, max_seconds=10000)
|
||||||
|
# {"a.com","www.a.com"} = 2
|
||||||
|
self.assertEqual(result, 30 + 25*2)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
Reference in New Issue
Block a user