vee
111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from cognitive_agent.infrastructure.can.access import (
|
|
HardwareAccessController,
|
|
HardwareAccessRequiredError,
|
|
)
|
|
|
|
|
|
CLIENT_A = "hwc-aaaaaaaa-1111"
|
|
CLIENT_B = "hwc-bbbbbbbb-2222"
|
|
|
|
|
|
class FakeClock:
|
|
def __init__(self) -> None:
|
|
self.value = 100.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.value
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
self.value += seconds
|
|
|
|
|
|
class HardwareAccessControllerTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.clock = FakeClock()
|
|
self.controller = HardwareAccessController(clock=self.clock)
|
|
|
|
def test_local_admin_can_choose_one_of_multiple_pending_terminals(self) -> None:
|
|
first = self.controller.request_access(CLIENT_A, "电脑 A", "203.0.113.10")
|
|
second = self.controller.request_access(CLIENT_B, "电脑 B", "203.0.113.11")
|
|
|
|
self.assertEqual(first["queue_position"], 1)
|
|
self.assertEqual(second["queue_position"], 2)
|
|
snapshot = self.controller.approve(CLIENT_B)
|
|
|
|
self.assertEqual(snapshot["active"]["client_id"], CLIENT_B)
|
|
self.assertEqual([item["client_id"] for item in snapshot["pending"]], [CLIENT_A])
|
|
self.assertEqual(self.controller.status(CLIENT_B)["state"], "approved")
|
|
self.assertEqual(self.controller.status(CLIENT_A)["state"], "pending")
|
|
|
|
def test_execution_refreshes_lease_and_idle_180_seconds_expires_it(self) -> None:
|
|
self.controller.request_access(CLIENT_A, "电脑 A", "203.0.113.10")
|
|
self.controller.approve(CLIENT_A)
|
|
self.clock.advance(179)
|
|
self.assertEqual(self.controller.status(CLIENT_A)["state"], "approved")
|
|
|
|
with self.controller.use(CLIENT_A):
|
|
self.clock.advance(5)
|
|
|
|
self.clock.advance(179)
|
|
self.assertEqual(self.controller.status(CLIENT_A)["state"], "approved")
|
|
self.clock.advance(1)
|
|
expired = self.controller.status(CLIENT_A)
|
|
self.assertEqual(expired["state"], "expired")
|
|
self.assertIsNone(self.controller.admin_snapshot()["active"])
|
|
with self.assertRaises(HardwareAccessRequiredError) as caught:
|
|
with self.controller.use(CLIENT_A):
|
|
pass
|
|
self.assertEqual(caught.exception.state, "expired")
|
|
|
|
def test_long_running_execution_is_not_revoked_while_it_is_active(self) -> None:
|
|
self.controller.request_access(CLIENT_A, "电脑 A", "203.0.113.10")
|
|
self.controller.approve(CLIENT_A)
|
|
|
|
with self.controller.use(CLIENT_A):
|
|
self.clock.advance(600)
|
|
status = self.controller.status(CLIENT_A)
|
|
self.assertEqual(status["state"], "approved")
|
|
self.assertTrue(status["active_execution"])
|
|
|
|
self.assertEqual(
|
|
self.controller.status(CLIENT_A)["idle_remaining_seconds"], 180
|
|
)
|
|
|
|
def test_user_release_requires_a_new_request(self) -> None:
|
|
self.controller.request_access(CLIENT_A, "电脑 A", "203.0.113.10")
|
|
self.controller.approve(CLIENT_A)
|
|
self.assertEqual(self.controller.release(CLIENT_A)["state"], "none")
|
|
with self.assertRaises(HardwareAccessRequiredError):
|
|
with self.controller.use(CLIENT_A):
|
|
pass
|
|
|
|
def test_approving_another_terminal_revokes_previous_lease(self) -> None:
|
|
self.controller.request_access(CLIENT_A, "电脑 A", "203.0.113.10")
|
|
self.controller.approve(CLIENT_A)
|
|
self.controller.request_access(CLIENT_B, "电脑 B", "203.0.113.11")
|
|
snapshot = self.controller.approve(CLIENT_B)
|
|
|
|
self.assertEqual(snapshot["active"]["client_id"], CLIENT_B)
|
|
self.assertEqual(self.controller.status(CLIENT_A)["state"], "none")
|
|
with self.assertRaises(HardwareAccessRequiredError):
|
|
with self.controller.use(CLIENT_A):
|
|
pass
|
|
|
|
def test_local_admin_can_deny_pending_and_revoke_active(self) -> None:
|
|
self.controller.request_access(CLIENT_A, "电脑 A", "203.0.113.10")
|
|
self.controller.deny(CLIENT_A)
|
|
self.assertEqual(self.controller.status(CLIENT_A)["state"], "none")
|
|
|
|
self.controller.request_access(CLIENT_A, "电脑 A", "203.0.113.10")
|
|
self.controller.approve(CLIENT_A)
|
|
snapshot = self.controller.revoke(CLIENT_A)
|
|
self.assertIsNone(snapshot["active"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|