458 lines
16 KiB
Python
458 lines
16 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("/notifications").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_clipboard_requires_login(self):
|
|
response = self.client.get("/clipboard")
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn("/login", response.headers["Location"])
|
|
|
|
def test_clipboard_page_shows_upload_form(self):
|
|
self.login()
|
|
response = self.client.get("/clipboard")
|
|
html = response.get_data(as_text=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('action="/clipboard/upload"', html)
|
|
self.assertIn('name="file"', html)
|
|
|
|
def test_clipboard_upload_download_delete_flow(self):
|
|
import io
|
|
from sqlalchemy import func
|
|
from user_models import ClipboardFile, User, user_db
|
|
|
|
self.login()
|
|
upload = self.client.post(
|
|
"/clipboard/upload",
|
|
data={
|
|
"file": (io.BytesIO(b"clipboard payload"), "hello.txt", "text/plain")
|
|
},
|
|
content_type="multipart/form-data",
|
|
follow_redirects=False,
|
|
)
|
|
self.assertEqual(upload.status_code, 302)
|
|
|
|
with self.main_app.app_context():
|
|
user = User.query.filter_by(username="tester").one()
|
|
stored = (
|
|
ClipboardFile.query.filter_by(user_id=user.id)
|
|
.order_by(ClipboardFile.id.desc())
|
|
.first()
|
|
)
|
|
self.assertIsNotNone(stored)
|
|
file_id = stored.id
|
|
original_name = stored.original_name
|
|
size = stored.size
|
|
total = (
|
|
user_db.session.query(func.coalesce(func.sum(ClipboardFile.size), 0))
|
|
.filter(ClipboardFile.user_id == user.id)
|
|
.scalar()
|
|
)
|
|
|
|
self.assertEqual(original_name, "hello.txt")
|
|
self.assertEqual(size, len(b"clipboard payload"))
|
|
self.assertEqual(total, len(b"clipboard payload"))
|
|
|
|
download = self.client.get(f"/clipboard/{file_id}/download")
|
|
self.assertEqual(download.status_code, 200)
|
|
self.assertEqual(download.data, b"clipboard payload")
|
|
|
|
delete = self.client.post(f"/clipboard/{file_id}/delete")
|
|
self.assertEqual(delete.status_code, 302)
|
|
|
|
with self.main_app.app_context():
|
|
user_db.session.expire_all()
|
|
self.assertEqual(ClipboardFile.query.filter_by(id=file_id).count(), 0)
|
|
|
|
def test_notifications_can_be_deleted(self):
|
|
from user_models import Notification, User, user_db
|
|
|
|
self.login()
|
|
with self.main_app.app_context():
|
|
user = User.query.filter_by(username="tester").one()
|
|
first = Notification(user_id=user.id, message="first notification")
|
|
second = Notification(user_id=user.id, message="second notification")
|
|
user_db.session.add_all([first, second])
|
|
user_db.session.commit()
|
|
first_id = first.id
|
|
second_id = second.id
|
|
|
|
delete = self.client.post(f"/notifications/{first_id}/delete")
|
|
self.assertEqual(delete.status_code, 302)
|
|
|
|
with self.main_app.app_context():
|
|
user_db.session.expire_all()
|
|
self.assertEqual(Notification.query.filter_by(id=first_id).count(), 0)
|
|
self.assertEqual(Notification.query.filter_by(id=second_id).count(), 1)
|
|
|
|
def test_calendar_requires_login(self):
|
|
response = self.client.get("/calendar")
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn("/login", response.headers["Location"])
|
|
|
|
def test_calendar_month_and_week_views_render(self):
|
|
self.login()
|
|
month_response = self.client.get("/calendar?view=month")
|
|
self.assertEqual(month_response.status_code, 200)
|
|
self.assertIn("Calendar", month_response.get_data(as_text=True))
|
|
|
|
week_response = self.client.get("/calendar?view=week")
|
|
self.assertEqual(week_response.status_code, 200)
|
|
self.assertIn("Calendar", week_response.get_data(as_text=True))
|
|
|
|
def test_calendar_readouts_range_endpoint(self):
|
|
self.login()
|
|
sensor_id = self.create_sensor_and_readout("calendar-test")
|
|
|
|
response = self.client.get(
|
|
f"/api/sensors/{sensor_id}/readouts/range"
|
|
"?from=2000-01-01T00:00:00&to=2100-01-01T00:00:00"
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.get_json()
|
|
self.assertEqual(len(data), 1)
|
|
self.assertEqual(data[0]["sensor_id"], sensor_id)
|
|
|
|
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}",
|
|
data={"text": "renamed first item"},
|
|
).status_code,
|
|
302,
|
|
)
|
|
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.assertIn('value="renamed first item"', html)
|
|
self.assertLess(html.index("second item"), html.index("renamed 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", "renamed first item"],
|
|
)
|
|
|
|
def test_notifications_badge_and_mark_read(self):
|
|
from user_models import Notification, User, user_db
|
|
|
|
self.login()
|
|
with self.main_app.app_context():
|
|
user = User.query.filter_by(username="tester").one()
|
|
other = User(username="other-user")
|
|
other.set_password("secret")
|
|
user_db.session.add(other)
|
|
user_db.session.flush()
|
|
|
|
first = Notification(user_id=user.id, message="first notification")
|
|
second = Notification(user_id=user.id, message="second notification")
|
|
foreign = Notification(user_id=other.id, message="other user notification")
|
|
user_db.session.add_all([first, second, foreign])
|
|
user_db.session.commit()
|
|
first_id = first.id
|
|
foreign_id = foreign.id
|
|
|
|
home = self.client.get("/home").get_data(as_text=True)
|
|
self.assertIn('nav-badge">2</span>', home)
|
|
|
|
page = self.client.get("/notifications")
|
|
html = page.get_data(as_text=True)
|
|
self.assertEqual(page.status_code, 200)
|
|
self.assertIn("first notification", html)
|
|
self.assertIn("second notification", html)
|
|
self.assertNotIn("other user notification", html)
|
|
self.assertEqual(html.count('class="notification-item new"'), 2)
|
|
|
|
read = self.client.get(f"/notifications/{first_id}/read")
|
|
self.assertEqual(read.status_code, 302)
|
|
|
|
with self.main_app.app_context():
|
|
user_db.session.expire_all()
|
|
notification = user_db.session.get(Notification, first_id)
|
|
self.assertTrue(notification.is_read)
|
|
self.assertIsNotNone(notification.read_at)
|
|
|
|
page = self.client.get("/notifications")
|
|
html = page.get_data(as_text=True)
|
|
self.assertIn("1 new notification.", html)
|
|
self.assertIn('nav-badge">1</span>', html)
|
|
self.assertEqual(html.count('class="notification-item read"'), 1)
|
|
self.assertEqual(html.count('class="notification-item new"'), 1)
|
|
|
|
self.assertEqual(
|
|
self.client.get(f"/notifications/{foreign_id}/read").status_code, 404
|
|
)
|
|
|
|
def test_navigation_links_are_present_on_all_pages(self):
|
|
self.login()
|
|
links = (
|
|
'href="/home"',
|
|
'href="/sensors"',
|
|
'href="/notes"',
|
|
'href="/clipboard"',
|
|
'href="/notifications"',
|
|
'href="/api/docs"',
|
|
'href="/logout"',
|
|
)
|
|
for path in ("/home", "/sensors", "/notes", "/clipboard", "/notifications", "/api/docs"):
|
|
html = self.client.get(path).get_data(as_text=True)
|
|
for link in links:
|
|
self.assertIn(link, html, f"{link} missing on {path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|