Fix tests so they run again

Tests that can not work because of missing endpoints in the controllers
were removed. Also tests that have no clear intent and were failing were
deleted.
This commit is contained in:
Philipp Grathwohl 2022-02-20 10:50:44 +00:00
parent 86b8d6fc62
commit 652fbd74be
36 changed files with 660 additions and 741 deletions

View file

@ -1,15 +1,12 @@
import logging import logging
import requests import requests
import yaml
from cbpi.configFolder import ConfigFolder from cbpi.configFolder import ConfigFolder
from cbpi.utils.utils import load_config from cbpi.utils.utils import load_config
from zipfile import ZipFile from zipfile import ZipFile
from cbpi.craftbeerpi import CraftBeerPi from cbpi.craftbeerpi import CraftBeerPi
import os import os
import pathlib
import pkgutil import pkgutil
import shutil import shutil
import yaml
import click import click
from subprocess import call from subprocess import call
from colorama import Fore, Back, Style from colorama import Fore, Back, Style
@ -18,120 +15,211 @@ from importlib_metadata import metadata
from tabulate import tabulate from tabulate import tabulate
from PyInquirer import prompt, print_json from PyInquirer import prompt, print_json
def create_home_folder_structure(configFolder):
pathlib.Path(os.path.join(".", 'logs/sensors')).mkdir(parents=True, exist_ok=True)
configFolder.create_folders()
print("Folder created")
class CraftBeerPiCli():
def __init__(self, config) -> None:
self.config = config
pass
def setup_one_wire(): def setup(self):
print("Setting up 1Wire") print("Setting up CraftBeerPi")
with open('/boot/config.txt', 'w') as f: self.config.create_home_folder_structure()
f.write("dtoverlay=w1-gpio,gpiopin=4,pullup=on") self.config.create_config_file()
print("/boot/config.txt created")
def list_one_wire(): def start(self):
print("List 1Wire") if self.config.check_for_setup() is False:
call(["modprobe", "w1-gpio"]) return
call(["modprobe", "w1-therm"]) print("START")
try: cbpi = CraftBeerPi(self.config)
for dirname in os.listdir('/sys/bus/w1/devices'): cbpi.start()
if (dirname.startswith("28") or dirname.startswith("10")):
print(dirname)
except Exception as e:
print(e)
def setup_one_wire(self):
print("Setting up 1Wire")
with open('/boot/config.txt', 'w') as f:
f.write("dtoverlay=w1-gpio,gpiopin=4,pullup=on")
print("/boot/config.txt created")
def plugins_list(): def list_one_wire(self):
result = [] print("List 1Wire")
print("") call(["modprobe", "w1-gpio"])
print(Fore.LIGHTYELLOW_EX,"List of active plugins", Style.RESET_ALL) call(["modprobe", "w1-therm"])
print("")
discovered_plugins = {
name: importlib.import_module(name)
for finder, name, ispkg
in pkgutil.iter_modules()
if name.startswith('cbpi') and len(name) > 4
}
for key, module in discovered_plugins.items():
try: try:
meta = metadata(key) for dirname in os.listdir('/sys/bus/w1/devices'):
result.append(dict(Name=meta["Name"], Version=meta["Version"], Author=meta["Author"], Homepage=meta["Home-page"], Summary=meta["Summary"])) if (dirname.startswith("28") or dirname.startswith("10")):
print(dirname)
except Exception as e: except Exception as e:
print(e) print(e)
print(Fore.LIGHTGREEN_EX, tabulate(result, headers="keys"), Style.RESET_ALL)
def plugins_list(self):
def plugin_create(): result = []
print("Plugin Creation") print("")
print("") print(Fore.LIGHTYELLOW_EX,"List of active plugins", Style.RESET_ALL)
print("")
questions = [ discovered_plugins = {
{ name: importlib.import_module(name)
'type': 'input', for finder, name, ispkg
'name': 'name', in pkgutil.iter_modules()
'message': 'Plugin Name:', if name.startswith('cbpi') and len(name) > 4
} }
] for key, module in discovered_plugins.items():
try:
meta = metadata(key)
result.append(dict(Name=meta["Name"], Version=meta["Version"], Author=meta["Author"], Homepage=meta["Home-page"], Summary=meta["Summary"]))
except Exception as e:
print(e)
print(Fore.LIGHTGREEN_EX, tabulate(result, headers="keys"), Style.RESET_ALL)
answers = prompt(questions)
name = "cbpi4_" + answers["name"] def plugin_create(self):
if os.path.exists(os.path.join(".", name)) is True: print("Plugin Creation")
print("Cant create Plugin. Folder {} already exists ".format(name)) print("")
return
url = 'https://github.com/Manuel83/craftbeerpi4-plugin-template/archive/main.zip' questions = [
r = requests.get(url) {
with open('temp.zip', 'wb') as f: 'type': 'input',
f.write(r.content) 'name': 'name',
'message': 'Plugin Name:',
}
]
with ZipFile('temp.zip', 'r') as repo_zip: answers = prompt(questions)
repo_zip.extractall()
os.rename("./craftbeerpi4-plugin-template-main", os.path.join(".", name)) name = "cbpi4_" + answers["name"]
os.rename(os.path.join(".", name, "src"), os.path.join(".", name, name)) if os.path.exists(os.path.join(".", name)) is True:
print("Cant create Plugin. Folder {} already exists ".format(name))
return
import jinja2 url = 'https://github.com/Manuel83/craftbeerpi4-plugin-template/archive/main.zip'
r = requests.get(url)
with open('temp.zip', 'wb') as f:
f.write(r.content)
templateLoader = jinja2.FileSystemLoader(searchpath=os.path.join(".", name)) with ZipFile('temp.zip', 'r') as repo_zip:
templateEnv = jinja2.Environment(loader=templateLoader) repo_zip.extractall()
TEMPLATE_FILE = "setup.py"
template = templateEnv.get_template(TEMPLATE_FILE)
outputText = template.render(name=name)
with open(os.path.join(".", name, "setup.py"), "w") as fh: os.rename("./craftbeerpi4-plugin-template-main", os.path.join(".", name))
fh.write(outputText) os.rename(os.path.join(".", name, "src"), os.path.join(".", name, name))
TEMPLATE_FILE = "MANIFEST.in" import jinja2
template = templateEnv.get_template(TEMPLATE_FILE)
outputText = template.render(name=name)
with open(os.path.join(".", name, "MANIFEST.in"), "w") as fh:
fh.write(outputText)
TEMPLATE_FILE = os.path.join("/", name, "config.yaml") templateLoader = jinja2.FileSystemLoader(searchpath=os.path.join(".", name))
operatingsystem = str(platform.system()).lower() templateEnv = jinja2.Environment(loader=templateLoader)
if operatingsystem.startswith("win"): TEMPLATE_FILE = "setup.py"
TEMPLATE_FILE=str(TEMPLATE_FILE).replace('\\','/') template = templateEnv.get_template(TEMPLATE_FILE)
outputText = template.render(name=name)
template = templateEnv.get_template(TEMPLATE_FILE)
outputText = template.render(name=name)
with open(os.path.join(".", name, name, "config.yaml"), "w") as fh: with open(os.path.join(".", name, "setup.py"), "w") as fh:
fh.write(outputText) fh.write(outputText)
print("") TEMPLATE_FILE = "MANIFEST.in"
print("") template = templateEnv.get_template(TEMPLATE_FILE)
print("Plugin {}{}{} created! ".format(Fore.LIGHTGREEN_EX, name, Style.RESET_ALL) ) outputText = template.render(name=name)
print("") with open(os.path.join(".", name, "MANIFEST.in"), "w") as fh:
print("Developer Documentation: https://openbrewing.gitbook.io/craftbeerpi4_support/readme/development") fh.write(outputText)
print("")
print("Happy developing! Cheers") TEMPLATE_FILE = os.path.join("/", name, "config.yaml")
print("") operatingsystem = str(platform.system()).lower()
print("") if operatingsystem.startswith("win"):
TEMPLATE_FILE=str(TEMPLATE_FILE).replace('\\','/')
template = templateEnv.get_template(TEMPLATE_FILE)
outputText = template.render(name=name)
with open(os.path.join(".", name, name, "config.yaml"), "w") as fh:
fh.write(outputText)
print("")
print("")
print("Plugin {}{}{} created! ".format(Fore.LIGHTGREEN_EX, name, Style.RESET_ALL) )
print("")
print("Developer Documentation: https://openbrewing.gitbook.io/craftbeerpi4_support/readme/development")
print("")
print("Happy developing! Cheers")
print("")
print("")
def autostart(self, name):
'''Enable or disable autostart'''
if(name == "status"):
if os.path.exists(os.path.join("/etc/systemd/system","craftbeerpi.service")) is True:
print("CraftBeerPi Autostart is {}ON{}".format(Fore.LIGHTGREEN_EX,Style.RESET_ALL))
else:
print("CraftBeerPi Autostart is {}OFF{}".format(Fore.RED,Style.RESET_ALL))
elif(name == "on"):
print("Add craftbeerpi.service to systemd")
try:
if os.path.exists(os.path.join("/etc/systemd/system","craftbeerpi.service")) is False:
srcfile = self.config.get_file_path("craftbeerpi.service")
destfile = os.path.join("/etc/systemd/system")
shutil.copy(srcfile, destfile)
print("Copied craftbeerpi.service to /etc/systemd/system")
os.system('systemctl enable craftbeerpi.service')
print('Enabled craftbeerpi service')
os.system('systemctl start craftbeerpi.service')
print('Started craftbeerpi.service')
else:
print("craftbeerpi.service is already located in /etc/systemd/system")
except Exception as e:
print(e)
return
return
elif(name == "off"):
print("Remove craftbeerpi.service from systemd")
try:
status = os.popen('systemctl list-units --type=service --state=running | grep craftbeerpi.service').read()
if status.find("craftbeerpi.service") != -1:
os.system('systemctl stop craftbeerpi.service')
print('Stopped craftbeerpi service')
os.system('systemctl disable craftbeerpi.service')
print('Removed craftbeerpi.service as service')
else:
print('craftbeerpi.service service is not running')
if os.path.exists(os.path.join("/etc/systemd/system","craftbeerpi.service")) is True:
os.remove(os.path.join("/etc/systemd/system","craftbeerpi.service"))
print("Deleted craftbeerpi.service from /etc/systemd/system")
else:
print("craftbeerpi.service is not located in /etc/systemd/system")
except Exception as e:
print(e)
return
return
def chromium(self, name):
'''Enable or disable autostart'''
if(name == "status"):
if os.path.exists(os.path.join("/etc/xdg/autostart/","chromium.desktop")) is True:
print("CraftBeerPi Chromium Desktop is {}ON{}".format(Fore.LIGHTGREEN_EX,Style.RESET_ALL))
else:
print("CraftBeerPi Chromium Desktop is {}OFF{}".format(Fore.RED,Style.RESET_ALL))
elif(name == "on"):
print("Add chromium.desktop to /etc/xdg/autostart/")
try:
if os.path.exists(os.path.join("/etc/xdg/autostart/","chromium.desktop")) is False:
srcfile = self.config.get_file_path("chromium.desktop")
destfile = os.path.join("/etc/xdg/autostart/")
shutil.copy(srcfile, destfile)
print("Copied chromium.desktop to /etc/xdg/autostart/")
else:
print("chromium.desktop is already located in /etc/xdg/autostart/")
except Exception as e:
print(e)
return
return
elif(name == "off"):
print("Remove chromium.desktop from /etc/xdg/autostart/")
try:
if os.path.exists(os.path.join("/etc/xdg/autostart/","chromium.desktop")) is True:
os.remove(os.path.join("/etc/xdg/autostart/","chromium.desktop"))
print("Deleted chromium.desktop from /etc/xdg/autostart/")
else:
print("chromium.desktop is not located in /etc/xdg/autostart/")
except Exception as e:
print(e)
return
return
@click.group() @click.group()
@ -143,139 +231,55 @@ def main(context, config_folder_path):
print("---------------------") print("---------------------")
level = logging.INFO level = logging.INFO
logging.basicConfig(level=level, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s') logging.basicConfig(level=level, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')
context.obj = ConfigFolder(config_folder_path) cbpi_cli = CraftBeerPiCli(ConfigFolder(config_folder_path))
context.obj = cbpi_cli
pass pass
@main.command() @main.command()
@click.pass_context @click.pass_context
def setup(context): def setup(context):
'''Create Config folder''' '''Create Config folder'''
print("Setting up CraftBeerPi") context.obj.setup()
create_home_folder_structure(context.obj)
context.obj.create_config_file()
@main.command() @main.command()
@click.pass_context
@click.option('--list', is_flag=True, help="List all 1Wire Devices") @click.option('--list', is_flag=True, help="List all 1Wire Devices")
@click.option('--setup', is_flag=True, help="Setup 1Wire on Raspberry Pi") @click.option('--setup', is_flag=True, help="Setup 1Wire on Raspberry Pi")
def onewire(list, setup): def onewire(context, list, setup):
'''Setup 1wire on Raspberry Pi''' '''Setup 1wire on Raspberry Pi'''
if setup is True: if setup is True:
setup_one_wire() context.obj.setup_one_wire()
if list is True: if list is True:
list_one_wire() context.obj.list_one_wire()
@main.command() @main.command()
@click.pass_context @click.pass_context
def start(context): def start(context):
'''Lets go brewing''' context.obj.start()
if context.obj.check_for_setup() is False:
return
print("Starting up CraftBeerPi ...")
cbpi = CraftBeerPi(context.obj)
cbpi.start()
@main.command() @main.command()
@click.pass_context @click.pass_context
def plugins(context): def plugins(context):
'''List active plugins''' '''List active plugins'''
plugins_list() context.obj.plugins_list()
return
@click.command() @click.command()
def create(): @click.pass_context
def create(context):
'''Create New Plugin''' '''Create New Plugin'''
plugin_create() context.obj.plugin_create()
@main.command() @main.command()
@click.argument('name')
@click.pass_context @click.pass_context
@click.argument('name')
def autostart(context, name): def autostart(context, name):
'''(on|off|status) Enable or disable autostart''' '''(on|off|status) Enable or disable autostart'''
if(name == "status"): context.obj.autostart(name)
if os.path.exists(os.path.join("/etc/systemd/system","craftbeerpi.service")) is True:
print("CraftBeerPi Autostart is {}ON{}".format(Fore.LIGHTGREEN_EX,Style.RESET_ALL))
else:
print("CraftBeerPi Autostart is {}OFF{}".format(Fore.RED,Style.RESET_ALL))
elif(name == "on"):
print("Add craftbeerpi.service to systemd")
try:
if os.path.exists(os.path.join("/etc/systemd/system","craftbeerpi.service")) is False:
srcfile = context.obj.get_file_path("craftbeerpi.service")
destfile = os.path.join("/etc/systemd/system")
shutil.copy(srcfile, destfile)
print("Copied craftbeerpi.service to /etc/systemd/system")
os.system('systemctl enable craftbeerpi.service')
print('Enabled craftbeerpi service')
os.system('systemctl start craftbeerpi.service')
print('Started craftbeerpi.service')
else:
print("craftbeerpi.service is already located in /etc/systemd/system")
except Exception as e:
print(e)
return
return
elif(name == "off"):
print("Remove craftbeerpi.service from systemd")
try:
status = os.popen('systemctl list-units --type=service --state=running | grep craftbeerpi.service').read()
if status.find("craftbeerpi.service") != -1:
os.system('systemctl stop craftbeerpi.service')
print('Stopped craftbeerpi service')
os.system('systemctl disable craftbeerpi.service')
print('Removed craftbeerpi.service as service')
else:
print('craftbeerpi.service service is not running')
if os.path.exists(os.path.join("/etc/systemd/system","craftbeerpi.service")) is True:
os.remove(os.path.join("/etc/systemd/system","craftbeerpi.service"))
print("Deleted craftbeerpi.service from /etc/systemd/system")
else:
print("craftbeerpi.service is not located in /etc/systemd/system")
except Exception as e:
print(e)
return
return
@main.command() @main.command()
@click.argument('name')
@click.pass_context @click.pass_context
@click.argument('name')
def chromium(context, name): def chromium(context, name):
'''(on|off|status) Enable or disable Kiosk mode''' '''(on|off|status) Enable or disable Kiosk mode'''
if(name == "status"): context.obj.chromium(name)
if os.path.exists(os.path.join("/etc/xdg/autostart/","chromium.desktop")) is True:
print("CraftBeerPi Chromium Desktop is {}ON{}".format(Fore.LIGHTGREEN_EX,Style.RESET_ALL))
else:
print("CraftBeerPi Chromium Desktop is {}OFF{}".format(Fore.RED,Style.RESET_ALL))
elif(name == "on"):
print("Add chromium.desktop to /etc/xdg/autostart/")
try:
if os.path.exists(os.path.join("/etc/xdg/autostart/","chromium.desktop")) is False:
srcfile = context.obj.get_file_path("chromium.desktop")
destfile = os.path.join("/etc/xdg/autostart/")
shutil.copy(srcfile, destfile)
print("Copied chromium.desktop to /etc/xdg/autostart/")
else:
print("chromium.desktop is already located in /etc/xdg/autostart/")
except Exception as e:
print(e)
return
return
elif(name == "off"):
print("Remove chromium.desktop from /etc/xdg/autostart/")
try:
if os.path.exists(os.path.join("/etc/xdg/autostart/","chromium.desktop")) is True:
os.remove(os.path.join("/etc/xdg/autostart/","chromium.desktop"))
print("Deleted chromium.desktop from /etc/xdg/autostart/")
else:
print("chromium.desktop is not located in /etc/xdg/autostart/")
except Exception as e:
print(e)
return
return

View file

@ -103,6 +103,11 @@ class ConfigFolder:
print("Config Folder created") print("Config Folder created")
def create_home_folder_structure(configFolder):
pathlib.Path(os.path.join(".", 'logs/sensors')).mkdir(parents=True, exist_ok=True)
configFolder.create_folders()
print("Folder created")
def create_folders(self): def create_folders(self):
pathlib.Path(self._rawPath).mkdir(parents=True, exist_ok=True) pathlib.Path(self._rawPath).mkdir(parents=True, exist_ok=True)

View file

@ -23,6 +23,32 @@ class ActorHttpEndpoints():
description: successful operation description: successful operation
""" """
return web.json_response(data=self.controller.get_state()) return web.json_response(data=self.controller.get_state())
@request_mapping(path="/{id:\w+}", auth_required=False)
async def http_get_one(self, request):
"""
---
description: Get one Actor
tags:
- Actor
parameters:
- name: "id"
in: "path"
description: "Actor ID"
required: true
type: "integer"
format: "int64"
responses:
"200":
description: successful operation
"404":
description: Actor not found
"""
actor = self.controller.find_by_id(request.match_info['id'])
if (actor is None):
return web.json_response(status=404)
return web.json_response(data=actor.to_dict(), status=200)
@request_mapping(path="/", method="POST", auth_required=False) @request_mapping(path="/", method="POST", auth_required=False)

View file

@ -0,0 +1,12 @@
{
"data": [
{
"id": "3CUJte4bkxDMFCtLX8eqsX",
"name": "SomeActor",
"power": 100,
"props": {},
"state": false,
"type": "DummyActor"
}
]
}

View file

@ -0,0 +1,6 @@
[Desktop Entry]
Type=Application
Name=Chromium
Comment=Chromium Webbrowser
NoDisplay=false
Exec=chromium-browser --noerrordialogs --disable-session-crashed-bubble --disable-infobars --force-device-scale-factor=1.00 --start-fullscreen "http://localhost:8000"

View file

@ -0,0 +1,148 @@
{
"AUTHOR": {
"description": "Author",
"name": "AUTHOR",
"options": null,
"type": "string",
"value": "John Doe"
},
"BREWERY_NAME": {
"description": "Brewery Name",
"name": "BREWERY_NAME",
"options": null,
"type": "string",
"value": "CraftBeerPi Brewery"
},
"MASH_TUN": {
"description": "Default Mash Tun",
"name": "MASH_TUN",
"options": null,
"type": "kettle",
"value": ""
},
"AddMashInStep": {
"description": "Add MashIn Step automatically if not defined in recipe",
"name": "AddMashInStep",
"options": [
{
"label": "Yes",
"value": "Yes"
},
{
"label": "No",
"value": "No"
}
],
"type": "select",
"value": "Yes"
},
"RECIPE_CREATION_PATH": {
"description": "API path to creation plugin. Default: empty",
"name": "RECIPE_CREATION_PATH",
"options": null,
"type": "string",
"value": ""
},
"brewfather_api_key": {
"description": "Brewfather API Kay",
"name": "brewfather_api_key",
"options": null,
"type": "string",
"value": ""
},
"brewfather_user_id": {
"description": "Brewfather User ID",
"name": "brewfather_user_id",
"options": null,
"type": "string",
"value": ""
},
"TEMP_UNIT": {
"description": "Temperature Unit",
"name": "TEMP_UNIT",
"options": [
{
"label": "C",
"value": "C"
},
{
"label": "F",
"value": "F"
}
],
"type": "select",
"value": "C"
},
"AutoMode": {
"description": "Use AutoMode in steps",
"name": "AutoMode",
"options": [
{
"label": "Yes",
"value": "Yes"
},
{
"label": "No",
"value": "No"
}
],
"type": "select",
"value": "Yes"
},
"steps_boil": {
"description": "Boil step type",
"name": "steps_boil",
"options": null,
"type": "step",
"value": "BoilStep"
},
"steps_boil_temp": {
"description": "Default Boil Temperature for Recipe Creation",
"name": "steps_boil_temp",
"options": null,
"type": "number",
"value": "99"
},
"steps_cooldown": {
"description": "Cooldown step type",
"name": "steps_cooldown",
"options": null,
"type": "step",
"value": "CooldownStep"
},
"steps_cooldown_sensor": {
"description": "Alternative Sensor to monitor temperature durring cooldown (if not selected, Kettle Sensor will be used)",
"name": "steps_cooldown_sensor",
"options": null,
"type": "sensor",
"value": ""
},
"steps_cooldown_temp": {
"description": "Cooldown temp will send notification when this temeprature is reached",
"name": "steps_cooldown_temp",
"options": null,
"type": "number",
"value": "20"
},
"steps_mash": {
"description": "Mash step type",
"name": "steps_mash",
"options": null,
"type": "step",
"value": "MashStep"
},
"steps_mashin": {
"description": "MashIn step type",
"name": "steps_mashin",
"options": null,
"type": "step",
"value": "MashInStep"
},
"steps_mashout": {
"description": "MashOut step type",
"name": "steps_mashout",
"options": null,
"type": "step",
"value": "NotificationStep"
}
}

View file

@ -0,0 +1,20 @@
name: CraftBeerPi
version: 4.0.8
index_url: /cbpi_ui/static/index.html
port: 8000
mqtt: false
mqtt_host: localhost
mqtt_port: 1883
mqtt_username: ""
mqtt_password: ""
username: cbpi
password: 123
plugins:
- cbpi4ui

View file

@ -0,0 +1,9 @@
[Unit]
Description=Craftbeer Pi
[Service]
WorkingDirectory=/home/pi
ExecStart=/usr/local/bin/cbpi start
[Install]
WantedBy=multi-user.target

View file

@ -0,0 +1,62 @@
#!/bin/sh
### BEGIN INIT INFO
# Provides: craftbeerpi
# Required-Start: $remote_fs $syslog
# Required-Stop: $remote_fs $syslog
# Default-Start: 2 3 4 5
# Default-Stop: 0 1 6
# Short-Description: Put a short description of the service here
# Description: Put a long description of the service here
### END INIT INFO
# Change the next 3 lines to suit where you install your script and what you want to call it
DIR=#DIR#
DAEMON=$DIR/cbpi
DAEMON_NAME=CraftBeerPI
# Add any command line options for your daemon here
DAEMON_OPTS=""
# This next line determines what user the script runs as.
# Root generally not recommended but necessary if you are using the Raspberry Pi GPIO from Python.
DAEMON_USER=root
# The process ID of the script when it runs is stored here:
PIDFILE=/var/run/$DAEMON_NAME.pid
. /lib/lsb/init-functions
do_start () {
log_daemon_msg "Starting system $DAEMON_NAME daemon"
start-stop-daemon --start --background --pidfile $PIDFILE --make-pidfile --user $DAEMON_USER --chuid $DAEMON_USER --chdir $DIR --startas $DAEMON -- $DAEMON_OPTS
log_end_msg $?
}
do_stop () {
log_daemon_msg "Stopping system $DAEMON_NAME daemon"
start-stop-daemon --stop --pidfile $PIDFILE --retry 10
log_end_msg $?
}
case "$1" in
start|stop)
do_${1}
;;
restart|reload|force-reload)
do_stop
do_start
;;
status)
status_of_proc "$DAEMON_NAME" "$DAEMON" && exit 0 || exit $?
;;
*)
echo "Usage: /etc/init.d/$DAEMON_NAME {start|stop|restart|status}"
exit 1
;;
esac
exit 0

View file

@ -0,0 +1,3 @@
{
"elements": []
}

View file

@ -0,0 +1,5 @@
{
"data": [
]
}

View file

@ -0,0 +1,5 @@
{
"data": [
]
}

View file

@ -0,0 +1,3 @@
cbpi4-ui:
installation_date: '2021-01-06 16:03:31'
version: '0.0.1'

View file

View file

@ -0,0 +1,5 @@
{
"data": [
]
}

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.9 MiB

View file

@ -0,0 +1,8 @@
{
"basic": {
"name": ""
},
"steps": [
]
}

View file

View file

@ -0,0 +1,23 @@
# content of conftest.py
from codecs import ignore_errors
from distutils.command.config import config
import os
from cbpi.configFolder import ConfigFolder
from cbpi.craftbeerpi import CraftBeerPi
from aiohttp.test_utils import AioHTTPTestCase
from distutils.dir_util import copy_tree
class CraftBeerPiTestCase(AioHTTPTestCase):
async def get_application(self):
self.config_folder = self.configuration()
self.cbpi = CraftBeerPi(self.config_folder)
await self.cbpi.init_serivces()
return self.cbpi.app
def configuration(self):
test_directory = os.path.dirname(__file__)
test_config_directory = os.path.join(test_directory, 'cbpi-test-config')
configFolder = ConfigFolder(test_config_directory)
return configFolder

View file

@ -1,66 +1,39 @@
import logging import logging
from unittest import mock from unittest import mock
from unittest.mock import MagicMock, Mock from aiohttp.test_utils import unittest_run_loop
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s') logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')
class ActorTestCase(AioHTTPTestCase): class ActorTestCase(CraftBeerPiTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop
async def test_actor_mock(self):
with mock.patch.object(self.cbpi.bus, 'fire', wraps=self.cbpi.bus.fire) as mock_obj:
# Send HTTP POST
resp = await self.client.request("POST", "/actor/1/on")
# Check Result
assert resp.status == 204
# Check if Event are fired
assert mock_obj.call_count == 2
@unittest_run_loop @unittest_run_loop
async def test_actor_switch(self): async def test_actor_switch(self):
resp = await self.client.post(path="/login", data={"username": "cbpi", "password": "123"}) resp = await self.client.post(path="/login", data={"username": "cbpi", "password": "123"})
assert resp.status == 200 assert resp.status == 200, "login should be successful"
resp = await self.client.request("POST", "/actor/1/on") resp = await self.client.request("POST", "/actor/3CUJte4bkxDMFCtLX8eqsX/on")
assert resp.status == 204 assert resp.status == 204, "switching actor on should work"
i = await self.cbpi.actor.get_one(1) i = self.cbpi.actor.find_by_id("3CUJte4bkxDMFCtLX8eqsX")
assert i.instance.state is True assert i.instance.state is True
resp = await self.client.request("POST", "/actor/1/off") resp = await self.client.request("POST", "/actor/3CUJte4bkxDMFCtLX8eqsX/off")
assert resp.status == 204 assert resp.status == 204
i = await self.cbpi.actor.get_one(1) i = self.cbpi.actor.find_by_id("3CUJte4bkxDMFCtLX8eqsX")
assert i.instance.state is False
resp = await self.client.request("POST", "/actor/1/toggle")
assert resp.status == 204
i = await self.cbpi.actor.get_one(1)
assert i.instance.state is True
resp = await self.client.request("POST", "/actor/1/toggle")
assert resp.status == 204
i = await self.cbpi.actor.get_one(1)
assert i.instance.state is False assert i.instance.state is False
@unittest_run_loop @unittest_run_loop
async def test_crud(self): async def test_crud(self):
data = { data = {
"name": "CustomActor", "name": "SomeActor",
"type": "CustomActor", "power": 100,
"config": { "props": {
"interval": 5 },
} "state": False,
"type": "DummyActor"
} }
# Add new sensor # Add new sensor

View file

@ -1,20 +1,16 @@
import logging import logging
import unittest import unittest
from cbpi.cli import CraftBeerPiCli
from cli import add, remove, list_plugins from cbpi.configFolder import ConfigFolder
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s') logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')
class CLITest(unittest.TestCase): class CLITest(unittest.TestCase):
def test_install(self):
assert add("cbpi4-ui-plugin") == True
assert add("cbpi4-ui-plugin") == False
assert remove("cbpi4-ui-plugin") == True
def test_list(self): def test_list(self):
list_plugins() cli = CraftBeerPiCli(ConfigFolder("./cbpi-test-config"))
cli.plugins_list()
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()

View file

@ -1,44 +1,31 @@
import time import time
import aiosqlite from aiohttp.test_utils import unittest_run_loop
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.api.config import ConfigType
from cbpi.craftbeerpi import CraftBeerPi
class ConfigTestCase(AioHTTPTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
class ConfigTestCase(CraftBeerPiTestCase):
@unittest_run_loop @unittest_run_loop
async def test_get(self): async def test_get(self):
assert self.cbpi.config.get("CBPI_TEST_1", 1) == "22" assert self.cbpi.config.get("steps_boil_temp", 1) == "99"
@unittest_run_loop @unittest_run_loop
async def test_set_get(self): async def test_set_get(self):
value = str(time.time()) value = 35
await self.cbpi.config.set("CBPI_TEST_2", value) await self.cbpi.config.set("steps_cooldown_temp", value)
assert self.cbpi.config.get("steps_cooldown_temp", 1) == value
assert self.cbpi.config.get("CBPI_TEST_2", 1) == value
@unittest_run_loop @unittest_run_loop
async def test_http_set(self): async def test_http_set(self):
value = str(time.time()) value = "Some New Brewery Name"
key = "CBPI_TEST_3" key = "BREWERY_NAME"
await self.cbpi.config.set(key, value)
assert self.cbpi.config.get(key, 1) == value
resp = await self.client.request("PUT", "/config/%s/" % key, json={'value': '1'}) resp = await self.client.request("PUT", "/config/%s/" % key, json={'value': value})
assert resp.status == 204 assert resp.status == 204
assert self.cbpi.config.get(key, -1) == "1"
assert self.cbpi.config.get(key, -1) == value
@unittest_run_loop @unittest_run_loop
async def test_http_get(self): async def test_http_get(self):
@ -47,5 +34,5 @@ class ConfigTestCase(AioHTTPTestCase):
@unittest_run_loop @unittest_run_loop
async def test_get_default(self): async def test_get_default(self):
value = self.cbpi.config.get("HELLO_WORLD", None) value = self.cbpi.config.get("HELLO_WORLD", "DefaultValue")
assert value == None assert value == "DefaultValue"

View file

@ -1,18 +1,10 @@
import aiohttp from aiohttp.test_utils import unittest_run_loop
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi from cbpi.craftbeerpi import CraftBeerPi
class DashboardTestCase(AioHTTPTestCase): class DashboardTestCase(CraftBeerPiTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_crud(self): async def test_crud(self):
@ -28,56 +20,12 @@ class DashboardTestCase(AioHTTPTestCase):
"config": {} "config": {}
} }
resp = await self.client.get(path="/dashboard") resp = await self.client.get(path="/dashboard/current")
assert resp.status == 200 assert resp.status == 200
# Add new dashboard dashboard_id = await resp.json()
resp = await self.client.post(path="/dashboard/", json=data)
assert resp.status == 200
m = await resp.json() # Add dashboard content
dashboard_id = m["id"]
# Get dashboard
resp = await self.client.get(path="/dashboard/%s" % dashboard_id)
assert resp.status == 200
m2 = await resp.json()
dashboard_id = m2["id"]
# Update dashboard
resp = await self.client.put(path="/dashboard/%s" % dashboard_id, json=m)
assert resp.status == 200
# Add dashboard content
dashboard_content["dbid"] = dashboard_id dashboard_content["dbid"] = dashboard_id
resp = await self.client.post(path="/dashboard/%s/content" % dashboard_id, json=dashboard_content) resp = await self.client.post(path="/dashboard/%s/content" % dashboard_id, json=dashboard_content)
assert resp.status == 200 assert resp.status == 204
m_content = await resp.json()
print("CONTENT", m_content)
content_id = m_content["id"]
# Get dashboard
resp = await self.client.get(path="/dashboard/%s/content" % (dashboard_id))
assert resp.status == 200
resp = await self.client.post(path="/dashboard/%s/content/%s/move" % (dashboard_id, content_id), json=dict(x=1,y=1))
assert resp.status == 200
resp = await self.client.delete(path="/dashboard/%s/content/%s" % (dashboard_id, content_id))
assert resp.status == 204
# Delete dashboard
resp = await self.client.delete(path="/dashboard/%s" % dashboard_id)
assert resp.status == 204
@unittest_run_loop
async def test_dashboard_controller(self):
result = await self.cbpi.dashboard.get_all()
print(result)
await self.cbpi.dashboard.add(**{"name":"Tewst"})
print(await self.cbpi.dashboard.get_one(1))
await self.cbpi.dashboard.add_content(dict(dbid=1,element_id=1,type="test",config={"name":"Manue"}))
await self.cbpi.dashboard.move_content(1,2,3)

View file

@ -1,116 +0,0 @@
import asyncio
from cbpi.api.dataclasses import Fermenter, FermenterStep, Props, Step
import logging
from unittest import mock
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from cbpi.craftbeerpi import CraftBeerPi
from cbpi.controller.fermentation_controller import FermenationController
import unittest
import json
from aiohttp import web
from unittest.mock import MagicMock, Mock
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')
class FermenterTest(AioHTTPTestCase):
async def get_application(self):
app = web.Application()
return app
def create_file(self):
data = [
{
"id": "f1",
"name": "Fermenter1",
"props": {},
"steps": [
{
"id": "f1s1",
"name": "Step1",
"props": {},
"state_text": "",
"status": "I",
"type": "T2"
},
{
"id": "f1s2",
"name": "Step2",
"props": {},
"state_text": "",
"status": "I",
"type": "T1"
},
],
"target_temp": 0
},
{
"id": "f2",
"name": "Fermenter2",
"props": {},
"steps": [
{
"id": "f2s1",
"name": "Step1",
"props": {},
"state_text": "",
"status": "I",
"type": "T1"
},
{
"id": "f2s2",
"name": "Step2",
"props": {},
"state_text": "",
"status": "I",
"type": "T2"
},
],
"target_temp": 0
}
]
with open("./config/fermenter_data.json", "w") as file:
json.dump(data, file, indent=4, sort_keys=True)
@unittest_run_loop
async def test_actor_mock(self):
self.create_file()
mock = Mock()
f = FermenationController(mock)
f.types = {
"T1": {"name": "T2", "class": FermenterStep, "properties": [], "actions": []},
"T2": {"name": "T2", "class": FermenterStep, "properties": [], "actions": []}
}
await f.load()
#ferm = Fermenter(name="Maneul")
# item = await f.create(ferm)
# await f.create_step(item.id, Step(name="Manuel"))
# await f.delete(item.id)
item = await f.get("f1")
await f.start("f1")
await f.start("f2")
await asyncio.sleep(3)
# await f.create_step(item.id, Step(name="MANUEL", props=Props()))
#await f.start(item.id)
#await asyncio.sleep(1)
#await f.next(item.id)
#await asyncio.sleep(1)
#await f.next(item.id)
#await asyncio.sleep(1)
#await f.next(item.id)
#await asyncio.sleep(1)
#await f.move_step("f1", "f1s1", 1)
# await f.reset(item.id)
await f.shutdown()
if __name__ == '__main__':
unittest.main()

View file

@ -15,13 +15,6 @@ except Exception:
patcher.start() patcher.start()
import RPi.GPIO as GPIO import RPi.GPIO as GPIO
class HelloWorld(object):
def test(self, a):
return a
class TestSwitch(unittest.TestCase): class TestSwitch(unittest.TestCase):
GPIO_NUM = 22 GPIO_NUM = 22
@ -35,15 +28,4 @@ class TestSwitch(unittest.TestCase):
@patch("RPi.GPIO.output") @patch("RPi.GPIO.output")
def test_switch_without_scheduler_starts_disabled(self, patched_output): def test_switch_without_scheduler_starts_disabled(self, patched_output):
GPIO.output(self.GPIO_NUM, GPIO.LOW) GPIO.output(self.GPIO_NUM, GPIO.LOW)
patched_output.assert_called_once_with(self.GPIO_NUM, GPIO.LOW) patched_output.assert_called_once_with(self.GPIO_NUM, GPIO.LOW)
def test_hello_world(self):
h = HelloWorld()
with mock.patch.object(HelloWorld, 'test', wraps=h.test) as fake_increment:
#print(h.test("HALLO"))
print(h.test("ABC"))
print(fake_increment.call_args)
print(h.test("HALLO"))
print(fake_increment.call_args_list)

View file

@ -1,14 +1,8 @@
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from aiohttp.test_utils import unittest_run_loop
from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi
class IndexTestCase(AioHTTPTestCase): class IndexTestCase(CraftBeerPiTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_index(self): async def test_index(self):

View file

@ -1,52 +1,16 @@
import asyncio from aiohttp.test_utils import unittest_run_loop
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi
class KettleTestCase(AioHTTPTestCase): class KettleTestCase(CraftBeerPiTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_get(self): async def test_get(self):
resp = await self.client.request("GET", "/kettle") resp = await self.client.request("GET", "/kettle")
assert resp.status == 200 assert resp.status == 200
print(await resp.json()) kettle = resp.json()
assert kettle != None
@unittest_run_loop
async def test_heater(self):
resp = await self.client.get("/kettle/1/heater/on")
assert resp.status == 204
resp = await self.client.get("/kettle/1/heater/off")
assert resp.status == 204
@unittest_run_loop
async def test_agitator(self):
resp = await self.client.get("/kettle/1/agitator/on")
assert resp.status == 204
resp = await self.client.get("/kettle/1/agitator/off")
assert resp.status == 204
@unittest_run_loop
async def test_temp(self):
resp = await self.client.get("/kettle/1/temp")
assert resp.status == 204
resp = await self.client.get("/kettle/1/targettemp")
assert resp.status == 200
@unittest_run_loop
async def test_automatic(self):
resp = await self.client.post("/kettle/1/automatic")
assert resp.status == 204
@unittest_run_loop @unittest_run_loop
async def test_crud(self): async def test_crud(self):
@ -69,21 +33,21 @@ class KettleTestCase(AioHTTPTestCase):
m = await resp.json() m = await resp.json()
sensor_id = m["id"] kettle_id = m["id"]
print("KETTLE", m["id"], m) print("KETTLE", m["id"], m)
# Get sensor
resp = await self.client.get(path="/kettle/%s" % sensor_id) # Update Kettle
resp = await self.client.put(path="/kettle/%s" % kettle_id, json=m)
assert resp.status == 200 assert resp.status == 200
m2 = await resp.json() # Set Kettle target temp
sensor_id = m2["id"] resp = await self.client.post(path="/kettle/%s/target_temp" % kettle_id, json={"temp":75})
assert resp.status == 204
# Update Sensor
resp = await self.client.put(path="/kettle/%s" % sensor_id, json=m)
assert resp.status == 200
# # Delete Sensor # # Delete Sensor
resp = await self.client.delete(path="/kettle/%s" % sensor_id) resp = await self.client.delete(path="/kettle/%s" % kettle_id)
assert resp.status == 204 assert resp.status == 204

View file

@ -1,21 +1,16 @@
import asyncio import asyncio
import glob import glob
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from aiohttp.test_utils import unittest_run_loop
from tests.cbpi_config_fixture import CraftBeerPiTestCase
import os
from cbpi.craftbeerpi import CraftBeerPi, load_config class LoggerTestCase(CraftBeerPiTestCase):
class UtilsTestCase(AioHTTPTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_log_data(self): async def test_log_data(self):
os.makedirs("./logs", exist_ok=True)
log_name = "test" log_name = "test"
#clear all logs #clear all logs
self.cbpi.log.clear_log(log_name) self.cbpi.log.clear_log(log_name)

View file

@ -1,16 +1,8 @@
import aiohttp from aiohttp.test_utils import unittest_run_loop
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi
class NotificationTestCase(AioHTTPTestCase): class NotificationTestCase(CraftBeerPiTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_actor_switch(self): async def test_actor_switch(self):

View file

@ -1,14 +1,8 @@
import asyncio from aiohttp.test_utils import unittest_run_loop
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi
class SensorTestCase(AioHTTPTestCase): class SensorTestCase(CraftBeerPiTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_crud(self): async def test_crud(self):
@ -28,12 +22,11 @@ class SensorTestCase(AioHTTPTestCase):
m = await resp.json() m = await resp.json()
sensor_id = m["id"] sensor_id = m["id"]
# Get sensor # Get sensor value
resp = await self.client.get(path="/sensor/%s"% sensor_id) resp = await self.client.get(path="/sensor/%s"% sensor_id)
assert resp.status == 200 assert resp.status == 200
m2 = await resp.json() m2 = await resp.json()
sensor_id = m2["id"]
# Update Sensor # Update Sensor
resp = await self.client.put(path="/sensor/%s" % sensor_id, json=m) resp = await self.client.put(path="/sensor/%s" % sensor_id, json=m)

View file

@ -1,23 +1,13 @@
import asyncio import asyncio
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from aiohttp.test_utils import unittest_run_loop
from cbpi.craftbeerpi import CraftBeerPi from tests.cbpi_config_fixture import CraftBeerPiTestCase
class StepTestCase(CraftBeerPiTestCase):
class StepTestCase(AioHTTPTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_get(self): async def test_get(self):
resp = await self.client.request("GET", "/step") resp = await self.client.request("GET", "/step2")
print(resp)
assert resp.status == 200
resp = await self.client.request("GET", "/step/types")
print(resp) print(resp)
assert resp.status == 200 assert resp.status == 200
@ -31,26 +21,19 @@ class StepTestCase(AioHTTPTestCase):
} }
# Add new step # Add new step
resp = await self.client.post(path="/step/", json=data) resp = await self.client.post(path="/step2/", json=data)
assert resp.status == 200 assert resp.status == 200
m = await resp.json() m = await resp.json()
print("Step", m) print("Step", m)
sensor_id = m["id"] sensor_id = m["id"]
# Get sensor # Update step
resp = await self.client.get(path="/step/%s" % sensor_id) resp = await self.client.put(path="/step2/%s" % sensor_id, json=m)
assert resp.status == 200 assert resp.status == 200
m2 = await resp.json() # # Delete step
sensor_id = m2["id"] resp = await self.client.delete(path="/step2/%s" % sensor_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 assert resp.status == 204
def create_wait_callback(self, topic): def create_wait_callback(self, topic):
@ -66,34 +49,4 @@ class StepTestCase(AioHTTPTestCase):
done, pending = await asyncio.wait({future}) done, pending = await asyncio.wait({future})
if future in done: if future in done:
pass 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())

View file

@ -1,46 +0,0 @@
import logging
from unittest import mock
import unittest
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from cbpi.craftbeerpi import CraftBeerPi
import pprint
import asyncio
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')
pp = pprint.PrettyPrinter(indent=4)
class ActorTestCase(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_all(self):
resp = await self.client.get(path="/step2")
assert resp.status == 200
@unittest_run_loop
async def test_add_step(self):
resp = await self.client.post(path="/step2", json=dict(name="Manuel"))
data = await resp.json()
assert resp.status == 200
@unittest_run_loop
async def test_delete(self):
resp = await self.client.post(path="/step2", json=dict(name="Manuel"))
data = await resp.json()
assert resp.status == 200
resp = await self.client.delete(path="/step2/%s" % data["id"])
assert resp.status == 204
'''
@unittest_run_loop
async def test_move(self):
await self.cbpi.step2.resume()
if __name__ == '__main__':
unittest.main()

View file

@ -1,19 +1,11 @@
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from aiohttp.test_utils import unittest_run_loop
from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi
class IndexTestCase(AioHTTPTestCase): class IndexTestCase(CraftBeerPiTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop @unittest_run_loop
async def test_endpoints(self): async def test_endpoints(self):
# Test Index Page # Test Index Page
resp = await self.client.post(path="/system/restart") resp = await self.client.post(path="/system/restart")
assert resp.status == 200 assert resp.status == 200
@ -21,9 +13,3 @@ class IndexTestCase(AioHTTPTestCase):
resp = await self.client.post(path="/system/shutdown") resp = await self.client.post(path="/system/shutdown")
assert resp.status == 200 assert resp.status == 200
resp = await self.client.get(path="/system/jobs")
assert resp.status == 200
resp = await self.client.get(path="/system/events")
assert resp.status == 200

View file

@ -1,16 +0,0 @@
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from cbpi.craftbeerpi import CraftBeerPi, load_config
class UtilsTestCase(AioHTTPTestCase):
async def get_application(self):
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
@unittest_run_loop
async def test_load_file(self):
assert load_config("") is None

View file

@ -1,56 +1,46 @@
import asyncio
import aiohttp import aiohttp
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop from aiohttp.test_utils import unittest_run_loop
from tests.cbpi_config_fixture import CraftBeerPiTestCase
from cbpi.craftbeerpi import CraftBeerPi # class WebSocketTestCase(CraftBeerPiTestCase):
# @unittest_run_loop
# async def test_brewing_process(self):
class WebSocketTestCase(AioHTTPTestCase): # count_step_done = 0
# async with self.client.ws_connect('/ws') as ws:
# await ws.send_json(data=dict(topic="step/stop"))
# await ws.send_json(data=dict(topic="step/start"))
# async for msg in ws:
# if msg.type == aiohttp.WSMsgType.TEXT:
# try:
# msg_obj = msg.json()
# topic = msg_obj.get("topic")
# if topic == "job/step/done":
# count_step_done = count_step_done + 1
# if topic == "step/brewing/finished":
# await ws.send_json(data=dict(topic="close"))
# except Exception as e:
# print(e)
# break
# elif msg.type == aiohttp.WSMsgType.ERROR:
# break
async def get_application(self): # assert count_step_done == 4
self.cbpi = CraftBeerPi()
await self.cbpi.init_serivces()
return self.cbpi.app
# @unittest_run_loop
# async def test_wrong_format(self):
@unittest_run_loop # async with self.client.ws_connect('/ws') as ws:
async def test_brewing_process(self): # await ws.send_json(data=dict(a="close"))
# async for msg in ws:
# print("MSG TYP", msg.type, msg.data)
# if msg.type == aiohttp.WSMsgType.TEXT:
# msg_obj = msg.json()
# if msg_obj["topic"] != "connection/success":
# print(msg.data)
# raise Exception()
count_step_done = 0 # else:
async with self.client.ws_connect('/ws') as ws: # raise Exception()
await ws.send_json(data=dict(topic="step/stop"))
await ws.send_json(data=dict(topic="step/start"))
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
try:
msg_obj = msg.json()
topic = msg_obj.get("topic")
if topic == "job/step/done":
count_step_done = count_step_done + 1
if topic == "step/brewing/finished":
await ws.send_json(data=dict(topic="close"))
except Exception as e:
print(e)
break
elif msg.type == aiohttp.WSMsgType.ERROR:
break
assert count_step_done == 4
@unittest_run_loop
async def test_wrong_format(self):
async with self.client.ws_connect('/ws') as ws:
await ws.send_json(data=dict(a="close"))
async for msg in ws:
print("MSG TYP", msg.type, msg.data)
if msg.type == aiohttp.WSMsgType.TEXT:
msg_obj = msg.json()
if msg_obj["topic"] != "connection/success":
print(msg.data)
raise Exception()
else:
raise Exception()