mirror of
https://github.com/PiBrewing/craftbeerpi4.git
synced 2024-09-19 22:07:28 +02:00
107 lines
No EOL
2.8 KiB
Python
107 lines
No EOL
2.8 KiB
Python
import asyncio
|
|
from unittest import mock
|
|
|
|
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
|
|
from cbpi.craftbeerpi import CraftBeerPi
|
|
|
|
|
|
class StepTestCase(AioHTTPTestCase):
|
|
|
|
|
|
async def get_application(self):
|
|
self.cbpi = CraftBeerPi()
|
|
await self.cbpi.init_serivces()
|
|
return self.cbpi.app
|
|
|
|
@unittest_run_loop
|
|
async def test_get(self):
|
|
|
|
resp = await self.client.request("GET", "/step")
|
|
print(resp)
|
|
assert resp.status == 200
|
|
|
|
resp = await self.client.request("GET", "/step/types")
|
|
print(resp)
|
|
assert resp.status == 200
|
|
|
|
|
|
@unittest_run_loop
|
|
async def test_crud(self):
|
|
data = {
|
|
"name": "Test",
|
|
"type": "CustomStepCBPi",
|
|
"config": {}
|
|
}
|
|
|
|
# Add new step
|
|
resp = await self.client.post(path="/step/", json=data)
|
|
assert resp.status == 200
|
|
|
|
m = await resp.json()
|
|
print("Step", m)
|
|
sensor_id = m["id"]
|
|
|
|
# Get sensor
|
|
resp = await self.client.get(path="/step/%s" % sensor_id)
|
|
assert resp.status == 200
|
|
|
|
m2 = await resp.json()
|
|
sensor_id = m2["id"]
|
|
|
|
# Update Sensor
|
|
resp = await self.client.put(path="/step/%s" % sensor_id, json=m)
|
|
assert resp.status == 200
|
|
|
|
# # Delete Sensor
|
|
resp = await self.client.delete(path="/step/%s" % sensor_id)
|
|
assert resp.status == 204
|
|
|
|
def create_wait_callback(self, topic):
|
|
future = self.cbpi.app.loop.create_future()
|
|
|
|
async def test(**kwargs):
|
|
print("GOON")
|
|
future.set_result("OK")
|
|
self.cbpi.bus.register(topic, test, once=True)
|
|
return future
|
|
|
|
async def wait(self, future):
|
|
done, pending = await asyncio.wait({future})
|
|
|
|
if future in done:
|
|
pass
|
|
|
|
|
|
|
|
@unittest_run_loop
|
|
async def test_process(self):
|
|
|
|
step_ctlr = self.cbpi.step
|
|
|
|
|
|
await step_ctlr.clear_all()
|
|
await step_ctlr.add(**{"name": "Kettle1", "type": "CustomStepCBPi", "config": {"name1": "1", "temp": 99}})
|
|
await step_ctlr.add(**{"name": "Kettle1", "type": "CustomStepCBPi", "config": {"name1": "1", "temp": 99}})
|
|
await step_ctlr.add(**{"name": "Kettle1", "type": "CustomStepCBPi", "config": {"name1": "1", "temp": 99}})
|
|
|
|
await step_ctlr.stop_all()
|
|
|
|
future = self.create_wait_callback("step/+/started")
|
|
await step_ctlr.start()
|
|
await self.wait(future)
|
|
|
|
for i in range(len(step_ctlr.cache)-1):
|
|
future = self.create_wait_callback("step/+/started")
|
|
await step_ctlr.next()
|
|
await self.wait(future)
|
|
|
|
await self.print_steps()
|
|
|
|
|
|
|
|
async def print_steps(self):
|
|
|
|
s = await self.cbpi.step.get_all()
|
|
print(s)
|
|
for k, v in s.items():
|
|
print(k, v.to_json()) |