This commit is contained in:
@@ -72,7 +72,11 @@ class AcmeDnsConfigurationTest(unittest.TestCase):
|
||||
"bongo.configuration.acme_dns._bongo_ids",
|
||||
return_value=(os.getuid(), os.getgid())):
|
||||
result = provision_acme_dns(paths, scenario)
|
||||
first = paths.acme_provider_config.read_bytes()
|
||||
repeated = provision_acme_dns(paths, scenario)
|
||||
self.assertEqual(result, paths.acme_provider_config)
|
||||
self.assertEqual(repeated, paths.acme_provider_config)
|
||||
self.assertEqual(paths.acme_provider_config.read_bytes(), first)
|
||||
document = json.loads(paths.acme_provider_config.read_text())
|
||||
self.assertEqual(document["providers"][0]["type"], "nsupdate")
|
||||
self.assertEqual(paths.acme_provider_config.stat().st_mode & 0o777,
|
||||
|
||||
@@ -84,8 +84,12 @@ class MailSecurityMaterialTest(unittest.TestCase):
|
||||
"bongo.configuration.security_material._bongo_ids",
|
||||
return_value=(os.getuid(), os.getgid())), mock.patch(
|
||||
"bongo.configuration.security_material._generate_private_key",
|
||||
return_value=(b"PRIVATE KEY\n", "PUBLIC")) as generate, \
|
||||
mock.patch(
|
||||
"bongo.configuration.security_material._private_key_data",
|
||||
return_value=(b"PRIVATE KEY\n", "PUBLIC")):
|
||||
results = provision_security_material(paths, scenario)
|
||||
repeated = provision_security_material(paths, scenario)
|
||||
private_key = paths.dkim_dir / "example.test" / "bongo.private"
|
||||
self.assertEqual(private_key.read_bytes(), b"PRIVATE KEY\n")
|
||||
self.assertEqual(private_key.stat().st_mode & 0o777, 0o640)
|
||||
@@ -97,6 +101,9 @@ class MailSecurityMaterialTest(unittest.TestCase):
|
||||
"v=spf1 ip4:192.0.2.5 -all")
|
||||
self.assertTrue(any(action.get("reason", "").startswith("pct")
|
||||
for action in results[0].actions))
|
||||
self.assertEqual(repeated[0].suggested, results[0].suggested)
|
||||
self.assertEqual(repeated[0].actions, results[0].actions)
|
||||
generate.assert_called_once_with("example.test", "bongo")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -23,6 +23,7 @@ from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
@@ -162,6 +163,52 @@ class ConfigurationStorageTest(unittest.TestCase):
|
||||
"mail.xn--bcher-kva.example")
|
||||
self.assertEqual(queue["domains"], ["xn--bcher-kva.example"])
|
||||
|
||||
def test_install_is_repeatable_and_preserves_existing_aliases(self):
|
||||
configurations = copy.deepcopy(self.store.live)
|
||||
invocations = []
|
||||
|
||||
def completed(command, *, pass_fds, check):
|
||||
self.assertFalse(check)
|
||||
password_fd, bundle_fd = pass_fds
|
||||
os.lseek(password_fd, 0, os.SEEK_SET)
|
||||
os.lseek(bundle_fd, 0, os.SEEK_SET)
|
||||
password = os.read(password_fd, 4096)
|
||||
bundle = json.loads(os.read(bundle_fd, 1024 * 1024))
|
||||
invocations.append((list(command), password, bundle))
|
||||
return subprocess.CompletedProcess(command, 0)
|
||||
|
||||
with mock.patch("subprocess.run", side_effect=completed):
|
||||
self.store.install(
|
||||
configurations, "mail.example.test", "192.0.2.10",
|
||||
["example.test"], "0123456789ab", ["mail.example.test"])
|
||||
first_documents = {
|
||||
name: self.paths.document(name).read_bytes()
|
||||
for name in STORE_DOCUMENTS
|
||||
}
|
||||
aliases = json.loads(
|
||||
(self.paths.aliases_dir / "default_config").read_text())
|
||||
aliases["aliases"]["abuse"] = "security"
|
||||
(self.paths.aliases_dir / "default_config").write_bytes(
|
||||
json_bytes(aliases))
|
||||
|
||||
self.store.install(
|
||||
configurations, "mail.example.test", "192.0.2.10",
|
||||
["example.test"], "0123456789ab", ["mail.example.test"])
|
||||
|
||||
self.assertEqual(len(invocations), 2)
|
||||
self.assertEqual(invocations[0], invocations[1])
|
||||
self.assertEqual(invocations[0][1], b"0123456789ab")
|
||||
self.assertEqual(
|
||||
set(invocations[0][2]), STORE_DOCUMENTS)
|
||||
self.assertEqual(first_documents, {
|
||||
name: self.paths.document(name).read_bytes()
|
||||
for name in STORE_DOCUMENTS
|
||||
})
|
||||
aliases = json.loads(
|
||||
(self.paths.aliases_dir / "default_config").read_text())
|
||||
self.assertEqual(aliases["aliases"]["abuse"], "security")
|
||||
self.assertEqual(aliases["aliases"]["dmarc-reports"], "admin")
|
||||
|
||||
def test_worker_defaults_append_new_registered_jobs(self):
|
||||
configurations = copy.deepcopy(self.store.live)
|
||||
configurations["worker"]["jobs"] = [
|
||||
|
||||
Reference in New Issue
Block a user