sensors is separate API with td
This commit is contained in:
@@ -0,0 +1,200 @@
|
||||
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("/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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user