docs: add administrator guide, update self-tests to 51 passing, update documentation and release notes for v0.3.5
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m17s
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m17s
This commit is contained in:
@@ -778,6 +778,110 @@ class TestServerComponent(unittest.TestCase):
|
||||
self.assertTrue(info["encrypt_at_rest"])
|
||||
self.assertEqual(info["flush_interval"], 4.0)
|
||||
|
||||
def test_active_licensed_features_helper(self):
|
||||
"""Tests helper function to extract active licensed capabilities from DB and certificates."""
|
||||
Server.init_db(self.test_db)
|
||||
Server.SERVER_STATE["ca_cert_pem"] = ""
|
||||
Server.SERVER_STATE["config"] = {"db_path": self.test_db, "cert_dir": "non_existent_test_certs"}
|
||||
|
||||
# 1. Unlicensed state
|
||||
features_unlicensed = Server.get_active_licensed_features(db_path=self.test_db)
|
||||
self.assertEqual(features_unlicensed, [])
|
||||
self.assertFalse(Server.is_feature_licensed("enterprise_db", db_path=self.test_db))
|
||||
|
||||
# 2. Add features via JSON string in DB
|
||||
conn = sqlite3.connect(self.test_db)
|
||||
conn.execute("UPDATE license_config SET features = ? WHERE id = 1", (json.dumps(["enterprise_db", "ai_heuristics"]),))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
features_licensed = Server.get_active_licensed_features(db_path=self.test_db)
|
||||
self.assertEqual(features_licensed, ["ai_heuristics", "enterprise_db"])
|
||||
self.assertTrue(Server.is_feature_licensed("enterprise_db", db_path=self.test_db))
|
||||
self.assertTrue(Server.is_feature_licensed("ai_heuristics", db_path=self.test_db))
|
||||
self.assertFalse(Server.is_feature_licensed("non_existent_feature", db_path=self.test_db))
|
||||
|
||||
# 3. Verify get_license_summary includes features list
|
||||
summary = Server.get_license_summary()
|
||||
self.assertIn("features", summary)
|
||||
self.assertIn("enterprise_db", summary["features"])
|
||||
|
||||
def test_webui_tab_rendering_and_feature_cards(self):
|
||||
"""Tests WebUI HTML generation for clean tab buttons and feature cards in PKI tab."""
|
||||
Server.init_db(self.test_db)
|
||||
Server.SERVER_STATE["config"] = {
|
||||
"db_path": self.test_db,
|
||||
"cert_dir": "certs",
|
||||
"tcp_port": 9443,
|
||||
"hermes_port": 8443,
|
||||
"server_name": "Test-Hub"
|
||||
}
|
||||
|
||||
# 1. Render WebUI in standard / unlicensed mode
|
||||
resp = Server.serve_webui(request=None)
|
||||
html_content = resp.body.decode("utf-8")
|
||||
|
||||
# Verify clean tab buttons without number suffixes
|
||||
self.assertIn('>💻 Fleet Forwarders</button>', html_content)
|
||||
self.assertIn('>🤖 Hermes AI & Tokens</button>', html_content)
|
||||
self.assertIn('>📊 Verified Telemetry</button>', html_content)
|
||||
self.assertIn('>📜 Sub-CA & Licensing</button>', html_content)
|
||||
self.assertIn('>💾 DB Engine</button>', html_content)
|
||||
self.assertNotIn('🟢 UNLOCKED', html_content)
|
||||
self.assertNotIn('Fleet Forwarders (', html_content)
|
||||
|
||||
# Verify Licensed Capabilities card is present
|
||||
self.assertIn('Licensed Capabilities & Feature Modules', html_content)
|
||||
self.assertIn('Standard Community Edition is running', html_content)
|
||||
|
||||
# 2. Render WebUI with enterprise_db unlocked
|
||||
conn = sqlite3.connect(self.test_db)
|
||||
conn.execute("UPDATE license_config SET features = ? WHERE id = 1", (json.dumps(["enterprise_db"]),))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
resp_ent = Server.serve_webui(request=None)
|
||||
html_ent = resp_ent.body.decode("utf-8")
|
||||
self.assertIn('enterprise_db', html_ent)
|
||||
self.assertIn('Enterprise Database & High-Throughput Engine', html_ent)
|
||||
self.assertIn('>💾 DB Engine</button>', html_ent)
|
||||
self.assertNotIn('🟢 UNLOCKED', html_ent)
|
||||
|
||||
def test_explicit_opt_in_guarantee(self):
|
||||
"""Tests that licensing does NOT automatically turn on encryption or buffering without explicit admin setting."""
|
||||
Server.init_db(self.test_db)
|
||||
priv_armored, pub_armored, fp = Server.generate_server_keypair("test-opt-in-node")
|
||||
priv_key, _ = pgpy.PGPKey.from_blob(priv_armored)
|
||||
pub_key, _ = pgpy.PGPKey.from_blob(pub_armored)
|
||||
|
||||
Server.SERVER_STATE["config"] = {
|
||||
"db_path": self.test_db,
|
||||
"cert_dir": "certs",
|
||||
"storage_engine": "sqlite",
|
||||
"cache_mode": "direct",
|
||||
"encrypt_at_rest": False,
|
||||
"private_key": priv_armored,
|
||||
"public_key": pub_armored,
|
||||
"server_fingerprint": fp
|
||||
}
|
||||
Server.SERVER_STATE["private_key_obj"] = priv_key
|
||||
Server.SERVER_STATE["public_key_obj"] = pub_key
|
||||
|
||||
# Unlock enterprise license
|
||||
conn = sqlite3.connect(self.test_db)
|
||||
conn.execute("UPDATE license_config SET features = ? WHERE id = 1", (json.dumps(["enterprise_db"]),))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# License is active, but config has encrypt_at_rest=False -> plaintext storage
|
||||
raw_msg = "Sample operational log event"
|
||||
result_msg = Server.encrypt_field_for_storage(raw_msg)
|
||||
self.assertEqual(result_msg, raw_msg)
|
||||
self.assertFalse(result_msg.startswith("-----BEGIN PGP MESSAGE-----"))
|
||||
|
||||
# Ingestion cache mode is direct
|
||||
self.assertEqual(Server.SERVER_STATE["config"]["cache_mode"], "direct")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user