Coverage for tests/streamwise/test_streamwise

1#!/usr/bin/env python3

2"""

3Unit tests for streamwise.py job submission endpoints.

4"""

6import sys

7import pytest

9from http import HTTPStatus

11from unittest.mock import patch

12from unittest.mock import MagicMock

14from tests.test_utils import temp_sys_path

15from tests.k8s_mock import K8sMock

16from tests.torch_mock import TorchMock

17from tests.numpy_mock import NumPyMock

19mock_k8s = K8sMock()

20mock_torch = TorchMock()

21mock_numpy = NumPyMock()

23mock_modules = {

24 "scipy": MagicMock(),

25 "scipy.io": MagicMock(),

26}

27mock_modules.update(mock_k8s.get_sub_modules())

28mock_modules.update(mock_torch.get_sub_modules())

29mock_modules.update(mock_numpy.get_sub_modules())

30with patch.dict(sys.modules, mock_modules):

31 with temp_sys_path("streamwise"):

32 from streamwise import streamwise

34 with temp_sys_path("apps"):

35 from apps.streamwise_job import StreamWiseJob

38@pytest.fixture(scope="function", autouse=True)

39def setup_k8s_cluster() -> None:

40 # for some reason k8s_config.load_kube_config() is not async mocked

41 streamwise.k8s_cluster = "unittest"

44@pytest.mark.asyncio

45async def test_submit_job() -> None:

46 app = streamwise.app

47 client = app.test_client()

49 # Non existing endpoint

50 response = await client.get("/job")

52 # GET job submission form

53 response = await client.get("/job/1.2.3.4/8080")

54 assert response.status_code == HTTPStatus.OK

55 assert "html" in response.content_type

56 response_html = await response.get_data(as_text=True)

57 assert response_html.startswith("<!DOCTYPE html>\n<html lang=\"en\">")

58 assert "<title>Submit Generation</title>" in response_html

61@pytest.mark.asyncio

62async def test_api_submit_job() -> None:

63 app = streamwise.app

64 client = app.test_client()

66 # Non existing GET endpoint

67 await client.get("/api/job/fantasytalking/1.2.3.4/8080")

69 # POST without JSON body

70 response = await client.post("/api/job/fantasytalking/1.2.3.4/8080")

71 assert response.status_code == HTTPStatus.BAD_REQUEST

72 response_json = await response.get_json()

73 assert response_json == {"error": "No job data provided"}

75 # POST with JSON body

76 response = await client.post(

77 "/api/job/yolo/10.20.3.4/8989",

78 json={"param1": "value1", "param2": 2}

79 )

80 assert response.status_code in (

81 HTTPStatus.GATEWAY_TIMEOUT,

82 HTTPStatus.INTERNAL_SERVER_ERROR

83 )

84 response_json = await response.get_json()

85 assert response_json in (

86 {"error": "Request timed out"},

87 {"error": "Event loop is closed"}

88 )

91@pytest.mark.asyncio

92async def test_get_quality() -> None:

93 job_id = "test_job_quality"

94 service_manager = MagicMock()

95 job = StreamWiseJob("test_app", job_id, service_manager)

96 assert job.get_num_steps() == 15

Coverage for tests/streamwise/test_streamwise_job.py: 100%

54 statements