From c29812475c7cd03fc1309dbcabb4d0511e0ec0c8 Mon Sep 17 00:00:00 2001 From: Mario Fetka Date: Wed, 22 Jul 2026 08:36:48 +0200 Subject: [PATCH] Add comprehensive web and collector fixtures --- contrib/testing/README.md | 75 +++ contrib/testing/cyrus-fixture.py | 390 ++++++++++++ docs/release-testing-0.7.md | 26 +- src/agents/collector/CMakeLists.txt | 18 +- .../tests/vtestmail-transport-test.c | 70 ++- src/www/CMakeLists.txt | 2 + src/www/tests/test_web_api.py | 586 ++++++++++++++++++ 7 files changed, 1129 insertions(+), 38 deletions(-) create mode 100644 contrib/testing/README.md create mode 100755 contrib/testing/cyrus-fixture.py create mode 100644 src/www/tests/test_web_api.py diff --git a/contrib/testing/README.md b/contrib/testing/README.md new file mode 100644 index 0000000..61b4de7 --- /dev/null +++ b/contrib/testing/README.md @@ -0,0 +1,75 @@ +# Local mail protocol fixtures + +These fixtures are intentionally bound to loopback and use non-standard ports. +They are development tools, not production configuration. + +## Cyrus IMAP, POP3, and LMTP + +Install `net-mail/cyrus-imapd`, then start the disposable fixture as root: + +```sh +sudo ./contrib/testing/cyrus-fixture.py start +``` + +It creates no files outside `/tmp/bongo-cyrus-fixture` and listens on: + +- IMAP: `127.0.0.1:18143` +- POP3: `127.0.0.1:18110` +- LMTP: `127.0.0.1:18024` + +The source mailboxes are `pop-source@cyrus.bongo.test` and +`imap-source@cyrus.bongo.test`, with the test-only password +`BongoCollector-Test-2026`. Insert a complete RFC 5322 message through Cyrus' +real LMTP delivery path: + +```sh +sudo ./contrib/testing/cyrus-fixture.py inject message.eml +``` + +Seed three different messages in each source mailbox and exercise Bongo's real +Collector POP3 and IMAP transport implementation: + +```sh +./contrib/testing/cyrus-fixture.py verify-collector \ + --probe ./build/src/agents/collector/collector-vtestmail-transport-test \ + --count 3 +``` + +The probe lists and fetches every message, checks its fixture marker, deletes +it through the selected protocol, and verifies that the Cyrus source mailbox +is empty. A live end-to-end run creates two Bongo external accounts: the POP3 +source delivers into Bongo user `test1`, while the IMAP source delivers into a +different Bongo user. Both use `after_import`, so Cyrus deletion happens only +after the Bongo Store accepts the complete message. + +The same LMTP listener is the forwarding target for the exact transport-domain +mapping below. This accepts `*@cyrus.bongo.test` without making the test domain +a local Bongo domain: + +```json +"lmtp_transports": ["cyrus.bongo.test=127.0.0.1:18024"] +``` + +The generated self-signed certificate at +`/tmp/bongo-cyrus-fixture/tls/server.crt` is the fixture trust anchor. Set +`BONGO_COLLECTOR_CAINFO` to that path for the Collector process during this +test; normal installations continue to use the operating-system trust store. + +Stop or remove all fixture state with: + +```sh +sudo ./contrib/testing/cyrus-fixture.py stop +sudo ./contrib/testing/cyrus-fixture.py reset +``` + +## Outbound SMTP capture + +Use `net-mail/xeams-devnullsmtp-bin` from the Bongo Gentoo overlay: + +```sh +xeams-devnullsmtp -p 2526 -d bongo.test -s /tmp/bongo-devnull-mail +``` + +Configure `127.0.0.1:2526` as Bongo's relay host before testing any non-local +recipient. Xeams DevNullSMTP has no listen-address option, so isolate it with a +firewall or network namespace; do not expose it on a production network. diff --git a/contrib/testing/cyrus-fixture.py b/contrib/testing/cyrus-fixture.py new file mode 100755 index 0000000..92481ea --- /dev/null +++ b/contrib/testing/cyrus-fixture.py @@ -0,0 +1,390 @@ +#!/usr/bin/env python3 +# /**************************************************************************** +# * +# * Copyright (c) 2001 Novell, Inc. All Rights Reserved. +# * +# * This program is free software; you can redistribute it and/or +# * modify it under the terms of version 2 of the GNU General Public License +# * as published by the Free Software Foundation. +# * +# * This program is distributed in the hope that it will be useful, +# * but WITHOUT ANY WARRANTY; without even the implied warranty of +# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# * GNU General Public License for more details. +# * +# * You should have received a copy of the GNU General Public License +# * along with this program; if not, contact Novell, Inc. +# * +# * To contact Novell about this file by physical or electronic mail, you +# * may find current contact information at www.novell.com. +# * +# ****************************************************************************/ + +"""Manage a disposable loopback-only Cyrus IMAP/POP3/LMTP test server.""" + +from __future__ import annotations + +import argparse +from email.message import EmailMessage +import grp +import imaplib +import os +from pathlib import Path +import pwd +import shutil +import signal +import smtplib +import socket +import subprocess +import sys +import time + + +DEFAULT_ROOT = Path("/tmp/bongo-cyrus-fixture") +DEFAULT_PASSWORD = "BongoCollector-Test-2026" +FIXTURE_DOMAIN = "cyrus.bongo.test" +SOURCE_USERS = ("pop-source", "imap-source") +PORTS = {"imap": 18143, "pop3": 18110, "lmtp": 18024} +BINARIES = { + "master": "/usr/libexec/cyrusmaster", + "imap": "/usr/libexec/imapd", + "pop3": "/usr/libexec/pop3d", + "lmtp": "/usr/libexec/lmtpd", + "recover": "/usr/sbin/ctl_cyrusdb", + "saslpasswd": "/usr/bin/saslpasswd2", + "certtool": "/usr/bin/certtool", +} + + +def checked_root(value: str) -> Path: + root = Path(value).resolve() + if root == Path("/") or not str(root).startswith("/tmp/bongo-cyrus-"): + raise argparse.ArgumentTypeError( + "fixture root must be below /tmp and start with bongo-cyrus-") + return root + + +def require_root() -> None: + if os.geteuid() != 0: + raise SystemExit("run this command as root (or through sudo)") + + +def paths(root: Path) -> dict[str, Path]: + return { + "config": root / "config", + "spool": root / "spool", + "run": root / "run", + "imapd": root / "imapd.conf", + "cyrus": root / "cyrus.conf", + "sasldb": root / "sasldb2", + "pid": root / "run" / "master.pid", + "ready": root / "run" / "master.ready", + "log": root / "run" / "master.log", + "certificate": root / "tls" / "server.crt", + "private_key": root / "tls" / "server.key", + "certificate_template": root / "tls" / "server.tmpl", + } + + +def write_configuration(root: Path) -> None: + values = paths(root) + imapd = f"""# Generated disposable Bongo test fixture. +configdirectory: {values['config']} +partition-default: {values['spool']} +defaultpartition: default +admins: testadmin +servername: {FIXTURE_DOMAIN} +allowplaintext: yes +virtdomains: off +altnamespace: yes +sasl_mech_list: PLAIN LOGIN +sasl_pwcheck_method: auxprop +sasl_auxprop_plugin: sasldb +sasl_sasldb_path: {values['sasldb']} +sasl_default_realm: {FIXTURE_DOMAIN} +tls_server_cert: {values['certificate']} +tls_server_key: {values['private_key']} +tls_versions: tls1_2 tls1_3 +master_pid_file: {values['pid']} +master_ready_file: {values['ready']} +statuscache: off +duplicate_db: skiplist +mboxlist_db: skiplist +subscription_db: flat +seenstate_db: flat +""" + config = f"""# Generated disposable Bongo test fixture. +START {{ + recover cmd=\"{BINARIES['recover']} -C {values['imapd']} -r\" +}} +SERVICES {{ + imap cmd=\"{BINARIES['imap']} -C {values['imapd']}\" listen=\"127.0.0.1:{PORTS['imap']}\" prefork=0 + pop3 cmd=\"{BINARIES['pop3']} -C {values['imapd']}\" listen=\"127.0.0.1:{PORTS['pop3']}\" prefork=0 + lmtp cmd=\"{BINARIES['lmtp']} -C {values['imapd']}\" listen=\"127.0.0.1:{PORTS['lmtp']}\" prefork=0 +}} +EVENTS {{ +}} +""" + values["imapd"].write_text(imapd, encoding="utf-8") + values["cyrus"].write_text(config, encoding="utf-8") + + +def set_password(database: Path, username: str, password: str) -> None: + subprocess.run( + [BINARIES["saslpasswd"], "-p", "-c", "-f", str(database), + "-u", FIXTURE_DOMAIN, username], + input=(password + "\n").encode("utf-8"), check=True) + + +def generate_certificate(root: Path) -> None: + values = paths(root) + template = f"""cn = {FIXTURE_DOMAIN} +dns_name = {FIXTURE_DOMAIN} +dns_name = localhost +ip_address = 127.0.0.1 +expiration_days = 2 +signing_key +encryption_key +tls_www_server +""" + values["certificate_template"].write_text(template, encoding="ascii") + subprocess.run( + [BINARIES["certtool"], "--generate-privkey", "--sec-param", "high", + "--outfile", str(values["private_key"])], check=True, + stdout=subprocess.DEVNULL) + subprocess.run( + [BINARIES["certtool"], "--generate-self-signed", + "--load-privkey", str(values["private_key"]), + "--template", str(values["certificate_template"]), + "--outfile", str(values["certificate"])], check=True, + stdout=subprocess.DEVNULL) + + +def prepare(root: Path, password: str) -> None: + require_root() + missing = [path for path in BINARIES.values() if not Path(path).is_file()] + if missing: + raise SystemExit("missing Cyrus tools: " + ", ".join(missing)) + values = paths(root) + for directory in (root, values["config"], values["spool"], values["run"], + root / "tls", + values["config"] / "socket", values["config"] / "proc", + values["config"] / "log", values["config"] / "msg"): + directory.mkdir(parents=True, exist_ok=True) + write_configuration(root) + generate_certificate(root) + set_password(values["sasldb"], "testadmin", password) + for username in SOURCE_USERS: + set_password(values["sasldb"], username, password) + cyrus = pwd.getpwnam("cyrus") + mail_gid = grp.getgrnam("mail").gr_gid + for directory, subdirs, filenames in os.walk(root): + os.chown(directory, cyrus.pw_uid, mail_gid) + os.chmod(directory, 0o750) + for filename in filenames: + target = Path(directory) / filename + os.chown(target, cyrus.pw_uid, mail_gid) + os.chmod(target, 0o640) + print(f"prepared Cyrus fixture in {root}") + + +def read_pid(root: Path) -> int | None: + try: + pid = int(paths(root)["pid"].read_text(encoding="ascii").strip()) + os.kill(pid, 0) + return pid + except (FileNotFoundError, ProcessLookupError, PermissionError, ValueError): + return None + + +def wait_for_port(port: int, timeout: float = 10.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.25): + return + except OSError: + time.sleep(0.1) + raise RuntimeError(f"Cyrus did not open 127.0.0.1:{port}") + + +def ensure_mailboxes(password: str) -> None: + client = imaplib.IMAP4("127.0.0.1", PORTS["imap"]) + try: + client.login("testadmin", password) + for username in SOURCE_USERS: + mailbox = f"user.{username}" + status, _ = client.create(mailbox) + if status not in ("OK", "NO"): + raise RuntimeError(f"could not create Cyrus mailbox {mailbox}") + client.setacl(mailbox, username, "lrswipkxtecdan") + finally: + try: + client.logout() + except OSError: + pass + + +def start(root: Path, password: str) -> None: + require_root() + if read_pid(root): + print("Cyrus fixture is already running") + return + prepare(root, password) + values = paths(root) + subprocess.run( + [BINARIES["master"], "-d", "-C", str(values["imapd"]), + "-M", str(values["cyrus"]), "-p", str(values["pid"]), + "-L", str(values["log"])], check=True) + for port in PORTS.values(): + wait_for_port(port) + ensure_mailboxes(password) + print("Cyrus fixture ready: IMAP 18143, POP3 18110, LMTP 18024") + print(f"TLS trust anchor: {values['certificate']}") + + +def stop(root: Path) -> None: + require_root() + pid = read_pid(root) + if pid is None: + print("Cyrus fixture is not running") + return + os.kill(pid, signal.SIGTERM) + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + os.kill(pid, 0) + except ProcessLookupError: + break + time.sleep(0.1) + else: + raise RuntimeError("Cyrus fixture did not stop") + print("Cyrus fixture stopped") + + +def inject(filename: str, sender: str, recipient: str) -> None: + if filename == "-": + payload = sys.stdin.buffer.read() + else: + payload = Path(filename).read_bytes() + if not payload: + message = EmailMessage() + message["From"] = sender + message["To"] = recipient + message["Subject"] = "Bongo collector fixture" + message.set_content("Bongo collector fixture message") + payload = message.as_bytes() + with smtplib.LMTP("127.0.0.1", PORTS["lmtp"], timeout=5) as client: + refused = client.sendmail(sender, [recipient], payload) + if refused: + raise RuntimeError(f"LMTP refused recipient: {refused}") + print(f"injected {len(payload)} bytes for {recipient}") + + +def seed(count: int) -> None: + if count < 2 or count > 100: + raise ValueError("seed count must be between 2 and 100") + for username in SOURCE_USERS: + for number in range(1, count + 1): + recipient = f"{username}@{FIXTURE_DOMAIN}" + message = EmailMessage() + message["From"] = "fixture-sender@bongo.test" + message["To"] = recipient + message["Subject"] = f"Bongo {username} collector test {number}" + message["Message-ID"] = ( + f"") + message["X-Bongo-Collector-Test"] = f"{username}-{number}" + message.set_content( + f"Cyrus collector fixture message {number} for {username}.\n") + with smtplib.LMTP("127.0.0.1", PORTS["lmtp"], timeout=5) as client: + refused = client.send_message( + message, from_addr=message["From"], to_addrs=[recipient]) + if refused: + raise RuntimeError(f"LMTP refused recipient: {refused}") + print(f"seeded {count} messages in each Cyrus source mailbox") + + +def verify_collector(root: Path, probe: str, password: str, count: int) -> None: + executable = Path(probe).resolve() + if not executable.is_file() or not os.access(executable, os.X_OK): + raise ValueError(f"Collector transport probe is not executable: {executable}") + seed(count) + environment = os.environ.copy() + environment["BONGO_COLLECTOR_CAINFO"] = str(paths(root)["certificate"]) + cases = ( + ("pop3", PORTS["pop3"], "pop-source"), + ("imap", PORTS["imap"], "imap-source"), + ) + for protocol, port, username in cases: + marker = f"X-Bongo-Collector-Test: {username}-" + subprocess.run( + [str(executable), protocol, str(port), username, password, + marker, str(count)], + env=environment, check=True) + print(f"verified Collector {protocol.upper()} fetch and deletion " + f"for {count} messages") + + +def status(root: Path) -> None: + pid = read_pid(root) + print(f"pid: {pid if pid else 'stopped'}") + for name, port in PORTS.items(): + try: + with socket.create_connection(("127.0.0.1", port), timeout=0.25): + state = "ready" + except OSError: + state = "closed" + print(f"{name}: 127.0.0.1:{port} {state}") + + +def reset(root: Path) -> None: + require_root() + stop(root) + if root.exists(): + shutil.rmtree(root) + print(f"removed {root}") + + +def parser() -> argparse.ArgumentParser: + result = argparse.ArgumentParser(description=__doc__) + result.add_argument("--root", type=checked_root, default=DEFAULT_ROOT) + result.add_argument("--password", default=DEFAULT_PASSWORD) + commands = result.add_subparsers(dest="command", required=True) + for name in ("prepare", "start", "stop", "status", "reset"): + commands.add_parser(name) + delivery = commands.add_parser("inject") + delivery.add_argument("file", help="RFC 5322 message file or - for stdin") + delivery.add_argument("--sender", default="sender@bongo.test") + delivery.add_argument( + "--recipient", default=f"pop-source@{FIXTURE_DOMAIN}") + seeding = commands.add_parser("seed") + seeding.add_argument("--count", type=int, default=3) + verification = commands.add_parser("verify-collector") + verification.add_argument("--probe", required=True) + verification.add_argument("--count", type=int, default=3) + return result + + +def main() -> int: + args = parser().parse_args() + if args.command == "prepare": + prepare(args.root, args.password) + elif args.command == "start": + start(args.root, args.password) + elif args.command == "stop": + stop(args.root) + elif args.command == "status": + status(args.root) + elif args.command == "reset": + reset(args.root) + elif args.command == "seed": + seed(args.count) + elif args.command == "verify-collector": + verify_collector(args.root, args.probe, args.password, args.count) + else: + inject(args.file, args.sender, args.recipient) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/release-testing-0.7.md b/docs/release-testing-0.7.md index 48f94b7..6f4fb1c 100644 --- a/docs/release-testing-0.7.md +++ b/docs/release-testing-0.7.md @@ -249,8 +249,8 @@ inputs and outputs is absent from the ZIP. | DAV-02 | CardDAV discovery, CRUD, sync, ETag, and vCard | | | | | DAV-03 | DAV master/app-password scope and TLS/proxy policy | | | | | EXT-01 | Provider presets and manual configuration validation | | | | -| EXT-02 | POP3 collection keep/delete and INBOX/custom target | | | | -| EXT-03 | IMAP collection keep/delete and INBOX/custom target | | | | +| EXT-02 | Multiple messages delivered by LMTP to `pop-source@cyrus.bongo.test`, POP3 collection into `test1`, successful-store-before-delete, empty Cyrus source | | | | +| EXT-03 | Multiple messages delivered by LMTP to `imap-source@cyrus.bongo.test`, IMAP collection into a second Bongo user, successful-store-before-delete, empty Cyrus source | | | | | EXT-04 | Multiple accounts and identities stay isolated per user | | | | | EXT-05 | Matching external SMTP identity from Web and SMTP client | | | | | EXT-06 | Collector TLS/failure/retry/deduplication/secret-safe logs | | | | @@ -282,12 +282,28 @@ inputs and outputs is absent from the ZIP. - temporary GnuTLS certificates and a temporary OpenDKIM signing key; - ClamAV, SpamAssassin, EICAR, and deterministic ham/spam mail; - Xeams DevNull SMTP for captured outbound relay messages; -- a status-aware LMTP daemon which records envelope/message and can return - per-recipient 2xx, 4xx, and 5xx responses; -- vtestmail for external POP3/IMAP collection tests; +- the loopback-only Cyrus fixture in `contrib/testing`, providing real IMAP, + POP3, and LMTP delivery for external collection and forwarding tests; +- a status-aware LMTP fault fixture which records envelope/message and can + return per-recipient 2xx, 4xx, and 5xx responses; - disposable LDAP/AD-compatible and ACME/DNS challenge fixtures; - old and current client probes, including a basic legacy IMAP4 path. DevNull and LMTP captures are checked for exact envelope, headers, body, DKIM signature, SRS address, client metadata, and duplicate or missing deliveries. Merely receiving a TCP connection is not a pass. + +### Web test delivery safety + +The automated HTTP-level web suite replaces the queue submission function with +an in-process fake, so it cannot send mail outside the test process. Live web +mail tests may use only recipients in the configured local test domain. Before +using any non-local recipient, configure Bongo's relay host to the Xeams DevNull +SMTP listener and verify that direct Internet delivery is disabled. The DevNull +capture must then be inspected for the envelope recipients (including Bcc) and +for absence of a `Bcc` message header. + +The web release matrix covers session login/logout, optional TOTP, mail folder +listing/opening/submission, contact and calendar create/list/update/delete, +filters, tasks, external accounts, sender identities and signatures, server +administration, and CalDAV/CardDAV CRUD and discovery. diff --git a/src/agents/collector/CMakeLists.txt b/src/agents/collector/CMakeLists.txt index fdcb884..ae5f3b2 100644 --- a/src/agents/collector/CMakeLists.txt +++ b/src/agents/collector/CMakeLists.txt @@ -30,6 +30,13 @@ if(BUILD_TESTING) target_link_libraries(collector-transport-test CURL::libcurl) add_test(NAME collector-external-transport COMMAND collector-transport-test) + add_executable(collector-remote-transport-probe + tests/vtestmail-transport-test.c + external_transport.c) + target_include_directories(collector-remote-transport-probe PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}) + target_link_libraries(collector-remote-transport-probe CURL::libcurl) + set(BONGO_VTESTMAIL_JAR "" CACHE FILEPATH "Path to vtestmail-core 1.0.0 for the optional Collector integration test") if(BONGO_VTESTMAIL_JAR) @@ -71,14 +78,7 @@ if(BUILD_TESTING) add_custom_target(collector-vtestmail-fixture-classes DEPENDS "${BONGO_VTESTMAIL_FIXTURE_CLASS}") - add_executable(collector-vtestmail-transport-test - tests/vtestmail-transport-test.c - external_transport.c) - target_include_directories(collector-vtestmail-transport-test PRIVATE - ${CMAKE_CURRENT_SOURCE_DIR}) - target_link_libraries(collector-vtestmail-transport-test - CURL::libcurl) - add_dependencies(collector-vtestmail-transport-test + add_dependencies(collector-remote-transport-probe collector-vtestmail-fixture-classes) add_test(NAME collector-vtestmail-integration @@ -86,7 +86,7 @@ if(BUILD_TESTING) "${CMAKE_CURRENT_SOURCE_DIR}/tests/vtestmail-integration.py" --java "${Java_JAVA_EXECUTABLE}" --classpath "${BONGO_VTESTMAIL_CLASSES}:${BONGO_VTESTMAIL_JAR}" - --probe "$") + --probe "$") set_tests_properties(collector-vtestmail-integration PROPERTIES LABELS "collector;integration" TIMEOUT 60) diff --git a/src/agents/collector/tests/vtestmail-transport-test.c b/src/agents/collector/tests/vtestmail-transport-test.c index 0cb91a5..09afc19 100644 --- a/src/agents/collector/tests/vtestmail-transport-test.c +++ b/src/agents/collector/tests/vtestmail-transport-test.c @@ -76,14 +76,27 @@ main(int argc, char **argv) FILE *message = NULL; size_t message_size = 0; int port; + long expected_count = 1; + size_t index; int result = 1; - if (argc != 6 || (strcmp(argv[1], "pop3") != 0 && + if ((argc != 6 && argc != 7) || (strcmp(argv[1], "pop3") != 0 && strcmp(argv[1], "imap") != 0)) { - fprintf(stderr, "usage: %s pop3|imap PORT USER PASSWORD MARKER\n", + fprintf(stderr, + "usage: %s pop3|imap PORT USER PASSWORD MARKER [COUNT]\n", argv[0]); return 2; } + if (argc == 7) { + char *end = NULL; + errno = 0; + expected_count = strtol(argv[6], &end, 10); + if (errno != 0 || end == argv[6] || *end != '\0' || + expected_count < 1 || expected_count > 100) { + fprintf(stderr, "invalid expected message count\n"); + return 2; + } + } port = ParsePort(argv[2]); if (port < 0 || CopyField(account.inbound_protocol, @@ -96,7 +109,7 @@ main(int argc, char **argv) sizeof(account.inbound_username), argv[3]) != 0 || CopyField(account.inbound_mailbox, sizeof(account.inbound_mailbox), "INBOX") != 0 || - snprintf(expected, sizeof(expected), "X-Bongo-Vtestmail: %s", argv[5]) >= + snprintf(expected, sizeof(expected), "%s", argv[5]) >= (int) sizeof(expected)) { fprintf(stderr, "invalid Collector probe arguments\n"); return 2; @@ -112,30 +125,39 @@ main(int argc, char **argv) fprintf(stderr, "%s list failed: %s\n", argv[1], error); goto done; } - if (ids.count != 1) { - fprintf(stderr, "%s list returned %zu messages instead of one\n", - argv[1], ids.count); + if (ids.count != (size_t) expected_count) { + fprintf(stderr, "%s list returned %zu messages instead of %ld\n", + argv[1], ids.count, expected_count); goto done; } - message = tmpfile(); - if (message == NULL) { - perror("tmpfile"); - goto done; + for (index = 0; index < ids.count; index++) { + message = tmpfile(); + if (message == NULL) { + perror("tmpfile"); + goto done; + } + if (CollectorRemoteFetch(&account, argv[4], &ids.items[index], message, + &message_size, error, sizeof(error)) != 0) { + fprintf(stderr, "%s fetch %zu failed: %s\n", + argv[1], index + 1, error); + goto done; + } + if (!MessageContains(message, message_size, expected)) { + fprintf(stderr, + "%s fetched message %zu does not contain fixture marker\n", + argv[1], index + 1); + goto done; + } + fclose(message); + message = NULL; } - if (CollectorRemoteFetch(&account, argv[4], &ids.items[0], message, - &message_size, error, sizeof(error)) != 0) { - fprintf(stderr, "%s fetch failed: %s\n", argv[1], error); - goto done; - } - if (!MessageContains(message, message_size, expected)) { - fprintf(stderr, "%s fetched message does not contain fixture marker\n", - argv[1]); - goto done; - } - if (CollectorRemoteDelete(&account, argv[4], &ids.items[0], - error, sizeof(error)) != 0) { - fprintf(stderr, "%s deletion failed: %s\n", argv[1], error); - goto done; + for (index = 0; index < ids.count; index++) { + if (CollectorRemoteDelete(&account, argv[4], &ids.items[index], + error, sizeof(error)) != 0) { + fprintf(stderr, "%s deletion %zu failed: %s\n", + argv[1], index + 1, error); + goto done; + } } if (CollectorRemoteList(&account, argv[4], &after_delete, error, sizeof(error)) != 0) { diff --git a/src/www/CMakeLists.txt b/src/www/CMakeLists.txt index caf740e..7b611e2 100644 --- a/src/www/CMakeLists.txt +++ b/src/www/CMakeLists.txt @@ -39,6 +39,8 @@ if(BUILD_TESTING) "PYTHONPATH=${CMAKE_CURRENT_SOURCE_DIR}:${CMAKE_SOURCE_DIR}/src/libs/python" "PYTHONPYCACHEPREFIX=${CMAKE_CURRENT_BINARY_DIR}/pycache" "PYTHONWARNINGS=error::ResourceWarning" + "BONGO_WEB_ENTRYPOINT=${CMAKE_CURRENT_BINARY_DIR}/bongo-web" + "BONGO_PYTHON_EXTENSION_DIR=$" ${Python3_EXECUTABLE} -m unittest discover -s ${CMAKE_CURRENT_SOURCE_DIR}/tests -v) endif() diff --git a/src/www/tests/test_web_api.py b/src/www/tests/test_web_api.py new file mode 100644 index 0000000..33209a4 --- /dev/null +++ b/src/www/tests/test_web_api.py @@ -0,0 +1,586 @@ +# /**************************************************************************** +# * +# * Copyright (c) 2001 Novell, Inc. All Rights Reserved. +# * +# * This program is free software; you can redistribute it and/or +# * modify it under the terms of version 2 of the GNU General Public License +# * as published by the Free Software Foundation. +# * +# * This program is distributed in the hope that it will be useful, +# * but WITHOUT ANY WARRANTY; without even the implied warranty of +# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# * GNU General Public License for more details. +# * +# * You should have received a copy of the GNU General Public License +# * along with this program; if not, contact Novell, Inc. +# * +# * To contact Novell about this file by physical or electronic mail, you +# * may find current contact information at www.novell.com. +# * +# ****************************************************************************/ + +"""HTTP-level coverage for every function exposed by the Bongo web UI. + +The protocol/storage helpers have their own focused unit tests. These tests +exercise the actual HTTP router, authentication boundary, CSRF checks and the +complete CRUD request flow while replacing external services with deterministic +in-process fakes. In particular, the send test can never deliver to the +Internet. +""" + +from __future__ import annotations + +import http.client +import importlib.machinery +import importlib.util +import json +import os +from pathlib import Path +import re +import tempfile +import threading +import unittest +from unittest import mock + +import libbongo + + +extension_dir = os.environ.get("BONGO_PYTHON_EXTENSION_DIR") +if extension_dir and extension_dir not in libbongo.__path__: + libbongo.__path__.append(extension_dir) + + +def _load_entrypoint(): + filename = os.environ.get("BONGO_WEB_ENTRYPOINT") + if not filename: + filename = str(Path(__file__).parents[1] / "bongo-web.py.in") + loader = importlib.machinery.SourceFileLoader("bongo_web_entrypoint_test", filename) + spec = importlib.util.spec_from_loader(loader.name, loader) + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + return module + + +web = _load_entrypoint() + + +class FakeSessions: + def __init__(self): + self.values = { + "valid": {"user": "admin", "store_cookie": "store-cookie", + "csrf_token": "csrf-token"}, + "user": {"user": "user", "store_cookie": "user-cookie", + "csrf_token": "user-csrf"}, + } + + def get(self, token): + return self.values.get(token) + + def create(self, user, store_cookie): + token = "created-session" + self.values[token] = {"user": user, "store_cookie": store_cookie, + "csrf_token": "created-csrf"} + return token, "created-csrf" + + def delete(self, token): + self.values.pop(token, None) + + +class FakeLimiter: + window = 300 + + def allowed(self, _key): + return True + + def failed(self, _key): + pass + + def succeeded(self, _key): + pass + + +class FakeSecrets: + def __init__(self): + self.values = {} + self.serial = 0 + + def create(self, value, source): + self.serial += 1 + token = f"secret-{self.serial}" + self.values[token] = (value, source) + return token + + def get(self, token, source): + value = self.values.get(token) + return value[0] if value and value[1] == source else None + + def pop(self, token, source): + value = self.get(token, source) + if value is not None: + self.values.pop(token, None) + return value + + def failed(self, token, source): + return self.pop(token, source) + + +class FakeTasks: + def __init__(self): + self.values = [{"id": 1, "title": "DNS", "state": "open"}] + self.states = [] + + def list(self, owner, include_closed=False): + return [{**item, "owner": owner, "include_closed": include_closed} + for item in self.values] + + def upsert(self, *_args, **_kwargs): + return 1 + + def set_state(self, owner, task_id, state): + if int(task_id) != 1: + raise FileNotFoundError(task_id) + self.states.append((owner, int(task_id), state)) + + +class FakeIdentities: + def __init__(self): + self.signatures = [] + self.preferences = [] + + def list_identities(self, _owner, available): + return [{**item, "display_name": "", "signature_id": None, + "default": index == 0} + for index, item in enumerate(available)] + + def list_signatures(self, _owner): + return list(self.signatures) + + def signature_for_sender(self, _owner, _sender): + return {"display_name": "", "signature": None} + + def save_signature(self, _owner, data): + signature_id = int(data.get("id") or len(self.signatures) + 1) + self.signatures = [{"id": signature_id, "name": data["name"]}] + return signature_id + + def delete_signature(self, _owner, signature_id): + if signature_id != 1: + raise FileNotFoundError(signature_id) + self.signatures = [] + + def save_preference(self, owner, data, available): + if data["address"] not in available: + raise PermissionError(data["address"]) + self.preferences.append((owner, data)) + + +class WebApiTests(unittest.TestCase): + API_ROUTES = { + ("GET", "/api/session"), ("POST", "/api/session"), + ("DELETE", "/api/session"), ("GET", "/api/i18n"), + ("GET", "/api/auth/security"), + ("POST", "/api/auth/totp/start"), + ("POST", "/api/auth/totp/enable"), + ("DELETE", "/api/auth/totp"), + ("GET", "/api/admin/status"), ("POST", "/api/admin/action"), + ("GET", "/api/mail/folders"), + ("GET", "/api/mail/conversations"), + ("GET", "/api/mail/conversation"), + ("POST", "/api/mail/send"), + ("GET", "/api/calendar/events"), + ("POST", "/api/calendar/events"), + ("DELETE", "/api/calendar/events"), + ("GET", "/api/contacts"), ("POST", "/api/contacts"), + ("DELETE", "/api/contacts"), + ("GET", "/api/filters"), ("POST", "/api/filters"), + ("DELETE", "/api/filters"), + ("GET", "/api/tasks"), ("POST", "/api/tasks/state"), + ("GET", "/api/external-accounts/providers"), + ("GET", "/api/external-accounts"), + ("POST", "/api/external-accounts"), + ("DELETE", "/api/external-accounts"), + ("GET", "/api/identities"), + ("POST", "/api/identities/preferences"), + ("POST", "/api/signatures"), ("DELETE", "/api/signatures"), + } + + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.addCleanup(self.tempdir.cleanup) + config = { + "login_domains": ["bongo.test"], + "administrators": ["admin"], + "admin_command": "/usr/bin/bongo-admin", + "public_base_url": "https://mail.bongo.test", + "external_https": False, + "maximum_request_size": 1024 * 1024, + } + self.server = web.BongoWebServer( + ("127.0.0.1", 0), web.BongoWebHandler, config, initialize=False) + self.server.sessions = FakeSessions() + self.server.login_limiter = FakeLimiter() + self.server.pending_logins = FakeSecrets() + self.server.pending_totp = FakeSecrets() + self.server.tasks = FakeTasks() + self.server.identities = FakeIdentities() + self.server.providers = {"providers": [{"id": "manual"}]} + self.thread = threading.Thread(target=self.server.serve_forever, + daemon=True) + self.thread.start() + self.addCleanup(self._stop_server) + self.cookie = f"{web.cookie_name(False)}=valid" + self.user_cookie = f"{web.cookie_name(False)}=user" + + def _stop_server(self): + self.server.shutdown() + self.thread.join(timeout=2) + self.server.server_close() + + def request(self, method, path, data=None, *, authenticated=True, + administrator=True, csrf=True, headers=None, raw=None): + request_headers = dict(headers or {}) + if authenticated: + request_headers["Cookie"] = (self.cookie if administrator + else self.user_cookie) + if csrf: + request_headers["X-Bongo-CSRF"] = ("csrf-token" if administrator + else "user-csrf") + body = raw + if data is not None: + body = json.dumps(data).encode("utf-8") + request_headers.setdefault("Content-Type", "application/json") + connection = http.client.HTTPConnection(*self.server.server_address, + timeout=3) + try: + connection.request(method, path, body=body, headers=request_headers) + response = connection.getresponse() + payload = response.read() + result_headers = dict(response.getheaders()) + return response.status, result_headers, payload + finally: + connection.close() + + @staticmethod + def json(payload): + return json.loads(payload.decode("utf-8")) + + def test_public_pages_health_locales_and_security_headers(self): + old_static = web.STATIC_DIR + web.STATIC_DIR = str(Path(__file__).parents[1] / "static") + self.addCleanup(setattr, web, "STATIC_DIR", old_static) + for path in ("/", "/mail", "/admin", "/settings"): + status, headers, payload = self.request("GET", path, + authenticated=False) + self.assertEqual(status, 200) + self.assertIn(b"", payload.lower()) + self.assertEqual(headers["X-Frame-Options"], "DENY") + status, _, payload = self.request("GET", "/health", authenticated=False) + self.assertEqual((status, self.json(payload)), (200, {"status": "ok"})) + self.assertEqual(self.request("HEAD", "/health", authenticated=False)[0], 200) + status, _, payload = self.request("GET", "/api/i18n?lang=de", + authenticated=False) + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["language"], "de") + self.assertEqual(self.request("GET", "/missing", authenticated=False)[0], 404) + + def test_ui_api_manifest_requires_a_tested_route(self): + app = (Path(__file__).parents[1] / "static" / "app.js").read_text( + encoding="utf-8") + ui_paths = set(re.findall(r'["`](/api/[a-z0-9?=/_${}-]+)', app, + flags=re.IGNORECASE)) + normalized = {path.split("?", 1)[0].split("${", 1)[0].rstrip("/") + for path in ui_paths} + covered = {path for _method, path in self.API_ROUTES} + self.assertFalse(normalized - covered, + f"Web UI routes without HTTP tests: {normalized - covered}") + + def test_session_login_read_logout_and_access_control(self): + self.assertEqual(self.request("GET", "/api/session", + authenticated=False)[0], 401) + with mock.patch.object(web, "policy", return_value={"totp_enabled": False}), \ + mock.patch.object(web, "authenticate", return_value="login-cookie"), \ + mock.patch.object(web, "resolve_user_alias", return_value="admin"): + status, headers, payload = self.request( + "POST", "/api/session", + {"user": "admin@bongo.test", "password": "secret"}, + authenticated=False) + self.assertEqual(status, 201) + self.assertIn(web.cookie_name(False), headers["Set-Cookie"]) + self.assertEqual(self.json(payload)["user"], "admin") + self.assertEqual(self.request("GET", "/api/session")[0], 200) + self.assertEqual(self.request("DELETE", "/api/session", csrf=False)[0], 403) + with mock.patch.object(web, "revoke") as revoke: + self.assertEqual(self.request("DELETE", "/api/session")[0], 204) + revoke.assert_called_once_with("admin", "store-cookie") + + def test_authentication_policy_and_complete_totp_lifecycle(self): + with mock.patch.object(web, "policy", return_value={"totp_enabled": False}): + status, _, payload = self.request("GET", "/api/auth/security") + self.assertEqual(status, 200) + self.assertFalse(self.json(payload)["totp_enabled"]) + + enrollment = {"secret": "TOTPSECRET", "svg": ""} + with mock.patch.object(web, "authenticate", return_value="proof"), \ + mock.patch.object(web, "revoke"), \ + mock.patch.object(web, "policy", return_value={"totp_enabled": False}), \ + mock.patch.object(web, "new_totp_enrollment", return_value=enrollment): + status, _, payload = self.request( + "POST", "/api/auth/totp/start", {"password": "secret"}) + self.assertEqual(status, 201) + token = self.json(payload)["enrollment_token"] + with mock.patch.object(web, "enable_totp", return_value=["recovery"]): + status, _, payload = self.request( + "POST", "/api/auth/totp/enable", + {"enrollment_token": token, "confirmation": "123456"}) + self.assertEqual(status, 201) + self.assertEqual(self.json(payload)["recovery_codes"], ["recovery"]) + with mock.patch.object(web, "authenticate", return_value="proof"), \ + mock.patch.object(web, "verify_second_factor", + return_value=web.SECOND_FACTOR_VALID), \ + mock.patch.object(web, "disable_totp") as disable, \ + mock.patch.object(web, "revoke"): + status, _, _ = self.request( + "DELETE", "/api/auth/totp", + {"password": "secret", "second_factor": "123456"}) + self.assertEqual(status, 204) + disable.assert_called_once_with("admin") + + def test_mail_folder_list_open_and_local_only_send(self): + with mock.patch.object(web, "list_folders", + return_value=[{"path": "/mail/INBOX"}]): + status, _, payload = self.request("GET", "/api/mail/folders") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["folders"][0]["path"], "/mail/INBOX") + with mock.patch.object(web, "list_conversations", return_value={ + "folder": "/mail/INBOX", "conversations": [{"id": "1"}]}): + status, _, payload = self.request( + "GET", "/api/mail/conversations?folder=%2Fmail%2FINBOX") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["conversations"][0]["id"], "1") + with mock.patch.object(web, "read_conversation", return_value={ + "id": "1", "messages": [{"subject": "Test"}]}): + status, _, payload = self.request("GET", "/api/mail/conversation?id=1") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["messages"][0]["subject"], "Test") + + message = {"from": "admin@bongo.test", "to": "admin@bongo.test", + "cc": "test1@bongo.test", "bcc": "test2@bongo.test", + "subject": "Web test", "text": "local delivery only"} + with mock.patch.object(web, "list_external_identities", return_value=[]), \ + mock.patch.object(web, "send_message") as send: + status, _, _ = self.request("POST", "/api/mail/send", message) + self.assertEqual(status, 204) + send.assert_called_once() + args, kwargs = send.call_args + self.assertEqual(args, ("admin", message)) + self.assertEqual(kwargs["sender"], "admin@bongo.test") + self.assertEqual(kwargs["authenticated_address"], "admin@bongo.test") + self.assertEqual(self.request("POST", "/api/mail/send", message, + csrf=False)[0], 403) + + def test_contact_create_list_update_delete(self): + with mock.patch.object(web, "list_cards", return_value=[b"card"]), \ + mock.patch.object(web, "card_to_dict", + return_value={"resource": "person.vcf", + "name": "Person"}): + status, _, payload = self.request("GET", "/api/contacts") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["contacts"][0]["name"], "Person") + + contact = {"name": "Person", "email": ["person@bongo.test"]} + with mock.patch.object(web, "make_card", return_value=b"VCARD"), \ + mock.patch.object(web, "write_card", return_value=(True, '"one"')): + status, _, payload = self.request("POST", "/api/contacts", contact) + self.assertEqual(status, 201) + created = self.json(payload) + self.assertRegex(created["resource"], r"^[0-9a-f]{32}\.vcf$") + + updated = {**contact, "resource": created["resource"], "etag": '"one"', + "phone": ["+43 1 234"]} + with mock.patch.object(web, "read_card", return_value=(b"OLD", '"one"')), \ + mock.patch.object(web, "make_card", return_value=b"UPDATED"), \ + mock.patch.object(web, "write_card", return_value=(False, '"two"')) as write: + status, _, payload = self.request("POST", "/api/contacts", updated) + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["etag"], '"two"') + self.assertEqual(write.call_args.kwargs["if_match"], '"one"') + + with mock.patch.object(web, "delete_card") as delete: + status, _, _ = self.request( + "DELETE", "/api/contacts?resource=" + created["resource"] + + "&etag=%22two%22") + self.assertEqual(status, 204) + delete.assert_called_once_with("admin", "store-cookie", created["resource"], + '"two"') + + def test_calendar_create_list_update_delete(self): + with mock.patch.object(web, "list_events", return_value=[b"event"]), \ + mock.patch.object(web, "event_to_dict", + return_value={"resource": "event.ics", + "title": "Meeting"}): + status, _, payload = self.request("GET", "/api/calendar/events") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["events"][0]["title"], "Meeting") + + event = {"title": "Meeting", "start": "2026-07-22T08:00:00Z", + "end": "2026-07-22T09:00:00Z"} + with mock.patch.object(web, "make_event", return_value=b"ICAL"), \ + mock.patch.object(web, "write_event", return_value=(True, '"one"')): + status, _, payload = self.request("POST", "/api/calendar/events", event) + self.assertEqual(status, 201) + created = self.json(payload) + self.assertRegex(created["resource"], r"^[0-9a-f]{32}\.ics$") + + updated = {**event, "resource": created["resource"], "etag": '"one"', + "location": "Lab"} + with mock.patch.object(web, "read_event", return_value=(b"OLD", '"one"')), \ + mock.patch.object(web, "make_event", return_value=b"UPDATED"), \ + mock.patch.object(web, "write_event", return_value=(False, '"two"')) as write: + status, _, payload = self.request("POST", "/api/calendar/events", updated) + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["etag"], '"two"') + self.assertEqual(write.call_args.kwargs["if_match"], '"one"') + + with mock.patch.object(web, "delete_event") as delete: + status, _, _ = self.request( + "DELETE", "/api/calendar/events?resource=" + created["resource"] + + "&etag=%22two%22") + self.assertEqual(status, 204) + delete.assert_called_once_with("admin", "store-cookie", created["resource"], + '"two"') + + def test_filter_create_list_update_delete(self): + with mock.patch.object(web, "list_filters", return_value=[{"id": "rule-1"}]): + status, _, payload = self.request("GET", "/api/filters") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["filters"][0]["id"], "rule-1") + rule = {"name": "Family", "condition": "from_contains", + "value": "family", "action": "move", "folder": "/mail/Family"} + with mock.patch.object(web, "save_filter", return_value={"id": "rule-1"}) as save: + self.assertEqual(self.request("POST", "/api/filters", rule)[0], 201) + self.assertEqual(self.request("POST", "/api/filters", + {**rule, "id": "rule-1"})[0], 201) + self.assertEqual(save.call_count, 2) + with mock.patch.object(web, "delete_filter") as delete: + self.assertEqual(self.request("DELETE", "/api/filters?id=rule-1")[0], 204) + delete.assert_called_once_with("admin", "rule-1") + + def test_task_list_and_state_change(self): + with mock.patch.object(web, "legacy_login_clients", return_value=[]): + status, _, payload = self.request("GET", "/api/tasks?include_closed=1") + self.assertEqual(status, 200) + self.assertTrue(self.json(payload)["tasks"][0]["include_closed"]) + self.assertEqual(self.request("POST", "/api/tasks/state", + {"id": 1, "state": "done"})[0], 204) + self.assertEqual(self.server.tasks.states, [("admin", 1, "done")]) + + def test_external_account_create_list_delete_and_provider_presets(self): + self.assertEqual(self.request("GET", "/api/external-accounts/providers")[0], + 200) + with mock.patch.object(web, "list_external_accounts", + return_value=[{"id": 7, "provider": "manual"}]): + status, _, payload = self.request("GET", "/api/external-accounts") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["accounts"][0]["id"], 7) + account = {"provider": "manual", "email_address": "user@example.test", + "incoming_protocol": "imap", "incoming_host": "imap.example.test", + "incoming_username": "user", "incoming_password": "secret"} + with mock.patch.object(web, "create_external_account", return_value=7) as create: + self.assertEqual(self.request("POST", "/api/external-accounts", + account)[0], 201) + create.assert_called_once() + with mock.patch.object(web, "delete_external_account") as delete: + self.assertEqual(self.request("DELETE", + "/api/external-accounts?id=7")[0], 204) + delete.assert_called_once_with("admin", 7) + + def test_identity_signature_and_preference_crud(self): + with mock.patch.object(web, "list_external_identities", return_value=[]): + status, _, payload = self.request("GET", "/api/identities") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["identities"][0]["address"], + "admin@bongo.test") + signature = {"name": "Default", "text": "Regards"} + self.assertEqual(self.request("POST", "/api/signatures", signature)[0], 201) + with mock.patch.object(web, "list_external_identities", return_value=[]): + self.assertEqual(self.request( + "POST", "/api/identities/preferences", + {"address": "admin@bongo.test", "display_name": "Admin", + "signature_id": 1})[0], 204) + self.assertEqual(self.request("DELETE", "/api/signatures?id=1")[0], 204) + + def test_administration_status_action_and_role_boundary(self): + snapshot = {"users": [{"name": "admin"}], "agents": [], + "server": {"manager": "running"}, "ldap": [], "errors": []} + with mock.patch.object(web, "admin_snapshot", return_value=snapshot): + status, _, payload = self.request("GET", "/api/admin/status") + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["server"]["manager"], "running") + self.assertEqual(self.request("GET", "/api/admin/status", + administrator=False)[0], 403) + with mock.patch.object(web, "perform_admin", return_value="checked") as perform: + status, _, payload = self.request( + "POST", "/api/admin/action", {"action": "config_check"}) + self.assertEqual(status, 200) + self.assertEqual(self.json(payload)["message"], "checked") + perform.assert_called_once() + + def test_caldav_and_carddav_http_crud_discovery_and_reports(self): + self.assertEqual(self.request("GET", "/.well-known/caldav", + authenticated=False)[0], 307) + self.assertEqual(self.request("GET", "/.well-known/carddav", + authenticated=False)[0], 307) + self.assertEqual(self.request("OPTIONS", "/calendars/admin/")[0], 200) + with mock.patch.object(web, "parse_calendar_path", + return_value={"kind": "event", "event": "event.ics"}), \ + mock.patch.object(web, "write_event", return_value=(True, '"e1"')): + status, _, _ = self.request( + "PUT", "/calendars/admin/event.ics", raw=b"BEGIN:VCALENDAR\r\nEND:VCALENDAR\r\n", + headers={"Content-Type": "text/calendar"}) + self.assertEqual(status, 201) + with mock.patch.object(web, "parse_calendar_path", + return_value={"kind": "event", "event": "event.ics"}), \ + mock.patch.object(web, "read_event", return_value=(b"ICAL", '"e1"')): + self.assertEqual(self.request("GET", "/calendars/admin/event.ics")[0], 200) + with mock.patch.object(web, "parse_calendar_path", + return_value={"kind": "event", "event": "event.ics"}), \ + mock.patch.object(web, "delete_event"): + self.assertEqual(self.request("DELETE", + "/calendars/admin/event.ics")[0], 204) + + with mock.patch.object(web, "parse_card_path", + return_value={"kind": "card", "card": "person.vcf"}), \ + mock.patch.object(web, "write_card", return_value=(True, '"c1"')): + status, _, _ = self.request( + "PUT", "/addressbooks/admin/person.vcf", + raw=b"BEGIN:VCARD\r\nFN:Person\r\nEND:VCARD\r\n", + headers={"Content-Type": "text/vcard"}) + self.assertEqual(status, 201) + with mock.patch.object(web, "parse_card_path", + return_value={"kind": "card", "card": "person.vcf"}), \ + mock.patch.object(web, "read_card", return_value=(b"VCARD", '"c1"')): + self.assertEqual(self.request("GET", "/addressbooks/admin/person.vcf")[0], 200) + with mock.patch.object(web, "parse_card_path", + return_value={"kind": "card", "card": "person.vcf"}), \ + mock.patch.object(web, "delete_card"): + self.assertEqual(self.request("DELETE", + "/addressbooks/admin/person.vcf")[0], 204) + + with mock.patch.object(web, "requested_properties", return_value=[]), \ + mock.patch.object(web, "multistatus", return_value=b""): + self.assertEqual(self.request("PROPFIND", "/calendars/admin/", + raw=b"")[0], 207) + calendar_report = {"kind": "query", "start": None, "end": None, + "hrefs": [], "properties": []} + with mock.patch.object(web, "parse_calendar_path", + return_value={"kind": "calendar"}), \ + mock.patch.object(web, "parse_report", return_value=calendar_report), \ + mock.patch.object(web, "list_events", return_value=[]), \ + mock.patch.object(web, "event_multistatus", + return_value=b""): + self.assertEqual(self.request("REPORT", "/calendars/admin/", + raw=b"")[0], 207) + + +if __name__ == "__main__": + unittest.main()