flowtrack/tests/test_app.py

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()