224 lines
11 KiB
Python
224 lines
11 KiB
Python
import base64
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from datetime import date, timedelta
|
|
from http.client import HTTPConnection
|
|
from pathlib import Path
|
|
|
|
import app
|
|
|
|
|
|
class StoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.store = app.Store(Path(self.temp.name) / "flowtrack.sqlite3")
|
|
self.store.initialize()
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_empty_log_deletes_the_day(self):
|
|
self.store.set_day("2026-10-01", {"flow": "heavy", "symptoms": ["cramps"]})
|
|
self.assertEqual(self.store.days(), {"2026-10-01": {"flow": "heavy", "symptoms": ["cramps"]}})
|
|
self.store.set_day("2026-10-01", {"flow": None, "symptoms": []})
|
|
self.assertEqual(self.store.days(), {})
|
|
|
|
def test_fill_never_overwrites(self):
|
|
self.store.set_day("2026-10-01", {"flow": "heavy", "symptoms": []})
|
|
added = self.store.fill_days({
|
|
"2026-10-01": {"flow": "period", "symptoms": []},
|
|
"2026-10-02": {"flow": "period", "symptoms": []},
|
|
})
|
|
self.assertEqual(added, 1)
|
|
self.assertEqual(self.store.days()["2026-10-01"]["flow"], "heavy")
|
|
|
|
def test_backup_rotates(self):
|
|
self.store.set_day("2026-10-01", {"flow": "light", "symptoms": []})
|
|
destination = Path(self.temp.name) / "backups"
|
|
destination.mkdir()
|
|
for stale in ("2026-01-01", "2026-01-02", "2026-01-03"):
|
|
(destination / f"flowtrack-{stale}.sqlite3").touch()
|
|
target = self.store.backup(destination, keep=2)
|
|
self.assertEqual(sorted(path.name for path in destination.iterdir()), ["flowtrack-2026-01-03.sqlite3", target.name])
|
|
with sqlite3.connect(target) as copy:
|
|
self.assertEqual(copy.execute("SELECT flow FROM days").fetchone()[0], "light")
|
|
|
|
|
|
class ValidationTests(unittest.TestCase):
|
|
def test_dates(self):
|
|
self.assertIsNotNone(app.parse_day(date.today().isoformat()))
|
|
self.assertIsNotNone(app.parse_day((date.today() + timedelta(days=1)).isoformat()))
|
|
self.assertIsNone(app.parse_day((date.today() + timedelta(days=2)).isoformat()))
|
|
self.assertIsNone(app.parse_day("2026-02-30"))
|
|
self.assertIsNone(app.parse_day("26-10-01"))
|
|
|
|
def test_entries(self):
|
|
self.assertEqual(app.clean_entry({"flow": "light", "symptoms": ["acne", "cramps", "acne"]}), {"flow": "light", "symptoms": ["acne", "cramps"]})
|
|
self.assertIsNone(app.clean_entry({"flow": "gushing"}))
|
|
self.assertIsNone(app.clean_entry({"flow": None, "symptoms": ["<script>"]}))
|
|
self.assertIsNone(app.clean_entry({"flow": None, "symptoms": "cramps"}))
|
|
|
|
|
|
class ServerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
store = app.Store(Path(self.temp.name) / "flowtrack.sqlite3")
|
|
store.initialize()
|
|
self.server = app.TrackerServer(("127.0.0.1", 0), store, {"kayla"}, {"fred"})
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.temp.cleanup()
|
|
|
|
def request(self, method, path, user=None, body=None):
|
|
connection = HTTPConnection("127.0.0.1", self.server.server_address[1], timeout=5)
|
|
headers = {"Remote-User": user} if user else {}
|
|
payload = None
|
|
if body is not None:
|
|
payload = json.dumps(body).encode()
|
|
headers["Content-Type"] = "application/json"
|
|
connection.request(method, path, body=payload, headers=headers)
|
|
response = connection.getresponse()
|
|
data = response.read()
|
|
connection.close()
|
|
try:
|
|
return response.status, json.loads(data)
|
|
except json.JSONDecodeError:
|
|
return response.status, data.decode()
|
|
|
|
def test_unknown_users_get_nothing(self):
|
|
self.assertEqual(self.request("GET", "/api/state")[0], 403)
|
|
self.assertEqual(self.request("GET", "/api/state", user="mallory")[0], 403)
|
|
self.assertEqual(self.request("PUT", "/api/days/2026-10-01", user="mallory", body={"flow": "light"})[0], 403)
|
|
|
|
def test_editor_round_trip_and_export(self):
|
|
status, _ = self.request("PUT", "/api/days/2026-10-01", user="kayla", body={"flow": "heavy", "symptoms": ["cramps"]})
|
|
self.assertEqual(status, 200)
|
|
status, state = self.request("GET", "/api/state", user="kayla")
|
|
self.assertEqual(state["role"], "editor")
|
|
self.assertEqual(state["days"]["2026-10-01"], {"flow": "heavy", "symptoms": ["cramps"]})
|
|
status, exported = self.request("GET", "/api/export.csv", user="kayla")
|
|
self.assertEqual(status, 200)
|
|
self.assertIn("2026-10-01,heavy,Cramps", exported)
|
|
|
|
def test_viewer_is_read_only_and_details_are_opt_in(self):
|
|
self.request("PUT", "/api/days/2026-10-01", user="kayla", body={"flow": "heavy", "symptoms": ["cramps"]})
|
|
self.request("PUT", "/api/days/2026-10-09", user="kayla", body={"flow": "spotting", "symptoms": []})
|
|
self.request("PUT", "/api/days/2026-10-10", user="kayla", body={"flow": None, "symptoms": ["acne"]})
|
|
|
|
status, state = self.request("GET", "/api/state", user="fred")
|
|
self.assertEqual((status, state["role"]), (200, "viewer"))
|
|
self.assertEqual(state["days"], {"2026-10-01": {"flow": "period", "symptoms": []}})
|
|
|
|
self.assertEqual(self.request("PUT", "/api/days/2026-10-02", user="fred", body={"flow": "light"})[0], 403)
|
|
self.assertEqual(self.request("PUT", "/api/settings", user="fred", body={"shareDetails": True})[0], 403)
|
|
self.assertEqual(self.request("POST", "/api/days", user="fred", body={"days": {"2026-10-02": {"flow": "period"}}})[0], 403)
|
|
self.assertEqual(self.request("GET", "/api/export.json", user="fred")[0], 403)
|
|
|
|
self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"shareDetails": True})[0], 200)
|
|
_, state = self.request("GET", "/api/state", user="fred")
|
|
self.assertEqual(len(state["days"]), 3)
|
|
self.assertEqual(state["days"]["2026-10-01"]["symptoms"], ["cramps"])
|
|
|
|
def test_rejects_bad_writes(self):
|
|
future = (date.today() + timedelta(days=5)).isoformat()
|
|
self.assertEqual(self.request("PUT", f"/api/days/{future}", user="kayla", body={"flow": "light"})[0], 400)
|
|
self.assertEqual(self.request("PUT", "/api/days/2026-10-01", user="kayla", body={"flow": "lots"})[0], 400)
|
|
self.assertEqual(self.request("POST", "/api/days", user="kayla", body={"days": {"bad": {"flow": "period"}}})[0], 400)
|
|
|
|
def test_seed_batch(self):
|
|
status, result = self.request("POST", "/api/days", user="kayla", body={"days": {
|
|
"2026-09-01": {"flow": "period", "symptoms": []},
|
|
"2026-09-02": {"flow": "period", "symptoms": []},
|
|
}})
|
|
self.assertEqual((status, result), (200, {"added": 2}))
|
|
|
|
def test_push_is_off_without_a_helper(self):
|
|
_, state = self.request("GET", "/api/state", user="kayla")
|
|
self.assertFalse(state["push"]["available"])
|
|
self.assertEqual(self.request("POST", "/api/push/test", user="kayla")[0], 503)
|
|
_, state = self.request("GET", "/api/state", user="fred")
|
|
self.assertIsNone(state["push"])
|
|
|
|
def test_remind_days_setting(self):
|
|
self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"remindDays": 3})[1], {"shareDetails": False, "remindDays": 3})
|
|
self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"remindDays": 9})[0], 400)
|
|
self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"remindDays": True})[0], 400)
|
|
self.assertEqual(self.request("PUT", "/api/settings", user="kayla", body={"other": 1})[0], 400)
|
|
_, state = self.request("GET", "/api/state", user="kayla")
|
|
self.assertEqual(state["push"]["remindDays"], 3)
|
|
|
|
def test_static_files(self):
|
|
status, body = self.request("GET", "/")
|
|
self.assertEqual(status, 200)
|
|
self.assertIn("Flowtrack", body)
|
|
self.assertEqual(self.request("GET", "/../app.py")[0], 404)
|
|
self.assertEqual(self.request("GET", "/manifest.webmanifest")[0], 200)
|
|
|
|
|
|
FAKE_NOTIFY = """#!{python}
|
|
import json, sqlite3, sys
|
|
command, database = sys.argv[1], sys.argv[3]
|
|
if command == "init":
|
|
with sqlite3.connect(database) as db:
|
|
db.execute("INSERT OR IGNORE INTO settings VALUES ('vapid_public', 'PUBLIC')")
|
|
print(json.dumps({{"publicKey": "PUBLIC"}}))
|
|
else:
|
|
print(json.dumps({{"sent": 1, "failed": 0, "removed": 0, "errors": [], "contact": sys.argv[5]}}))
|
|
"""
|
|
|
|
APPLE_SUBSCRIPTION = {
|
|
"endpoint": "https://web.push.apple.com/QGuQyavXutnMH",
|
|
"keys": {"p256dh": base64.urlsafe_b64encode(b"\x04" + b"k" * 64).decode().rstrip("="), "auth": base64.urlsafe_b64encode(b"a" * 16).decode().rstrip("=")},
|
|
}
|
|
|
|
|
|
class PushServerTests(ServerTests):
|
|
def setUp(self):
|
|
super().setUp()
|
|
script = Path(self.temp.name) / "notify"
|
|
script.write_text(FAKE_NOTIFY.format(python=sys.executable))
|
|
script.chmod(0o755)
|
|
self.server.notify = str(script)
|
|
self.server.push_contact = "https://flowtrack.example.org"
|
|
self.server.run_notify("init")
|
|
|
|
# Every other server test also runs again with push switched on.
|
|
test_push_is_off_without_a_helper = None
|
|
|
|
def test_push_state(self):
|
|
_, state = self.request("GET", "/api/state", user="kayla")
|
|
self.assertEqual(state["push"], {"available": True, "publicKey": "PUBLIC", "remindDays": 2, "devices": 0})
|
|
|
|
def test_subscribe_test_and_unsubscribe(self):
|
|
self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=APPLE_SUBSCRIPTION), (200, {"devices": 1}))
|
|
self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=APPLE_SUBSCRIPTION), (200, {"devices": 1}))
|
|
status, result = self.request("POST", "/api/push/test", user="kayla")
|
|
self.assertEqual((status, result["sent"], result["contact"]), (200, 1, "https://flowtrack.example.org"))
|
|
self.assertEqual(self.request("POST", "/api/push/test", user="fred")[0], 403)
|
|
self.assertEqual(self.request("POST", "/api/push/unsubscribe", user="kayla", body={"endpoint": APPLE_SUBSCRIPTION["endpoint"]}), (200, {"devices": 0}))
|
|
|
|
def test_only_real_push_services(self):
|
|
for endpoint in (
|
|
"https://127.0.0.1/push",
|
|
"http://web.push.apple.com/x",
|
|
"https://evilpush.apple.com.example.org/x",
|
|
"https://router.nordhammer.it/api",
|
|
):
|
|
body = {**APPLE_SUBSCRIPTION, "endpoint": endpoint}
|
|
self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=body)[0], 400, endpoint)
|
|
bad_keys = {**APPLE_SUBSCRIPTION, "keys": {"p256dh": "short", "auth": APPLE_SUBSCRIPTION["keys"]["auth"]}}
|
|
self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body=bad_keys)[0], 400)
|
|
self.assertEqual(self.request("POST", "/api/push/subscribe", user="fred", body=APPLE_SUBSCRIPTION)[0], 403)
|
|
self.assertEqual(self.request("POST", "/api/push/subscribe", user="kayla", body={**APPLE_SUBSCRIPTION, "endpoint": "https://fcm.googleapis.com/fcm/send/abc"})[0], 200)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|