-
Notifications
You must be signed in to change notification settings - Fork 12
/
metrics_test.py
64 lines (48 loc) · 2.14 KB
/
metrics_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import pytest
from httpx import AsyncClient
from metrics import app
@pytest.mark.asyncio
async def test_base_uri():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/")
assert response.status_code == 200
assert "Welcome to Tasmota prometheus exporter!" in response.text
assert "Use the /probe endpoint" in response.text
@pytest.mark.asyncio
async def test_probe_scrape(monkeypatch):
async def mock_fetch(self):
return {"Power": "10 W", "Voltage": "230 V", "Total": "15.6 kWh"}
monkeypatch.setattr("metrics.TasmotaCollector.fetch", mock_fetch)
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/probe?target=192.168.1.34")
assert response.status_code == 200
assert "tasmota_power_W" in response.text
assert "tasmota_voltage_V" in response.text
assert "tasmota_kWh_total" in response.text
@pytest.mark.asyncio
async def test_probe_fail(monkeypatch):
async def mock_fetch(self):
raise Exception("Failed to fetch data")
monkeypatch.setattr("metrics.TasmotaCollector.fetch", mock_fetch)
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/probe?target=192.168.1.38")
assert response.status_code == 500
assert "Failed to fetch data" in response.json()["detail"]
@pytest.mark.asyncio
async def test_probe_missing():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/probe")
assert response.status_code == 422
@pytest.mark.asyncio
async def test_probe_with_auth(monkeypatch):
async def mock_fetch(self):
return {"Power": "10 W", "Voltage": "230 V", "Total": "15.6 kWh"}
monkeypatch.setattr("metrics.TasmotaCollector.fetch", mock_fetch)
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get(
"/probe?target=192.168.1.34&user=admin&password=password"
)
assert response.status_code == 200
assert "tasmota_power_W" in response.text
assert "tasmota_voltage_V" in response.text
assert "tasmota_kWh_total" in response.text