Coverage for tests/streamwise/test_streamwise_files.py: 100%
80 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1#!/usr/bin/env python3
2"""
3Unit tests for streamwise.py file-related endpoints.
4"""
6import sys
7import pytest
9from http import HTTPStatus
11from unittest.mock import patch
12from unittest.mock import MagicMock
14from quart.typing import TestClientProtocol
16from tests.test_utils import temp_sys_path
17from tests.k8s_mock import K8sMock
18from tests.torch_mock import TorchMock
19from tests.numpy_mock import NumPyMock
21mock_k8s = K8sMock()
22mock_torch = TorchMock()
23mock_numpy = NumPyMock()
25mock_modules = {
26 "scipy": MagicMock(),
27 "scipy.io": MagicMock(),
28}
29mock_modules.update(mock_k8s.get_sub_modules())
30mock_modules.update(mock_torch.get_sub_modules())
31mock_modules.update(mock_numpy.get_sub_modules())
32with patch.dict(sys.modules, mock_modules):
33 with temp_sys_path("streamwise"):
34 from streamwise import streamwise
36 from file_manager import get_file_info
39def _get_client() -> TestClientProtocol:
40 app = streamwise.app
41 return app.test_client()
44@pytest.mark.asyncio
45async def test_video_info() -> None:
46 client = _get_client()
47 response = await client.get("/video/10.1.1.1/8080/video.mp4")
48 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
49 response_json = await response.get_json()
50 assert response_json in (
51 {"error": "Connection timeout to host http://10.1.1.1:8080/file/video.mp4"},
52 {"error": "Event loop is closed"}
53 )
56@pytest.mark.asyncio
57async def test_get_file_info() -> None:
58 file_info = await get_file_info("127.0.0.1", 9980, "file.txt")
59 assert file_info is None
62@pytest.mark.asyncio
63async def test_get_audio_waveform() -> None:
64 client = _get_client()
65 response = await client.get("/audio_waveform/1.2.3.4/8080/audio.wav")
66 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
67 response_json = await response.get_json()
68 assert response_json == {"error": "Event loop is closed"}
71@pytest.mark.asyncio
72async def test_list_files() -> None:
73 client = _get_client()
74 response = await client.get("/files")
75 assert response.status_code == HTTPStatus.OK
76 response_json = await response.get_json()
77 assert "files" in response_json
80@pytest.mark.asyncio
81async def test_download_local_file() -> None:
82 client = _get_client()
83 response = await client.get("/file/nonexisting.mp4")
84 assert response.status_code == HTTPStatus.NOT_FOUND
85 response_json = await response.get_json()
86 assert response_json == {"error": "File not found"}
89@pytest.mark.asyncio
90async def test_download_service_file() -> None:
91 client = _get_client()
92 response = await client.get("/file_download/10.1.1.1/8080/video.mp4")
93 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
94 response_json = await response.get_json()
95 # TODO
96 assert response_json == {"error": "Event loop is closed"}
99@pytest.mark.asyncio
100async def test_file_stream() -> None:
101 client = _get_client()
102 response = await client.get("/file_stream/11.11.11.11/8080/video.mp4")
103 assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR
104 response_json = await response.get_json()
105 # TODO
106 assert response_json == {"error": "Event loop is closed"}
109@pytest.mark.asyncio
110async def test_file_view() -> None:
111 client = _get_client()
112 response = await client.get("/file_view/flux/10.1.2.3/8080/video.mp4")
113 assert response.status_code == HTTPStatus.OK
114 response_data = await response.get_data(as_text=True)
115 assert response_data.startswith("<!DOCTYPE html>\n<html>")
116 assert "flux" in response_data
117 response_json = await response.get_json()
118 assert response_json is None