tests/test_approval_gated_calendar.py — Approval-Gated Calendar

"""Synthetic command tests.

By Zero Width Lab (zerowidthlab.com)
"""
# By Zero Width Lab (zerowidthlab.com)
from __future__ import annotations

import json
import io
import os
import subprocess
import sys
import tempfile
import unittest
import urllib.parse
from pathlib import Path

__author__ = "Zero Width Lab"
ROOT = Path(__file__).resolve().parent.parent
SCRIPT = ROOT / "scripts" / "approval_gated_calendar.py"
FIXTURE = ROOT / "tests" / "fixtures" / "graph-responses.json"
sys.path.insert(0, str(ROOT / "scripts"))
import approval_gated_calendar as calendar
WINDOW = ("--start", "2027-03-01T10:00:00+00:00", "--end", "2027-03-01T11:00:00+00:00")


def invoke(*args: str, fixture: Path = FIXTURE, credentials: bool = True) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    for key in ("M365_TENANT_ID", "M365_CLIENT_ID", "M365_CLIENT_SECRET", "M365_DEFAULT_USER", "M365_CALENDAR_FAKE_TRANSPORT", "M365_CALENDAR_ALLOW_FAKE"):
        env.pop(key, None)
    if credentials:
        env.update(M365_TENANT_ID="fixture-tenant", M365_CLIENT_ID="fixture-client", M365_CLIENT_SECRET="fixture-secret")
    env.update(M365_DEFAULT_USER="calendar@example.com", M365_CALENDAR_FAKE_TRANSPORT=str(fixture), M365_CALENDAR_ALLOW_FAKE="zero-width-lab-test-suite-only")
    return subprocess.run([sys.executable, str(SCRIPT), *args], cwd=ROOT, env=env, text=True, capture_output=True, check=False)


class CalendarTests(unittest.TestCase):
    def test_pagination_equivalent_endpoint_and_interval(self) -> None:
        from unittest.mock import patch
        query = urllib.parse.urlencode({"startDateTime": "2027-03-01T10:00:00+00:00", "endDateTime": "2027-03-01T11:00:00+00:00"})
        original = calendar.GRAPH + "/users/calendar%40example.com/calendarView?" + query
        rewritten_query = urllib.parse.urlencode({"startDateTime": "2027-03-01T11:00:00+01:00", "endDateTime": "2027-03-01T11:00:00Z"})
        for path in ("/users('calendar%40example.com')/calendarView", "/users/calendar@example.com/calendarView"):
            target = calendar.GRAPH + path + "?" + rewritten_query
            with self.subTest(path=path), patch.object(calendar, "request", side_effect=[{"value": [{"id": "one"}], "@odata.nextLink": target}, {"value": [{"id": "two"}]}]) as mocked:
                result = calendar.collect(original, "fixture")
                self.assertFalse(result["incomplete"], result)
                self.assertEqual(len(result["events"]), 2)
                self.assertEqual(mocked.call_count, 2)
        targets = [original.replace("calendarView", "events"), original.replace("calendar%40example.com", "other@example.com"), original.replace("11%3A00%3A00", "12%3A00%3A00"), original.replace("graph.microsoft.com", "login.microsoftonline.com")]
        for target in targets:
            with self.subTest(target=target), patch.object(calendar, "request", return_value={"value": [], "@odata.nextLink": target}) as mocked:
                result = calendar.collect(original, "fixture")
                self.assertTrue(result["incomplete"])
                self.assertEqual(mocked.call_count, 1)

    def test_html_http_error_keeps_status_and_read_retry(self) -> None:
        from unittest.mock import patch
        import urllib.error
        url = calendar.GRAPH + "/users/selected/events"
        class Response(io.BytesIO):
            status = 200
        gateway = urllib.error.HTTPError(url, 503, "gateway", {}, io.BytesIO(b"<html>unavailable</html>"))
        with patch.object(calendar, "fake", return_value=None), patch.object(calendar.OPENER, "open", side_effect=[gateway, Response(b'{"value":[]}')]) as mocked:
            self.assertEqual(calendar.request("GET", url), {"value": []})
            self.assertEqual(mocked.call_count, 2)
        for write in (False, True):
            error = urllib.error.HTTPError(url, 500, "gateway", {}, io.BytesIO(b"<html>error</html>"))
            with patch.object(calendar, "fake", return_value=None), patch.object(calendar.OPENER, "open", side_effect=error):
                with self.assertRaisesRegex(calendar.CalendarError, "outcome unknown: write returned HTTP 500" if write else "GET failed with HTTP 500"):
                    calendar.request("POST" if write else "GET", url, write=write)

    def test_write_redirect_meaning_is_preserved(self) -> None:
        from unittest.mock import patch
        import urllib.request
        url = calendar.GRAPH + "/users/selected/events"
        def redirected(req: urllib.request.Request, **kwargs: object) -> None:
            calendar.GuardedRedirect().redirect_request(req, None, 307, "", {}, url)
        message = "outcome unknown: write redirected; inspect the event before retrying"
        with patch.object(calendar, "fake", return_value=None), patch.object(calendar.OPENER, "open", side_effect=redirected):
            with self.assertRaises(calendar.CalendarError) as raised:
                calendar.request("POST", url, write=True)
            self.assertEqual(str(raised.exception), message)

    def test_http_response_limits_and_unknown_write_outcomes(self) -> None:
        from unittest.mock import patch
        import urllib.error
        url = "https://graph.microsoft.com/v1.0/users/selected/events"
        class Response(io.BytesIO):
            status = 200
        for write in (False, True):
            for raw in (b"bad-json", b"x" * (calendar.MAX_INPUT_BYTES + 1)):
                with patch.object(calendar, "fake", return_value=None), patch.object(calendar.OPENER, "open", return_value=Response(raw)):
                    with self.assertRaisesRegex(calendar.CalendarError, "outcome unknown" if write else "JSON|limit"):
                        calendar.request("POST" if write else "GET", url, write=write)
                error = urllib.error.HTTPError(url, 500, "server error", {}, io.BytesIO(raw))
                with patch.object(calendar, "fake", return_value=None), patch.object(calendar.OPENER, "open", side_effect=error):
                    with self.assertRaisesRegex(calendar.CalendarError, "outcome unknown" if write else "HTTP 500"):
                        calendar.request("POST" if write else "GET", url, write=write)

    def test_hygiene_forbidden_constructs(self) -> None:
        import ast
        forbidden = {"eval", "exec", "compile", "os.system", "re.compile"}
        for path in [SCRIPT, *sorted((ROOT / "tests").glob("*.py"))]:
            tree = ast.parse(path.read_text())
            for node in ast.walk(tree):
                if isinstance(node, ast.Call):
                    name = node.func.id if isinstance(node.func, ast.Name) else (f"{node.func.value.id}.{node.func.attr}" if isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name) else "")
                    self.assertNotIn(name, forbidden)
                    self.assertFalse(any(keyword.arg == "shell" and isinstance(keyword.value, ast.Constant) and keyword.value.value is True for keyword in node.keywords))
                if isinstance(node, ast.Import):
                    self.assertFalse(any(alias.name.split('.')[0] in ('pickle', 'ctypes') for alias in node.names))
                if isinstance(node, ast.ImportFrom):
                    self.assertNotIn((node.module or '').split('.')[0], ('pickle', 'ctypes'))

    def test_traversal_rejected_before_any_fixture_or_auth(self) -> None:
        absent = Path(tempfile.gettempdir()) / "calendar-fixture-does-not-exist.json"
        for target in ("..", "../other", "/absolute", "a\\b", "%2e%2e", "%252e%252e", "a%2fb", "a%252fb", "%255cb"):
            for args in (("show", target), ("list", *WINDOW, "--user", target)):
                result = invoke(*args, fixture=absent)
                self.assertEqual(result.returncode, 1, result.stdout)
                self.assertNotIn("fake fixture", result.stderr)
        for tenant in ("..", "%2e%2e", "a\\b", "/absolute"):
            from unittest.mock import patch
            with patch.dict(os.environ, {"M365_TENANT_ID": tenant, "M365_CLIENT_ID": "fixture", "M365_CLIENT_SECRET": "fixture"}), patch.object(calendar, "request", side_effect=AssertionError("request forbidden")):
                with self.assertRaises(calendar.CalendarError):
                    calendar.token()

    def test_bounded_and_malformed_json(self) -> None:
        for raw, message in ((b"x" * (calendar.MAX_INPUT_BYTES + 1), "limit"), (b"not-json", "valid JSON"), (b"[NaN,Infinity]", "valid JSON"), (b"[" * 100000 + b"]" * 100000, "valid JSON")):
            with self.assertRaisesRegex(calendar.CalendarError, message):
                calendar.bounded_json(io.BytesIO(raw), "input")
        with tempfile.TemporaryDirectory() as temporary:
            path = Path(temporary) / "fixture.json"
            for raw in ('{"responses":[{"method":"GET","status":"bad"}]}', 'x' * (calendar.MAX_INPUT_BYTES + 1)):
                path.write_text(raw)
                result = invoke("list", *WINDOW, fixture=path)
                self.assertEqual(result.returncode, 1)
                self.assertNotIn("Traceback", result.stderr)

    def test_malformed_pages_and_scope_links(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            path = Path(temporary) / "responses.json"
            for page in ({"value": [1]}, {"value": [{"id": "one", "subject": 1}]}, {"value": [], "@odata.nextLink": 7}, {"value": [], "@odata.nextLink": "https://graph.microsoft.com/v1.0/users/other/events"}, {"value": [], "@odata.nextLink": "https://graph.microsoft.com/v1.0/%2e%2e/users/other/events"}):
                data = json.loads(FIXTURE.read_text())
                data["responses"][1]["json"] = page
                path.write_text(json.dumps(data))
                result = invoke("search", *WINDOW, "--query", "plan", "--json", fixture=path)
                self.assertEqual(result.returncode, 2, result.stderr)
                self.assertTrue(json.loads(result.stdout)["incomplete"])
                self.assertNotIn("Traceback", result.stderr)

    def test_malformed_success_shapes(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            path = Path(temporary) / "responses.json"
            for args, index, response in ((("show", "event-one"), 2, []), (("availability", *WINDOW, "--attendee", "guest@example.com"), 3, {"value": [None]}), (("cancel", "event-one", "--comment", "Cancel", "--dry-run"), 2, {"id": "one", "changeKey": 42})):
                data = json.loads(FIXTURE.read_text())
                data["responses"][index]["json"] = response
                path.write_text(json.dumps(data))
                result = invoke(*args, fixture=path)
                self.assertEqual(result.returncode, 1)
                self.assertNotIn("Traceback", result.stderr)

    def test_redirect_and_bad_url_validation(self) -> None:
        for url in ("https://[invalid", "https://graph.microsoft.com/a/%2e%2e/b", "https://graph.microsoft.com/a\\b"):
            with self.assertRaises(calendar.CalendarError):
                calendar.guard(url)
        import urllib.request
        req = urllib.request.Request("https://graph.microsoft.com/v1.0/users/selected/events")
        for url in ("https://login.microsoftonline.com/path", "https://graph.microsoft.com/v1.0/users/other/events"):
            with self.assertRaises(calendar.CalendarError):
                calendar.GuardedRedirect().redirect_request(req, None, 302, "", {}, url)
        original = "https://graph.microsoft.com/v1.0/users/selected/calendarView?startDateTime=2027-03-01T10%3A00%3A00Z&endDateTime=2027-03-01T11%3A00%3A00Z"
        for target in (original.replace("10%3A00", "09%3A00"), original.split('?')[0]):
            with self.assertRaisesRegex(calendar.CalendarError, "interval"):
                calendar.preserve_interval(original, target)
            from unittest.mock import patch
            with patch.object(calendar, "request", return_value={"value": [], "@odata.nextLink": target}) as mocked:
                data = calendar.collect(original, "fixture")
                self.assertTrue(data["incomplete"])
                self.assertEqual(mocked.call_count, 1)

    def test_create_preview_is_complete_and_offline(self) -> None:
        result = invoke("create", *WINDOW, "--subject", "Planning", "--body", "Agenda", "--location", "Room", "--attendee", "guest@example.com", "--online-meeting", "--dry-run", "--json", credentials=False)
        self.assertEqual(result.returncode, 0, result.stderr)
        data = json.loads(result.stdout)
        self.assertEqual(data["plan"]["event"]["organizer"], "calendar@example.com")
        self.assertEqual(data["plan"]["event"]["attendees"][0]["emailAddress"]["address"], "guest@example.com")
        self.assertTrue(data["plan"]["event"]["isOnlineMeeting"])
        self.assertNotIn("fake transport engaged", result.stderr)

    def test_every_create_field_binds_digest(self) -> None:
        baseline = json.loads(invoke("create", *WINDOW, "--subject", "Planning", "--body", "Agenda", "--location", "Room", "--attendee", "guest@example.com", "--online-meeting", "--dry-run", "--json").stdout)["approval_digest"]
        variants = [
            ("--subject", "Different"), ("--body", "Changed"), ("--location", "Other"),
            ("--attendee", "other@example.com"), ("--time-zone", "Pacific Standard Time"),
            ("--end", "2027-03-01T12:00:00+00:00"),
        ]
        for flag, value in variants:
            args = ["create", *WINDOW, "--subject", "Planning", "--body", "Agenda", "--location", "Room", "--attendee", "guest@example.com", "--online-meeting", "--dry-run", "--json"]
            if flag in args:
                position = args.index(flag)
                args[position + 1] = value
            else:
                args.extend((flag, value))
            self.assertNotEqual(json.loads(invoke(*args).stdout)["approval_digest"], baseline, flag)
        without_online = invoke("create", *WINDOW, "--subject", "Planning", "--body", "Agenda", "--location", "Room", "--attendee", "guest@example.com", "--dry-run", "--json")
        self.assertNotEqual(json.loads(without_online.stdout)["approval_digest"], baseline)
        changed_mailbox = invoke("create", *WINDOW, "--subject", "Planning", "--body", "Agenda", "--location", "Room", "--attendee", "guest@example.com", "--online-meeting", "--user", "other@example.com", "--dry-run", "--json")
        self.assertNotEqual(json.loads(changed_mailbox.stdout)["approval_digest"], baseline)

    def test_missing_or_wrong_approval_before_fixture_read(self) -> None:
        absent = Path(tempfile.gettempdir()) / "calendar-fixture-does-not-exist.json"
        for approval in ((), ("--approve", "wrong")):
            result = invoke("create", *WINDOW, "--subject", "Planning", *approval, fixture=absent, credentials=False)
            self.assertEqual(result.returncode, 1)
            self.assertIn("approval digest", result.stderr)
            self.assertNotIn("fake fixture", result.stderr)

    def test_create_live(self) -> None:
        args = ("create", *WINDOW, "--subject", "Planning")
        approval = json.loads(invoke(*args, "--dry-run", "--json").stdout)["approval_digest"]
        result = invoke(*args, "--approve", approval, "--json")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout)["event"]["id"], "new-event")
        self.assertIn("fake transport engaged", result.stderr)

    def test_read_commands(self) -> None:
        for args, key in [(("list", *WINDOW), "events"), (("search", *WINDOW, "--query", "plan"), "events"), (("show", "event-one"), "id"), (("availability", *WINDOW, "--attendee", "guest@example.com"), "value")]:
            with self.subTest(args=args):
                result = invoke(*args, "--json")
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertIn(key, json.loads(result.stdout))
        result = invoke("search", *WINDOW, "--query", "plan", "--json")
        self.assertEqual(len(json.loads(result.stdout)["events"]), 1)

    def test_cancel_gate_and_conflict(self) -> None:
        preview = invoke("cancel", "event-one", "--comment", "Reschedule", "--dry-run", "--json")
        self.assertEqual(preview.returncode, 0, preview.stderr)
        self.assertEqual(json.loads(preview.stdout)["current_event"]["subject"], "Planning")
        digest = json.loads(preview.stdout)["approval_digest"]
        absent = Path(tempfile.gettempdir()) / "calendar-fixture-does-not-exist.json"
        wrong = invoke("cancel", "event-one", "--comment", "Changed", "--change-key", "key-one", "--approve", digest, fixture=absent, credentials=False)
        self.assertIn("approval digest", wrong.stderr)
        for status in (409, 412):
            with self.subTest(status=status), tempfile.TemporaryDirectory() as temporary:
                data = json.loads(FIXTURE.read_text())
                data["responses"][-1]["status"] = status
                path = Path(temporary) / "responses.json"
                path.write_text(json.dumps(data))
                result = invoke("cancel", "event-one", "--comment", "Reschedule", "--change-key", "key-one", "--approve", digest, fixture=path)
                self.assertEqual(result.returncode, 1)
                self.assertIn(f"cancel conflicts with a concurrent change (HTTP {status}); rerun --dry-run for a fresh approval", result.stderr)

    def test_fake_redirects_recheck_host_and_refuse_write(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            path = Path(temporary) / "redirects.json"
            data = json.loads(FIXTURE.read_text())
            view = "https://graph.microsoft.com/v1.0/users/calendar%40example.com/calendarView"
            redirected_view = view + "?" + urllib.parse.urlencode({"startDateTime": WINDOW[1], "endDateTime": WINDOW[3], "page": "2"})
            data["responses"][1] = {"method": "GET", "url_prefix": view, "status": 302, "location": redirected_view}
            data["responses"].append({"method": "GET", "url_prefix": redirected_view, "status": 200, "json": {"value": [{"id": "redirected-event"}]}})
            path.write_text(json.dumps(data))
            allowed = invoke("list", *WINDOW, "--json", fixture=path)
            self.assertEqual(allowed.returncode, 0, allowed.stderr)
            self.assertEqual(json.loads(allowed.stdout)["events"][0]["id"], "redirected-event")

            data["responses"][1]["location"] = "https://unlisted.example.com/events"
            path.write_text(json.dumps(data))
            blocked = invoke("list", *WINDOW, "--json", fixture=path)
            self.assertEqual(blocked.returncode, 2)
            self.assertTrue(json.loads(blocked.stdout)["incomplete"])
            self.assertIn("destination must use HTTPS at a declared host", json.loads(blocked.stdout)["skipped"])

            preview = json.loads(invoke("create", *WINDOW, "--subject", "Planning", "--dry-run", "--json").stdout)
            for status in (302, 307):
                with self.subTest(status=status):
                    data = json.loads(FIXTURE.read_text())
                    data["responses"][4] = {"method": "POST", "url_prefix": "https://graph.microsoft.com/v1.0/users/calendar%40example.com/events", "status": status, "location": redirected_view}
                    path.write_text(json.dumps(data))
                    refused = invoke("create", *WINDOW, "--subject", "Planning", "--approve", preview["approval_digest"], fixture=path)
                    self.assertEqual(refused.returncode, 1)
                    self.assertIn("outcome unknown: write redirected", refused.stderr)

    def test_cancel_digest_binds_target_and_change_key(self) -> None:
        preview = json.loads(invoke("cancel", "event-one", "--comment", "Reschedule", "--dry-run", "--json").stdout)
        approval = preview["approval_digest"]
        absent = Path(tempfile.gettempdir()) / "calendar-fixture-does-not-exist.json"
        for args in (
            ("event-one", "--comment", "Changed", "--change-key", "key-one"),
            ("event-one", "--comment", "Reschedule", "--change-key", "other-key"),
            ("event-two", "--comment", "Reschedule", "--change-key", "key-one"),
            ("event-one", "--user", "other@example.com", "--comment", "Reschedule", "--change-key", "key-one"),
        ):
            with self.subTest(args=args):
                result = invoke("cancel", *args, "--approve", approval, fixture=absent, credentials=False)
                self.assertEqual(result.returncode, 1)
                self.assertIn("approval digest", result.stderr)
                self.assertNotIn("fake fixture", result.stderr)

    def test_cancel_refuses_stale_change_key(self) -> None:
        approval = json.loads(invoke("cancel", "event-one", "--comment", "Reschedule", "--dry-run", "--json").stdout)["approval_digest"]
        with tempfile.TemporaryDirectory() as temporary:
            data = json.loads(FIXTURE.read_text())
            data["responses"][2]["json"]["changeKey"] = "updated-key"
            path = Path(temporary) / "responses.json"
            path.write_text(json.dumps(data))
            result = invoke("cancel", "event-one", "--comment", "Reschedule", "--change-key", "key-one", "--approve", approval, fixture=path)
            self.assertEqual(result.returncode, 1)
            self.assertIn("approval expired", result.stderr)

    def test_escaping_and_missing_fixture(self) -> None:
        result = invoke("create", *WINDOW, "--subject", "Line\nnext", "--dry-run")
        self.assertEqual(result.returncode, 0)
        self.assertIn("Line\\u000anext", result.stdout)
        self.assertNotIn("Line\nnext", result.stdout)
        absent = Path(tempfile.gettempdir()) / "calendar-fixture-does-not-exist.json"
        result = invoke("list", *WINDOW, fixture=absent)
        self.assertIn("fake fixture unavailable", result.stderr)

    def test_partial_requires_opt_in(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            path = Path(temporary) / "partial.json"
            data = json.loads(FIXTURE.read_text())
            data["responses"][1]["json"]["@odata.nextLink"] = "https://graph.microsoft.com/v1.0/missing"
            path.write_text(json.dumps(data))
            result = invoke("list", *WINDOW, "--json", fixture=path)
            self.assertEqual(result.returncode, 2)
            self.assertTrue(json.loads(result.stdout)["incomplete"])
            allowed = invoke("list", *WINDOW, "--json", "--allow-partial", fixture=path)
            self.assertEqual(allowed.returncode, 0)


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

License: MIT (read it).

Back to the listing