coreci
7f7868f7cf
---ci--- phase: 5 milestone: v0.1 status: complete requirements: covered: [REQ-01, REQ-02, REQ-03, REQ-04, REQ-05, REQ-06, REQ-07, REQ-08, REQ-09, REQ-10, REQ-11, REQ-12, REQ-13, REQ-14, REQ-15, REQ-16, REQ-17, REQ-18, REQ-19, REQ-20] partial: [] ---/ci---
105 lines
4.4 KiB
Python
105 lines
4.4 KiB
Python
import base64
|
|
import contextlib
|
|
import io
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))
|
|
|
|
import coreci_broker
|
|
|
|
|
|
class LoadConfigTests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.env_path = os.path.join(tmp.name, ".env")
|
|
|
|
def _boot(self, env):
|
|
buffer = io.StringIO()
|
|
with contextlib.redirect_stdout(buffer):
|
|
config = coreci_broker.load_config(env)
|
|
return config, buffer.getvalue()
|
|
|
|
def _read_env_file(self):
|
|
values = {}
|
|
with open(self.env_path, "r", encoding="utf-8") as handle:
|
|
for raw in handle:
|
|
line = raw.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
values[key.strip()] = value.strip()
|
|
return values
|
|
|
|
def test_first_boot_writes_env_and_prints_pass_once(self):
|
|
config, stdout = self._boot({"BROKER_ENV_PATH": self.env_path})
|
|
self.assertTrue(os.path.isfile(self.env_path))
|
|
stored = self._read_env_file()
|
|
admin_pass = stored["BROKER_ADMIN_PASS"]
|
|
pepper = stored["BROKER_INSTANCE_PEPPER"]
|
|
self.assertGreaterEqual(len(admin_pass), 32)
|
|
self.assertEqual(32, len(base64.b64decode(pepper)))
|
|
self.assertEqual(admin_pass, config["BROKER_ADMIN_PASS"])
|
|
self.assertEqual(pepper, config["BROKER_INSTANCE_PEPPER"])
|
|
self.assertEqual(1, stdout.count(admin_pass))
|
|
self.assertNotIn(pepper, stdout)
|
|
self.assertEqual(0o600, os.stat(self.env_path).st_mode & 0o777)
|
|
|
|
def test_second_boot_prints_nothing(self):
|
|
first, first_stdout = self._boot({"BROKER_ENV_PATH": self.env_path})
|
|
second, second_stdout = self._boot({"BROKER_ENV_PATH": self.env_path})
|
|
self.assertEqual(1, first_stdout.count(first["BROKER_ADMIN_PASS"]))
|
|
self.assertEqual("", second_stdout)
|
|
self.assertEqual(first["BROKER_ADMIN_PASS"], second["BROKER_ADMIN_PASS"])
|
|
self.assertEqual(first["BROKER_INSTANCE_PEPPER"], second["BROKER_INSTANCE_PEPPER"])
|
|
|
|
def test_env_overrides_dotenv(self):
|
|
with open(self.env_path, "w", encoding="utf-8") as handle:
|
|
handle.write("BROKER_BIND=10.1.2.3:9999\n")
|
|
handle.write('BROKER_ADMIN_USER="ceo"\n')
|
|
handle.write("BROKER_ADMIN_PASS=oldpass\n")
|
|
handle.write("BROKER_TOKEN_TTL=100\n")
|
|
env = {
|
|
"BROKER_ENV_PATH": self.env_path,
|
|
"BROKER_ADMIN_PASS": "envpass",
|
|
"BROKER_TOKEN_TTL": "55",
|
|
"BROKER_DB_PATH": "/tmp/elsewhere/broker.db",
|
|
}
|
|
config, stdout = self._boot(env)
|
|
self.assertEqual("envpass", config["BROKER_ADMIN_PASS"])
|
|
self.assertEqual("55", config["BROKER_TOKEN_TTL"])
|
|
self.assertEqual("/tmp/elsewhere/broker.db", config["BROKER_DB_PATH"])
|
|
self.assertEqual("10.1.2.3:9999", config["BROKER_BIND"])
|
|
self.assertEqual("ceo", config["BROKER_ADMIN_USER"])
|
|
self.assertEqual("300", config["BROKER_SESSION_TTL"])
|
|
self.assertEqual("", stdout)
|
|
stored = self._read_env_file()
|
|
self.assertEqual("oldpass", stored["BROKER_ADMIN_PASS"])
|
|
self.assertIn("BROKER_INSTANCE_PEPPER", stored)
|
|
|
|
def test_defaults_applied(self):
|
|
config, _ = self._boot({"BROKER_ENV_PATH": self.env_path})
|
|
self.assertEqual(self.env_path, config["BROKER_ENV_PATH"])
|
|
self.assertEqual("127.0.0.1:8765", config["BROKER_BIND"])
|
|
self.assertEqual("admin", config["BROKER_ADMIN_USER"])
|
|
self.assertEqual("86400", config["BROKER_TOKEN_TTL"])
|
|
self.assertEqual("300", config["BROKER_SESSION_TTL"])
|
|
self.assertEqual("broker.db", config["BROKER_DB_PATH"])
|
|
self.assertEqual("broker.audit.log", config["BROKER_AUDIT_LOG_PATH"])
|
|
|
|
def test_reads_process_environ_by_default(self):
|
|
env = {"BROKER_ENV_PATH": self.env_path, "BROKER_BIND": "127.0.0.1:9000"}
|
|
with mock.patch.dict(os.environ, env, clear=True):
|
|
buffer = io.StringIO()
|
|
with contextlib.redirect_stdout(buffer):
|
|
config = coreci_broker.load_config()
|
|
self.assertEqual("127.0.0.1:9000", config["BROKER_BIND"])
|
|
self.assertEqual(self.env_path, config["BROKER_ENV_PATH"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |