|
1 | 1 | import asyncio |
| 2 | +from http.server import HTTPServer, SimpleHTTPRequestHandler |
| 3 | +from pathlib import Path |
| 4 | +from threading import Thread |
2 | 5 | from unittest.mock import AsyncMock, patch |
3 | 6 |
|
4 | 7 | import pytest |
@@ -37,22 +40,42 @@ def _set_url(mock_oav_to_redis_forwarder: OAVToRedisForwarder, url: str): |
37 | 40 | set_mock_value(mock_oav_to_redis_forwarder.selected_source, Source.FULL_SCREEN) |
38 | 41 |
|
39 | 42 |
|
40 | | -@pytest.mark.requires(external="internet") # depends on external webpage. See |
41 | | -# https://github.com/DiamondLightSource/mx-bluesky/issues/183 |
| 43 | +@pytest.fixture |
| 44 | +def static_http_server(tmp_path: Path): |
| 45 | + class HandlerInTestDirectory(SimpleHTTPRequestHandler): |
| 46 | + def __init__(self, *args, **kwargs): |
| 47 | + super().__init__(*args, directory=str(tmp_path), **kwargs) |
| 48 | + |
| 49 | + server_address = ("", 9876) |
| 50 | + httpd = HTTPServer(server_address, HandlerInTestDirectory) |
| 51 | + |
| 52 | + def run_server(): |
| 53 | + httpd.serve_forever() |
| 54 | + httpd.server_close() |
| 55 | + |
| 56 | + server_thread = Thread( |
| 57 | + group=None, target=run_server, name="Test server", daemon=True |
| 58 | + ) |
| 59 | + server_thread.start() |
| 60 | + try: |
| 61 | + yield |
| 62 | + finally: |
| 63 | + httpd.shutdown() |
| 64 | + server_thread.join() |
| 65 | + |
| 66 | + |
42 | 67 | async def test_given_stream_url_is_not_a_real_webpage_when_kickoff_then_error( |
43 | 68 | mock_oav_to_redis_forwarder: OAVToRedisForwarder, |
44 | 69 | ): |
45 | | - _set_url(mock_oav_to_redis_forwarder, "http://www.this_is_not_a_valid_webpage.com/") |
| 70 | + _set_url(mock_oav_to_redis_forwarder, "http://localhost:9875/") |
46 | 71 | with pytest.raises(ClientConnectorError): |
47 | 72 | await mock_oav_to_redis_forwarder.kickoff() |
48 | 73 |
|
49 | 74 |
|
50 | | -@pytest.mark.requires(external="internet") # depends on external webpage. |
51 | | -# See https://github.com/DiamondLightSource/mx-bluesky/issues/183 |
52 | 75 | async def test_given_stream_url_is_real_webpage_but_not_mjpg_when_kickoff_then_error( |
53 | | - mock_oav_to_redis_forwarder: OAVToRedisForwarder, |
| 76 | + mock_oav_to_redis_forwarder: OAVToRedisForwarder, static_http_server |
54 | 77 | ): |
55 | | - url = "https://www.google.com/" |
| 78 | + url = "http://localhost:9876/" |
56 | 79 | _set_url(mock_oav_to_redis_forwarder, url) |
57 | 80 | with pytest.raises(ValueError) as e: |
58 | 81 | await mock_oav_to_redis_forwarder.kickoff() |
|
0 commit comments