jfm/rapt-cloud-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved to uv and aiohttp

Browse Source
This commit is contained in:
Jesper Fussing Mørk
2025-10-29 14:52:18 +01:00
parent 50e85f3d5b
commit 450d583421
11 changed files with 653 additions and 1086 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
tests/test_fermentation_chamber.py
View File

@@ -1,81 +1,72 @@
import pytest
import json
import responses
from aiohttp import web
from aioresponses import aioresponses
from rapt.client import Client
from rapt.fermentation_chamber import FermentationChamber
@pytest.fixture
def client():
async def client(aiohttp_client):
app = web.Application()
await aiohttp_client(app)
yield Client("test", "test")
@responses.activate
def test_get_fermentation_chambers(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.get("https://api.rapt.io/api/fermentationchambers/getfermentationchambers", json=json_reader("./tests/json/get_fermentation_chambers_response.json"))
fermentation_chamber = FermentationChamber(client)
ferms = fermentation_chamber.get_fermentation_chambers()
async def test_get_fermentation_chambers(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/fermentationchambers/getfermentationchambers", payload=json_reader("./tests/json/get_fermentation_chambers_response.json"))
fermentation_chamber = FermentationChamber(client)
ferms = await fermentation_chamber.get_fermentation_chambers()
assert ferms is not None
assert len(ferms) == 1
assert ferms is not None
assert len(ferms) == 1
@responses.activate
def test_get_fermentation_chambers_500(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.add(
responses.GET,
"https://api.rapt.io/api/fermentationchambers/getfermentationchambers",
status=500,
)
fermentation_chamber = FermentationChamber(client)
ferms = fermentation_chamber.get_fermentation_chambers()
async def test_get_fermentation_chambers_500(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/fermentationchambers/getfermentationchambers", status=500)
fermentation_chamber = FermentationChamber(client)
ferms = await fermentation_chamber.get_fermentation_chambers()
assert ferms is None
assert ferms is None
@responses.activate
def test_get_fermentation_chamber(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.get("https://api.rapt.io/api/fermentationchambers/getfermentationchamber", json=json_reader("./tests/json/get_fermentation_chamber_response.json"))
fermentation_chamber = FermentationChamber(client)
ferms = fermentation_chamber.get_fermentation_chamber("")
async def test_get_fermentation_chamber(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/fermentationchambers/getfermentationchamber", payload=json_reader("./tests/json/get_fermentation_chamber_response.json"))
fermentation_chamber = FermentationChamber(client)
ferms = await fermentation_chamber.get_fermentation_chamber("")
assert ferms is not None
assert ferms is not None
@responses.activate
def test_get_fermentation_chamber_500(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.add(
responses.GET,
"https://api.rapt.io/api/fermentationchambers/getfermentationchamber",
status=500,
)
fermentation_chamber = FermentationChamber(client)
ferms = fermentation_chamber.get_fermentation_chamber("")
async def test_get_fermentation_chamber_500(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/fermentationchambers/getfermentationchamber", status=500)
fermentation_chamber = FermentationChamber(client)
ferms = await fermentation_chamber.get_fermentation_chamber("")
assert ferms is None
assert ferms is None
@responses.activate
def test_get_telemetry(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.get("https://api.rapt.io/api/fermentationchambers/gettelemetry", json=json_reader("./tests/json/get_fermentation_chamber_telemetry_response.json"))
fermentation_chamber = FermentationChamber(client)
telemetry = fermentation_chamber.get_telemetry("", "", "", "")
async def test_get_telemetry(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/fermentationchambers/gettelemetry", payload=json_reader("./tests/json/get_fermentation_chamber_telemetry_response.json"))
fermentation_chamber = FermentationChamber(client)
telemetry = await fermentation_chamber.get_telemetry("", "", "", "")
assert telemetry is not None
assert len(telemetry) == 1
assert telemetry is not None
assert len(telemetry) == 1
@responses.activate
def test_get_telemetry_500(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.add(
responses.GET,
"https://api.rapt.io/api/fermentationchambers/gettelemetry",
status=500,
)
fermentation_chamber = FermentationChamber(client)
telemetry = fermentation_chamber.get_telemetry("", "", "", "")
async def test_get_telemetry_500(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/fermentationchambers/gettelemetry", status=500)
fermentation_chamber = FermentationChamber(client)
telemetry = await fermentation_chamber.get_telemetry("", "", "", "")
assert telemetry is None
assert telemetry is None
def json_reader(path):

106
tests/test_hydrometer.py
View File

@@ -1,81 +1,71 @@
import pytest
import json
import responses
from aiohttp import web
from aioresponses import aioresponses
from rapt.client import Client
from rapt.hydrometer import Hydrometer
@pytest.fixture
def client():
async def client(aiohttp_client):
app = web.Application()
await aiohttp_client(app)
yield Client("test", "test")
@responses.activate
def test_get_hydrometers(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.get("https://api.rapt.io/api/hydrometers/gethydrometers", json=json_reader("./tests/json/get_hydrometers_response.json"))
hydrometer = Hydrometer(client)
hydros = hydrometer.get_hydrometers()
async def test_get_hydrometers(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/hydrometers/gethydrometers", payload=json_reader("./tests/json/get_hydrometers_response.json"))
hydrometer = Hydrometer(client)
hydros = await hydrometer.get_hydrometers()
assert hydros is not None
assert len(hydros) == 1
assert hydros is not None
assert len(hydros) == 1
@responses.activate
def test_get_hydrometers_500(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.add(
responses.GET,
"https://api.rapt.io/api/hydrometers/gethydrometers",
status=500,
)
hydrometer = Hydrometer(client)
hydros = hydrometer.get_hydrometers()
async def test_get_hydrometers_500(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/hydrometers/gethydrometers", status=500)
hydrometer = Hydrometer(client)
hydros = await hydrometer.get_hydrometers()
assert hydros is None
assert hydros is None
@responses.activate
def test_get_hydrometer(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.get("https://api.rapt.io/api/hydrometers/gethydrometer", json=json_reader("./tests/json/get_hydrometer_response.json"))
hydrometer = Hydrometer(client)
hydros = hydrometer.get_hydrometer("")
async def test_get_hydrometer(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/hydrometers/gethydrometer", payload=json_reader("./tests/json/get_hydrometer_response.json"))
hydrometer = Hydrometer(client)
hydros = await hydrometer.get_hydrometer("")
assert hydros is not None
assert hydros is not None
@responses.activate
def test_get_hydrometer_500(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.add(
responses.GET,
"https://api.rapt.io/api/hydrometers/gethydrometer",
status=500,
)
hydrometer = Hydrometer(client)
hydros = hydrometer.get_hydrometer("")
async def test_get_hydrometer_500(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/hydrometers/gethydrometer", status=500)
hydrometer = Hydrometer(client)
hydros = await hydrometer.get_hydrometer("")
assert hydros is None
assert hydros is None
async def test_get_telemetry(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/hydrometers/gettelemetry", payload=json_reader("./tests/json/get_hydrometer_telemetry_response.json"))
hydrometer = Hydrometer(client)
telemetry = await hydrometer.get_telemetry("", "", "", "")
@responses.activate
def test_get_telemetry(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.get("https://api.rapt.io/api/hydrometers/gettelemetry", json=json_reader("./tests/json/get_hydrometer_telemetry_response.json"))
hydrometer = Hydrometer(client)
telemetry = hydrometer.get_telemetry("", "", "", "")
assert telemetry is not None
assert telemetry is not None
async def test_get_telemetry_500(client):
with aioresponses() as responses:
responses.post("https://id.rapt.io/connect/token", payload=json_reader("./tests/json/token_response.json"))
responses.get("/api/hydrometers/gettelemetry", status=500)
hydrometer = Hydrometer(client)
telemetry = await hydrometer.get_telemetry("", "", "", "")
@responses.activate
def test_get_telemetry_500(client):
responses.post("https://id.rapt.io/connect/token", json=json_reader("./tests/json/token_response.json"))
responses.add(
responses.GET,
"https://api.rapt.io/api/hydrometers/gettelemetry",
status=500,
)
hydrometer = Hydrometer(client)
telemetry = hydrometer.get_telemetry("", "", "", "")
assert telemetry is None
assert telemetry is None
def json_reader(path):