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