docker-images/timepoll/tests/test_timepoll.py

import sqlite3
from datetime import datetime, timezone
from itertools import pairwise

import pytest
from domain import make_slots, remap_votes, resize_slots, summaries, votes_lost
from fastapi.testclient import TestClient
from server import create_app, normalize_code


def stamp(value):
    return int(datetime.fromisoformat(value).replace(tzinfo=timezone.utc).timestamp())


def test_dst_spring_skips_nonexistent_hour():
    slots = make_slots(["2026-03-29"], "02:00", "05:00", 30, "Europe/Helsinki")
    assert len(slots) == 4
    assert all(b - a == 1800 for a, b in pairwise(slots))


def test_dst_fall_keeps_both_repeated_hours():
    slots = make_slots(["2026-10-25"], "02:00", "05:00", 30, "Europe/Helsinki")
    assert len(slots) == 8
    assert len(set(slots)) == 8
    assert all(b - a == 1800 for a, b in pairwise(slots))


def test_slot_ending_at_spring_transition_is_not_lost():
    assert len(make_slots(["2026-03-29"], "02:00", "03:00", 30, "Europe/Helsinki")) == 2


def test_half_hour_dst_change_never_overlaps_slots_or_escapes_range():
    slots = make_slots(["2026-04-05"], "01:30", "03:30", 60, "Australia/Lord_Howe")
    assert len(slots) == 2
    assert all(b - a >= 3600 for a, b in pairwise(slots))
    # The first repeated 01:45 precedes a reset to 01:30, outside this range.
    slots = make_slots(["2026-04-05"], "01:45", "02:45", 60, "Australia/Lord_Howe")
    assert slots == [stamp("2026-04-04T15:15")]


def test_timezone_uses_date_not_current_offset():
    winter = make_slots(["2026-01-05"], "09:00", "10:00", 60, "Europe/Helsinki")
    summer = make_slots(["2026-07-05"], "09:00", "10:00", 60, "Europe/Helsinki")
    assert datetime.fromtimestamp(winter[0], timezone.utc).hour == 7
    assert datetime.fromtimestamp(summer[0], timezone.utc).hour == 6


def test_midnight_and_quarter_hour_zone():
    slots = make_slots(["2026-01-05"], "23:00", "24:00", 30, "Asia/Kathmandu")
    assert slots == [stamp("2026-01-05T17:15"), stamp("2026-01-05T17:45")]


@pytest.mark.parametrize(
    "start,end", [("17:00", "09:00"), ("09:00", "09:00"), ("09:00", "10:15")]
)
def test_invalid_ranges(start, end):
    with pytest.raises(ValueError):
        make_slots(["2026-01-01"], start, end, 30, "UTC")


def test_vote_resize():
    assert remap_votes({"0": "yes", "1800": "yes"}, 30, [0], 60) == {"0": "yes"}
    assert remap_votes({"0": "yes"}, 30, [0], 60) == {"0": "maybe"}
    assert remap_votes({"0": "maybe"}, 30, [0], 60) == {"0": "maybe"}
    assert remap_votes({"0": "yes", "1800": "maybe"}, 30, [0], 60) == {"0": "maybe"}
    assert remap_votes({"0": "yes"}, 60, [0, 1800], 30) == {"0": "yes", "1800": "yes"}
    assert resize_slots([0, 1800, 7200, 9000], 30, 60) == [0, 7200]
    assert votes_lost({"0": "yes"}, 60, [0], 30)
    assert not votes_lost({"0": "yes"}, 60, [0, 1800], 30)


def test_overlap_requires_same_people_and_counts_empty_voters():
    people = [
        {"id": 1, "votes": {"0": "yes", "1800": "yes"}},
        {"id": 2, "votes": {"0": "yes"}},
        {"id": 3, "votes": {"1800": "maybe"}},
        {"id": 4, "votes": {"0": "yes", "1800": "yes"}},
    ]
    result = summaries([0, 1800], 30, people)
    assert result["yes"][0]["seconds"] == 0
    assert result["yes"][1]["seconds"] == 1800
    assert len(result["inclusive"][1]["stretches"]) == 2
    assert (
        result["inclusive"][1]["stretches"][0]["people"]
        != result["inclusive"][1]["stretches"][1]["people"]
    )


def test_overlap_gap_and_midnight():
    people = [{"id": 1, "votes": {"84600": "yes", "86400": "yes", "91800": "yes"}}]
    result = summaries([84600, 86400, 91800], 30, people)
    assert result["yes"][0]["seconds"] == 3600


@pytest.fixture
def client(tmp_path):
    with TestClient(create_app(tmp_path)) as client:
        yield client


def settings(**kwargs):
    return dict(
        title="Weekend",
        timezone="Europe/Helsinki",
        fixed_timezone=False,
        minutes=30,
        slots=[1800000000, 1800001800, 1800003600, 1800005400],
        blocked=[],
        **kwargs,
    )


def create(client):
    response = client.post("/api/polls", json=settings())
    assert response.status_code == 201, response.text
    result = response.json()
    return result["code"], {"Authorization": "Bearer " + result["admin_token"]}


def add(client, code, name="Alice", votes=None):
    response = client.post(
        f"/api/polls/{code}/people",
        json={"name": name, "votes": votes or {}, "schedule_revision": 1},
    )
    assert response.status_code == 201, response.text
    return response.json()["id"]


def test_codes_and_private_token_are_not_public(client, tmp_path):
    code, headers = create(client)
    assert len(code) == 9 and len(headers["Authorization"].split()[1]) == 12
    assert normalize_code("ab-oil xyz") == "AB011XYZ"
    response = client.get(f"/api/polls/{code.lower()}")
    assert response.status_code == 200
    assert "admin" not in response.text
    with sqlite3.connect(tmp_path / "timepoll.sqlite3") as db:
        assert headers["Authorization"].split()[1] not in str(
            db.execute("SELECT * FROM polls").fetchall()
        )
    assert client.get(f"/api/polls/{code}/admin").status_code == 403
    assert client.get(f"/api/polls/{code}/admin", headers=headers).status_code == 200


def test_empty_duplicate_and_unavailable_votes(client):
    code, _ = create(client)
    add(client, code)
    duplicate = client.post(
        f"/api/polls/{code}/people",
        json={"name": " ALICE ", "votes": {}, "schedule_revision": 1},
    )
    assert duplicate.status_code == 409
    invalid = client.post(
        f"/api/polls/{code}/people",
        json={"name": "Bob", "votes": {"123": "yes"}, "schedule_revision": 1},
    )
    assert invalid.status_code == 422
    assert len(client.get(f"/api/polls/{code}").json()["people"]) == 1


def test_conflicts_and_unrelated_edits(client):
    code, headers = create(client)
    alice = add(client, code)
    add(client, code, "Bob")
    body = {
        "name": "Alice",
        "votes": {"1800000000": "yes"},
        "schedule_revision": 1,
        "revision": 1,
    }
    assert client.put(f"/api/polls/{code}/people/{alice}", json=body).status_code == 200
    assert client.put(f"/api/polls/{code}/people/{alice}", json=body).status_code == 409
    edit = settings(settings_revision=1)
    edit.update(title="Renamed", timezone="America/New_York")
    assert (
        client.put(f"/api/polls/{code}", json=edit, headers=headers).status_code == 200
    )
    assert client.get(f"/api/polls/{code}").json()["schedule_revision"] == 1
    body["revision"] = 2
    assert client.put(f"/api/polls/{code}/people/{alice}", json=body).status_code == 200
    assert (
        client.delete(f"/api/polls/{code}/people/{alice}?revision=2").status_code == 409
    )
    assert (
        client.delete(f"/api/polls/{code}/people/{alice}?revision=3").status_code == 200
    )


def test_admin_vote_loss_confirmation_and_conversion(client):
    code, headers = create(client)
    add(
        client,
        code,
        votes={"1800000000": "yes", "1800001800": "yes", "1800003600": "yes"},
    )
    edit = settings(settings_revision=1)
    edit.update(minutes=60, slots=[1800000000, 1800003600])
    response = client.put(f"/api/polls/{code}", json=edit, headers=headers)
    assert response.status_code == 200, response.text
    poll = client.get(f"/api/polls/{code}").json()
    assert poll["people"][0]["votes"] == {"1800000000": "yes", "1800003600": "maybe"}
    edit.update(settings_revision=2, blocked=[1800000000])
    response = client.put(f"/api/polls/{code}", json=edit, headers=headers)
    assert (
        response.status_code == 409 and response.json()["detail"]["code"] == "vote_loss"
    )
    assert client.get(f"/api/polls/{code}").json()["revision"] == poll["revision"]
    edit["confirm_loss"] = True
    assert (
        client.put(f"/api/polls/{code}", json=edit, headers=headers).status_code == 200
    )
    assert client.get(f"/api/polls/{code}").json()["people"][0]["votes"] == {
        "1800003600": "maybe"
    }


def test_blocked_votes_and_changed_schedule_conflict(client):
    code, headers = create(client)
    edit = settings(settings_revision=1)
    edit["blocked"] = [1800000000]
    client.put(f"/api/polls/{code}", json=edit, headers=headers)
    body = {"name": "Alice", "votes": {"1800000000": "yes"}, "schedule_revision": 1}
    assert client.post(f"/api/polls/{code}/people", json=body).status_code == 409
    body["schedule_revision"] = 2
    assert client.post(f"/api/polls/{code}/people", json=body).status_code == 422


def test_persistence(tmp_path):
    with TestClient(create_app(tmp_path)) as client:
        code, _ = create(client)
        add(client, code)
    with TestClient(create_app(tmp_path)) as restarted:
        assert (
            restarted.get(f"/api/polls/{code}").json()["people"][0]["name"] == "Alice"
        )


def test_security_validation_and_health(client):
    assert client.get("/healthz").json() == {"status": "ok"}
    assert client.get("/").headers["x-frame-options"] == "DENY"
    assert (
        client.post(
            "/api/polls",
            json=settings(),
            headers={"Origin": "https://elsewhere.invalid"},
        ).status_code
        == 403
    )
    assert (
        client.post("/api/polls", content=b"x" * (1024 * 1024 + 1)).status_code == 413
    )
    invalid = settings()
    invalid["slots"] = [1800000000, 1800000060]
    assert client.post("/api/polls", json=invalid).status_code == 422
    invalid.update(timezone="../etc/passwd")
    assert client.post("/api/polls", json=invalid).status_code == 422


def test_schedule_preview_and_resize(client):
    response = client.post(
        "/api/schedule",
        json={
            "dates": ["2026-10-25"],
            "start": "02:00",
            "end": "05:00",
            "minutes": 30,
            "timezone": "Europe/Helsinki",
        },
    )
    assert response.status_code == 200
    assert len(response.json()["slots"]) == 8
    body = settings()
    body["minutes"] = 60
    response = client.post("/api/resize?old_minutes=30", json=body)
    assert response.json()["slots"] == [1800000000, 1800003600]


def test_random_code_collision_is_retried(client, monkeypatch):
    code, headers = create(client)
    token = headers["Authorization"].split()[1]
    values = iter([code, token, "123456789", "123456789ABC"])
    monkeypatch.setattr("server.new_code", lambda length: next(values))
    response = client.post("/api/polls", json=settings())
    assert response.status_code == 201
    assert response.json()["code"] == "123456789"


def test_root_public_and_private_links(client):
    code, headers = create(client)
    token = headers["Authorization"].split()[1]
    assert client.get(f"/{code}").status_code == 200
    assert client.get(f"/{token}").status_code == 200
    response = client.post("/api/admin/resolve", headers=headers)
    assert response.json() == {"code": code}
    assert client.post("/api/admin/resolve").status_code == 403
    assert (
        client.post(
            "/api/admin/resolve", headers={"Authorization": "Bearer " + code}
        ).status_code
        == 403
    )
    assert client.get(f"/p/{code}").status_code == 404
    assert client.get(f"/admin/{code}").status_code == 404
    assert token not in client.get(f"/api/polls/{code}").text


def test_resizing_rounds_local_range_outward():
    slots = make_slots(["2026-09-15"], "09:15", "10:45", 15, "Europe/Helsinki")
    assert resize_slots(slots, 15, 30, "Europe/Helsinki") == make_slots(
        ["2026-09-15"], "09:00", "11:00", 30, "Europe/Helsinki"
    )
    # A non-whole-hour UTC offset must not change the local rounding origin.
    slots = make_slots(["2026-09-15"], "09:15", "10:45", 15, "Asia/Kathmandu")
    assert resize_slots(slots, 15, 60, "Asia/Kathmandu") == make_slots(
        ["2026-09-15"], "09:00", "11:00", 60, "Asia/Kathmandu"
    )


def test_resizing_merges_ranges_that_round_into_one_another():
    assert resize_slots([900, 2700], 15, 60) == [0]