mirror of
https://github.com/flobz/psa_car_controller.git
synced 2026-08-26 10:17:18 +00:00
fix: ha ingress compat
This commit is contained in:
@@ -7,8 +7,7 @@ from flask import Flask
|
||||
from werkzeug import run_simple
|
||||
from werkzeug.middleware.proxy_fix import ProxyFix
|
||||
|
||||
from psa_car_controller.web.dash_custom import DashCustom
|
||||
|
||||
from dash import Dash
|
||||
try:
|
||||
from werkzeug.middleware.dispatcher import DispatcherMiddleware
|
||||
except ImportError:
|
||||
@@ -28,30 +27,26 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MyProxyFix(ProxyFix):
|
||||
def __init__(self, dashapp):
|
||||
def __init__(self, dashapp, static_prefix):
|
||||
self.flask_app = dashapp.server
|
||||
self.dash_app = dash_app
|
||||
self.dash_app = dashapp
|
||||
self.static_prefix = static_prefix
|
||||
super().__init__(self.flask_app.wsgi_app, x_host=1, x_port=1, x_prefix=1)
|
||||
|
||||
def __call__(self, environ, start_response):
|
||||
prefix = environ.get("HTTP_X_INGRESS_PATH")
|
||||
if prefix:
|
||||
environ["HTTP_X_FORWARDED_PREFIX"] = prefix
|
||||
self.flask_app.config['APPLICATION_ROOT'] = environ['SCRIPT_NAME'] = prefix
|
||||
prefix += "/"
|
||||
self.dash_app.requests_pathname_external_prefix = prefix
|
||||
self.dash_app.config.assets_external_path = prefix
|
||||
prefix = environ.get("HTTP_X_INGRESS_PATH") or environ.get("HTTP_X_FORWARDED_PREFIX") or self.static_prefix
|
||||
self.dash_app.config.__dict__["_read_only"] = []
|
||||
|
||||
if prefix == "/":
|
||||
self.dash_app.config.requests_pathname_prefix = ""
|
||||
self.dash_app.config.url_base_pathname = None
|
||||
else:
|
||||
# In Dash 4.0, requests_pathname_prefix is "/" when base_path is "/"
|
||||
# but APPLICATION_ROOT should be None (not "/") for root-mounted apps
|
||||
rpp = self.dash_app.config.requests_pathname_prefix
|
||||
# APPLICATION_ROOT should not have trailing slash (Flask convention)
|
||||
if rpp and rpp != "/":
|
||||
self.flask_app.config['APPLICATION_ROOT'] = rpp.rstrip('/')
|
||||
else:
|
||||
self.flask_app.config['APPLICATION_ROOT'] = None
|
||||
# requests_pathname_external_prefix should be empty string for root
|
||||
self.dash_app.requests_pathname_external_prefix = rpp if rpp and rpp != "/" else ""
|
||||
if not prefix.endswith("/"):
|
||||
prefix += "/"
|
||||
self.dash_app.config.requests_pathname_prefix = prefix
|
||||
self.dash_app.config.url_base_pathname = prefix
|
||||
self.dash_app.config.assets_external_path = prefix
|
||||
|
||||
return super().__call__(environ, start_response)
|
||||
|
||||
|
||||
@@ -85,12 +80,12 @@ def config_flask(title, base_path, debug: bool, host, port, reloader=False,
|
||||
else:
|
||||
application = DispatcherMiddleware(Flask('dummy_app'), {base_path: app})
|
||||
requests_pathname_prefix = base_path + "/"
|
||||
dash_app = DashCustom(external_stylesheets=[dbc.themes.BOOTSTRAP, dbc.icons.BOOTSTRAP],
|
||||
external_scripts=locale_url, title=title,
|
||||
server=app, requests_pathname_prefix=requests_pathname_prefix,
|
||||
suppress_callback_exceptions=True)
|
||||
dash_app = Dash(external_stylesheets=[dbc.themes.BOOTSTRAP, dbc.icons.BOOTSTRAP],
|
||||
external_scripts=locale_url, title=title,
|
||||
server=app, requests_pathname_prefix=requests_pathname_prefix,
|
||||
suppress_callback_exceptions=True)
|
||||
dash_app.enable_dev_tools(debug)
|
||||
app.wsgi_app = MyProxyFix(dash_app)
|
||||
app.wsgi_app = MyProxyFix(dash_app, base_path)
|
||||
# keep this line
|
||||
importlib.import_module(view)
|
||||
if reload_view:
|
||||
|
||||
@@ -1,23 +0,0 @@
|
||||
from dash import Dash
|
||||
|
||||
|
||||
class DashCustom(Dash):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.requests_pathname_external_prefix = self.config.requests_pathname_prefix
|
||||
|
||||
def _config(self):
|
||||
try:
|
||||
config = super()._config()
|
||||
# pieces of config needed by the front end
|
||||
config.update({
|
||||
"requests_pathname_prefix": self.requests_pathname_external_prefix,
|
||||
})
|
||||
if hasattr(self, "_dev_tools"):
|
||||
config.update({
|
||||
"ui": getattr(self._dev_tools, "ui", True),
|
||||
"props_check": getattr(self._dev_tools, "props_check", False),
|
||||
})
|
||||
return config
|
||||
except Exception:
|
||||
return self.config
|
||||
@@ -57,11 +57,11 @@ def add_header(el):
|
||||
color="secondary",
|
||||
className="me-1 bi bi-github",
|
||||
external_link=True, href=github_url)
|
||||
return dbc.Row([dbc.Col(dcc.Link(html.H1('My car info'), href=dash_app.requests_pathname_external_prefix,
|
||||
return dbc.Row([dbc.Col(dcc.Link(html.H1('My car info'), href=dash_app.config.requests_pathname_prefix,
|
||||
style={"TextDecoration": "none"})),
|
||||
dbc.Col(html.Div([dbc_version,
|
||||
dcc.Link(html.Img(src="assets/images/settings.svg", width="30veh"),
|
||||
href=dash_app.requests_pathname_external_prefix + "config",
|
||||
href=dash_app.config.requests_pathname_prefix + "config",
|
||||
className="float-end")],
|
||||
className="d-grid gap-2 d-md-flex justify-content-md-end",))],
|
||||
className='align-items-center'), el
|
||||
@@ -71,7 +71,7 @@ def add_header(el):
|
||||
[Input('url', 'pathname'),
|
||||
Input('url', 'search')])
|
||||
def display_page(pathname, search):
|
||||
prefix = dash_app.requests_pathname_external_prefix or "/"
|
||||
prefix = dash_app.config.requests_pathname_prefix or "/"
|
||||
pathname = pathname[len(prefix) - 1:]
|
||||
query_params = parse_qs(urlparse(search).query)
|
||||
no_header = query_params.get("header", None) == ["false"]
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
{
|
||||
"abrp": {
|
||||
"abrp_enable_vin": [],
|
||||
"token": null
|
||||
},
|
||||
"client_id": "cid",
|
||||
@@ -7,7 +8,10 @@
|
||||
"co2_signal_api": null,
|
||||
"country_code": "FR",
|
||||
"customer_id": "AP-ACNT1234",
|
||||
"proxies": null,
|
||||
"proxies": {
|
||||
"http": "",
|
||||
"https": ""
|
||||
},
|
||||
"realm": "clientsB2CPeugeot",
|
||||
"refresh_token": "aasds-aaa",
|
||||
"remote_refresh_token": "aa",
|
||||
|
||||
@@ -8,10 +8,12 @@ the HTML response contain the prefix.
|
||||
Run with: python -m unittest tests.test_ha_ingress_integration -v
|
||||
or: python tests/test_ha_ingress_integration.py
|
||||
"""
|
||||
import psa_car_controller
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import re
|
||||
import socket
|
||||
import requests
|
||||
import threading
|
||||
from unittest import TestCase, skipIf
|
||||
@@ -21,14 +23,16 @@ from unittest.mock import patch, MagicMock
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
|
||||
try:
|
||||
import psa_car_controller
|
||||
HAS_APP = True
|
||||
except ImportError:
|
||||
HAS_APP = False
|
||||
def is_port_free(port, host='127.0.0.1'):
|
||||
"""Check if a port is free on the specified host."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
try:
|
||||
s.bind((host, port))
|
||||
return True
|
||||
except socket.error:
|
||||
return False
|
||||
|
||||
|
||||
@skipIf(not HAS_APP, "App dependencies not available")
|
||||
class TestHomeAssistantIngressIntegration(TestCase):
|
||||
"""
|
||||
Integration tests for PSA Car Controller with Home Assistant X-Ingress-Path header.
|
||||
@@ -42,6 +46,10 @@ class TestHomeAssistantIngressIntegration(TestCase):
|
||||
cls.server_port = 18080
|
||||
cls.server_url = f"http://127.0.0.1:{cls.server_port}"
|
||||
|
||||
# Check if port is free before starting server
|
||||
if not is_port_free(cls.server_port):
|
||||
raise RuntimeError(f"Port {cls.server_port} is already in use")
|
||||
|
||||
# Start the PSA app in a thread
|
||||
def start_server():
|
||||
from psa_car_controller.__main__ import main
|
||||
@@ -49,6 +57,7 @@ class TestHomeAssistantIngressIntegration(TestCase):
|
||||
|
||||
# Override sys.argv to pass the required arguments
|
||||
original_argv = sys.argv
|
||||
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "config.json")
|
||||
try:
|
||||
# Set up command line arguments
|
||||
sys.argv = [
|
||||
@@ -60,7 +69,7 @@ class TestHomeAssistantIngressIntegration(TestCase):
|
||||
'-l', '127.0.0.1',
|
||||
'-p', str(cls.server_port),
|
||||
'-b', '/',
|
||||
'-f', 'config/config.json'
|
||||
'-f', config_path
|
||||
]
|
||||
|
||||
# Run the app
|
||||
@@ -107,21 +116,10 @@ class TestHomeAssistantIngressIntegration(TestCase):
|
||||
except requests.exceptions.RequestException:
|
||||
self.skipTest("Server not responding")
|
||||
|
||||
def test_root_route_with_ingress_path(self):
|
||||
"""
|
||||
Test that / route returns HTML with hrefs containing the prefix
|
||||
when X-Ingress-Path header is present.
|
||||
"""
|
||||
prefix = '/api/ingress/a93a74ea_psacc'
|
||||
headers = {
|
||||
'X-Ingress-Path': prefix,
|
||||
'Host': f'127.0.0.1:{self.server_port}'
|
||||
}
|
||||
|
||||
url = f"{self.server_url}/"
|
||||
def validate_path(self, headers, expected_prefix):
|
||||
|
||||
try:
|
||||
response = requests.get(url, headers=headers, timeout=10)
|
||||
response = requests.get(self.server_url, headers=headers, timeout=10)
|
||||
|
||||
self.assertEqual(response.status_code, 200,
|
||||
f"Expected 200, got {response.status_code}")
|
||||
@@ -130,54 +128,39 @@ class TestHomeAssistantIngressIntegration(TestCase):
|
||||
|
||||
# Check for prefix in HTML
|
||||
|
||||
scripts = re.findall(r'src="([^"]+)"', html)
|
||||
scripts = re.findall(r'src="(/[^"]+)"', html)
|
||||
for script in scripts:
|
||||
assert (not script.startswith("/")
|
||||
) or script.startswith(prefix), f"script {script} doesn't contain prefix"
|
||||
# The Dash app's initial HTML should contain the prefix in its config
|
||||
# Look for the Dash config which includes requests_pathname_prefix
|
||||
if 'requests_pathname_prefix' in html:
|
||||
import json
|
||||
# Try to extract the Dash config
|
||||
config_match = re.search(r'id="_dash-config"[^>]*>([^<]+)</script>', html)
|
||||
if config_match:
|
||||
config_str = config_match.group(1)
|
||||
try:
|
||||
config = json.loads(config_str)
|
||||
requests_prefix = config.get('requests_pathname_prefix', '')
|
||||
print(f"Dash config requests_pathname_prefix: {requests_prefix}")
|
||||
|
||||
# Check if it matches our prefix
|
||||
if prefix in requests_prefix:
|
||||
print(f"✓ Dash config has correct prefix")
|
||||
return
|
||||
else:
|
||||
print(f"✗ Dash config prefix doesn't match")
|
||||
print(f" Expected: {prefix}")
|
||||
print(f" Got: {requests_prefix}")
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
assert re.match(expected_prefix, script), f"Script {script} does not match {expected_prefix}"
|
||||
# Check if prefix appears in hrefs
|
||||
hrefs = re.findall(r'href="([^"]+)"', html)
|
||||
prefix_hrefs = [h for h in hrefs if prefix in h]
|
||||
|
||||
if prefix_hrefs:
|
||||
print(f"✓ Found {len(prefix_hrefs)} hrefs with prefix")
|
||||
else:
|
||||
# Check if prefix appears anywhere
|
||||
if prefix in html:
|
||||
print(f"✓ Prefix found in HTML")
|
||||
else:
|
||||
print(f"✗ Prefix NOT found in HTML")
|
||||
print(f"Sample hrefs: {hrefs[:10]}")
|
||||
with open('/tmp/test_root_response.html', 'w') as f:
|
||||
f.write(html)
|
||||
self.fail(f"Prefix '{prefix}' not found in HTML")
|
||||
hrefs = re.findall(f'href="(/[^"]+)"', html)
|
||||
for href in hrefs:
|
||||
assert re.match(expected_prefix, href), f"Href {href} does not match {expected_prefix}"
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
self.fail(f"Request failed: {e}")
|
||||
|
||||
def test_root_route_with_ingress_path(self):
|
||||
"""
|
||||
Test that / route returns HTML with hrefs containing the prefix
|
||||
when X-Ingress-Path header is present.
|
||||
"""
|
||||
prefix = r'/api/ingress/a93a74ea_psacc/.+'
|
||||
headers = {
|
||||
'X-Ingress-Path': prefix,
|
||||
'Host': f'127.0.0.1:{self.server_port}'
|
||||
}
|
||||
self.validate_path(headers, prefix)
|
||||
|
||||
def test_root_route_without(self):
|
||||
"""
|
||||
Test that / route returns HTML with hrefs containing the prefix
|
||||
when X-Ingress-Path header is present.
|
||||
"""
|
||||
headers = {
|
||||
'Host': f'127.0.0.1:{self.server_port}'
|
||||
}
|
||||
self.validate_path(headers, r"/(assets|_dash|_favicon).+")
|
||||
|
||||
def test_api_route_with_ingress_path(self):
|
||||
"""
|
||||
Test that Flask API routes (from api.py) work with prefix header.
|
||||
|
||||
Reference in New Issue
Block a user