vacationplanner2ics/tests/e2e.py

"""Four HTTP end-to-end scenarios; all credentials and vacation records are fake."""

import copy
import datetime as dt
import http.server
import json
import os
import subprocess
import threading
import time
import unittest
import urllib.error
import urllib.parse
import urllib.request

from icalendar import Calendar

TOKEN = "e2e-only-token+with/reserved=characters"
BASE = "http://127.0.0.1:8080"
WORKER = "Иван, Петров; \\" + "Длинное имя " * 12
COMMENT = "First line\r\nSecond; line, with \\ slash\rBEGIN:VEVENT\r\nSUMMARY:injection"


def vacation(identifier=1, start="2026-12-30", end="2027-01-03", **extra):
    return dict(
        id=identifier,
        workerName=WORKER,
        dateStart=start,
        dateEnd=end,
        departmentName="Engineering",
        comment=COMMENT,
        moderationStatus="approved",
        **extra,
    )


def request(path="/calendar.ics", parameters=None, method="GET"):
    url = BASE + path
    if parameters is not None:
        url += "?" + urllib.parse.urlencode(parameters)
    try:
        response = urllib.request.urlopen(urllib.request.Request(url, method=method), timeout=5)
    except urllib.error.HTTPError as error:
        response = error
    with response:
        return response.status, response.headers, response.read()


class Upstream(http.server.BaseHTTPRequestHandler):
    calls = []
    responses = {}

    def log_message(self, *_):
        pass

    def do_GET(self):
        self.calls.append((self.command, self.path, self.headers.get("Authorization")))
        response = self.responses.get(self.path, [])
        if callable(response):
            response = response(self)
        status, headers, payload = response if isinstance(response, tuple) else (200, {}, response)
        body = payload if isinstance(payload, bytes) else json.dumps(payload).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        for name, value in headers.items():
            self.send_header(name, value)
        self.end_headers()
        try:
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError):
            pass


class CalendarE2E(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.upstream = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Upstream)
        cls.thread = threading.Thread(target=cls.upstream.serve_forever, daemon=True)
        cls.thread.start()
        cls.origin = f"http://127.0.0.1:{cls.upstream.server_port}"

    @classmethod
    def tearDownClass(cls):
        cls.upstream.shutdown()
        cls.upstream.server_close()
        cls.thread.join()

    def setUp(self):
        Upstream.calls.clear()
        Upstream.responses.clear()
        self.process = subprocess.Popen(
            ["vacationplanner2ics"],
            env={**os.environ, "API_URL": self.origin, "UPSTREAM_TIMEOUT_SECONDS": "1"},
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )
        self.addCleanup(self.stop_service)
        for _ in range(100):
            if self.process.poll() is not None:
                self.fail("service exited before becoming ready")
            try:
                if request("/healthz")[0] == 200:
                    return
            except urllib.error.URLError:
                pass
            time.sleep(0.02)
        self.fail("service did not become ready")

    def stop_service(self):
        self.process.terminate()
        try:
            output, _ = self.process.communicate(timeout=5)
        except subprocess.TimeoutExpired:
            self.process.kill()
            self.process.communicate()
            self.fail("service did not shut down gracefully")
        self.assertEqual(self.process.returncode, 0)
        self.assertNotIn(TOKEN.encode(), output)
        self.assertNotIn(urllib.parse.quote_plus(TOKEN).encode(), output)
        self.assertNotIn(WORKER.encode(), output)

    def get_calendar(self, **parameters):
        status, headers, body = request(parameters={"token": TOKEN, **parameters})
        self.assertEqual(status, 200)
        self.assertEqual(headers["Content-Type"], "text/calendar; charset=utf-8")
        self.assertIn("no-store", headers["Cache-Control"])
        self.assertIn("vacations.ics", headers["Content-Disposition"])
        self.assertNotIn(TOKEN.encode(), body)
        self.assertTrue(body.endswith(b"END:VCALENDAR\r\n"))
        self.assertNotIn(b"\n", body.replace(b"\r\n", b""))
        for line in body.split(b"\r\n"):
            self.assertLessEqual(len(line), 75)
            line.decode("utf-8")
        calendar = Calendar.from_ical(body)
        self.assertEqual(str(calendar["VERSION"]), "2.0")
        self.assertIn("PRODID", calendar)
        for component in calendar.walk():
            self.assertFalse(component.errors)
        return calendar.walk("VEVENT")

    def test_default_years_and_icalendar_compatibility(self):
        year = dt.datetime.now(dt.UTC).year
        first = vacation(start=f"{year}-12-30", end=f"{year + 1}-01-03")
        pending = vacation(2, "2024-02-29", "2024-02-29")
        pending.update(moderationStatus="on_moderation", comment=None, departmentName=None)
        rejected = vacation(3)
        rejected["moderationStatus"] = "rejected"
        Upstream.responses = {
            f"/v1/vacations/year/{year}": [first, pending, rejected],
            f"/v1/vacations/year/{year + 1}": [first],
        }
        events = self.get_calendar()
        self.assertEqual(len(events), 3)
        self.assertEqual(events[0].decoded("DTSTART"), dt.date(year, 12, 30))
        self.assertEqual(events[0].decoded("DTEND"), dt.date(year + 1, 1, 4))
        self.assertEqual(events[1].decoded("DTEND"), dt.date(2024, 3, 1))
        self.assertEqual(str(events[0]["SUMMARY"]), WORKER + " — Vacation")
        self.assertIn(COMMENT.replace("\r\n", "\n").replace("\r", "\n"), str(events[0]["DESCRIPTION"]))
        self.assertEqual([str(event["STATUS"]) for event in events], ["CONFIRMED", "TENTATIVE", "CANCELLED"])
        self.assertTrue(all(event["DTSTART"].params["VALUE"] == "DATE" for event in events))
        self.assertTrue(all(event["DTSTAMP"].dt.utcoffset() == dt.timedelta() for event in events))
        self.assertEqual(Upstream.calls, [
            ("GET", f"/v1/vacations/year/{value}", "Bearer " + TOKEN) for value in (year, year + 1)
        ])

    def test_explicit_years_and_request_validation(self):
        for years, expected in [("2026", [2026]), ("2030,2026,2028,2027,2029,2026", list(range(2026, 2031)))]:
            with self.subTest(years=years):
                Upstream.calls.clear()
                self.assertEqual(self.get_calendar(years=years), [])
                self.assertEqual([call[1] for call in Upstream.calls], [f"/v1/vacations/year/{year}" for year in expected])
        Upstream.calls.clear()
        cases = [
            ({}, 401), ({"token": ""}, 401), ({"token": "bad\r\nheader"}, 401),
            ({"token": TOKEN, "years": ""}, 400),
            *[({"token": TOKEN, "years": value}, 400) for value in ["2026,nope", "2026,", "0", "9999", "2026-2027"]],
            ({"token": TOKEN, "year": "2026"}, 400),
            ({"token": TOKEN, "url": "https://example.invalid"}, 400),
            ([("token", TOKEN), ("token", TOKEN)], 400),
        ]
        for parameters, expected in cases:
            with self.subTest(parameters=parameters):
                status, headers, body = request(parameters=parameters)
                self.assertEqual(status, expected)
                self.assertIn("no-store", headers["Cache-Control"])
                self.assertNotIn(TOKEN.encode(), body)
        self.assertEqual(request(parameters={"token": TOKEN}, method="POST")[0], 405)
        self.assertEqual(request("/v1/workers")[0], 404)
        self.assertEqual(request("/healthz")[2], b"ok")
        self.assertEqual(Upstream.calls, [])

    def test_upstream_failures_never_return_partial_calendars(self):
        invalid_date = vacation()
        invalid_date["dateEnd"] = "not-a-date"
        reversed_dates = vacation(start="2026-02-10", end="2026-02-01")

        def slow(_):
            time.sleep(1.5)
            return []

        cases = [
            ((401, {}, TOKEN.encode()), 401),
            ((403, {}, TOKEN.encode()), 403),
            ((429, {}, TOKEN.encode()), 503),
            ((500, {}, TOKEN.encode()), 502),
            ((302, {"Location": self.origin + "/should-not-be-called"}, b""), 502),
            ((200, {}, b"not json " + TOKEN.encode()), 502),
            ((200, {}, {"result": "error", "message": TOKEN}), 502),
            ((200, {"Content-Length": str(10 * 1024 * 1024 + 1)}, b""), 502),
            ([invalid_date], 502), ([reversed_dates], 502), ([{"id": 1}], 502),
            (slow, 504),
        ]
        for response, expected in cases:
            with self.subTest(expected=expected, case=cases.index((response, expected))):
                Upstream.calls.clear()
                Upstream.responses = {
                    "/v1/vacations/year/2026": [vacation()],
                    "/v1/vacations/year/2027": response,
                }
                status, headers, body = request(parameters={"token": TOKEN, "years": "2026,2027"})
                self.assertEqual(status, expected)
                self.assertIn("no-store", headers["Cache-Control"])
                self.assertNotIn(b"BEGIN:VCALENDAR", body)
                self.assertNotIn(TOKEN.encode(), body)
                self.assertEqual(len(Upstream.calls), 2)

    def test_subscription_refresh_and_token_isolation(self):
        original = vacation()
        Upstream.responses = {"/v1/vacations/year/2026": [original]}
        first = self.get_calendar(years="2026")[0]
        changed = copy.deepcopy(original)
        changed.update(workerName="New name", dateEnd="2027-01-05", moderationStatus="on_moderation")
        Upstream.responses["/v1/vacations/year/2026"] = [changed]
        updated = self.get_calendar(years="2026")[0]
        self.assertEqual(first["UID"], updated["UID"])
        self.assertNotEqual(first["SUMMARY"], updated["SUMMARY"])
        self.assertEqual(updated.decoded("DTEND"), dt.date(2027, 1, 6))
        self.assertEqual(str(updated["STATUS"]), "TENTATIVE")
        Upstream.responses["/v1/vacations/year/2026"] = (
            lambda handler: [changed] if handler.headers["Authorization"] == "Bearer " + TOKEN else []
        )
        self.assertEqual(self.get_calendar(token="another-fake-token", years="2026"), [])
        self.assertEqual(self.get_calendar(years="2026")[0]["UID"], first["UID"])
        Upstream.responses["/v1/vacations/year/2026"] = []
        self.assertEqual(self.get_calendar(years="2026"), [])


if __name__ == "__main__":
    unittest.main(verbosity=2)