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}", 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"], ) if __name__ == "__main__": unittest.main()