Mirrors/EOS
1
0
Fork 0
mirror of https://github.com/Akkudoktor-EOS/EOS.git synced 2026-11-13 01:03:45 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity
Files
4475774c061e84ce109923faca83d4d86d6dd7dc
EOS/tests/test_server.py
T
Bobby NoelteandClaude Opus 5.5 4475774c06 test(eosdash): run EOSdash in a headless Chromium browser 
Run EOSdash in a headless Chromium browser for testing. Every menu tab must
load with HTTP 200 and render content, in light and dark mode, without
JavaScript errors.

Additional changes
Enable the browser tests by --browser-test (marker browser), skip them otherwise
Add the CI job browser-test that installs Chromium and runs them
Add the make target test-browser, run with --extra dev so playwright is installed
Add playwright to the dev dependencies
Document how to run the browser tests with --extra dev, incl. the browser system
libraries on Linux
Accept a child PID of the started server in the server startup checks, on Windows
only: there the venv python.exe runs the real interpreter as a child process
Share one Chromium per test class, so Playwright stops before other tests run
Request chromium before servers so a missing browser fails before EOS starts
Make the missing-browser hint platform neutral (Linux: add --with-deps)
Set PYTHONIOENCODING=utf-8 for the test server to avoid encoding errors on Windows

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-09 12:39:30 +01:00

562 lines
23 KiB
Python
Raw Blame History

import asyncio
import json
import os
import re
import time
from http import HTTPStatus
from pathlib import Path
import psutil
import pytest
import requests
from conftest import cleanup_eos_eosdash, is_owned_pid
from loguru import logger
from pydantic import ValidationError
from akkudoktoreos.config.configabc import is_home_assistant_addon
from akkudoktoreos.core.version import __version__
from akkudoktoreos.server.server import (
ServerCommonSettings,
get_default_host,
get_default_port,
get_host_ip,
wait_for_port_free,
)
@pytest.mark.skipif(os.name == "nt", reason="Directory symlinks require privileges on Windows")
def test_temp_directory_alias_is_contained(tmp_path, is_in_test_dir):
"""Resolved path containment accepts aliases but rejects sibling directories."""
target = tmp_path / "server"
target.mkdir()
alias = tmp_path / "alias"
alias.symlink_to(target, target_is_directory=True)
assert is_in_test_dir(alias / "config.json", str(target))
assert is_in_test_dir(target / "config.json", str(alias))
assert not is_in_test_dir(tmp_path / "server-other" / "config.json", str(target))
class TestServer:
def test_server_setup_for_class(self, server_setup_for_class, is_in_test_dir):
"""Ensure server is started."""
server = server_setup_for_class["server"]
eos_dir = server_setup_for_class["eos_dir"]
# Assure server is running
result = requests.get(f"{server}/v1/health", timeout=2)
assert result.status_code == HTTPStatus.OK
health = result.json()
assert health["status"] == "alive"
assert health["version"] == __version__
result = requests.get(f"{server}/v1/config", timeout=2)
assert result.status_code == HTTPStatus.OK
# Get testing config
config_json = result.json()
config_folder_path = Path(config_json["general"]["config_folder_path"])
config_file_path = Path(config_json["general"]["config_file_path"])
data_folder_path = Path(config_json["general"]["data_folder_path"])
data_ouput_path = Path(config_json["general"]["data_output_path"])
# Assure we are working in test environment
assert is_in_test_dir(config_folder_path, eos_dir)
assert is_in_test_dir(config_file_path, eos_dir)
assert is_in_test_dir(data_folder_path, eos_dir)
assert is_in_test_dir(data_ouput_path, eos_dir)
class TestServerSettingsValidation:
"""Test the port restrictions in ServerCommonSettings when running as HA addon."""
def test_ha_addon_default_ports_ok(self, config_eos, monkeypatch):
"""Default ports are accepted in HA addon mode."""
monkeypatch.setattr("akkudoktoreos.server.server.is_home_assistant_addon", lambda: True)
assert config_eos.server.port == get_default_port() # 8503
assert config_eos.server.eosdash_port == get_default_port() + 1 # 8504
def test_server_port_restriction_in_ha_addon(self, config_eos, monkeypatch):
"""Server port must be the default (8503) in HA addon mode."""
monkeypatch.setattr("akkudoktoreos.server.server.is_home_assistant_addon", lambda: True)
with pytest.raises(ValidationError) as excinfo:
config_eos.server.port = 9000
assert "Server port number `8503` for Home Assistant add-on can not be changed" in str(
excinfo.value
)
def test_eosdash_port_restriction_in_ha_addon(self, config_eos, monkeypatch):
"""EOSdash port must be the default (8504) in HA addon mode."""
monkeypatch.setattr("akkudoktoreos.server.server.is_home_assistant_addon", lambda: True)
with pytest.raises(ValidationError) as excinfo:
config_eos.server.eosdash_port = 9001
assert "EOSdash port number `8504` for Home Assistant add-on can not be changed" in str(
excinfo.value
)
def test_ports_allowed_when_not_ha_addon(self, config_eos):
"""Custom ports are allowed when not in HA addon mode."""
# is_home_assistant_addon() returns False by default in this test environment
config_eos.server.port = 9000
config_eos.server.eosdash_port = 9001
assert config_eos.server.port == 9000
assert config_eos.server.eosdash_port == 9001
class TestServerStartStop:
@pytest.mark.asyncio
async def test_forward_stream_truncates_very_long_line(self, monkeypatch, tmp_path):
"""Test logging from EOSdash can also handle very long lines."""
eos_dir = tmp_path
monkeypatch.setenv("EOS_DIR", str(eos_dir))
monkeypatch.setenv("EOS_CONFIG_DIR", str(eos_dir))
# Import after env vars are set
from akkudoktoreos.server.rest.starteosdash import (
EOSDASH_LOG_MAX_LINE_BYTES,
_eosdash_log_worker,
eosdash_log_queue,
forward_stream,
)
# ---- Ensure queue + worker are initialized ----
if eosdash_log_queue is None:
from akkudoktoreos.server.rest import starteosdash
starteosdash.eosdash_log_queue = asyncio.Queue(maxsize=10)
worker_task = asyncio.create_task(_eosdash_log_worker())
else:
worker_task = None
long_message = "X" * (EOSDASH_LOG_MAX_LINE_BYTES + 10_000)
raw_line = f"INFO some.module:123 some_func - {long_message}\n"
raw_bytes = raw_line.encode()
reader = asyncio.StreamReader()
reader.feed_data(raw_bytes)
reader.feed_eof()
# ---- Capture Loguru output ----
records = []
def sink(message):
records.append(message.record)
logger_id = logger.add(sink, level="INFO")
try:
await forward_stream(reader)
# Allow log worker to flush queue
await asyncio.sleep(0)
finally:
logger.remove(logger_id)
# Clean shutdown of worker (important for pytest)
if worker_task:
from akkudoktoreos.server.rest import starteosdash
if starteosdash.eosdash_log_queue:
starteosdash.eosdash_log_queue.put_nowait(None)
await worker_task
# ---- Assert ----
assert len(records) == 1, "Expected exactly one log record"
record = records[0]
msg = record["message"]
assert msg.endswith("[TRUNCATED]"), "Expected truncation marker"
assert len(msg) <= EOSDASH_LOG_MAX_LINE_BYTES + 20
@pytest.mark.asyncio
async def test_server_start_eosdash(self, config_eos, monkeypatch, tmp_path):
"""Test the EOSdash server startup from EOS.
Do not use any fixture as this will make pytest the owner of the EOSdash port.
Tests that:
1. EOSdash starts via the supervisor
2. The /eosdash/health endpoint returns OK
3. EOSdash reports correct status and version
4. EOSdash can be terminated cleanly
"""
eos_dir = tmp_path
monkeypatch.setenv("EOS_DIR", str(eos_dir))
monkeypatch.setenv("EOS_CONFIG_DIR", str(eos_dir))
# Import with environment vars set to prevent creation of EOS.config.json in wrong dir.
from akkudoktoreos.server.rest import starteosdash
config_eos.server.host = get_default_host()
config_eos.server.port = 8503
config_eos.server.eosdash_host = config_eos.server.host
config_eos.server.eosdash_port = 8504
timeout = 120
eosdash_server = f"http://{config_eos.server.eosdash_host}:{config_eos.server.eosdash_port}"
# Port may be blocked
assert wait_for_port_free(
config_eos.server.eosdash_port, timeout=120, waiting_app_name="EOSdash"
)
owned_processes: list[psutil.Process] = []
try:
await starteosdash.supervise_eosdash()
process = starteosdash.eosdash_proc
assert process is not None, "EOSdash supervisor did not start a process"
owned_processes.append(psutil.Process(process.pid))
startup = False
error = ""
for _ in range(int(timeout / 3)):
try:
resp = requests.get(f"{eosdash_server}/eosdash/health", timeout=2)
if resp.status_code == HTTPStatus.OK:
startup = True
break
error = f"{resp.status_code}, {str(resp.content)}"
except requests.RequestException as ex:
error = str(ex)
await asyncio.sleep(3)
assert startup, f"Connection to {eosdash_server}/eosdash/health failed: {error}"
health = resp.json()
assert health.get("status") == "alive"
assert health.get("version") == __version__
assert is_owned_pid(owned_processes[0], health.get("pid"))
# Terminate the process started by this test, then reap it via asyncio.
process.terminate()
await asyncio.wait_for(process.wait(), timeout=timeout)
try:
resp = requests.get(f"{eosdash_server}/eosdash/health", timeout=2)
except requests.RequestException:
pass
else:
assert resp.status_code != HTTPStatus.OK
finally:
cleanup_eos_eosdash(
host=config_eos.server.host,
port=config_eos.server.port,
eosdash_host=config_eos.server.eosdash_host,
eosdash_port=config_eos.server.eosdash_port,
server_timeout=timeout,
owned_processes=owned_processes,
config_dir=str(config_eos.general.config_folder_path),
)
if starteosdash.eosdash_proc is not None:
await asyncio.wait_for(starteosdash.eosdash_proc.wait(), timeout=timeout)
@pytest.mark.skipif(os.name == "nt", reason="Server restart not supported on Windows")
def test_server_restart(self, server_setup_for_function, is_system_test, is_in_test_dir):
"""Test server restart."""
server = server_setup_for_function["server"]
eos_dir = server_setup_for_function["eos_dir"]
timeout = server_setup_for_function["timeout"]
result = requests.get(f"{server}/v1/config")
assert result.status_code == HTTPStatus.OK
# Get testing config
config_json = result.json()
config_folder_path = Path(config_json["general"]["config_folder_path"])
config_file_path = Path(config_json["general"]["config_file_path"])
data_folder_path = Path(config_json["general"]["data_folder_path"])
data_ouput_path = Path(config_json["general"]["data_output_path"])
cache_file_path = data_folder_path.joinpath(config_json["cache"]["subpath"]).joinpath(
"cachefilestore.json"
)
# Assure we are working in test environment
assert is_in_test_dir(config_folder_path, eos_dir)
assert is_in_test_dir(config_file_path, eos_dir)
assert is_in_test_dir(data_folder_path, eos_dir)
assert is_in_test_dir(data_ouput_path, eos_dir)
if is_system_test:
# Prepare cache entry and get cached data
result = requests.put(f"{server}/v1/config/weather/provider", json="BrightSky")
assert result.status_code == HTTPStatus.OK
result = requests.post(f"{server}/v1/prediction/update/BrightSky")
assert result.status_code == HTTPStatus.OK
result = requests.get(f"{server}/v1/prediction/series?key=weather_temp_air")
assert result.status_code == HTTPStatus.OK
data = result.json()
assert data["data"] != {}
result = requests.put(f"{server}/v1/config/file")
assert result.status_code == HTTPStatus.OK
# Save cache
result = requests.post(f"{server}/v1/admin/cache/save")
assert result.status_code == HTTPStatus.OK
cache = result.json()
assert cache_file_path.exists()
result = requests.get(f"{server}/v1/admin/cache")
assert result.status_code == HTTPStatus.OK
cache = result.json()
result = requests.get(f"{server}/v1/health")
assert result.status_code == HTTPStatus.OK
pid = result.json()["pid"]
result = requests.post(f"{server}/v1/admin/server/restart")
assert result.status_code == HTTPStatus.OK
assert "Restarting EOS.." in result.json()["message"]
new_pid = result.json()["pid"]
# Wait for server to shut down
for retries in range(10):
try:
result = requests.get(f"{server}/v1/health", timeout=2)
if result.status_code == HTTPStatus.OK:
pid = result.json()["pid"]
if pid == new_pid:
# Already started
break
else:
break
except Exception as ex:
break
time.sleep(3)
# Assure EOS is up again
startup = False
error = ""
for retries in range(int(timeout / 5)):
try:
result = requests.get(f"{server}/v1/health", timeout=2)
if result.status_code == HTTPStatus.OK:
startup = True
break
error = f"{result.status_code}, {str(result.content)}"
except Exception as ex:
error = str(ex)
time.sleep(5)
assert startup, f"Connection to {server}/v1/health failed: {error}"
assert result.json()["status"] == "alive"
pid = result.json()["pid"]
assert pid == new_pid
result = requests.get(f"{server}/v1/admin/cache")
assert result.status_code == HTTPStatus.OK
new_cache = result.json()
assert cache.items() <= new_cache.items()
if is_system_test:
result = requests.get(f"{server}/v1/config")
assert result.status_code == HTTPStatus.OK
assert result.json()["weather"]["provider"] == "BrightSky"
# Wait for initialisation task to have finished
time.sleep(5)
result = requests.get(f"{server}/v1/prediction/series?key=weather_temp_air")
assert result.status_code == HTTPStatus.OK
assert result.json() == data
# Shutdown the newly created server
result = requests.post(f"{server}/v1/admin/server/shutdown")
assert result.status_code == HTTPStatus.OK
assert "Stopping EOS.." in result.json()["message"]
new_pid = result.json()["pid"]
class TestServerWithEnv:
eos_env = {
"EOS_SERVER__EOSDASH_PORT": "8555",
}
def test_server_setup_for_class(self, server_setup_for_class):
"""Ensure server is started with environment passed to configuration."""
server = server_setup_for_class["server"]
assert server_setup_for_class["eosdash_port"] == int(
self.eos_env["EOS_SERVER__EOSDASH_PORT"]
)
result = requests.get(f"{server}/v1/config")
assert result.status_code == HTTPStatus.OK
# Get testing config
config_json = result.json()
# Assure config got configuration from environment
assert config_json["server"]["eosdash_port"] == int(
self.eos_env["EOS_SERVER__EOSDASH_PORT"]
)
class TestEosdashRedirect:
"""Redirects to EOSdash must target an address the client can reach.
See https://github.com/Akkudoktor-EOS/EOS/issues/1320: the redirect was built from
the EOSdash bind address, so remote clients were sent to their own localhost.
"""
@pytest.fixture
def client(self, config_eos):
from fastapi.testclient import TestClient
from akkudoktoreos.server.eos import app
config_eos.server.eosdash_host = "127.0.0.1"
config_eos.server.eosdash_port = 8504
return TestClient(app, follow_redirects=False)
@pytest.mark.parametrize("host", ["localhost:8503", "127.0.0.1:8503"])
def test_root_redirect_uses_request_host(self, client, host):
"""The root redirect points to the host the client used, not to the bind address."""
response = client.get("/", headers={"Host": host})
assert response.status_code == HTTPStatus.SEE_OTHER
assert response.headers["location"] == f"http://{host.split(':')[0]}:8504/"
def test_root_redirect_uses_host_ip_of_the_eos_machine(self, client):
"""Access by the IP address of the EOS machine redirects to that address."""
host_ip = get_host_ip()
response = client.get("/", headers={"Host": f"{host_ip}:8503"})
assert response.status_code == HTTPStatus.SEE_OTHER
assert response.headers["location"] == f"http://{host_ip}:8504/"
def test_root_redirect_ignores_untrusted_forwarded_headers(self, client):
"""Raw forwarding headers cannot override the public dashboard address."""
response = client.get(
"/",
headers={
"Host": "localhost",
"X-Forwarded-Host": "eos.example.com",
"X-Forwarded-Proto": "https",
},
)
assert response.status_code == HTTPStatus.SEE_OTHER
assert response.headers["location"] == "http://localhost:8504/"
def test_root_redirect_keeps_local_host(self, client):
"""Local access still redirects to the local EOSdash."""
response = client.get("/", headers={"Host": "127.0.0.1:8503"})
assert response.status_code == HTTPStatus.SEE_OTHER
assert response.headers["location"] == "http://127.0.0.1:8504/"
def test_untrusted_request_host_is_not_reflected(self, client):
"""An unknown Host header must not become the redirect target."""
response = client.get("/", headers={"Host": "attacker.example"})
assert response.status_code == HTTPStatus.NOT_FOUND
assert "attacker.example:8504" not in response.text
assert "eosdash_public_url" in response.text
def test_untrusted_request_host_on_unknown_path(self, client):
"""The 404 page offers no link for an unknown Host header."""
response = client.get("/no-such-page", headers={"Host": "attacker.example"})
assert response.status_code == HTTPStatus.NOT_FOUND
assert "attacker.example:8504" not in response.text
def test_eosdash_path_redirect_keeps_path(self, client):
"""The path is preserved when redirecting to EOSdash."""
response = client.get("/eosdash/health", headers={"Host": "localhost:8503"})
assert response.status_code == HTTPStatus.SEE_OTHER
assert response.headers["location"] == "http://localhost:8504/eosdash/health"
@pytest.mark.parametrize(
"path",
[
"/eosdash:payload",
"/eosdash+custom:payload",
"/eosdash:%09payload",
"/eosdash:%0Apayload",
],
)
def test_scheme_bearing_path_is_rejected(self, client, path):
"""User paths must not replace the configured dashboard URL with a scheme."""
response = client.get(path, headers={"Host": "localhost:8503"})
assert response.status_code == HTTPStatus.NOT_FOUND
assert "location" not in response.headers
@pytest.mark.parametrize(
"path",
["eosdash:payload", "./eosdash:payload", "eosdash:\tpayload", "eosdash:\npayload"],
)
def test_sanitizer_rejects_schemes_after_normalization(self, path):
"""Normalization must not turn an accepted path into a scheme-bearing URL."""
from akkudoktoreos.server.eos import _sanitize_redirect_path
assert _sanitize_redirect_path(path) is None
def test_unknown_path_error_page_links_to_request_host(self, client):
"""The 404 page links to EOSdash on the host the client used."""
response = client.get("/no-such-page", headers={"Host": "localhost:8503"})
assert response.status_code == HTTPStatus.NOT_FOUND
# The link must be real HTML, the error page escapes the message it is given.
assert "&lt;a href" not in response.text
# Compare the whole link target, a substring check would also accept a foreign host.
hrefs = [href for href in re.findall(r'href="([^"]*)"', response.text) if href != "/docs"]
assert hrefs == ["http://localhost:8504/"]
def test_error_page_escapes_request_url(self, client):
"""A crafted URL is shown as text, never as markup."""
response = client.get(
"/%3Cscript%3Ealert(1)%3C/script%3E", headers={"Host": "localhost:8503"}
)
assert response.status_code == HTTPStatus.NOT_FOUND
assert "<script>alert(1)</script>" not in response.text
@pytest.mark.parametrize("host", ["[::1]", "[::1]:8503"])
def test_direct_ipv6_preserves_address(self, client, host):
"""An IPv6 address keeps its brackets in the redirect."""
response = client.get("/", headers={"Host": host})
assert response.headers["location"] == "http://[::1]:8504/"
def test_untrusted_ipv6_host_is_not_reflected(self, client):
"""An IPv6 address that the configuration does not know is not reflected."""
response = client.get("/", headers={"Host": "[2001:db8::1234]:8503"})
assert response.status_code == HTTPStatus.NOT_FOUND
assert "2001:db8::1234" not in response.headers.get("location", "")
@pytest.mark.parametrize(
"public_url",
[
"https://energy.example.com",
"https://energy.example.com:443",
"https://energy.example.com:9443/dashboard",
"https://[2001:db8::1234]/dashboard",
],
)
def test_public_url_preserves_proxy_port_and_prefix(self, client, config_eos, public_url):
config_eos.server.eosdash_public_url = public_url + "/"
headers = {"Host": "internal:8503", "X-Forwarded-Host": "wrong.example"}
response = client.get("/", headers=headers)
assert response.headers["location"] == public_url + "/"
response = client.get("/eosdash/health", headers=headers)
assert response.headers["location"] == public_url + "/eosdash/health"
response = client.get("/missing", headers=headers)
hrefs = [href for href in re.findall(r'href="([^"]*)"', response.text) if href != "/docs"]
assert hrefs == [public_url + "/"]
@pytest.mark.parametrize(
"value",
[
"",
"/dashboard",
"//example.com",
"ftp://example.com",
"https://user:password@example.com",
"https://example.com?token=a",
"https://example.com#fragment",
"https://example.com:99999",
"https://example.com:0",
"https://example.com\\evil",
"https://example.com/\nheader",
"https://example.com/a b",
],
)
def test_invalid_public_url_rejected(self, config_eos, value):
with pytest.raises(ValueError):
config_eos.server.eosdash_public_url = value
Reference in New Issue View Git Blame Copy Permalink