mia/bumper
1
0
Fork 0
Code Issues 47 Pull requests 11 Projects Releases Packages Wiki Activity Actions

no more start_bumper

Browse source 
remove start_bumper
- built into __main__ + __init__
- add tests
This commit is contained in:
Brian Martin 2019-06-08 16:45:52 -04:00
parent 7637dfca6e
commit 1f0e417760
8 changed files with 316 additions and 133 deletions
Download patch file Download diff file Expand all files Collapse all files

8
README.md
View file

@ -55,7 +55,6 @@ But seriously, there are a several reasons for eliminating the central server:
## Quick Start Usage
- Download bumper then run `pipenv install` to install dependencies
- Generate certificates for Bumper - See the [Creating Certs](docs/Create_Certs.md) doc
- Configure your Ecovacs vacuum using the official mobile app (if you haven't done this already)
- Configure your DNS server as described in the [DNS Setup](docs/DNS_Setup.md) doc.
- Start bumper - see the [Starting Bumper](#starting-bumper) section.
@ -65,7 +64,12 @@ But seriously, there are a several reasons for eliminating the central server:
### Starting Bumper
- Start Bumper with `pipenv run python start_bumper.py`
Bumper requires certificates to function. If certificates aren't found it will prompt to generate them for you.
For more information on generating certificates manually, see the [Creating Certs](docs/Create_Certs.md) doc
- Start Bumper with `pipenv run python -m bumper`
- If prompted to generate certificates choose yes or no
- Reboot your robot
- **Note:** Some models may require removing and re-inserting the battery pack.

156
bumper/__init__.py
View file

@ -12,6 +12,8 @@ from logging.handlers import RotatingFileHandler
from tinydb import TinyDB, Query
from tinydb.storages import MemoryStorage
import socket
import sys
def strtobool(strbool):
@ -34,13 +36,6 @@ server_key = os.environ.get("BUMPER_KEY") or os.path.join(
bumper_dir, "certs", "bumper.key"
)
if not (
os.path.exists(ca_cert)
and os.path.exists(server_cert)
and os.path.exists(server_key)
):
logging.log(logging.FATAL, "Certificate(s) don't exist at paths specified")
# Folders
logs_dir = os.environ.get("BUMPER_LOGS") or os.path.join(bumper_dir, "logs")
os.makedirs(logs_dir, exist_ok=True) # Ensure logs directory exists or create
@ -51,9 +46,7 @@ os.makedirs(data_dir, exist_ok=True) # Ensure data directory exists or create
bumper_listen = os.environ.get("BUMPER_LISTEN") or socket.gethostbyname(
socket.gethostname()
)
if not bumper_listen:
logging.log(logging.FATAL, "No listen address configured!")
os._exit(1)
bumper_announce_ip = os.environ.get("BUMPER_ANNOUNCE_IP") or bumper_listen
@ -121,6 +114,14 @@ xmppserverlog.addHandler(xmpp_rotate)
logging.getLogger("asyncio").setLevel(logging.CRITICAL + 1) # Ignore this logger
mqtt_listen_address = bumper_listen
mqtt_listen_port = 8883
conf1_listen_address = bumper_listen
conf1_listen_port = 443
conf2_listen_address = bumper_listen
conf2_listen_port = 8007
xmpp_listen_address = bumper_listen
xmpp_listen_port = 5223
async def start():
@ -142,29 +143,33 @@ async def start():
format="[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s",
)
if not bumper_listen:
logging.log(logging.FATAL, "No listen address configured")
return
if not (
os.path.exists(ca_cert)
and os.path.exists(server_cert)
and os.path.exists(server_key)
):
logging.log(logging.FATAL, "Certificate(s) don't exist at paths specified")
os._exit(1)
return
bumperlog.info("Starting Bumper")
global mqtt_server
mqtt_server = MQTTServer((bumper_listen, 8883))
mqtt_server = MQTTServer((mqtt_listen_address, mqtt_listen_port))
global mqtt_helperbot
mqtt_helperbot = MQTTHelperBot((bumper_listen, 8883))
mqtt_helperbot = MQTTHelperBot((mqtt_listen_address, mqtt_listen_port))
global conf_server
conf_server = ConfServer(
(bumper_listen, 443), usessl=True, helperbot=mqtt_helperbot
(conf1_listen_address, conf1_listen_port), usessl=True, helperbot=mqtt_helperbot
)
global conf_server_2
conf_server_2 = ConfServer(
(bumper_listen, 8007), usessl=False, helperbot=mqtt_helperbot
(conf2_listen_address, conf2_listen_port), usessl=False, helperbot=mqtt_helperbot
)
global xmpp_server
xmpp_server = XMPPServer((bumper_listen, 5223))
xmpp_server = XMPPServer((xmpp_listen_address, xmpp_listen_port))
# Start web servers
conf_server.confserver_app()
@ -203,8 +208,9 @@ async def shutdown():
elif mqtt_server.broker.transitions.state == "starting":
while mqtt_server.broker.transitions.state == "starting":
await asyncio.sleep(0.1)
await mqtt_server.broker.shutdown()
await mqtt_helperbot.Client.disconnect()
if mqtt_server.broker.transitions.state == "started":
await mqtt_server.broker.shutdown()
await mqtt_helperbot.Client.disconnect()
if xmpp_server.server:
if xmpp_server.server._serving:
xmpp_server.server.close()
@ -432,12 +438,6 @@ class VacBotDevice(object):
"xmpp_connection": self.xmpp_connection,
}
def toJSON(self):
return json.dumps(
self, default=lambda o: o.__dict__, sort_keys=False
) # , indent=4)
class GlobalVacBotDevice(VacBotDevice): # EcoVacs Home
UILogicId = ""
ota = True
@ -943,3 +943,111 @@ API_ERRORS = {
ERR_WRONG_EMAIL_ADDRESS: "1008",
ERR_WRONG_PWD_FROMATE: "1009",
}
def create_certs():
import platform
import os
import subprocess
import sys
path = os.path.dirname(sys.modules[__name__].__file__)
path = os.path.join(path, "..")
sys.path.insert(0, path)
print("Creating certificates")
odir = os.path.realpath(os.curdir)
os.chdir("certs")
if str(platform.system()).lower() == "windows":
# run for win
subprocess.run(
[os.path.join("..", "create_certs", "create_certs_windows.exe")]
)
elif str(platform.system()).lower() == "darwin":
# run on mac
subprocess.run([os.path.join("..", "create_certs", "create_certs_osx")])
elif str(platform.system()).lower() == "linux":
if "arm" in platform.machine().lower():
# run for pi
subprocess.run([os.path.join("..", "create_certs", "create_certs_rpi")])
else:
# run for linux
subprocess.run(
[os.path.join("..", "create_certs", "create_certs_linux")]
)
else:
logging.log(logging.FATAL, "Can't determine platform. Create certs manually and try again.")
return
print("Certificates created")
os.chdir(odir)
print(os.path.realpath(os.curdir))
if "__main__.py" in sys.argv[0]:
os.execv(
sys.executable, ["python", "-m", "bumper"] + sys.argv[1:]
) # Start again
else:
os.execv(sys.executable, ["python"] + sys.argv) # Start again
def firstrun_input():
return input(
"No certificates found, would you like to create them automatically? (y/n): "
).lower()
def first_run():
yes = {"yes", "y", "ye", ""}
print("")
if firstrun_input() in yes:
create_certs()
else:
logging.log(logging.FATAL, "Can't continue without certificates, please create some then try again.")
def main(argv=None):
import argparse
global bumper_debug
global bumper_listen
global bumper_announce_ip
try:
if not (
os.path.exists(ca_cert)
and os.path.exists(server_cert)
and os.path.exists(server_key)
):
first_run()
return
parser = argparse.ArgumentParser()
parser.add_argument(
"--listen", type=str, default=None, help="start serving on address"
)
parser.add_argument(
"--announce", type=str, default=None, help="announce address to bots on checkin"
)
parser.add_argument("--debug", action="store_true", help="enable debug logs")
args = parser.parse_args(args=argv)
if args.debug:
bumper_debug = True
if args.listen:
bumper_listen = args.listen
if args.announce:
bumper_announce_ip = args.announce
asyncio.run(start())
except KeyboardInterrupt:
bumperlog.info("Keyboard Interrupt!")
pass
except Exception as e:
bumperlog.exception(e)
pass
finally:
asyncio.run(shutdown())

4
bumper/__main__.py Normal file
View file

@ -0,0 +1,4 @@
import bumper
if __name__ == "__main__":
bumper.main()

100
start_bumper.py
View file

@ -1,100 +0,0 @@
#!/usr/bin/env python3
import bumper
import argparse
import asyncio
import platform
import os
import subprocess
import sys
def first_run():
yes = {"yes", "y", "ye", ""}
print("")
create_cert = input(
"No certificates found, would you like to create them automatically? (y/n): "
).lower()
if create_cert in yes:
print("Creating certificates")
odir = os.path.dirname(os.path.realpath(__file__))
os.chdir("certs")
if platform.system().lower() == "windows":
# run for win
subprocess.run(
[os.path.join("..", "create_certs", "create_certs_windows.exe")]
)
elif platform.system().lower() == "darwin":
# run on mac
subprocess.run([os.path.join("..", "create_certs", "create_certs_osx")])
elif platform.system().lower() == "linux":
if "arm" in platform.machine().lower():
# run for pi
subprocess.run([os.path.join("..", "create_certs", "create_certs_rpi")])
else:
# run for linux
subprocess.run(
[os.path.join("..", "create_certs", "create_certs_linux")]
)
else:
print("Can't determine platform. Create certs manually and try again.")
exit(1)
print("Certificates created")
os.chdir(odir)
os.execv(sys.executable, ["python"] + sys.argv) # Start again
else:
print("Can't continue without certificates, please create some then try again.")
exit(1)
def main():
try:
if not (
os.path.exists(bumper.ca_cert)
and os.path.exists(bumper.server_cert)
and os.path.exists(bumper.server_key)
):
first_run()
parser = argparse.ArgumentParser()
parser.add_argument(
"--listen", type=str, default=None, help="start serving on address"
)
parser.add_argument(
"--announce",
type=str,
default=None,
help="announce address to bots on checkin",
)
parser.add_argument("--debug", action="store_true", help="enable debug logs")
args = parser.parse_args()
if args.debug:
bumper.bumper_debug = True
if args.listen:
bumper.bumper_listen = args.listen
if args.announce:
bumper.bumper_announce_ip = args.announce
asyncio.run(bumper.start())
except KeyboardInterrupt:
bumper.bumperlog.info("Keyboard Interrupt!")
pass
except Exception as e:
bumper.bumperlog.exception(e)
pass
finally:
asyncio.run(bumper.shutdown())
if __name__ == "__main__":
main()

6
tests/pytest.ini
View file

@ -1,5 +1,5 @@
[pytest]
env =
BUMPER_CA=tests/test_certs/ca.crt
BUMPER_CERT=tests/test_certs/bumper.crt
BUMPER_KEY=tests/test_certs/bumper.key
D:BUMPER_CA=tests/test_certs/ca.crt
D:BUMPER_CERT=tests/test_certs/bumper.crt
D:BUMPER_KEY=tests/test_certs/bumper.key

60
tests/test_init.py
View file

@ -1,4 +1,5 @@
import mock
from mock import patch
import pytest
from tinydb.storages import MemoryStorage
from tinydb import TinyDB, Query
@ -6,6 +7,9 @@ import bumper
import os
import datetime, time
import platform
import json
import asyncio
from testfixtures import LogCapture
def test_get_milli_time():
@ -19,9 +23,61 @@ def test_get_milli_time():
)
def test_strtobool():
assert bumper.strtobool("t") == True
assert bumper.strtobool("f") == False
assert bumper.strtobool(0) == False
async def test_start_stop():
with LogCapture() as l:
if os.path.exists("tests/tmp.db"):
os.remove("tests/tmp.db") # Remove existing db
b = bumper
b.db = "tests/tmp.db" # Set db location for testing
b.conf1_listen_address = "0.0.0.0"
b.conf1_listen_port = 443
asyncio.create_task(b.start())
await asyncio.sleep(0.1)
l.check_present(("bumper", "INFO", "Starting Bumper"))
l.clear()
assert b.shutting_down == False
asyncio.create_task(b.shutdown())
await asyncio.sleep(0.1)
l.check_present(
("bumper", "INFO", "Shutting down"), ("bumper", "INFO", "Shutdown complete")
)
assert b.shutting_down == True
async def test_start_stop_debug():
with LogCapture() as l:
if os.path.exists("tests/tmp.db"):
os.remove("tests/tmp.db") # Remove existing db
b = bumper
b.db = "tests/tmp.db" # Set db location for testing
b.bumper_listen = "0.0.0.0"
b.bumper_debug = True
asyncio.create_task(b.start())
await asyncio.sleep(0.1)
asyncio.create_task(b.shutdown())
l.check_present(("bumper", "INFO", "Starting Bumper"))
l.clear()
await asyncio.sleep(0.1)
l.check_present(
("bumper", "INFO", "Shutting down"), ("bumper", "INFO", "Shutdown complete")
)
assert b.shutting_down == True
def test_db_path():
bumper.db = None
assert bumper.db_file() == os.path.join(bumper.data_dir, "bumper.db")
def test_user_db():
if os.path.exists("tests/tmp.db"):
os.remove("tests/tmp.db") # Remove existing db
bumper.db = "tests/tmp.db" # Set db location for testing
bumper.user_add("testuser") # Add testuser

0
tests/test_runbumper.py
View file

111
tests/test_z_problem.py Normal file
View file

@ -0,0 +1,111 @@
import mock
from mock import patch
import pytest
from tinydb.storages import MemoryStorage
from tinydb import TinyDB, Query
import bumper
import os
import datetime, time
import platform
import json
import asyncio
from testfixtures import LogCapture
import sys
@patch("bumper.firstrun_input")
@patch("bumper.create_certs")
def test_firstrun(mock_input, mock_create):
with LogCapture() as l:
bumper.firstrun_input.return_value = "n"
bumper.first_run()
l.check_present(
(
"root",
"CRITICAL",
"Can't continue without certificates, please create some then try again.",
)
)
bumper.firstrun_input.return_value = "y"
bumper.first_run()
assert mock_create.called == True
def mock_subrun(*args):
return args
@patch("bumper.start")
def test_argparse(mock_start):
bumper.ca_cert = "tests/test_certs/ca.crt"
bumper.server_cert = "tests/test_certs/bumper.crt"
bumper.server_key = "tests/test_certs/bumper.key"
bumper.main(["--debug"])
assert bumper.bumper_debug == True
assert mock_start.called == True
bumper.main(["--listen", "127.0.0.1"])
assert bumper.bumper_listen == "127.0.0.1"
assert mock_start.called == True
bumper.main(["--announce", "127.0.0.1"])
assert bumper.bumper_announce_ip == "127.0.0.1"
assert mock_start.called == True
bumper.main(["--debug", "--listen", "127.0.0.1", "--announce", "127.0.0.1"])
assert bumper.bumper_debug == True
assert bumper.bumper_announce_ip == "127.0.0.1"
assert bumper.bumper_listen == "127.0.0.1"
assert mock_start.called == True
@patch("subprocess.run")
@patch("platform.system.lower")
@patch("platform.machine")
@patch("os.execv")
def test_createcert(mock_run, mock_platform, mock_machine, mock_exec):
mock_run.side_effect = mock_subrun
platform.system.return_value = "darwin"
bumper.create_certs()
assert mock_run.called == True
assert "../create_certs/create_certs_osx" in mock_exec.call_args.args[0]
platform.system.return_value = "windows"
bumper.create_certs()
assert mock_run.called == True
assert "../create_certs/create_certs_windows.exe" in mock_exec.call_args.args[0]
platform.system.return_value = "linux"
bumper.create_certs()
assert mock_run.called == True
assert "../create_certs/create_certs_linux" in mock_exec.call_args.args[0]
platform.system.return_value = "linux"
platform.machine.return_value = "arm"
bumper.create_certs()
assert mock_run.called == True
assert "../create_certs/create_certs_rpi" in mock_exec.call_args.args[0]
with LogCapture() as l:
platform.system.return_value = "nixbad"
bumper.create_certs()
l.check_present(
(
"root",
"CRITICAL",
"Can't determine platform. Create certs manually and try again.",
)
)
@patch("bumper.first_run")
def test_main(mock_firstrun):
bumper.ca_cert = "sf"
bumper.main()
assert mock_firstrun.called == True
bumper.ca_cert = "tests/test_certs/ca.crt"