Build guided media issue workflow
Magent CI/CD / verify (push) Successful in 10m42s
Magent CI/CD / deploy-prod (push) Skipped
Magent CI/CD / deploy-beta (push) Successful in 19s

This commit is contained in:
2026-08-31 21:46:43 +12:00
parent 0e04d219a0
commit f8770cb44a
5 changed files with 1124 additions and 37 deletions
+61
View File
@@ -1166,6 +1166,67 @@ class AuthFlowTests(TempDatabaseMixin, unittest.IsolatedAsyncioTestCase):
)
class PortalMediaStatusTests(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
portal_router._MEDIA_STATUS_CACHE.update(expires_at=0.0, payload=None)
async def test_media_status_is_live_and_removes_session_identity(self) -> None:
client = SimpleNamespace(
configured=lambda: True,
get_system_info=AsyncMock(
return_value={
"Version": "10.10.7",
"HasPendingRestart": False,
"WanAddress": "https://private.example",
}
),
get_sessions=AsyncMock(
return_value=[
{
"UserName": "private-user",
"DeviceName": "Living room television",
"NowPlayingItem": {"Name": "Private title"},
"TranscodingInfo": {"VideoCodec": "h264"},
}
]
),
)
with (
patch.object(portal_router, "get_runtime_settings", return_value=SimpleNamespace(
jellyfin_base_url="http://jellyfin",
jellyfin_api_key="secret",
)),
patch.object(portal_router, "JellyfinClient", return_value=client),
):
result = await portal_router.portal_media_status()
self.assertEqual(result["status"], "up")
self.assertEqual(result["activity"]["active_streams"], 1)
self.assertEqual(result["activity"]["transcoding_streams"], 1)
serialized = str(result)
self.assertNotIn("private-user", serialized)
self.assertNotIn("Living room television", serialized)
self.assertNotIn("Private title", serialized)
self.assertNotIn("private.example", serialized)
async def test_media_status_reports_unavailable_without_exposing_exception(self) -> None:
client = SimpleNamespace(
configured=lambda: True,
get_system_info=AsyncMock(side_effect=RuntimeError("secret upstream failure")),
)
with (
patch.object(portal_router, "get_runtime_settings", return_value=SimpleNamespace(
jellyfin_base_url="http://jellyfin",
jellyfin_api_key="secret",
)),
patch.object(portal_router, "JellyfinClient", return_value=client),
):
result = await portal_router.portal_media_status()
self.assertEqual(result["status"], "down")
self.assertNotIn("secret upstream failure", str(result))
class PortalWorkflowTests(TempDatabaseMixin, unittest.TestCase):
def test_legacy_request_status_maps_to_workflow(self) -> None:
item = {"kind": "request", "status": "in_progress"}