Fix unit tests

This commit is contained in:
2025-08-31 21:46:14 +00:00
parent c76a77cb0c
commit 71d5b378d9
3 changed files with 87 additions and 35 deletions

View File

@@ -28,11 +28,11 @@ class User:
} }
@classmethod @classmethod
def from_dict(cls, gid: str, data: dict): def from_dict(cls, uid: str, data: dict):
credentials = Credential.from_dict(data.get("credentials", {})) credentials = Credential.from_dict(data.get("credentials", {}))
grants = {gate: Grant.from_dict(grant) for gate, grant in data.get("grants", {}).items()} grants = {gate: Grant.from_dict(grant) for gate, grant in data.get("grants", {}).items()}
return cls( return cls(
uid=gid, uid=uid,
role=Role(data.get("role", Role.GUEST)), role=Role(data.get("role", Role.GUEST)),
credentials=credentials, credentials=credentials,
grants=grants, grants=grants,

View File

@@ -1,4 +1,5 @@
import unittest import unittest
from pathlib import Path
from src.models import Gate, Status from src.models import Gate, Status
from src.repository import GatesRepository from src.repository import GatesRepository
@@ -6,7 +7,7 @@ from src.repository import GatesRepository
class TestGatesRepository(unittest.TestCase): class TestGatesRepository(unittest.TestCase):
def setUp(self): def setUp(self):
self.mock_json_path = "../resources/mock_gates.json" self.mock_json_path = str(Path("./tests/resources/mock_gates.json").resolve())
self.repo = GatesRepository(self.mock_json_path) self.repo = GatesRepository(self.mock_json_path)
def test_get_by_key_returns_gate_when_key_exists(self): def test_get_by_key_returns_gate_when_key_exists(self):
@@ -38,7 +39,6 @@ class TestGatesRepository(unittest.TestCase):
for gate in result: for gate in result:
self.assertTrue(gate.status == Status.ENABLED) self.assertTrue(gate.status == Status.ENABLED)
print(gate.status)
# New Test # New Test
def test_load_gates_handles_invalid_json_file_gracefully(self): def test_load_gates_handles_invalid_json_file_gracefully(self):

View File

@@ -1,69 +1,121 @@
# tests/test_gates_service.py
import unittest import unittest
from unittest.mock import MagicMock from unittest.mock import MagicMock
from src.models import Credential, Gate, Status from src.models import Gate, Status
from src.repository import GatesRepository, UsersRepository from src.repository import GatesRepository
from src.services import AVConnectService, GatesService, UsersService from src.services import AVConnectService, GatesService, UsersService
class TestGatesService(unittest.TestCase): class TestGatesService(unittest.TestCase):
def setUp(self): def setUp(self):
self.mock_users_json_path = "../resources/mock_users.json"
self.gates_repo = MagicMock(spec=GatesRepository) self.gates_repo = MagicMock(spec=GatesRepository)
self.users_repo = UsersRepository(self.mock_users_json_path)
self.avconnect_service = MagicMock(spec=AVConnectService) self.avconnect_service = MagicMock(spec=AVConnectService)
self.users_service = UsersService(self.users_repo) self.users_service = MagicMock(spec=UsersService)
self.service = GatesService(self.gates_repo, self.avconnect_service, self.users_service) self.service = GatesService(self.gates_repo, self.avconnect_service, self.users_service)
def test_open_gate_fails_when_gate_is_disabled(self): def test_open_gate_returns_false_when_gate_does_not_exist(self):
gate_key = "gate1" gate_key = "missing_gate"
credential = Credential(username="user", password="pass") uid = "user-1"
gate = Gate(gid="1", name="Test Gate", status=Status.DISABLED) self.gates_repo.get_by_key.return_value = None
self.gates_repo.get_by_key.return_value = gate
result = self.service.open_gate(gate_key, credential) result = self.service.open_gate(gate_key, uid)
self.assertFalse(result) self.assertFalse(result)
self.gates_repo.get_by_key.assert_called_once_with(gate_key) self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.users_service.get_credentials.assert_not_called()
self.avconnect_service.open_gate_by_id.assert_not_called() self.avconnect_service.open_gate_by_id.assert_not_called()
def test_open_gate_succeeds_with_valid_enabled_gate(self): def test_open_gate_returns_false_when_gate_is_disabled(self):
gate_key = "gate2" gate_key = "gate-disabled"
credential = Credential(username="user", password="pass") uid = "user-2"
gate = Gate(gid="2", name="Enabled Gate", status=Status.ENABLED) gate = Gate(gid="G-1", name="Disabled Gate", status=Status.DISABLED)
self.gates_repo.get_by_key.return_value = gate self.gates_repo.get_by_key.return_value = gate
result = self.service.open_gate(gate_key, uid)
self.assertFalse(result)
self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.users_service.get_credentials.assert_not_called()
self.avconnect_service.open_gate_by_id.assert_not_called()
def test_open_gate_calls_services_and_returns_true_when_avconnect_succeeds(self):
gate_key = "gate-enabled"
uid = "user-3"
credentials = {"u": "name", "p": "pwd"}
gate = Gate(gid="G-2", name="Enabled Gate", status=Status.ENABLED)
self.gates_repo.get_by_key.return_value = gate
self.users_service.get_credentials.return_value = credentials
self.avconnect_service.open_gate_by_id.return_value = True self.avconnect_service.open_gate_by_id.return_value = True
result = self.service.open_gate(gate_key, credential) result = self.service.open_gate(gate_key, uid)
self.assertTrue(result) self.assertTrue(result)
self.gates_repo.get_by_key.assert_called_once_with(gate_key) self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.gid, credential) self.users_service.get_credentials.assert_called_once_with(uid, gate_key)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.gid, credentials)
def test_open_gate_handles_exception_gracefully(self): def test_open_gate_calls_services_and_returns_false_when_avconnect_fails(self):
gate_key = "gate3" gate_key = "gate-enabled"
credential = Credential(username="user", password="pass") uid = "user-4"
gate = Gate(gid="3", name="Test Gate", status=Status.ENABLED) credentials = {"u": "name", "p": "pwd"}
gate = Gate(gid="G-3", name="Enabled Gate", status=Status.ENABLED)
self.gates_repo.get_by_key.return_value = gate self.gates_repo.get_by_key.return_value = gate
self.avconnect_service.open_gate_by_id.side_effect = Exception("Test Exception") self.users_service.get_credentials.return_value = credentials
self.avconnect_service.open_gate_by_id.return_value = False
result = self.service.open_gate(gate_key, credential) result = self.service.open_gate(gate_key, uid)
self.assertFalse(result) self.assertFalse(result)
self.gates_repo.get_by_key.assert_called_once_with(gate_key) self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.gid, credential) self.users_service.get_credentials.assert_called_once_with(uid, gate_key)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.gid, credentials)
def test_open_gate_returns_false_when_gate_does_not_exist(self): def test_open_gate_propagates_exception_from_avconnect(self):
gate_key = "nonexistent_gate" gate_key = "gate-exception"
credential = Credential(username="user", password="pass") uid = "user-5"
credentials = {"u": "name", "p": "pwd"}
gate = Gate(gid="G-4", name="Enabled Gate", status=Status.ENABLED)
self.gates_repo.get_by_key.return_value = gate
self.users_service.get_credentials.return_value = credentials
self.avconnect_service.open_gate_by_id.side_effect = RuntimeError("boom")
with self.assertRaises(RuntimeError):
self.service.open_gate(gate_key, uid)
self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.users_service.get_credentials.assert_called_once_with(uid, gate_key)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.gid, credentials)
def test_get_name_returns_gate_name_when_gate_exists(self):
gate_key = "gate-name"
gate = Gate(gid="G-5", name="Main Entrance", status=Status.ENABLED)
self.gates_repo.get_by_key.return_value = gate
name = self.service.get_name(gate_key)
self.assertEqual("Main Entrance", name)
self.gates_repo.get_by_key.assert_called_once_with(gate_key)
def test_get_name_returns_none_when_gate_missing(self):
gate_key = "missing"
self.gates_repo.get_by_key.return_value = None self.gates_repo.get_by_key.return_value = None
result = self.service.open_gate(gate_key, credential) name = self.service.get_name(gate_key)
self.assertFalse(result) self.assertIsNone(name)
self.gates_repo.get_by_key.assert_called_once_with(gate_key) self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.avconnect_service.open_gate_by_id.assert_not_called()
def test_get_all_enabled_delegates_to_repository(self):
gates = [
Gate(gid="G-6", name="G1", status=Status.ENABLED),
Gate(gid="G-7", name="G2", status=Status.ENABLED),
]
self.gates_repo.get_all_enabled.return_value = gates
result = self.service.get_all_enabled()
self.assertEqual(gates, result)
self.gates_repo.get_all_enabled.assert_called_once()
if __name__ == "__main__": if __name__ == "__main__":