Document parity plan and add engine unit tests

This commit is contained in:
2025-10-22 14:22:51 -06:00
parent fcaf072d44
commit 95c4c6e0ff
9 changed files with 292 additions and 18 deletions

View File

@@ -0,0 +1 @@
"""Test suite for the Borealis Engine."""

View File

@@ -0,0 +1,74 @@
import unittest
from Data.Engine.builders.device_auth import (
DeviceAuthRequestBuilder,
RefreshTokenRequestBuilder,
)
from Data.Engine.domain.device_auth import DeviceAuthErrorCode, DeviceAuthFailure
class DeviceAuthRequestBuilderTests(unittest.TestCase):
def test_build_successful_request(self) -> None:
request = (
DeviceAuthRequestBuilder()
.with_authorization("Bearer abc123")
.with_http_method("post")
.with_htu("https://example.test/api")
.with_service_context("currentUser")
.with_dpop_proof("proof")
.build()
)
self.assertEqual(request.access_token, "abc123")
self.assertEqual(request.http_method, "POST")
self.assertEqual(request.htu, "https://example.test/api")
self.assertEqual(request.service_context, "CURRENTUSER")
self.assertEqual(request.dpop_proof, "proof")
def test_missing_authorization_raises_failure(self) -> None:
builder = (
DeviceAuthRequestBuilder()
.with_http_method("GET")
.with_htu("/health")
)
with self.assertRaises(DeviceAuthFailure) as ctx:
builder.build()
self.assertEqual(ctx.exception.code, DeviceAuthErrorCode.MISSING_AUTHORIZATION)
class RefreshTokenRequestBuilderTests(unittest.TestCase):
def test_refresh_request_requires_all_fields(self) -> None:
request = (
RefreshTokenRequestBuilder()
.with_payload({"guid": "de305d54-75b4-431b-adb2-eb6b9e546014", "refresh_token": "tok"})
.with_http_method("post")
.with_htu("https://example.test/api")
.with_dpop_proof("proof")
.build()
)
self.assertEqual(request.guid.value, "DE305D54-75B4-431B-ADB2-EB6B9E546014")
self.assertEqual(request.refresh_token, "tok")
self.assertEqual(request.http_method, "POST")
self.assertEqual(request.htu, "https://example.test/api")
self.assertEqual(request.dpop_proof, "proof")
def test_refresh_request_missing_guid_raises_failure(self) -> None:
builder = (
RefreshTokenRequestBuilder()
.with_payload({"refresh_token": "tok"})
.with_http_method("POST")
.with_htu("https://example.test/api")
)
with self.assertRaises(DeviceAuthFailure) as ctx:
builder.build()
self.assertEqual(ctx.exception.code, DeviceAuthErrorCode.INVALID_CLAIMS)
self.assertIn("missing guid", ctx.exception.detail)
if __name__ == "__main__": # pragma: no cover - convenience for local runs
unittest.main()

View File

@@ -0,0 +1,59 @@
import unittest
from Data.Engine.domain.device_auth import (
DeviceAuthErrorCode,
DeviceAuthFailure,
DeviceFingerprint,
DeviceGuid,
sanitize_service_context,
)
class DeviceGuidTests(unittest.TestCase):
def test_guid_normalization_accepts_braces_and_lowercase(self) -> None:
guid = DeviceGuid("{de305d54-75b4-431b-adb2-eb6b9e546014}")
self.assertEqual(guid.value, "DE305D54-75B4-431B-ADB2-EB6B9E546014")
def test_guid_rejects_empty_string(self) -> None:
with self.assertRaises(ValueError):
DeviceGuid("")
class DeviceFingerprintTests(unittest.TestCase):
def test_fingerprint_normalization_trims_and_lowercases(self) -> None:
fingerprint = DeviceFingerprint(" AA:BB:CC ")
self.assertEqual(fingerprint.value, "aa:bb:cc")
def test_fingerprint_rejects_blank_input(self) -> None:
with self.assertRaises(ValueError):
DeviceFingerprint(" ")
class ServiceContextTests(unittest.TestCase):
def test_sanitize_service_context_returns_uppercase_only(self) -> None:
self.assertEqual(sanitize_service_context("system"), "SYSTEM")
def test_sanitize_service_context_filters_invalid_chars(self) -> None:
self.assertEqual(sanitize_service_context("sys tem!"), "SYSTEM")
def test_sanitize_service_context_returns_none_for_empty_result(self) -> None:
self.assertIsNone(sanitize_service_context("@@@"))
class DeviceAuthFailureTests(unittest.TestCase):
def test_to_dict_includes_retry_after_and_detail(self) -> None:
failure = DeviceAuthFailure(
DeviceAuthErrorCode.RATE_LIMITED,
http_status=429,
retry_after=30,
detail="too many attempts",
)
payload = failure.to_dict()
self.assertEqual(
payload,
{"error": "rate_limited", "retry_after": 30.0, "detail": "too many attempts"},
)
if __name__ == "__main__": # pragma: no cover - convenience for local runs
unittest.main()

View File

@@ -0,0 +1,32 @@
import sqlite3
import unittest
from Data.Engine.repositories.sqlite import migrations
class MigrationTests(unittest.TestCase):
def test_apply_all_creates_expected_tables(self) -> None:
conn = sqlite3.connect(":memory:")
try:
migrations.apply_all(conn)
cursor = conn.cursor()
tables = {
row[0]
for row in cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
)
}
self.assertIn("devices", tables)
self.assertIn("refresh_tokens", tables)
self.assertIn("enrollment_install_codes", tables)
self.assertIn("device_approvals", tables)
self.assertIn("scheduled_jobs", tables)
self.assertIn("scheduled_job_runs", tables)
self.assertIn("github_token", tables)
finally:
conn.close()
if __name__ == "__main__": # pragma: no cover - convenience for local runs
unittest.main()