tommal/fiscality-maschera
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Aggiunto un log migliore

Browse Source
This commit is contained in:
Tomas Mali
2026-07-01 16:02:31 +02:00
parent 4b47a65fbc
commit 8891e78b1d
2 changed files with 46 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
manager-solution/manager/test_manager.py
View File

@@ -192,14 +192,13 @@ class ManagerConfigTests(unittest.TestCase):
)
users = {"user2": {"id": "user2"}}
user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users)
user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users)
self.assertEqual(user["id"], "user2")
self.assertEqual(rewritten, request)
self.assertIsNone(redirect_target)
self.assertFalse(selected_via_path)
def test_extract_user_from_path_rewrites_request_for_direct_proxying(self):
def test_extract_user_from_path_returns_redirect_target(self):
request = (
b"GET /user1/websockify?token=abc HTTP/1.1\r\n"
b"Host: localhost:8080\r\n"
@@ -208,112 +207,11 @@ class ManagerConfigTests(unittest.TestCase):
)
users = {"user1": {"id": "user1"}}
user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users)
self.assertEqual(user["id"], "user1")
self.assertEqual(
rewritten,
(
b"GET /websockify?token=abc HTTP/1.1\r\n"
b"Host: localhost:8080\r\n"
b"Upgrade: websocket\r\n"
b"\r\n"
),
)
self.assertIsNone(redirect_target)
self.assertTrue(selected_via_path)
def test_extract_user_from_bare_user_path_rewrites_to_root(self):
request = (
b"GET /user1 HTTP/1.1\r\n"
b"Host: 10.200.200.113:8080\r\n"
b"Connection: keep-alive\r\n"
b"\r\n"
)
users = {"user1": {"id": "user1"}}
user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users)
self.assertEqual(user["id"], "user1")
self.assertEqual(
rewritten,
(
b"GET / HTTP/1.1\r\n"
b"Host: 10.200.200.113:8080\r\n"
b"Connection: keep-alive\r\n"
b"\r\n"
),
)
self.assertIsNone(redirect_target)
self.assertTrue(selected_via_path)
def test_extract_user_from_absolute_form_request_rewrites_to_root(self):
request = (
b"GET http://10.200.200.113:8080/user1 HTTP/1.1\r\n"
b"Host: 10.200.200.113:8080\r\n"
b"Connection: keep-alive\r\n"
b"\r\n"
)
users = {"user1": {"id": "user1"}}
user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users)
self.assertEqual(user["id"], "user1")
self.assertEqual(
rewritten,
(
b"GET / HTTP/1.1\r\n"
b"Host: 10.200.200.113:8080\r\n"
b"Connection: keep-alive\r\n"
b"\r\n"
),
)
self.assertIsNone(redirect_target)
self.assertTrue(selected_via_path)
def test_extract_user_from_cookie_route(self):
request = (
b"GET / HTTP/1.1\r\n"
b"Host: 10.200.200.113:8080\r\n"
b"Cookie: fis_user_route=user2\r\n"
b"Connection: keep-alive\r\n"
b"\r\n"
)
users = {"user2": {"id": "user2"}}
user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users)
self.assertEqual(user["id"], "user2")
self.assertEqual(rewritten, request)
self.assertIsNone(redirect_target)
self.assertFalse(selected_via_path)
def test_remembered_gateway_route_returns_user_within_ttl(self):
self.manager.users_by_id = {"user1": {"id": "user1"}}
self.manager.gateway_client_routes.clear()
self.manager.time = types.SimpleNamespace(time=lambda: 100.0)
self.manager.GATEWAY_ROUTE_TTL_SECONDS = 60
self.manager.remember_gateway_route("10.200.200.136", "user1")
user = self.manager.get_remembered_gateway_user("10.200.200.136")
self.assertEqual(user["id"], "user1")
def test_remembered_gateway_route_expires_after_ttl(self):
now = {"value": 100.0}
self.manager.users_by_id = {"user1": {"id": "user1"}}
self.manager.gateway_client_routes.clear()
self.manager.time = types.SimpleNamespace(time=lambda: now["value"])
self.manager.GATEWAY_ROUTE_TTL_SECONDS = 60
self.manager.remember_gateway_route("10.200.200.136", "user1")
now["value"] = 161.0
user = self.manager.get_remembered_gateway_user("10.200.200.136")
user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users)
self.assertIsNone(user)
self.assertEqual(self.manager.gateway_client_routes, {})
self.assertEqual(rewritten, request)
self.assertEqual(redirect_target, "http://user1.localhost:8080/websockify?token=abc")
def test_wait_until_ready_applies_stabilization_delay(self):
sleep_calls = []