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---
114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import urllib.request
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)))
|
|
|
|
import coreci_broker
|
|
|
|
|
|
class AcceptanceP1Tests(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.tmp = tmp.name
|
|
self.env = {
|
|
"BROKER_ENV_PATH": os.path.join(self.tmp, ".env"),
|
|
"BROKER_DB_PATH": os.path.join(self.tmp, "broker.db"),
|
|
"BROKER_AUDIT_LOG_PATH": os.path.join(self.tmp, "broker.audit.log"),
|
|
}
|
|
|
|
def _boot(self):
|
|
buffer = io.StringIO()
|
|
with contextlib.redirect_stdout(buffer):
|
|
broker = coreci_broker.boot(dict(self.env))
|
|
return broker, buffer.getvalue()
|
|
|
|
def test_ac14_coreci_broker_line_count_under_1400(self):
|
|
source_path = os.path.join(os.path.dirname(__file__), "..", "coreci_broker.py")
|
|
with open(source_path, "r", encoding="utf-8") as handle:
|
|
line_count = sum(1 for _ in handle)
|
|
self.assertLess(line_count, 1400)
|
|
|
|
def test_first_boot_prints_pass_once_second_boot_silent_with_0600_state(self):
|
|
first = io.StringIO()
|
|
with contextlib.redirect_stdout(first):
|
|
broker = coreci_broker.boot(dict(self.env))
|
|
server = coreci_broker.serve("127.0.0.1", 0, broker)
|
|
db_path = self.env["BROKER_DB_PATH"]
|
|
audit_path = self.env["BROKER_AUDIT_LOG_PATH"]
|
|
env_path = self.env["BROKER_ENV_PATH"]
|
|
self.assertTrue(os.path.isfile(db_path))
|
|
self.assertTrue(os.path.isfile(audit_path))
|
|
self.assertTrue(os.path.isfile(env_path))
|
|
self.assertEqual(0o600, os.stat(db_path).st_mode & 0o777)
|
|
self.assertEqual(0o600, os.stat(audit_path).st_mode & 0o777)
|
|
admin_pass = broker.config["BROKER_ADMIN_PASS"]
|
|
pepper = broker.config["BROKER_INSTANCE_PEPPER"]
|
|
self.assertIn(admin_pass, first.getvalue())
|
|
self.assertEqual(1, first.getvalue().count(admin_pass))
|
|
self.assertNotIn(pepper, first.getvalue())
|
|
server.shutdown()
|
|
server.server_close()
|
|
broker.audit.close()
|
|
broker.store.close()
|
|
second = io.StringIO()
|
|
with contextlib.redirect_stdout(second):
|
|
rebooted = coreci_broker.boot(dict(self.env))
|
|
self.assertEqual("", second.getvalue())
|
|
rebooted.audit.close()
|
|
rebooted.store.close()
|
|
|
|
def test_restart_preserves_allowlist_row_and_audit_history(self):
|
|
broker, _ = self._boot()
|
|
server = coreci_broker.serve("127.0.0.1", 0, broker)
|
|
user_id = "acceptance-restart-user"
|
|
broker.store.conn.execute(
|
|
"INSERT INTO user_allowlist (user_id, email, status, created_at) VALUES (?, ?, ?, ?)",
|
|
(user_id, "ceo@coreci.dev", "active", coreci_broker._utc_now_iso()),
|
|
)
|
|
broker.store.conn.commit()
|
|
broker.audit.write("add_user", user_id=user_id)
|
|
server.shutdown()
|
|
server.server_close()
|
|
broker.audit.close()
|
|
broker.store.close()
|
|
rebooted, _ = self._boot()
|
|
self.addCleanup(rebooted.audit.close)
|
|
self.addCleanup(rebooted.store.close)
|
|
row = rebooted.store.conn.execute(
|
|
"SELECT email, status FROM user_allowlist WHERE user_id = ?", (user_id,)
|
|
).fetchone()
|
|
self.assertEqual(("ceo@coreci.dev", "active"), row)
|
|
audit_path = self.env["BROKER_AUDIT_LOG_PATH"]
|
|
with open(audit_path, "r", encoding="utf-8") as handle:
|
|
lines = [line for line in handle if line.strip()]
|
|
self.assertEqual(1, len(lines))
|
|
entry = json.loads(lines[0])
|
|
self.assertEqual("add_user", entry["action"])
|
|
self.assertEqual(user_id, entry["user_id"])
|
|
self.assertEqual(0o600, os.stat(audit_path).st_mode & 0o777)
|
|
|
|
def test_healthz_end_to_end(self):
|
|
broker, _ = self._boot()
|
|
server = coreci_broker.serve("127.0.0.1", 0, broker)
|
|
self.addCleanup(broker.audit.close)
|
|
self.addCleanup(broker.store.close)
|
|
self.addCleanup(server.server_close)
|
|
self.addCleanup(server.shutdown)
|
|
port = server.server_address[1]
|
|
with urllib.request.urlopen(
|
|
"http://127.0.0.1:%d/healthz" % port, timeout=5
|
|
) as response:
|
|
self.assertEqual(200, response.status)
|
|
payload = json.loads(response.read().decode("utf-8"))
|
|
self.assertEqual({"ok": True}, payload)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |