bunny-lab/Borealis-Github-Replica
1
0
Fork 0
mirror of https://github.com/bunny-lab-io/Borealis.git synced 2025-12-15 00:35:47 -07:00
Code Issues Packages Projects Releases Wiki Activity
Files
e68b52ef5a83552696ff438444c582c716257dd3
Borealis-Github-Replica/Data/Engine/Unit_Tests/conftest.py

337 lines
9.7 KiB
Python
Raw Blame History

# ======================================================
# Data\Engine\Unit_Tests\conftest.py
# Description: Pytest fixtures that seed the Engine app, SQLite schema, TLS materials, and WebUI assets.
#
# API Endpoints (if applicable): None
# ======================================================
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
import pytest
from flask import Flask
from Data.Engine.server import create_app
_SCHEMA_DEFINITION = """CREATE TABLE IF NOT EXISTS devices (
guid TEXT PRIMARY KEY,
hostname TEXT,
description TEXT,
created_at INTEGER,
agent_hash TEXT,
memory TEXT,
network TEXT,
software TEXT,
storage TEXT,
cpu TEXT,
device_type TEXT,
domain TEXT,
external_ip TEXT,
internal_ip TEXT,
last_reboot TEXT,
last_seen INTEGER,
last_user TEXT,
operating_system TEXT,
uptime INTEGER,
agent_id TEXT,
ansible_ee_ver TEXT,
connection_type TEXT,
connection_endpoint TEXT,
ssl_key_fingerprint TEXT,
token_version INTEGER,
status TEXT,
key_added_at TEXT
);
CREATE TABLE IF NOT EXISTS refresh_tokens (
id TEXT PRIMARY KEY,
guid TEXT,
token_hash TEXT,
dpop_jkt TEXT,
created_at TEXT,
expires_at TEXT,
revoked_at TEXT,
last_used_at TEXT
);
CREATE TABLE IF NOT EXISTS enrollment_install_codes (
id TEXT PRIMARY KEY,
code TEXT UNIQUE,
expires_at TEXT,
created_by_user_id TEXT,
used_at TEXT,
used_by_guid TEXT,
max_uses INTEGER,
use_count INTEGER,
last_used_at TEXT
);
CREATE TABLE IF NOT EXISTS enrollment_install_codes_persistent (
id TEXT PRIMARY KEY,
code TEXT UNIQUE,
created_at TEXT NOT NULL,
expires_at TEXT NOT NULL,
created_by_user_id TEXT,
used_at TEXT,
used_by_guid TEXT,
max_uses INTEGER NOT NULL DEFAULT 1,
last_known_use_count INTEGER NOT NULL DEFAULT 0,
last_used_at TEXT,
is_active INTEGER NOT NULL DEFAULT 1,
archived_at TEXT,
consumed_at TEXT
);
CREATE TABLE IF NOT EXISTS device_approvals (
id TEXT PRIMARY KEY,
approval_reference TEXT UNIQUE,
guid TEXT,
hostname_claimed TEXT,
ssl_key_fingerprint_claimed TEXT,
enrollment_code_id TEXT,
status TEXT,
client_nonce TEXT,
server_nonce TEXT,
agent_pubkey_der BLOB,
created_at TEXT,
updated_at TEXT,
approved_by_user_id TEXT
);
CREATE TABLE IF NOT EXISTS device_keys (
id TEXT PRIMARY KEY,
guid TEXT,
ssl_key_fingerprint TEXT,
added_at TEXT,
retired_at TEXT
);
CREATE TABLE IF NOT EXISTS device_list_views (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT UNIQUE,
columns_json TEXT,
filters_json TEXT,
created_at INTEGER,
updated_at INTEGER
);
CREATE TABLE IF NOT EXISTS sites (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT,
description TEXT,
created_at INTEGER
);
CREATE TABLE IF NOT EXISTS device_sites (
device_hostname TEXT PRIMARY KEY,
site_id INTEGER,
assigned_at INTEGER
);
CREATE TABLE IF NOT EXISTS github_token (
id INTEGER PRIMARY KEY,
token TEXT
);
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE,
display_name TEXT,
password_sha512 TEXT,
role TEXT,
last_login INTEGER,
created_at INTEGER,
updated_at INTEGER
);
"""
@dataclass
class EngineTestHarness:
app: Flask
db_path: Path
bundle_contents: str
def _initialise_legacy_schema(db_path: Path) -> None:
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path))
try:
conn.executescript(_SCHEMA_DEFINITION)
conn.commit()
finally:
conn.close()
@pytest.fixture()
def engine_harness(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[EngineTestHarness]:
project_root = Path(__file__).resolve().parents[3]
monkeypatch.setenv("BOREALIS_PROJECT_ROOT", str(project_root))
runtime_dir = tmp_path / "runtime"
runtime_dir.mkdir()
cert_root = tmp_path / "certificates"
cert_root.mkdir()
monkeypatch.setenv("BOREALIS_SERVER_ROOT", str(runtime_dir))
monkeypatch.setenv("BOREALIS_CERT_ROOT", str(cert_root))
monkeypatch.setenv("BOREALIS_SERVER_CERT_ROOT", str(cert_root / "Server"))
monkeypatch.setenv("BOREALIS_AGENT_CERT_ROOT", str(cert_root / "Agent"))
monkeypatch.setenv("BOREALIS_ENGINE_DISABLE_LEGACY_PROXY", "1")
db_path = tmp_path / "database" / "engine.sqlite3"
_initialise_legacy_schema(db_path)
conn = sqlite3.connect(str(db_path))
try:
cur = conn.cursor()
cur.execute(
"""
INSERT INTO devices (
guid,
hostname,
description,
created_at,
agent_hash,
memory,
network,
software,
storage,
cpu,
device_type,
domain,
external_ip,
internal_ip,
last_reboot,
last_seen,
last_user,
operating_system,
uptime,
agent_id,
ansible_ee_ver,
connection_type,
connection_endpoint,
ssl_key_fingerprint,
token_version,
status,
key_added_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"GUID-TEST-0001",
"test-device",
"Test device for Engine API",
1_700_000_000,
"hash-123",
json.dumps([{"slot": "DIMM1", "size_gb": 16}]),
json.dumps([{"iface": "eth0", "mac": "00:11:22:33:44:55"}]),
json.dumps(["sample-app"]),
json.dumps([{"drive": "C", "size_gb": 256}]),
json.dumps({"name": "Intel", "cores": 8}),
"Workstation",
"example.local",
"203.0.113.5",
"10.0.0.10",
"2025-10-01T00:00:00Z",
1_700_000_500,
"Alice",
"Windows 11 Pro",
7200,
"test-device-agent",
"1.0.0",
"",
"",
"FF:FF:FF",
1,
"active",
"2025-10-01T00:00:00Z",
),
)
cur.execute(
"INSERT INTO sites (id, name, description, created_at) VALUES (?, ?, ?, ?)",
(1, "Main Lab", "Primary integration site", 1_700_000_000),
)
cur.execute(
"INSERT INTO device_sites (device_hostname, site_id, assigned_at) VALUES (?, ?, ?)",
("test-device", 1, 1_700_000_500),
)
cur.execute(
"""
INSERT INTO users (id, username, display_name, password_sha512, role, last_login, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(1, "admin", "Administrator", "test", "Admin", 0, 0, 0),
)
cur.execute(
"""
INSERT INTO device_approvals (
id,
approval_reference,
guid,
hostname_claimed,
ssl_key_fingerprint_claimed,
enrollment_code_id,
status,
client_nonce,
server_nonce,
agent_pubkey_der,
created_at,
updated_at,
approved_by_user_id
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"approval-1",
"APP-REF-1",
None,
"pending-device",
"aa:bb:cc:dd",
None,
"pending",
"client-nonce",
"server-nonce",
None,
"2025-01-01T00:00:00Z",
"2025-01-01T00:00:00Z",
None,
),
)
conn.commit()
finally:
conn.close()
tls_dir = tmp_path / "tls"
tls_dir.mkdir()
bundle_contents = "-----BEGIN CERTIFICATE-----\nengine-test\n-----END CERTIFICATE-----\n"
cert_path = tls_dir / "server-cert.pem"
key_path = tls_dir / "server-key.pem"
bundle_path = tls_dir / "server-bundle.pem"
cert_path.write_text(bundle_contents, encoding="utf-8")
key_path.write_text("test-key", encoding="utf-8")
bundle_path.write_text(bundle_contents, encoding="utf-8")
logs_dir = tmp_path / "logs"
logs_dir.mkdir(parents=True, exist_ok=True)
log_path = logs_dir / "server.log"
error_log_path = logs_dir / "error.log"
static_dir = tmp_path / "static"
static_dir.mkdir(parents=True, exist_ok=True)
(static_dir / "index.html").write_text("<html><body>Engine Test UI</body></html>", encoding="utf-8")
assets_dir = static_dir / "assets"
assets_dir.mkdir(parents=True, exist_ok=True)
(assets_dir / "example.txt").write_text("asset", encoding="utf-8")
config = {
"DATABASE_PATH": str(db_path),
"TLS_CERT_PATH": str(cert_path),
"TLS_KEY_PATH": str(key_path),
"TLS_BUNDLE_PATH": str(bundle_path),
"LOG_FILE": str(log_path),
"ERROR_LOG_FILE": str(error_log_path),
"STATIC_FOLDER": str(static_dir),
"API_GROUPS": ("core", "auth", "tokens", "enrollment", "devices"),
}
app, _socketio, _context = create_app(config)
app.config.update(TESTING=True)
yield EngineTestHarness(app=app, db_path=db_path, bundle_contents=bundle_contents)
Reference in New Issue View Git Blame Copy Permalink