mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-06-11 00:34:46 +02:00
Lloyd
45a44eb47b
- Updated byte representations in tests to use lowercase hex format for consistency. - Reformatted code for better readability, including line breaks and indentation adjustments. - Consolidated multiple lines into single lines where appropriate to enhance clarity. - Ensured that all test cases maintain consistent formatting and style across the test suite.
141 lines
4.6 KiB
Python
141 lines
4.6 KiB
Python
import io
|
|
import logging
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import cherrypy
|
|
import pytest
|
|
|
|
from repeater.web import http_server as hs
|
|
|
|
|
|
def test_log_buffer_emit_collects_messages():
|
|
buf = hs.LogBuffer(max_lines=2)
|
|
rec1 = logging.LogRecord("x", logging.INFO, __file__, 1, "hello", (), None)
|
|
rec2 = logging.LogRecord("x", logging.ERROR, __file__, 2, "boom", (), None)
|
|
rec3 = logging.LogRecord("x", logging.WARNING, __file__, 3, "warn", (), None)
|
|
|
|
buf.emit(rec1)
|
|
buf.emit(rec2)
|
|
buf.emit(rec3)
|
|
|
|
assert len(buf.logs) == 2
|
|
assert buf.logs[-1]["level"] == "WARNING"
|
|
assert "warn" in buf.logs[-1]["message"]
|
|
|
|
|
|
def test_doc_endpoint_routes_and_openapi_json_paths(monkeypatch):
|
|
api = SimpleNamespace(docs=lambda: "docs-html")
|
|
doc = hs.DocEndpoint(api)
|
|
|
|
assert doc.index() == "docs-html"
|
|
assert doc.docs() == "docs-html"
|
|
|
|
monkeypatch.setattr(
|
|
cherrypy, "response", SimpleNamespace(headers={}, status=200), raising=False
|
|
)
|
|
|
|
# success path
|
|
monkeypatch.setattr("builtins.open", lambda *args, **kwargs: io.StringIO("openapi: 3.0.0\n"))
|
|
out = doc.openapi_json()
|
|
assert cherrypy.response.headers["Content-Type"] == "application/json"
|
|
assert b"openapi" in out
|
|
|
|
# not found
|
|
def _missing(*_args, **_kwargs):
|
|
raise FileNotFoundError
|
|
|
|
monkeypatch.setattr("builtins.open", _missing)
|
|
out = doc.openapi_json()
|
|
assert cherrypy.response.status == 404
|
|
assert b"not found" in out
|
|
|
|
# generic error
|
|
def _err(*_args, **_kwargs):
|
|
raise RuntimeError("bad")
|
|
|
|
monkeypatch.setattr("builtins.open", _err)
|
|
out = doc.openapi_json()
|
|
assert cherrypy.response.status == 500
|
|
assert b"Error loading OpenAPI spec" in out
|
|
|
|
|
|
def test_stats_app_index_and_default_routing(monkeypatch, tmp_path):
|
|
index_html = tmp_path / "index.html"
|
|
index_html.write_text("<html>ok</html>", encoding="utf-8")
|
|
|
|
fake_api = SimpleNamespace(config_manager=object(), docs=lambda: "d")
|
|
monkeypatch.setattr(hs, "APIEndpoints", lambda *args, **kwargs: fake_api)
|
|
|
|
app = hs.StatsApp(config={"web": {"web_path": str(tmp_path)}})
|
|
|
|
monkeypatch.setattr(cherrypy, "request", SimpleNamespace(method="GET"), raising=False)
|
|
assert app.index() == "<html>ok</html>"
|
|
|
|
monkeypatch.setattr(cherrypy, "request", SimpleNamespace(method="OPTIONS"), raising=False)
|
|
assert app.default("anything") == ""
|
|
|
|
monkeypatch.setattr(cherrypy, "request", SimpleNamespace(method="GET"), raising=False)
|
|
with pytest.raises(cherrypy.NotFound):
|
|
app.default("api")
|
|
|
|
assert app.default("ws", "packets") == ""
|
|
assert app.default("route") == "<html>ok</html>"
|
|
|
|
|
|
def test_stats_app_index_error_paths(monkeypatch, tmp_path):
|
|
fake_api = SimpleNamespace(config_manager=object(), docs=lambda: "d")
|
|
monkeypatch.setattr(hs, "APIEndpoints", lambda *args, **kwargs: fake_api)
|
|
|
|
app = hs.StatsApp(config={"web": {"web_path": str(tmp_path)}})
|
|
|
|
with pytest.raises(cherrypy.HTTPError):
|
|
app.index()
|
|
|
|
# Force generic open() exception branch
|
|
def _explode(*_args, **_kwargs):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr("builtins.open", _explode)
|
|
(tmp_path / "index.html").write_text("ignored", encoding="utf-8")
|
|
with pytest.raises(cherrypy.HTTPError):
|
|
app.index()
|
|
|
|
|
|
def test_http_server_utility_methods(monkeypatch, tmp_path):
|
|
def _fake_init_auth(self):
|
|
self.jwt_handler = object()
|
|
self.token_manager = object()
|
|
|
|
monkeypatch.setattr(hs.HTTPStatsServer, "_init_auth_handlers", _fake_init_auth)
|
|
monkeypatch.setattr(
|
|
hs,
|
|
"StatsApp",
|
|
lambda *args, **kwargs: SimpleNamespace(api=SimpleNamespace(config_manager=object())),
|
|
)
|
|
monkeypatch.setattr(hs, "AuthEndpoints", lambda *args, **kwargs: object())
|
|
monkeypatch.setattr(hs, "DocEndpoint", lambda *_args, **_kwargs: object())
|
|
|
|
server = hs.HTTPStatsServer(
|
|
config={"web": {"cors_enabled": False}}, config_path=str(Path(tmp_path) / "cfg.yml")
|
|
)
|
|
|
|
monkeypatch.setattr(cherrypy, "response", SimpleNamespace(headers={}), raising=False)
|
|
out = server._json_error_handler(401, "no", "", "")
|
|
assert '"success": false' in out
|
|
|
|
install_called = {"v": False}
|
|
monkeypatch.setattr(hs.cherrypy_cors, "install", lambda: install_called.__setitem__("v", True))
|
|
server._setup_server_cors()
|
|
assert install_called["v"] is True
|
|
|
|
exited = {"v": False}
|
|
monkeypatch.setattr(
|
|
cherrypy,
|
|
"engine",
|
|
SimpleNamespace(exit=lambda: exited.__setitem__("v", True)),
|
|
raising=False,
|
|
)
|
|
server.stop()
|
|
assert exited["v"] is True
|