ettore.dreucci/lagomareGateKeeperBot
1
0
Fork 0
mirror of https://github.com/Noettore/lagomareGateKeeperBot.git synced 2025-10-15 19:46:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

Still refactoring and unit tests

Browse Source
This commit is contained in:
Ettore Dreucci
2025-08-30 22:37:17 +00:00
parent 1d423c5ea2
commit c76a77cb0c
16 changed files with 78 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
bot.py
View File

@@ -1,16 +1,17 @@
from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler
from functools import partial
from config import BotConfig
from services import GatesService, AVConnectService, UsersService
from repository import GatesRepository, UsersRepository
from handlers import *
from src.services import GatesService, AVConnectService, UsersService
from src.repository import GatesRepository, UsersRepository
from src.handlers import *
bot_config = BotConfig("lagomareGateKeeperBot")
gates_repository = GatesRepository()
avconnect_service = AVConnectService()
gates_service = GatesService(gates_repository, avconnect_service)
users_repository = UsersRepository()
users_service = UsersService(users_repository)
avconnect_service = AVConnectService()
gates_repository = GatesRepository()
gates_service = GatesService(gates_repository, avconnect_service, users_service)
def main():
app = Application.builder().token(bot_config.token).post_init(partial(post_init, bot_config=bot_config)).build()

11
src/handlers/access.py
View File

@@ -1,9 +1,10 @@
from telegram import Update
from telegram.ext import ContextTypes
from datetime import datetime
from models import Users, Role
from src.models import Role
from src.services import UsersService
async def requestaccess(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Users):
async def requestaccess(update: Update, context: ContextTypes.DEFAULT_TYPE, users: UsersService):
assert update.effective_user is not None
user_id = str(update.effective_user.id)
@@ -13,8 +14,8 @@ async def requestaccess(update: Update, context: ContextTypes.DEFAULT_TYPE, user
await update.callback_query.answer("Only guests can request access.")
elif update.message:
return await update.message.reply_text("Only guests can request access.")
requester = users.get_fullname(user_id) or users.get_username(user_id)
text = (f"Access request: {requester} ({user_id}) requests access.\nUse `/grantaccess {user_id} <gate_id|all> YYYY-MM-DDTHH:MM:SSZ` to grant access.")
requester = update.effective_user.full_name or update.effective_user.username
text = f"Access request: {requester} ({user_id}) requests access.\nUse `/grantaccess {user_id} <gate_id|all> YYYY-MM-DDTHH:MM:SSZ` to grant access."
if update.callback_query:
await update.callback_query.answer("Your request has been submitted.")
elif update.message:
@@ -26,7 +27,7 @@ async def requestaccess(update: Update, context: ContextTypes.DEFAULT_TYPE, user
except Exception as e:
print(f"Failed to notify {admin_id} that guest {user_id} requested access: {e}")
async def grantaccess(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Users):
async def grantaccess(update: Update, context: ContextTypes.DEFAULT_TYPE, users: UsersService):
assert update.effective_user is not None
assert update.message is not None
assert context.args is not None

5
src/handlers/credentials.py
View File

@@ -1,8 +1,9 @@
from telegram import Update
from telegram.ext import ContextTypes
from models import Users, Credential, Role
from src.models import Credential, Role
from src.services import UsersService
async def setcredentials(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Users):
async def setcredentials(update: Update, context: ContextTypes.DEFAULT_TYPE, users: UsersService):
assert update.effective_user is not None
assert update.message is not None
assert context.args is not None

26
src/handlers/gates.py
View File

@@ -1,6 +1,7 @@
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import ContextTypes
from services import GatesService, UsersService
from src.services import GatesService, UsersService
from src.models import Role
async def opengate(update: Update, context: ContextTypes.DEFAULT_TYPE, gates_service: GatesService):
assert update.effective_user is not None
@@ -16,7 +17,7 @@ async def opengate(update: Update, context: ContextTypes.DEFAULT_TYPE, gates_ser
return await update.message.reply_text(f"Gate {gate_name} opened!")
await update.message.reply_text(f"ERROR: Cannot open gate {gate_name}")
async def open_gate_menu(update: Update, context: ContextTypes.DEFAULT_TYPE, users_service: UsersService, gates: Gates):
async def open_gate_menu(update: Update, context: ContextTypes.DEFAULT_TYPE, users_service: UsersService, gates_service: GatesService):
assert update.effective_user is not None
user_id = str(update.effective_user.id)
@@ -28,11 +29,11 @@ async def open_gate_menu(update: Update, context: ContextTypes.DEFAULT_TYPE, use
return await update.message.reply_text("You have no gates to open.")
if 'all' in granted_gates:
granted_gates = gates.get_all_enabled()
granted_gates = gates_service.get_all_enabled()
# Show a list of available gates as buttons
keyboard = []
for gate_id in granted_gates:
gate_name = gates.get_name(gate_id)
gate_name = gates_service.get_name(gate_id)
assert gate_name is not None
keyboard.append([InlineKeyboardButton(gate_name, callback_data=f"opengate_{gate_id}")])
@@ -45,7 +46,7 @@ async def open_gate_menu(update: Update, context: ContextTypes.DEFAULT_TYPE, use
elif update.message:
await update.message.reply_text("Select a gate to open:", reply_markup=reply_markup)
async def handle_gate_open_callback(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Users, gates: Gates):
async def handle_gate_open_callback(update: Update, context: ContextTypes.DEFAULT_TYPE, users: UsersService, gates: GatesService):
assert update.callback_query is not None
query = update.callback_query
@@ -55,19 +56,14 @@ async def handle_gate_open_callback(update: Update, context: ContextTypes.DEFAUL
if data.startswith("opengate_"):
gate_id = data[len("opengate_") :]
gate_name = gates.get_name(gate_id)
user = users.get_user(user_id)
role = users.get_role(user_id)
if role in (Role.ADMIN, Role.MEMBER):
creds = users.get_credentials(user_id)
if not creds:
await query.answer("Please set your credentials with /setcredentials first", show_alert=True)
return
effective_user = user
elif role == Role.GUEST and users.can_open_gate(user_id, gate_id):
grantor = users.get_grantor(user_id, gate_id)
grantor = users.get_grantor(user, gate_id)
assert grantor is not None
creds = users.get_credentials(grantor)
if not grantor or not creds:
await query.answer("No valid grantor credentials available.", show_alert=True)
return
effective_user = grantor
users.update_grant_last_used(user_id, gate_id)
try:
await context.bot.send_message(chat_id=grantor, text=f"Guest {user_id} opened {gate_name}")
@@ -78,7 +74,7 @@ async def handle_gate_open_callback(update: Update, context: ContextTypes.DEFAUL
await query.answer("Access denied.", show_alert=True)
return
if gates.open_gate(gate_id, creds):
if gates.open_gate(gate_id, effective_user.uid):
await query.answer(f"Gate {gate_name} opened!", show_alert=True)
await query.edit_message_text(f"Gate {gate_name} opened!")
else:

9
src/handlers/main.py
View File

@@ -1,11 +1,12 @@
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, ContextTypes
from models import Role
from src.models import Role
from .gates import open_gate_menu
from .access import requestaccess
from config import BotConfig
from services import UsersService
from src.services import UsersService, GatesService
async def post_init(application: Application, bot_config: BotConfig) -> None:
await application.bot.set_my_name(bot_config.name)
@@ -43,11 +44,11 @@ async def start(update: Update, context: ContextTypes.DEFAULT_TYPE, users_servic
reply_markup=reply_markup
)
async def handle_main_menu_callback(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Users, gates: Gates):
async def handle_main_menu_callback(update: Update, context: ContextTypes.DEFAULT_TYPE, users: UsersService, gates: GatesService):
query = update.callback_query
assert query is not None
data = query.data
if data == "open_gate_menu":
await open_gate_menu(update, context, users=users, gates=gates)
await open_gate_menu(update, context, users, gates)
elif data == "request_access":
await requestaccess(update, context, users=users)

2
src/handlers/users.py
View File

@@ -1,6 +1,6 @@
from telegram import Update
from telegram.ext import ContextTypes
from services import UsersService
from src.services import UsersService
async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE, users_service: UsersService):
assert update.effective_user is not None

8
src/models/gate.py
View File

@@ -1,13 +1,13 @@
from .status import Status
class Gate:
def __init__(self, id: str, name: str, status: Status = Status.ENABLED):
self.id = id
def __init__(self, gid: str, name: str, status: Status = Status.ENABLED):
self.gid = gid
self.name = name
self.status = status if isinstance(status, Status) else Status(status)
self.status = status
def to_dict(self):
return {"id": self.id, "name": self.name, "status": self.status.value}
return {"id": self.gid, "name": self.name, "status": self.status.value}
@classmethod
def from_dict(cls, data: dict):

2
src/models/grant.py
View File

@@ -1,6 +1,6 @@
from datetime import datetime, timezone
from models import Status
from .status import Status
class Grant:

18
src/models/user.py
View File

@@ -1,6 +1,4 @@
from enum import Enum
from . import Grant
from .grant import Grant
from .role import Role
from .status import Status
from .credential import Credential
@@ -8,21 +6,21 @@ from .credential import Credential
class User:
def __init__(
self,
id: str,
uid: str,
role: Role = Role.GUEST,
credentials: Credential | None = None,
grants: dict[str, Grant] | None = None,
status: Status = Status.ENABLED
):
self.id = id
self.role = role if isinstance(role, Role) else Role(role)
self.uid = uid
self.role = role
self.credentials = credentials or Credential("", "")
self.grants = grants or {}
self.status = status if isinstance(status, Status) else Status(status)
self.status = status
def to_dict(self) -> dict:
return {
"id": self.id,
"id": self.uid,
"role": self.role.value,
"credentials": self.credentials.to_dict(),
"grants": {gate: grant.to_dict() for gate, grant in self.grants.items()},
@@ -30,11 +28,11 @@ class User:
}
@classmethod
def from_dict(cls, id: str, data: dict):
def from_dict(cls, gid: str, data: dict):
credentials = Credential.from_dict(data.get("credentials", {}))
grants = {gate: Grant.from_dict(grant) for gate, grant in data.get("grants", {}).items()}
return cls(
id=id,
uid=gid,
role=Role(data.get("role", Role.GUEST)),
credentials=credentials,
grants=grants,

2
src/repository/gates_repository.py
View File

@@ -1,5 +1,5 @@
import json
from models import Status, Gate
from src.models import Status, Gate
class GatesRepository:
def __init__(self, json_path: str = "./data/gates.json"):

4
src/repository/users_repository.py
View File

@@ -1,5 +1,5 @@
import json
from models import User
from src.models import User
class UsersRepository:
def __init__(self, json_path: str = "./data/users.json"):
@@ -23,7 +23,7 @@ class UsersRepository:
json.dump({uid: user.to_dict() for uid, user in self._users.items()}, f, indent=2)
def delete(self, user: User) -> None:
del self._users[user.id]
del self._users[user.uid]
self._dump_users()
def get_by_uid(self, uid: str) -> User | None:

2
src/services/avconnect_service.py
View File

@@ -1,7 +1,7 @@
from fake_useragent import UserAgent
from requests import Session
from models import Credential
from src.models import Credential
class AVConnectService:

14
src/services/gates_service.py
View File

@@ -1,6 +1,7 @@
from models import Credential, Status
from repository import GatesRepository
from services import AVConnectService, UsersService
from src.models import Gate, Status
from src.repository import GatesRepository
from .avconnect_service import AVConnectService
from .users_service import UsersService
class GatesService:
@@ -15,8 +16,11 @@ class GatesService:
if not gate or gate.status == Status.DISABLED:
return False
credentials = self._user_service.get_credentials(uid, gate_key)
return self._avconnect_service.open_gate_by_id(gate.id, credentials)
return self._avconnect_service.open_gate_by_id(gate.gid, credentials)
def get_name(self, gate_key: str) -> str | None:
gate = self._gate_repository.get_by_key(gate_key)
return gate.name if gate else None
return gate.name if gate else None
def get_all_enabled(self) -> list[Gate]:
return self._gate_repository.get_all_enabled()

6
src/services/users_service.py
View File

@@ -1,7 +1,7 @@
from datetime import datetime, timezone
from models import Status, Role, Credential, Grant, User
from repository import UsersRepository
from src.models import Status, Role, Credential, Grant, User
from src.repository import UsersRepository
class UsersService:
@@ -81,7 +81,7 @@ class UsersService:
return self._users_repository.get_by_uid(grant.grantor) if grant else None
def get_admins(self) -> list[str]:
return [user.id for user in self._users_repository.get_all() if user.role == Role.ADMIN]
return [user.uid for user in self._users_repository.get_all() if user.role == Role.ADMIN]
def grant_access(self, uid: str, gate: str, expires_at: datetime, grantor_id: str) -> bool:
user = self._users_repository.get_by_uid(uid)

8
tests/repository/test_gates_repository.py
View File

@@ -1,7 +1,7 @@
import unittest
from models import Gate, Status
from repository import GatesRepository
from src.models import Gate, Status
from src.repository import GatesRepository
class TestGatesRepository(unittest.TestCase):
@@ -15,7 +15,7 @@ class TestGatesRepository(unittest.TestCase):
result = self.repo.get_by_key(gate_key)
self.assertEqual(expected_gate.id, result.id)
self.assertEqual(expected_gate.gid, result.gid)
self.assertEqual(expected_gate.name, result.name)
self.assertEqual(expected_gate.status, result.status)
@@ -33,7 +33,7 @@ class TestGatesRepository(unittest.TestCase):
self.assertEqual(len(enabled_gate_ids), len(result))
self.assertEqual(
sorted(enabled_gate_ids), sorted([gate.id for gate in result])
sorted(enabled_gate_ids), sorted([gate.gid for gate in result])
)
for gate in result:

21
tests/services/test_gates_service.py
View File

@@ -2,21 +2,24 @@
import unittest
from unittest.mock import MagicMock
from models import Credential, Gate, Status
from repository import GatesRepository
from services import AVConnectService, GatesService
from src.models import Credential, Gate, Status
from src.repository import GatesRepository, UsersRepository
from src.services import AVConnectService, GatesService, UsersService
class TestGatesService(unittest.TestCase):
def setUp(self):
self.mock_users_json_path = "../resources/mock_users.json"
self.gates_repo = MagicMock(spec=GatesRepository)
self.users_repo = UsersRepository(self.mock_users_json_path)
self.avconnect_service = MagicMock(spec=AVConnectService)
self.service = GatesService(self.gates_repo, self.avconnect_service)
self.users_service = UsersService(self.users_repo)
self.service = GatesService(self.gates_repo, self.avconnect_service, self.users_service)
def test_open_gate_fails_when_gate_is_disabled(self):
gate_key = "gate1"
credential = Credential(username="user", password="pass")
gate = Gate(id="1", name="Test Gate", status=Status.DISABLED)
gate = Gate(gid="1", name="Test Gate", status=Status.DISABLED)
self.gates_repo.get_by_key.return_value = gate
result = self.service.open_gate(gate_key, credential)
@@ -28,7 +31,7 @@ class TestGatesService(unittest.TestCase):
def test_open_gate_succeeds_with_valid_enabled_gate(self):
gate_key = "gate2"
credential = Credential(username="user", password="pass")
gate = Gate(id="2", name="Enabled Gate", status=Status.ENABLED)
gate = Gate(gid="2", name="Enabled Gate", status=Status.ENABLED)
self.gates_repo.get_by_key.return_value = gate
self.avconnect_service.open_gate_by_id.return_value = True
@@ -36,12 +39,12 @@ class TestGatesService(unittest.TestCase):
self.assertTrue(result)
self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.id, credential)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.gid, credential)
def test_open_gate_handles_exception_gracefully(self):
gate_key = "gate3"
credential = Credential(username="user", password="pass")
gate = Gate(id="3", name="Test Gate", status=Status.ENABLED)
gate = Gate(gid="3", name="Test Gate", status=Status.ENABLED)
self.gates_repo.get_by_key.return_value = gate
self.avconnect_service.open_gate_by_id.side_effect = Exception("Test Exception")
@@ -49,7 +52,7 @@ class TestGatesService(unittest.TestCase):
self.assertFalse(result)
self.gates_repo.get_by_key.assert_called_once_with(gate_key)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.id, credential)
self.avconnect_service.open_gate_by_id.assert_called_once_with(gate.gid, credential)
def test_open_gate_returns_false_when_gate_does_not_exist(self):
gate_key = "nonexistent_gate"