100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
import httpx
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import main
|
|
|
|
|
|
class StubAsyncClient:
|
|
def __init__(self, response=None, error=None):
|
|
self.response = response
|
|
self.error = error
|
|
self.requested_url = None
|
|
self.timeout = None
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return None
|
|
|
|
async def get(self, url, timeout):
|
|
self.requested_url = url
|
|
self.timeout = timeout
|
|
if self.error:
|
|
raise self.error
|
|
return self.response
|
|
|
|
|
|
def test_health_is_independent_of_btcdata():
|
|
response = TestClient(main.app).get("/api/health")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "up"}
|
|
|
|
|
|
def test_home_updates_title_from_latest_price():
|
|
response = TestClient(main.app).get("/")
|
|
|
|
assert response.status_code == 200
|
|
assert "new MutationObserver(updateTitleFromPrice)" in response.text
|
|
assert "document.title = price + ' | BTC Price'" in response.text
|
|
|
|
|
|
def test_price_fragment_exposes_formatted_price_for_title():
|
|
html = main.price_json_to_html(
|
|
'{"price": 67432.1, "prev_price": 67400, "recorded_at": ""}'
|
|
)
|
|
|
|
assert 'data-title-price="$67.432,10"' in html
|
|
assert ">$67.432,10</p>" in html
|
|
|
|
|
|
def test_ready_when_btcdata_is_healthy(monkeypatch):
|
|
request = httpx.Request("GET", "http://btcdata.test/api/health")
|
|
upstream = httpx.Response(200, request=request)
|
|
client = StubAsyncClient(response=upstream)
|
|
monkeypatch.setattr(main, "BTCDATA_URL", "http://btcdata.test")
|
|
monkeypatch.setattr(main.httpx, "AsyncClient", lambda: client)
|
|
|
|
response = TestClient(main.app).get("/api/ready")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ready"}
|
|
assert client.requested_url == "http://btcdata.test/api/health"
|
|
assert client.timeout == main.READINESS_TIMEOUT_SECONDS
|
|
|
|
|
|
def test_not_ready_when_btcdata_is_unavailable(monkeypatch):
|
|
request = httpx.Request("GET", "http://btcdata.test/api/health")
|
|
error = httpx.ConnectError("connection refused", request=request)
|
|
client = StubAsyncClient(error=error)
|
|
monkeypatch.setattr(main, "BTCDATA_URL", "http://btcdata.test")
|
|
monkeypatch.setattr(main.httpx, "AsyncClient", lambda: client)
|
|
|
|
response = TestClient(main.app).get("/api/ready")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"status": "not ready"}
|
|
assert "btcdata.test" not in response.text
|
|
|
|
|
|
def test_not_ready_when_btcdata_returns_an_error(monkeypatch):
|
|
request = httpx.Request("GET", "http://btcdata.test/api/health")
|
|
upstream = httpx.Response(503, request=request)
|
|
client = StubAsyncClient(response=upstream)
|
|
monkeypatch.setattr(main.httpx, "AsyncClient", lambda: client)
|
|
|
|
response = TestClient(main.app).get("/api/ready")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"status": "not ready"}
|
|
|
|
|
|
def test_version_endpoint(monkeypatch):
|
|
monkeypatch.setattr(main, "APP_VERSION", "abc1234")
|
|
|
|
response = TestClient(main.app).get("/api/version")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"version": "abc1234"}
|