supervisor/tests/test_ingress.py
Florian Bachmann acc0e5c989
Allows the supervisor to send a session's user to addon with header X-Remote-User (#4152)
* Working draft for x-remote-user

* Renames prop to remote_user

* Allows to set in addon description whether it requests the username

* Fixes addon-options schema

* Sends user ID instead of username to addons

* Adds tests

* Removes configurability of remote-user forwarding

* Update const.py

* Also adds username header

* Fetches full user info object from homeassistant

* Cleaner validation and dataclasses

* Fixes linting

* Fixes linting

* Tries to fix test

* Updates tests

* Updates tests

* Updates tests

* Updates tests

* Updates tests

* Updates tests

* Updates tests

* Updates tests

* Resolves PR comments

* Linting

* Fixes tests

* Update const.py

* Removes header keys if not required

* Moves ignoring user ID headers if no session_data is given

* simplify

* fix lint with new job

---------

Co-authored-by: Pascal Vizeli <pvizeli@syshack.ch>
Co-authored-by: Pascal Vizeli <pascal.vizeli@syshack.ch>
2023-08-22 10:11:13 +02:00

65 lines
1.8 KiB
Python

"""Test ingress."""
from datetime import timedelta
from supervisor.const import ATTR_SESSION_DATA_USER_ID
from supervisor.utils.dt import utc_from_timestamp
def test_session_handling(coresys):
"""Create and test session."""
session = coresys.ingress.create_session()
validate = coresys.ingress.sessions[session]
assert session
assert validate
assert coresys.ingress.validate_session(session)
assert coresys.ingress.sessions[session] != validate
not_valid = utc_from_timestamp(validate) - timedelta(minutes=20)
coresys.ingress.sessions[session] = not_valid.timestamp()
assert not coresys.ingress.validate_session(session)
assert not coresys.ingress.validate_session("invalid session")
session_data = coresys.ingress.get_session_data(session)
assert session_data is None
def test_session_handling_with_session_data(coresys):
"""Create and test session."""
session = coresys.ingress.create_session(
dict([(ATTR_SESSION_DATA_USER_ID, "some-id")])
)
assert session
session_data = coresys.ingress.get_session_data(session)
assert session_data[ATTR_SESSION_DATA_USER_ID] == "some-id"
async def test_save_on_unload(coresys):
"""Test called save on unload."""
coresys.ingress.create_session()
await coresys.ingress.unload()
assert coresys.ingress.save_data.called
def test_dynamic_ports(coresys):
"""Test dyanmic port handling."""
port_test1 = coresys.ingress.get_dynamic_port("test1")
assert port_test1
assert coresys.ingress.save_data.called
assert port_test1 == coresys.ingress.get_dynamic_port("test1")
port_test2 = coresys.ingress.get_dynamic_port("test2")
assert port_test2
assert port_test2 != port_test1
assert port_test2 > 62000
assert port_test2 < 65500
assert port_test1 > 62000
assert port_test1 < 65500