273 lines
9.1 KiB
Python
273 lines
9.1 KiB
Python
import os
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import inspect
|
|
from werkzeug.serving import make_server
|
|
|
|
ENV_KEYS = (
|
|
"SENSORS_DATABASE_URL",
|
|
"USERS_DATABASE_URL",
|
|
"SECRET_KEY",
|
|
"SENSORS_API_URL",
|
|
)
|
|
|
|
|
|
def _save_env(keys):
|
|
return {key: os.environ.get(key) for key in keys}
|
|
|
|
|
|
def _restore_env(saved):
|
|
for key, value in saved.items():
|
|
if value is None:
|
|
os.environ.pop(key, None)
|
|
else:
|
|
os.environ[key] = value
|
|
|
|
|
|
class SensorApiTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmpdir = tempfile.TemporaryDirectory()
|
|
cls._env = _save_env(ENV_KEYS)
|
|
os.environ["SENSORS_DATABASE_URL"] = (
|
|
f"sqlite:///{Path(cls.tmpdir.name) / 'sensors.db'}"
|
|
)
|
|
|
|
from models import db
|
|
from sensors_app import create_app
|
|
|
|
cls.app = create_app()
|
|
with cls.app.app_context():
|
|
db.create_all()
|
|
|
|
cls.client = cls.app.test_client()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
from models import db
|
|
|
|
with cls.app.app_context():
|
|
db.session.remove()
|
|
db.drop_all()
|
|
cls.tmpdir.cleanup()
|
|
_restore_env(cls._env)
|
|
|
|
def test_sensors_db_has_no_user_table(self):
|
|
from models import db
|
|
|
|
with self.app.app_context():
|
|
tables = set(inspect(db.engine).get_table_names())
|
|
self.assertEqual(tables, {"sensor", "sensor_readout"})
|
|
|
|
def test_all_api_endpoints_are_public(self):
|
|
created = self.client.post("/api/sensors", json={"name": "api-test-sensor"})
|
|
self.assertEqual(created.status_code, 201)
|
|
sensor_id = created.get_json()["id"]
|
|
|
|
readout = self.client.post(
|
|
f"/api/sensors/{sensor_id}/readouts",
|
|
json={"type": "temperature", "value": 20.5},
|
|
)
|
|
self.assertEqual(readout.status_code, 201)
|
|
|
|
self.assertEqual(self.client.get("/api/sensors").status_code, 200)
|
|
types = self.client.get(f"/api/sensors/{sensor_id}/types")
|
|
self.assertEqual(types.status_code, 200)
|
|
self.assertEqual(types.get_json(), ["temperature"])
|
|
|
|
readouts = self.client.get(f"/api/sensors/{sensor_id}/readouts")
|
|
self.assertEqual(readouts.status_code, 200)
|
|
self.assertEqual(readouts.get_json()[0]["value"], 20.5)
|
|
|
|
hourly = self.client.get(
|
|
f"/api/sensors/{sensor_id}/readouts/hourly?type=temperature&range=all"
|
|
)
|
|
self.assertEqual(hourly.status_code, 200)
|
|
self.assertEqual(len(hourly.get_json()), 1)
|
|
|
|
|
|
class MainAppTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.tmpdir = tempfile.TemporaryDirectory()
|
|
cls._env = _save_env(ENV_KEYS)
|
|
base = Path(cls.tmpdir.name)
|
|
os.environ["SENSORS_DATABASE_URL"] = f"sqlite:///{base / 'sensors.db'}"
|
|
os.environ["USERS_DATABASE_URL"] = f"sqlite:///{base / 'main_app.db'}"
|
|
os.environ["SECRET_KEY"] = "test-secret"
|
|
|
|
from models import db
|
|
from sensors_app import create_app as create_sensors_app
|
|
from user_models import User, user_db
|
|
|
|
cls.sensors_app = create_sensors_app()
|
|
with cls.sensors_app.app_context():
|
|
db.create_all()
|
|
|
|
cls.server = make_server("127.0.0.1", 0, cls.sensors_app)
|
|
cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
|
|
cls.thread.start()
|
|
os.environ["SENSORS_API_URL"] = f"http://127.0.0.1:{cls.server.server_port}"
|
|
|
|
from app import create_app as create_main_app
|
|
|
|
cls.main_app = create_main_app()
|
|
with cls.main_app.app_context():
|
|
user_db.create_all()
|
|
user = User(username="tester")
|
|
user.set_password("secret")
|
|
user_db.session.add(user)
|
|
user_db.session.commit()
|
|
|
|
def setUp(self):
|
|
self.client = self.main_app.test_client()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
from models import db
|
|
from user_models import user_db
|
|
|
|
cls.server.shutdown()
|
|
cls.thread.join()
|
|
with cls.sensors_app.app_context():
|
|
db.session.remove()
|
|
db.drop_all()
|
|
with cls.main_app.app_context():
|
|
user_db.session.remove()
|
|
user_db.drop_all()
|
|
cls.tmpdir.cleanup()
|
|
_restore_env(cls._env)
|
|
|
|
def login(self):
|
|
return self.client.post(
|
|
"/login", data={"username": "tester", "password": "secret"}
|
|
)
|
|
|
|
def create_sensor_and_readout(self, name):
|
|
created = self.client.post("/api/sensors", json={"name": name})
|
|
self.assertEqual(created.status_code, 201)
|
|
sensor_id = created.get_json()["id"]
|
|
|
|
readout = self.client.post(
|
|
f"/api/sensors/{sensor_id}/readouts",
|
|
json={"type": "temperature", "value": 21.5},
|
|
)
|
|
self.assertEqual(readout.status_code, 201)
|
|
return sensor_id
|
|
|
|
def test_views_and_login_are_owned_by_main_app(self):
|
|
response = self.client.get("/sensors")
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn("/login", response.headers["Location"])
|
|
|
|
self.assertEqual(self.client.get("/login").status_code, 200)
|
|
self.assertEqual(self.login().status_code, 302)
|
|
self.assertEqual(self.client.get("/home").status_code, 200)
|
|
self.assertEqual(self.client.get("/sensors").status_code, 200)
|
|
self.assertEqual(self.client.get("/notes").status_code, 200)
|
|
self.assertEqual(self.client.get("/api/docs").status_code, 200)
|
|
|
|
def test_public_device_api_is_proxied_to_sensors(self):
|
|
sensor_id = self.create_sensor_and_readout("proxy-test")
|
|
|
|
listed = self.client.get("/api/sensors").get_json()
|
|
self.assertIn({"id": sensor_id, "name": "proxy-test"}, listed)
|
|
|
|
def test_dashboard_reads_require_login_and_are_proxied(self):
|
|
sensor_id = self.create_sensor_and_readout("dashboard-test")
|
|
|
|
self.assertEqual(self.client.get(f"/api/sensors/{sensor_id}/types").status_code, 401)
|
|
|
|
self.login()
|
|
types = self.client.get(f"/api/sensors/{sensor_id}/types")
|
|
self.assertEqual(types.status_code, 200)
|
|
self.assertEqual(types.get_json(), ["temperature"])
|
|
|
|
readouts = self.client.get(f"/api/sensors/{sensor_id}/readouts?type=temperature")
|
|
self.assertEqual(readouts.status_code, 200)
|
|
self.assertEqual(readouts.get_json()[0]["value"], 21.5)
|
|
|
|
hourly = self.client.get(
|
|
f"/api/sensors/{sensor_id}/readouts/hourly?type=temperature&range=all"
|
|
)
|
|
self.assertEqual(hourly.status_code, 200)
|
|
self.assertEqual(len(hourly.get_json()), 1)
|
|
|
|
def test_notes_require_login(self):
|
|
response = self.client.get("/notes")
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn("/login", response.headers["Location"])
|
|
|
|
def test_notes_and_checklist_are_stored_in_main_db(self):
|
|
from user_models import Note, NoteItem, User, user_db
|
|
|
|
self.login()
|
|
created = self.client.post(
|
|
"/notes", data={"title": "Test title", "content": "My short note"}
|
|
)
|
|
self.assertEqual(created.status_code, 302)
|
|
|
|
with self.main_app.app_context():
|
|
user = User.query.filter_by(username="tester").one()
|
|
note = Note.query.filter_by(user_id=user.id).one()
|
|
note_id = note.id
|
|
self.assertEqual(note.title, "Test title")
|
|
self.assertEqual(note.content, "My short note")
|
|
|
|
tables = set(inspect(user_db.engine).get_table_names())
|
|
self.assertIn("note", tables)
|
|
self.assertIn("note_item", tables)
|
|
|
|
self.assertEqual(
|
|
self.client.post(
|
|
f"/notes/{note_id}/items", data={"text": "first item"}
|
|
).status_code,
|
|
302,
|
|
)
|
|
self.assertEqual(
|
|
self.client.post(
|
|
f"/notes/{note_id}/items", data={"text": "second item"}
|
|
).status_code,
|
|
302,
|
|
)
|
|
|
|
with self.main_app.app_context():
|
|
first, second = (
|
|
NoteItem.query.filter_by(note_id=note_id).order_by(NoteItem.id).all()
|
|
)
|
|
first_id = first.id
|
|
|
|
self.assertEqual(
|
|
self.client.post(
|
|
f"/notes/{note_id}/items/{first_id}/toggle",
|
|
data={"is_done": "1"},
|
|
).status_code,
|
|
302,
|
|
)
|
|
|
|
page = self.client.get("/notes")
|
|
html = page.get_data(as_text=True)
|
|
self.assertEqual(page.status_code, 200)
|
|
self.assertIn('class="done"', html)
|
|
self.assertIn('value="Test title"', html)
|
|
self.assertLess(html.index("second item"), html.index("first item"))
|
|
|
|
with self.main_app.app_context():
|
|
user_db.session.expire_all()
|
|
items = (
|
|
NoteItem.query.filter_by(note_id=note_id).order_by(NoteItem.id).all()
|
|
)
|
|
self.assertTrue(items[0].is_done)
|
|
self.assertFalse(items[1].is_done)
|
|
self.assertEqual(
|
|
[item.text for item in items[0].note.ordered_items],
|
|
["second item", "first item"],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|