Refactor web health checker & domain expectations (filter-based)

- Move all domain→expected-status mapping to filter `web_health_expectations`.
- Require explicit app selection via non-empty `group_names`; only those apps are included.
- Add `www_enabled` flag (wired via `WWW_REDIRECT_ENABLED`) to generate/force www.* → 301.
- Support `redirect_maps` to include manual redirects (sources forced to 301), independent of app selection.
- Aliases always 301; canonicals use per-key override or `server.status_codes.default`, else [200,302,301].
- Remove legacy fallbacks (`server.status_codes.home` / `landingpage`).
- Wire filter output into systemd ExecStart script as JSON expectations.
- Normalize various templates to use `to_json` and minor spacing fixes.
- Update app configs (e.g., YOURLS default=301; Confluence default=302; Bluesky web=405; MediaWiki/Confluence canonical/aliases).
- Constructor now uses `WWW_REDIRECT_ENABLED` for domain generation.

Tests:
- Add comprehensive unit tests for filter: selection by group, keyed/default codes, aliases, www handling, redirect_maps, input sanitization.
- Add unit tests for the standalone checker script (JSON parsing, OK/mismatch counting, sanitization).

See conversation: https://chatgpt.com/share/68c2b93e-de58-800f-8c16-ea05755ba776
This commit is contained in:
2025-09-11 13:58:16 +02:00
parent 6418a462ec
commit cbfb096cdb
35 changed files with 717 additions and 106 deletions

View File

@@ -29,6 +29,9 @@ WEB_PORT: "{{ 443 if WEB_PROTOCOL == 'https' else 80 }}" # Defaul
# Websocket
WEBSOCKET_PROTOCOL: "{{ 'wss' if WEB_PROTOCOL == 'https' else 'ws' }}"
# WWW-Redirect to None WWW-Domains enabled
WWW_REDIRECT_ENABLED: "{{ ('web-opt-rdr-www' in group_names) | bool }}"
# Domain
PRIMARY_DOMAIN: "localhost" # Primary Domain of the server

View File

@@ -3,4 +3,3 @@ BACKUPS_FOLDER_PATH: "/Backups/" # Path to the backups folder
# Storage Space-Related Configurations
SIZE_PERCENT_MAXIMUM_BACKUP: 75 # Maximum storage space in percent for backups
SIZE_PERCENT_CLEANUP_DISC_SPACE: 85 # Threshold for triggering cleanup actions
SIZE_PERCENT_DISC_SPACE_WARNING: 90 # Warning threshold in percent for free disk space

View File

View File

@@ -0,0 +1,72 @@
#!/usr/bin/env python3
"""
Ultra-thin checker: consume a JSON mapping of {domain: [expected_status_codes]}
and verify HTTP HEAD responses. All mapping logic is done in the filter
`web_health_expectations`.
"""
import argparse
import json
import sys
from typing import Dict, List
import requests
def parse_args(argv=None):
p = argparse.ArgumentParser(description="Web health checker (expects precomputed domain→codes mapping).")
p.add_argument("--web-protocol", default="https", choices=["http", "https"], help="Protocol to use")
p.add_argument("--expectations", required=True, help="JSON STRING: {\"domain\": [codes], ...}")
return p.parse_args(argv)
def _parse_json_mapping(name: str, value: str) -> Dict[str, List[int]]:
try:
obj = json.loads(value)
except json.JSONDecodeError as e:
raise SystemExit(f"--{name} must be a valid JSON string: {e}")
if not isinstance(obj, dict):
raise SystemExit(f"--{name} must be a JSON object (mapping)")
# sanitize list-of-ints shape
clean = {}
for k, v in obj.items():
if isinstance(v, list):
try:
clean[k] = [int(x) for x in v]
except Exception:
clean[k] = []
else:
clean[k] = []
return clean
def main(argv=None) -> int:
args = parse_args(argv)
expectations = _parse_json_mapping("expectations", args.expectations)
errors = 0
for domain in sorted(expectations.keys()):
expected = expectations[domain] or []
url = f"{args.web_protocol}://{domain}"
try:
r = requests.head(url, allow_redirects=False, timeout=10)
if expected and r.status_code in expected:
print(f"{domain}: OK")
elif not expected:
# If somehow empty list slipped through, treat as failure to be explicit
print(f"{domain}: ERROR: No expectations provided. Got {r.status_code}.")
errors += 1
else:
print(f"{domain}: ERROR: Expected {expected}. Got {r.status_code}.")
errors += 1
except requests.RequestException as e:
print(f"{domain}: error due to {e}")
errors += 1
if errors:
print(f"Warning: {errors} domains responded with an unexpected https status code.")
return errors
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,186 @@
# roles/sys-ctl-hlth-webserver/filter_plugins/web_health_expectations.py
import os
import sys
from collections.abc import Mapping
# Make repo-level module_utils importable (go up three levels from this file)
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..')))
from module_utils.config_utils import get_app_conf # reuse existing helper
DEFAULT_OK = [200, 302, 301]
def _to_list(x, *, allow_mapping: bool = True):
"""Normalize into a flat list of **strings only**."""
if x is None:
return []
if isinstance(x, bytes):
try:
return [x.decode("utf-8")]
except Exception:
return []
if isinstance(x, str):
return [x]
if isinstance(x, (list, tuple, set)):
out = []
for v in x:
if isinstance(v, (list, tuple, set)):
out.extend(_to_list(v, allow_mapping=False))
elif isinstance(v, bytes):
try:
out.append(v.decode("utf-8"))
except Exception:
pass
elif isinstance(v, str):
out.append(v)
elif isinstance(v, Mapping):
continue
return out
if isinstance(x, Mapping) and allow_mapping:
out = []
for v in x.values():
out.extend(_to_list(v, allow_mapping=True))
return out
return []
def _valid_http_code(x):
"""Return int(x) if 100 <= code <= 599 else None."""
try:
v = int(x)
except (TypeError, ValueError):
return None
return v if 100 <= v <= 599 else None
def _extract_redirect_sources(redirect_maps):
"""Extract a set of source domains from redirect maps."""
sources = set()
if not redirect_maps:
return sources
def _add_one(obj):
if isinstance(obj, str) and obj:
sources.add(obj)
elif isinstance(obj, Mapping):
s = obj.get("source")
if isinstance(s, str) and s:
sources.add(s)
if isinstance(redirect_maps, (list, tuple, set)):
for item in redirect_maps:
_add_one(item)
else:
_add_one(redirect_maps)
return sources
def _normalize_selection(group_names):
"""Return a non-empty set of group names, or raise ValueError."""
if isinstance(group_names, (list, set, tuple)):
sel = {str(x) for x in group_names if str(x)}
elif isinstance(group_names, str):
sel = {g.strip() for g in group_names.split(",") if g.strip()}
else:
sel = set()
if not sel:
raise ValueError("web_health_expectations: 'group_names' must be provided and non-empty")
return sel
def web_health_expectations(applications, www_enabled: bool = False, group_names=None, redirect_maps=None):
"""Produce a **flat mapping**: domain -> [expected_status_codes].
Selection (REQUIRED):
- `group_names` must be provided and non-empty.
- Only include applications whose key is in `group_names`.
Rules:
- Canonical domains (dict-key overrides, else default, else DEFAULT_OK).
- Flat canonical (default, else DEFAULT_OK).
- Aliases always [301].
- No legacy fallbacks (ignore 'home'/'landingpage').
- `redirect_maps`: force <source> -> [301] and override app-derived entries.
- If `www_enabled`: add and/or force www.* -> [301] for all domains.
"""
if not isinstance(applications, Mapping):
return {}
selection = _normalize_selection(group_names)
expectations = {}
for app_id in applications.keys():
if app_id not in selection:
continue
canonical_raw = get_app_conf(
applications, app_id, 'server.domains.canonical',
strict=False, default=[]
)
aliases_raw = get_app_conf(
applications, app_id, 'server.domains.aliases',
strict=False, default=[]
)
aliases = _to_list(aliases_raw, allow_mapping=True)
sc_raw = get_app_conf(
applications, app_id, 'server.status_codes',
strict=False, default={}
)
sc_map = {}
if isinstance(sc_raw, Mapping):
for k, v in sc_raw.items():
code = _valid_http_code(v)
if code is not None:
sc_map[str(k)] = code
if isinstance(canonical_raw, Mapping) and canonical_raw:
for key, domains in canonical_raw.items():
domains_list = _to_list(domains, allow_mapping=False)
code = _valid_http_code(sc_map.get(key))
if code is None:
code = _valid_http_code(sc_map.get("default"))
expected = [code] if code is not None else list(DEFAULT_OK)
for d in domains_list:
if d:
expectations[d] = expected
else:
for d in _to_list(canonical_raw, allow_mapping=True):
if not d:
continue
code = _valid_http_code(sc_map.get("default"))
expectations[d] = [code] if code is not None else list(DEFAULT_OK)
for d in aliases:
if d:
expectations[d] = [301]
for src in _extract_redirect_sources(redirect_maps):
expectations[src] = [301]
if www_enabled:
add = {}
for d in expectations.keys():
if not d.startswith("www."):
add[f"www.{d}"] = [301]
expectations.update(add)
for d in list(expectations.keys()):
if d.startswith("www."):
expectations[d] = [301]
return expectations
class FilterModule(object):
def filters(self):
return {
'web_health_expectations': web_health_expectations,
}

View File

@@ -20,3 +20,7 @@
system_service_timer_enabled: true
system_service_tpl_on_failure: "{{ SYS_SERVICE_ON_FAILURE_COMPOSE }}"
system_service_tpl_timeout_start_sec: "{{ CURRENT_PLAY_DOMAINS_ALL | timeout_start_sec_for_domains }}"
system_service_tpl_exec_start: >
{{ system_service_script_exec }}
--web-protocol {{ WEB_PROTOCOL }}
--expectations '{{ applications | web_health_expectations(www_enabled=WWW_REDIRECT_ENABLED, group_names=group_names) | to_json }}'

View File

@@ -1,69 +0,0 @@
import os
import requests
import sys
import re
def get_expected_statuses(domain: str, parts: list[str], redirected_domains: set[str]) -> list[int]:
"""
Determine the expected HTTP status codes based on the domain name.
Args:
domain: The full domain string (e.g. 'example.com').
parts: The domain split into its subcomponents (e.g. ['www', 'example', 'com']).
redirected_domains: A set of domains that should trigger a redirect.
Returns:
A list of expected HTTP status codes.
"""
if domain == '{{ domains | get_domain('web-app-listmonk') }}':
return [404]
if (parts and parts[0] == 'www') or (domain in redirected_domains):
return [301]
if domain == '{{ domains | get_domain('web-app-yourls') }}':
return [{{ applications | get_app_conf('web-app-yourls', 'server.status_codes.landingpage') }}]
return [200, 302, 301]
# file in which fqdn server configs are deposit
config_path = '{{ NGINX.DIRECTORIES.HTTP.SERVERS }}'
# Initialize the error counter
error_counter = 0
# Regex pattern to match domain.tld or *.domain.tld
pattern = re.compile(r"^(?:[\w-]+\.)*[\w-]+\.[\w-]+\.conf$")
# Iterate over each file in the configuration directory
for filename in os.listdir(config_path):
if filename.endswith('.conf') and pattern.match(filename):
# Extract the domain and subdomain from the filename
domain = filename.replace('.conf', '')
parts = domain.split('.')
# Prepare the URL and expected status codes
url = f"{{ WEB_PROTOCOL }}://{domain}"
redirected_domains = [domain['source'] for domain in {{ redirect_domain_mappings }}]
redirected_domains.append("{{domains | get_domain('web-app-mailu') }}")
expected_statuses = get_expected_statuses(domain, parts, redirected_domains)
try:
# Send a HEAD request to get only the response header
response = requests.head(url)
# Check if the status code matches the expected statuses
if response.status_code in expected_statuses:
print(f"{domain}: OK")
else:
print(f"{domain}: ERROR: Expected {expected_statuses}. Got {response.status_code}.")
error_counter += 1
except requests.RequestException as e:
# Handle exceptions for requests like connection errors
print(f"{domain}: error due to {e}")
error_counter += 1
if error_counter > 0:
print(f"Warning: {error_counter} domains responded with an unexpected https status code.")
# Exit the script with the number of errors as the exit code
sys.exit(error_counter)

View File

@@ -12,6 +12,7 @@
src: "{{ system_service_script_src }}"
dest: "{{ [system_service_script_dir, (system_service_script_src | basename | regex_replace('\\.j2$', ''))] | path_join }}"
mode: "0755"
notify: refresh systemctl service # Just refresh service for testing that script is working in debug mode. In productive due to async it shouldn't make a difference
when: system_service_script_src.endswith('.j2')
- name: "copy raw file"
@@ -20,4 +21,5 @@
dest: "{{ [system_service_script_dir, (system_service_script_src | basename)] | path_join }}"
mode: "0755"
when: not system_service_script_src.endswith('.j2')
notify: refresh systemctl service # Just refresh service for testing that script is working in debug mode. In productive due to async it shouldn't make a difference
when: system_service_copy_files | bool

View File

@@ -6,11 +6,13 @@ features:
logout: true
server:
config_upstream_url: "https://ip.bsky.app/config"
status_codes:
web: 405
domains:
canonical:
web: "bskyweb.{{ PRIMARY_DOMAIN }}"
api: "bluesky.{{ PRIMARY_DOMAIN }}"
view: "view.bluesky.{{ PRIMARY_DOMAIN }}"
# view: "view.bluesky.{{ PRIMARY_DOMAIN }}"
csp:
whitelist:
connect-src:

View File

@@ -2,13 +2,13 @@ credentials: {}
docker:
services:
database:
enabled: true
enabled: true
application:
image: atlassian/confluence
version: latest
name: confluence
image: atlassian/confluence
version: latest
name: confluence
volumes:
data: "confluence_data"
data: "confluence_data"
features:
matomo: true
css: true
@@ -18,6 +18,8 @@ features:
oidc: false # Not enabled for demo version
ldap: false # Not enabled for demo version
server:
status_codes:
default: 302
csp:
whitelist: {}
flags:
@@ -27,7 +29,10 @@ server:
unsafe-inline: true
domains:
canonical:
- "c.wiki.{{ PRIMARY_DOMAIN }}"
aliases:
- "confluence.{{ PRIMARY_DOMAIN }}"
- "confluence.wiki.{{ PRIMARY_DOMAIN }}"
rbac:
roles: {}
truststore_enabled: false

View File

@@ -8,7 +8,10 @@ galaxy_info:
Kevin Veen-Birkenbach
Consulting & Coaching Solutions
https://www.veen.world
galaxy_tags: []
galaxy_tags:
- wiki
- documentation
- confluence
repository: "https://s.infinito.nexus/code"
issue_tracker_url: "https://s.infinito.nexus/issues"
documentation: "https://s.infinito.nexus/code/"

View File

@@ -14,11 +14,9 @@ galaxy_info:
versions:
- latest
galaxy_tags:
- docker
- homepage
- portfolio
- ansible
- flask
- web
- landingpage
repository: "https://github.com/kevinveenbirkenbach/portfolio"
issue_tracker_url: "https://github.com/kevinveenbirkenbach/portfolio/issues"
documentation: "https://github.com/kevinveenbirkenbach/portfolio#readme"

View File

@@ -104,12 +104,20 @@ portfolio_menu_categories:
- yourls
Presentation:
description: "Presentation and Documentation Tools"
description: "Presentation Tools"
icon: "fas fa-tools"
tags:
- presentation
- sphinx
- portfolio
Documentation:
description: "Documentation and Wiki Applications"
icon: "fa fa-book"
tags:
- confluence
- xwiki
- mediawiki
Finance & Accounting:
description: "Financial and accounting software"
@@ -174,5 +182,4 @@ portfolio_menu_categories:
- publishing
- website
- joomla
- mediawiki
- wordpress

View File

@@ -3,13 +3,13 @@ credentials: {}
docker:
services:
database:
enabled: true
enabled: true
application:
image: atlassian/jira-software
version: latest
name: jira
image: atlassian/jira-software
version: latest
name: jira
volumes:
data: "jira_data"
data: "jira_data"
features:
matomo: true
css: true
@@ -31,5 +31,7 @@ server:
domains:
canonical:
- "jira.{{ PRIMARY_DOMAIN }}"
status_codes:
default: 405
rbac:
roles: {}

View File

@@ -245,7 +245,7 @@
{# Build objectClasses from structural + auxiliary definitions #}
"userObjectClasses": [
{{ KEYCLOAK_LDAP_USER_OBJECT_CLASSES | trim | tojson }}
{{ KEYCLOAK_LDAP_USER_OBJECT_CLASSES | trim | to_json }}
],
"rdnLDAPAttribute": [ "{{ LDAP.USER.ATTRIBUTES.ID }}" ],

View File

@@ -55,7 +55,7 @@
"providerId": "declarative-user-profile",
"subComponents": {},
"config": {
"kc.user.profile.config": [{{ (user_profile | tojson) | tojson }}]
"kc.user.profile.config": [{{ (user_profile | to_json) | to_json }}]
}
}
]

View File

@@ -10,6 +10,8 @@ server:
domains:
canonical:
- "newsletter.{{ PRIMARY_DOMAIN }}"
status_codes:
default: 404
docker:
services:
database:

View File

@@ -13,7 +13,7 @@
- name: add config.toml
template:
src: "config.toml.j2"
dest: "{{docker_compose.directories.config}}config.toml"
dest: "{{ docker_compose.directories.config }}config.toml"
notify: docker compose up
- meta: flush_handlers

View File

@@ -7,7 +7,7 @@
ports:
- "127.0.0.1:{{ ports.localhost.http[application_id] }}:{{ container_port }}"
volumes:
- {{docker_compose.directories.config}}config.toml:/listmonk/config.toml
- {{ docker_compose.directories.config }}config.toml:/listmonk/config.toml
{% include 'roles/docker-container/templates/networks.yml.j2' %}
{% include 'roles/docker-container/templates/depends_on/dmbs_excl.yml.j2' %}
{% include 'roles/docker-container/templates/healthcheck/wget.yml.j2' %}

View File

@@ -25,7 +25,7 @@ report_stats: true
macaroon_secret_key: "{{ applications | get_app_conf(application_id, 'credentials.macaroon_secret_key') }}"
form_secret: "{{ applications | get_app_conf(application_id, 'credentials.form_secret') }}"
signing_key_path: "/data/{{ MATRIX_SYNAPSE_DOMAIN }}.signing.key"
web_client_location: "{{ WEB_PROTOCOL }}://{{domains[application_id].element}}"
web_client_location: "{{ WEB_PROTOCOL }}://{{ domains[application_id].element}}"
public_baseurl: "{{ MATRIX_SYNAPSE_URL }}"
trusted_key_servers:
- server_name: "matrix.org"

View File

@@ -2,6 +2,8 @@ sitename: "Wiki on {{ PRIMARY_DOMAIN | upper }}"
server:
domains:
canonical:
- "m.wiki.{{ PRIMARY_DOMAIN }}"
aliases:
- "media.wiki.{{ PRIMARY_DOMAIN }}"
docker:
services:

View File

@@ -10,8 +10,6 @@ galaxy_info:
https://www.veen.world
galaxy_tags:
- mediawiki
- docker
- cms
- wiki
- documentation
repository: "https://s.infinito.nexus/code"

View File

@@ -10,7 +10,7 @@ postgres_gis_enabled: true
docker_compose_flush_handlers: false
# Mobilizon
mobilizon_host_conf_exs_file: "{{docker_compose.directories.config}}config.exs"
mobilizon_host_conf_exs_file: "{{ docker_compose.directories.config }}config.exs"
mobilizon_version: "{{ applications | get_app_conf(application_id, 'docker.services.mobilizon.version', True) }}"
mobilizon_image: "{{ applications | get_app_conf(application_id, 'docker.services.mobilizon.image', True) }}"
mobilizon_container: "{{ applications | get_app_conf(application_id, 'docker.services.mobilizon.name', True) }}"

View File

@@ -42,7 +42,7 @@ plugin_configuration:
-
appid: "user_ldap"
configkey: "s01ldap_base_users"
configvalue: "{{LDAP.DN.OU.USERS}}"
configvalue: "{{ LDAP.DN.OU.USERS }}"
-
appid: "user_ldap"
@@ -67,7 +67,7 @@ plugin_configuration:
-
appid: "user_ldap"
configkey: "s01ldap_dn"
configvalue: "{{LDAP.DN.ADMINISTRATOR.DATA}}"
configvalue: "{{ LDAP.DN.ADMINISTRATOR.DATA }}"
-
appid: "user_ldap"
configkey: "s01ldap_email_attr"

View File

@@ -17,7 +17,7 @@ provider_display_name = "{{ OIDC.BUTTON_TEXT }}"
{# role based restrictions #}
scope = "openid email profile {{ RBAC.GROUP.CLAIM }}"
oidc_groups_claim = "{{ RBAC.GROUP.CLAIM }}"
allowed_groups = {{ applications | get_app_conf(oauth2_proxy_application_id, 'oauth2_proxy.allowed_groups', True) | tojson }}
allowed_groups = {{ applications | get_app_conf(oauth2_proxy_application_id, 'oauth2_proxy.allowed_groups', True) | to_json }}
email_domains = ["*"]
{% else %}
email_domains = "{{ PRIMARY_DOMAIN }}"

View File

@@ -30,7 +30,7 @@ server:
locations:
admin: "/admin/"
status_codes:
landingpage: 301
default: 301
docker:
services:
database:

View File

@@ -12,7 +12,7 @@ YOURLS_VERSION: "{{ applications | get_app_conf(application_id,
YOURLS_IMAGE: "{{ applications | get_app_conf(application_id, 'docker.services.yourls.image') }}"
YOURLS_CONTAINER: "{{ applications | get_app_conf(application_id, 'docker.services.yourls.name') }}"
YOURLS_ADMIN_LOCATION: "{{ applications | get_app_conf(application_id, 'server.locations.admin') }}"
YOURLS_LANDINGPAGE_STATUS_CODE: "{{ applications | get_app_conf(application_id, 'server.status_codes.landingpage') }}"
YOURLS_LANDINGPAGE_STATUS_CODE: "{{ applications | get_app_conf(application_id, 'server.status_codes.default') }}"
# Container
container_port: 8080

View File

@@ -1,4 +1,4 @@
application_id: web-svc-simpleicons
container_port: 3000
simpleicons_host_server_file: "{{docker_compose.directories.config}}server.js"
simpleicons_host_package_file: "{{docker_compose.directories.config}}package.json"
simpleicons_host_server_file: "{{ docker_compose.directories.config }}server.js"
simpleicons_host_package_file: "{{ docker_compose.directories.config }}package.json"

View File

@@ -89,9 +89,7 @@
items2dict(key_name='source', value_name='source'),
recursive=True
)) |
generate_all_domains(
('web-opt-rdr-www' in group_names)
)
generate_all_domains(WWW_REDIRECT_ENABLED)
}}
- name: Merge networks definitions

View File

@@ -0,0 +1,119 @@
# tests/unit/roles/sys-ctl-hlth-webserver/files/test_script.py
import os
import unittest
import importlib.util
from unittest.mock import patch
def load_module_from_path(mod_name: str, path: str):
"""Dynamically load a module from a filesystem path."""
spec = importlib.util.spec_from_file_location(mod_name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # type: ignore[attr-defined]
return module
class TestStandaloneCheckerScript(unittest.TestCase):
@classmethod
def setUpClass(cls):
# Compute repo root: tests/unit/roles/sys-ctl-hlth-webserver/files/test_script.py → 5 levels up
here = os.path.abspath(os.path.dirname(__file__))
repo_root = os.path.abspath(os.path.join(here, "..", "..", "..", "..", ".."))
cls.script_path = os.path.join(
repo_root, "roles", "sys-ctl-hlth-webserver", "files", "script.py"
)
if not os.path.isfile(cls.script_path):
raise FileNotFoundError(f"Cannot find script.py at {cls.script_path}")
cls.script = load_module_from_path("health_script", cls.script_path)
# ------------- JSON parsing ------------------
def test_rejects_invalid_json(self):
with self.assertRaises(SystemExit):
self.script.main([
"--expectations", '{"bad json": [200, 301]', # missing closing brace
])
def test_rejects_non_mapping_json(self):
with self.assertRaises(SystemExit):
self.script.main([
"--expectations", '["not", "a", "mapping"]',
])
# ------------- Happy path / mismatches -------
@patch("requests.head")
def test_all_ok_returns_zero(self, mock_head):
def head_side_effect(url, allow_redirects=False, timeout=10):
class R: pass
r = R()
domain = url.split("://", 1)[1]
# both match expectations exactly
mapping = {"ok1.example.org": 200, "ok2.example.org": 301}
r.status_code = mapping.get(domain, 200)
return r
mock_head.side_effect = head_side_effect
exp = {
"ok1.example.org": [200, 302, 301],
"ok2.example.org": [301],
}
exit_code = self.script.main([
"--web-protocol", "https",
"--expectations", self._to_json(exp),
])
self.assertEqual(exit_code, 0)
@patch("requests.head")
def test_mismatches_counted(self, mock_head):
def head_side_effect(url, allow_redirects=False, timeout=10):
class R: pass
r = R()
domain = url.split("://", 1)[1]
mapping = {"bad.example.org": 200, "ok301.example.org": 301}
r.status_code = mapping.get(domain, 200)
return r
mock_head.side_effect = head_side_effect
exp = {
"bad.example.org": [404], # mismatch (got 200)
"ok301.example.org": [301], # OK
"never.example.org": [200], # will default to 200 in side effect? No mapping -> 200 -> OK
}
# Adjust side effect to ensure "never.example.org" is OK 200
exit_code = self.script.main([
"--expectations", self._to_json(exp),
])
# only 'bad.example.org' mismatched
self.assertEqual(exit_code, 1)
@patch("requests.head")
def test_non_list_values_sanitize_to_empty_and_fail(self, mock_head):
# If a domain maps to a non-list, it becomes [] and is treated as a failure
def head_side_effect(url, allow_redirects=False, timeout=10):
class R: pass
r = R()
r.status_code = 200
return r
mock_head.side_effect = head_side_effect
exp_json = '{"foo.example.org": "not-a-list", "bar.example.org": 200}'
# Both entries get empty expectations -> 2 errors
exit_code = self.script.main([
"--expectations", exp_json,
])
self.assertEqual(exit_code, 2)
# ------------- Helpers -----------------------
@staticmethod
def _to_json(obj) -> str:
import json
return json.dumps(obj, separators=(",", ":"))
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,278 @@
# tests/unit/roles/sys-ctl-hlth-webserver/filter_plugins/test_web_health_expectations.py
import os
import unittest
import importlib.util
from unittest.mock import patch
def load_module_from_path(mod_name: str, path: str):
"""Dynamically load a module from a filesystem path."""
spec = importlib.util.spec_from_file_location(mod_name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # type: ignore[attr-defined]
return module
class TestWebHealthExpectationsFilter(unittest.TestCase):
@classmethod
def setUpClass(cls):
# Compute repo root from this test file location
here = os.path.abspath(os.path.dirname(__file__))
# tests/unit/roles/sys-ctl-hlth-webserver/filter_plugins/ -> repo root is 5 levels up
cls.ROOT = os.path.abspath(os.path.join(here, "..", "..", "..", "..", ".."))
cls.module_path = os.path.join(
cls.ROOT, "roles", "sys-ctl-hlth-webserver", "filter_plugins", "web_health_expectations.py"
)
if not os.path.isfile(cls.module_path):
raise FileNotFoundError(f"Cannot find web_health_expectations.py at {cls.module_path}")
# Load the filter module once for all tests
cls.mod = load_module_from_path("web_health_expectations", cls.module_path)
def setUp(self):
# Fresh mock for get_app_conf per test
self.get_app_conf_patch = patch.object(self.mod, "get_app_conf")
self.mock_get_app_conf = self.get_app_conf_patch.start()
def tearDown(self):
self.get_app_conf_patch.stop()
def _configure_returns(self, mapping):
"""
Provide a dict keyed by (app_id, key) -> value.
get_app_conf(...) will return mapping.get((app_id, key), default)
"""
def side_effect(applications, app_id, key, strict=False, default=None):
return mapping.get((app_id, key), default)
self.mock_get_app_conf.side_effect = side_effect
# ------------ Required selection --------------
def test_raises_when_group_names_missing(self):
apps = {"app-a": {}}
with self.assertRaises(ValueError):
self.mod.web_health_expectations(apps, group_names=None)
def test_raises_when_group_names_empty_variants(self):
apps = {"app-a": {}}
with self.assertRaises(ValueError):
self.mod.web_health_expectations(apps, group_names=[])
with self.assertRaises(ValueError):
self.mod.web_health_expectations(apps, group_names="")
with self.assertRaises(ValueError):
self.mod.web_health_expectations(apps, group_names=" , ")
# ---- Non-mapping apps short-circuit (but group_names still required) ----
def test_non_mapping_returns_empty_dict(self):
expectations = self.mod.web_health_expectations(applications=["not", "a", "mapping"], group_names=["any"])
self.assertEqual(expectations, {})
# ------------ Flat canonical -----------------
def test_flat_canonical_with_default_status(self):
apps = {"app-a": {}}
self._configure_returns({
("app-a", "server.domains.canonical"): ["a.example.org"],
("app-a", "server.domains.aliases"): [],
("app-a", "server.status_codes"): {"default": 405},
})
out = self.mod.web_health_expectations(apps, group_names=["app-a"])
self.assertEqual(out["a.example.org"], [405])
def test_flat_canonical_invalid_default_falls_back_to_DEFAULT_OK(self):
apps = {"app-x": {}}
self._configure_returns({
("app-x", "server.domains.canonical"): ["x.example.org"],
("app-x", "server.domains.aliases"): [],
("app-x", "server.status_codes"): {"default": 700}, # invalid HTTP code
})
out = self.mod.web_health_expectations(apps, group_names=["app-x"])
self.assertEqual(out["x.example.org"], [200, 302, 301])
# ------------ Keyed canonical ----------------
def test_keyed_canonical_with_per_key_overrides_and_default(self):
apps = {"app-d": {}}
self._configure_returns({
("app-d", "server.domains.canonical"): {
"api": "api.d.example.org",
"web": "web.d.example.org",
"view": ["v1.d.example.org", "v2.d.example.org"],
},
("app-d", "server.domains.aliases"): ["alias.d.example.org"],
("app-d", "server.status_codes"): {"api": 404, "default": 405},
})
out = self.mod.web_health_expectations(apps, group_names=["app-d"])
self.assertEqual(out["api.d.example.org"], [404]) # per-key override wins
self.assertEqual(out["web.d.example.org"], [405]) # default used
self.assertEqual(out["v1.d.example.org"], [405]) # default used
self.assertEqual(out["v2.d.example.org"], [405]) # default used
self.assertEqual(out["alias.d.example.org"], [301]) # aliases always redirect
def test_keyed_canonical_invalid_key_and_default_falls_back(self):
apps = {"app-y": {}}
self._configure_returns({
("app-y", "server.domains.canonical"): {"web": ["y.example.org"]},
("app-y", "server.domains.aliases"): [],
("app-y", "server.status_codes"): {"web": 999}, # invalid; default missing
})
out = self.mod.web_health_expectations(apps, group_names=["app-y"])
self.assertEqual(out["y.example.org"], [200, 302, 301])
# ------------ Selection by group_names -------
def test_selection_by_group_names_list(self):
apps = {"app-a": {}, "app-b": {}, "app-c": {}}
self._configure_returns({
("app-a", "server.domains.canonical"): ["a.example.org"],
("app-a", "server.domains.aliases"): [],
("app-a", "server.status_codes"): {"default": 200},
("app-b", "server.domains.canonical"): ["b.example.org"],
("app-b", "server.domains.aliases"): [],
("app-b", "server.status_codes"): {"default": 405},
("app-c", "server.domains.canonical"): ["c.example.org"],
("app-c", "server.domains.aliases"): ["alias.c.example.org"],
("app-c", "server.status_codes"): {},
})
out = self.mod.web_health_expectations(apps, group_names=["app-a", "app-c"])
self.assertIn("a.example.org", out)
self.assertIn("c.example.org", out)
self.assertIn("alias.c.example.org", out)
self.assertNotIn("b.example.org", out)
def test_selection_by_group_names_string(self):
apps = {"app-a": {}, "app-b": {}}
self._configure_returns({
("app-a", "server.domains.canonical"): ["a.example.org"],
("app-a", "server.domains.aliases"): [],
("app-a", "server.status_codes"): {"default": 200},
("app-b", "server.domains.canonical"): ["b.example.org"],
("app-b", "server.domains.aliases"): [],
("app-b", "server.status_codes"): {"default": 405},
})
out = self.mod.web_health_expectations(apps, group_names="app-a, app-c ")
self.assertIn("a.example.org", out)
self.assertNotIn("b.example.org", out)
# ------------ Aliases & filtering ------------
def test_aliases_are_always_301(self):
apps = {"app-f": {}}
self._configure_returns({
("app-f", "server.domains.canonical"): ["f.example.org"],
("app-f", "server.domains.aliases"): ["alias1.example.org", "alias2.example.org"],
("app-f", "server.status_codes"): {"default": 200},
})
out = self.mod.web_health_expectations(apps, group_names=["app-f"])
self.assertEqual(out["alias1.example.org"], [301])
self.assertEqual(out["alias2.example.org"], [301])
self.assertEqual(out["f.example.org"], [200])
def test_non_string_entries_in_lists_are_dropped(self):
apps = {"app-g": {}}
self._configure_returns({
("app-g", "server.domains.canonical"): ["ok.g.example.org", None, 123, {"x": "y"}],
("app-g", "server.domains.aliases"): [{"bad": "obj"}, "alias.g.example.org", None],
("app-g", "server.status_codes"): {}, # → fallback
})
out = self.mod.web_health_expectations(apps, group_names=["app-g"])
self.assertIn("ok.g.example.org", out)
self.assertEqual(out["alias.g.example.org"], [301])
self.assertNotIn(123, out)
# ------------ WWW mapping (flag) --------------
def test_www_mapping_is_added_and_forced_to_301_when_enabled(self):
apps = {"app-h": {}}
# includes a canonical that already starts with www.
self._configure_returns({
("app-h", "server.domains.canonical"): ["h.example.org", "www.keep301.example.org"],
("app-h", "server.domains.aliases"): ["alias.h.example.org"],
("app-h", "server.status_codes"): {"default": 405},
})
out = self.mod.web_health_expectations(apps, group_names=["app-h"], www_enabled=True)
# base domains
self.assertEqual(out["h.example.org"], [405])
self.assertEqual(out["alias.h.example.org"], [301])
# auto-generated www.* entries always 301
self.assertEqual(out["www.h.example.org"], [301])
self.assertEqual(out["www.alias.h.example.org"], [301])
# any pre-existing www.* must be forced to 301 too
self.assertEqual(out["www.keep301.example.org"], [301])
def test_no_www_mapping_when_disabled(self):
apps = {"app-i": {}}
self._configure_returns({
("app-i", "server.domains.canonical"): ["i.example.org"],
("app-i", "server.domains.aliases"): [],
("app-i", "server.status_codes"): {"default": 200},
})
out = self.mod.web_health_expectations(apps, group_names=["app-i"], www_enabled=False)
self.assertIn("i.example.org", out)
self.assertNotIn("www.i.example.org", out)
# ------------ redirect_maps -------------------
def test_redirect_maps_sources_are_included_as_301(self):
apps = {}
out = self.mod.web_health_expectations(
apps,
group_names=["any"], # required, even if no apps
redirect_maps=[{"source": "mail.example.org"}, "legacy.example.org"]
)
self.assertEqual(out["mail.example.org"], [301])
self.assertEqual(out["legacy.example.org"], [301])
def test_redirect_maps_override_app_expectations(self):
apps = {"conflict-app": {}}
self._configure_returns({
("conflict-app", "server.domains.canonical"): ["conflict.example.org"],
("conflict-app", "server.domains.aliases"): [],
("conflict-app", "server.status_codes"): {"default": 200},
})
out = self.mod.web_health_expectations(
apps,
group_names=["conflict-app"],
redirect_maps=[{"source": "conflict.example.org"}]
)
self.assertEqual(out["conflict.example.org"], [301])
def test_redirect_maps_get_www_when_enabled(self):
apps = {}
out = self.mod.web_health_expectations(
apps,
group_names=["any"],
www_enabled=True,
redirect_maps=[{"source": "redir.example.org"}]
)
self.assertEqual(out["redir.example.org"], [301])
self.assertEqual(out["www.redir.example.org"], [301])
def test_redirect_maps_independent_of_group_filter(self):
apps = {"ignored-app": {}}
self._configure_returns({
("ignored-app", "server.domains.canonical"): ["ignored.example.org"],
("ignored-app", "server.domains.aliases"): [],
("ignored-app", "server.status_codes"): {"default": 200},
})
out = self.mod.web_health_expectations(
apps,
group_names=["some-other-app"], # excludes the only app
redirect_maps=[{"source": "manual.example.org"}]
)
self.assertNotIn("ignored.example.org", out)
self.assertEqual(out["manual.example.org"], [301])
if __name__ == "__main__":
unittest.main()