diff --git a/.gitignore b/.gitignore index b1bf1a0..1a89fed 100644 --- a/.gitignore +++ b/.gitignore @@ -59,3 +59,6 @@ docs/_build/ target/ data/ +/.idea/ +/.venv/ +/out/ diff --git a/bot.py b/bot.py index 6184e1d..2314569 100644 --- a/bot.py +++ b/bot.py @@ -1,11 +1,15 @@ from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler from functools import partial from config import BotConfig -from models import Gates, Users +from models import Users +from services import GatesService, AVConnectService +from repository import GatesRepository from handlers import * bot_config = BotConfig("lagomareGateKeeperBot") -gates = Gates() +gates_repository = GatesRepository() +avconnect_service = AVConnectService() +gates_service = GatesService(gates_repository, avconnect_service) users = Users() def main(): @@ -13,11 +17,11 @@ def main(): app.add_handler(MessageHandler(None, partial(updateuser, users=users)), group=1) app.add_handler(CommandHandler("start", partial(start, users=users))) app.add_handler(CommandHandler("setcredentials", partial(setcredentials, users=users))) - app.add_handler(CommandHandler("opengate", partial(opengate, users=users, gates=gates))) + app.add_handler(CommandHandler("opengate", partial(opengate, users=users, gates=gates_service))) app.add_handler(CommandHandler("requestaccess", partial(requestaccess, users=users))) app.add_handler(CommandHandler("grantaccess", partial(grantaccess, users=users))) - app.add_handler(CallbackQueryHandler(partial(handle_main_menu_callback, users=users, gates=gates), pattern="^(open_gate_menu|request_access)$")) - app.add_handler(CallbackQueryHandler(partial(handle_gate_open_callback, users=users, gates=gates), pattern="^opengate_")) + app.add_handler(CallbackQueryHandler(partial(handle_main_menu_callback, users=users, gates=gates_service), pattern="^(open_gate_menu|request_access)$")) + app.add_handler(CallbackQueryHandler(partial(handle_gate_open_callback, users=users, gates=gates_service), pattern="^opengate_")) app.run_polling() if __name__ == "__main__": diff --git a/services/__init__.py b/services/__init__.py deleted file mode 100644 index 583106e..0000000 --- a/services/__init__.py +++ /dev/null @@ -1,5 +0,0 @@ -from .avconnect import AVConnectAPI - -__all__ = [ - "AVConnectAPI" -] \ No newline at end of file diff --git a/services/avconnect.py b/services/avconnect.py deleted file mode 100644 index cae25f7..0000000 --- a/services/avconnect.py +++ /dev/null @@ -1,55 +0,0 @@ -import requests -from fake_useragent import UserAgent -from models import Credential - -class AVConnectAPI: - _BASE_URL = "https://www.avconnect.it" - - def __init__(self, credentials: Credential): - self._ua = UserAgent(browsers=["Chrome Mobile"], os=["Android"], platforms=["mobile"]).random - self._credentials = credentials - self._session = requests.Session() - self._authenticated = False - - if credentials.sessionid: - self._session.cookies.set("PHPSESSID", credentials.sessionid) - self._authenticated = True - - def _authenticate(self) -> bool: - login_url = f"{self._BASE_URL}/loginone.php" - headers = { - "User-Agent": self._ua, - "Content-Type": "application/x-www-form-urlencoded" - } - payload = f"userid={self._credentials.username}&password={self._credentials.password}&entra=Login" - response = self._session.post(login_url, data=payload, headers=headers) - if response.ok and "PHPSESSID" in self._session.cookies: - self._authenticated = True - print("Authenticated") - return True - return False - - def _check_sessionid(self) -> bool: - if not self._authenticated or not self._credentials.sessionid: - return False - exec_url = f"{self._BASE_URL}/exemacrocom.php" - headers = { - "User-Agent": self._ua, - } - response = self._session.get(exec_url, headers=headers) - print(response.ok) - return response.ok - - def exec_gate_macro(self, id_macro) -> bool: - if (not self._authenticated or not self._check_sessionid()) and not self._authenticate(): - raise Exception("Authentication failed.") - exec_url = f"{self._BASE_URL}/exemacrocom.php" - headers = { - "User-Agent": self._ua, - "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8" - } - payload = f"idmacrocom={id_macro}&nome=16" - # Uncomment for real request: - # response = self._session.post(exec_url, data=payload, headers=headers) - # return response.ok - return True # For testing \ No newline at end of file diff --git a/services/gates.py b/services/gates.py deleted file mode 100644 index baebb40..0000000 --- a/services/gates.py +++ /dev/null @@ -1,14 +0,0 @@ -from models import Credential, Status - -class GatesService: - def open_gate(self, gate: str, credentials: Credential) -> bool: - if gate not in self._gates: - return False - if self._gates[gate].status == Status.DISABLED: - return False - try: - api = AVConnectAPI(credentials) - return api.exec_gate_macro(self._gates[gate].id) - except Exception as e: - print(f"Failed to open gate {gate}: {e}") - return False \ No newline at end of file diff --git a/handlers/__init__.py b/src/handlers/__init__.py similarity index 100% rename from handlers/__init__.py rename to src/handlers/__init__.py diff --git a/handlers/access.py b/src/handlers/access.py similarity index 100% rename from handlers/access.py rename to src/handlers/access.py diff --git a/handlers/credentials.py b/src/handlers/credentials.py similarity index 100% rename from handlers/credentials.py rename to src/handlers/credentials.py diff --git a/handlers/gates.py b/src/handlers/gates.py similarity index 94% rename from handlers/gates.py rename to src/handlers/gates.py index 15363f9..743a3be 100644 --- a/handlers/gates.py +++ b/src/handlers/gates.py @@ -1,8 +1,9 @@ from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import ContextTypes -from models import Gates, Users, Role +from models import Users, Role +from services import GatesService -async def opengate(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Users, gates: Gates): +async def opengate(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Users, gates_service: GatesService): assert update.effective_user is not None assert update.message is not None @@ -11,7 +12,7 @@ async def opengate(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Us if not args: return await update.message.reply_text("Usage: `/opengate `") gate = args[0] - gate_name = gates.get_name(gate) + gate_name = gates_service.get_name(gate) role = users.get_role(user_id) if role in (Role.ADMIN, Role.MEMBER): creds = users.get_credentials(user_id) @@ -34,7 +35,7 @@ async def opengate(update: Update, context: ContextTypes.DEFAULT_TYPE, users: Us else: return await update.message.reply_text("Access denied.") - if gates.open_gate(gate, creds): + if gates_service.open_gate(gate, creds): return await update.message.reply_text(f"Gate {gate_name} opened!") await update.message.reply_text(f"ERROR: Cannot open gate {gate_name}") @@ -50,7 +51,7 @@ 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_gates_id() + granted_gates = gates.get_all_enabled() # Show a list of available gates as buttons keyboard = [] for gate_id in granted_gates: diff --git a/handlers/main.py b/src/handlers/main.py similarity index 100% rename from handlers/main.py rename to src/handlers/main.py diff --git a/handlers/users.py b/src/handlers/users.py similarity index 100% rename from handlers/users.py rename to src/handlers/users.py diff --git a/models/__init__.py b/src/models/__init__.py similarity index 100% rename from models/__init__.py rename to src/models/__init__.py diff --git a/models/credential.py b/src/models/credential.py similarity index 100% rename from models/credential.py rename to src/models/credential.py diff --git a/models/gate.py b/src/models/gate.py similarity index 85% rename from models/gate.py rename to src/models/gate.py index 0a47b4e..2f8d823 100644 --- a/models/gate.py +++ b/src/models/gate.py @@ -1,7 +1,4 @@ -import json -from services import AVConnectAPI from .status import Status -from .credential import Credential class Gate: def __init__(self, id: str, name: str, status: Status = Status.ENABLED): diff --git a/models/status.py b/src/models/status.py similarity index 100% rename from models/status.py rename to src/models/status.py diff --git a/models/users.py b/src/models/users.py similarity index 100% rename from models/users.py rename to src/models/users.py diff --git a/src/repository/__init__.py b/src/repository/__init__.py new file mode 100644 index 0000000..200d2f9 --- /dev/null +++ b/src/repository/__init__.py @@ -0,0 +1,5 @@ +from .gates_repository import GatesRepository + +__all__ = [ + "GatesRepository" +] \ No newline at end of file diff --git a/repository/gates_repository.py b/src/repository/gates_repository.py similarity index 81% rename from repository/gates_repository.py rename to src/repository/gates_repository.py index 7ef2263..ea7b485 100644 --- a/repository/gates_repository.py +++ b/src/repository/gates_repository.py @@ -17,5 +17,5 @@ class GatesRepository: def get_by_key(self, key: str) -> Gate | None: return self._gates[key] if key in self._gates else None - def get_all_gates_id(self) -> list[str]: - return [gate for gate in self._gates.keys() if self._gates[gate].status == Status.ENABLED] \ No newline at end of file + def get_all_enabled(self) -> list[Gate]: + return [value for key, value in self._gates.items() if self._gates[key].status == Status.ENABLED] \ No newline at end of file diff --git a/src/services/__init__.py b/src/services/__init__.py new file mode 100644 index 0000000..b47e531 --- /dev/null +++ b/src/services/__init__.py @@ -0,0 +1,7 @@ +from .avconnect_service import AVConnectService +from .gates_service import GatesService + +__all__ = [ + "AVConnectService", + "GatesService" +] \ No newline at end of file diff --git a/src/services/avconnect_service.py b/src/services/avconnect_service.py new file mode 100644 index 0000000..dfa6212 --- /dev/null +++ b/src/services/avconnect_service.py @@ -0,0 +1,54 @@ +from fake_useragent import UserAgent +from requests import Session + +from models import Credential + + +class AVConnectService: + _BASE_URL = "https://www.avconnect.it" + _USER_AGENT = UserAgent(browsers=["Chrome Mobile"], os=["Android"], platforms=["mobile"]).random + _CONTENT_TYPE = "application/x-www-form-urlencoded; charset=UTF-8" + + def open_gate_by_id(self, gate_id: str, credentials: Credential) -> bool: + session = self._get_session(credentials) + if not self._is_sessionid_valid(session) and not self._authenticate(session, credentials): + raise Exception("Authentication failed.") + exec_url = f"{self._BASE_URL}/exemacrocom.php" + headers = { + "User-Agent": self._USER_AGENT, + "Content-Type": self._CONTENT_TYPE + } + payload = f"idmacrocom={gate_id}&nome=16" + # Uncomment for real request: + # response = self._session.post(exec_url, data=payload, headers=headers) + # return response.ok + return True # For testing + + def _authenticate(self, session: Session, credentials: Credential) -> bool: + login_url = f"{self._BASE_URL}/loginone.php" + headers = { + "User-Agent": self._USER_AGENT, + "Content-Type": self._CONTENT_TYPE + } + payload = f"userid={credentials.username}&password={credentials.password}&entra=Login" + response = session.post(login_url, data=payload, headers=headers) + if response.ok and "PHPSESSID" in session.cookies: + print("Authenticated") + return True + return False + + def _is_sessionid_valid(self, session: Session) -> bool: + exec_url = f"{self._BASE_URL}/exemacrocom.php" + headers = { + "User-Agent": self._USER_AGENT, + } + response = session.get(exec_url, headers=headers) + print(response.ok) + return response.ok + + @staticmethod + def _get_session(credentials: Credential) -> Session: + session = Session() + if credentials.sessionid: + session.cookies.set("PHPSESSID", credentials.sessionid) + return session diff --git a/src/services/gates_service.py b/src/services/gates_service.py new file mode 100644 index 0000000..a93965f --- /dev/null +++ b/src/services/gates_service.py @@ -0,0 +1,23 @@ +from models import Credential, Status +from repository import GatesRepository +from services import AVConnectService + +class GatesService: + + def __init__(self, gate_repository: GatesRepository, avconnect_service: AVConnectService): + self.gate_repository = gate_repository + self.avconnect_service = avconnect_service + + def open_gate(self, gate_key: str, credentials: Credential) -> bool: + gate = self.gate_repository.get_by_key(gate_key) + if not gate or gate.status == Status.DISABLED: + return False + try: + return self.avconnect_service.open_gate_by_id(gate.id, credentials) + except Exception as e: + print(f"Failed to open gate {gate.name}: {e}") + return False + + 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 \ No newline at end of file diff --git a/tests/repository/test_gates_repository.py b/tests/repository/test_gates_repository.py new file mode 100644 index 0000000..41a5b6c --- /dev/null +++ b/tests/repository/test_gates_repository.py @@ -0,0 +1,51 @@ +import unittest + +from models import Gate, Status +from repository import GatesRepository + + +class TestGatesRepository(unittest.TestCase): + def setUp(self): + self.mock_json_path = "../resources/mock_gates.json" + self.repo = GatesRepository(self.mock_json_path) + + def test_get_by_key_returns_gate_when_key_exists(self): + gate_key = "gateA" + expected_gate = Gate("1", "Gate A", Status.ENABLED) + + result = self.repo.get_by_key(gate_key) + + self.assertEqual(expected_gate.id, result.id) + self.assertEqual(expected_gate.name, result.name) + self.assertEqual(expected_gate.status, result.status) + + def test_get_by_key_returns_none_when_key_does_not_exist(self): + gate_key = "invalid_key" + + result = self.repo.get_by_key(gate_key) + + self.assertIsNone(result) + + def test_get_all_enabled_returns_only_enabled_gates(self): + enabled_gate_ids = ["1", "3", "5"] + + result = self.repo.get_all_enabled() + + self.assertEqual(len(enabled_gate_ids), len(result)) + self.assertEqual( + sorted(enabled_gate_ids), sorted([gate.id for gate in result]) + ) + + for gate in result: + self.assertTrue(gate.status == Status.ENABLED) + print(gate.status) + + # New Test + def test_load_gates_handles_invalid_json_file_gracefully(self): + repo = GatesRepository("./tests/invalid_mock_gates.json") + + self.assertEqual(repo._gates, {}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/resources/mock_gates.json b/tests/resources/mock_gates.json new file mode 100644 index 0000000..cf29116 --- /dev/null +++ b/tests/resources/mock_gates.json @@ -0,0 +1,24 @@ +{ + "gateA": { + "name": "Gate A", + "id": "1" + }, + "gateB": { + "name": "Gate B", + "id": "2", + "status": 0 + }, + "gateC": { + "name": "Gate C", + "id": "3" + }, + "gateD": { + "name": "Gate D", + "id": "4", + "status": 0 + }, + "gateE": { + "name": "Gate E", + "id": "5" + } +} \ No newline at end of file diff --git a/tests/services/test_gates_service.py b/tests/services/test_gates_service.py new file mode 100644 index 0000000..706c03a --- /dev/null +++ b/tests/services/test_gates_service.py @@ -0,0 +1,67 @@ +# tests/test_gates_service.py +import unittest +from unittest.mock import MagicMock + +from models import Credential, Gate, Status +from repository import GatesRepository +from services import AVConnectService, GatesService + + +class TestGatesService(unittest.TestCase): + def setUp(self): + self.gates_repo = MagicMock(spec=GatesRepository) + self.avconnect_service = MagicMock(spec=AVConnectService) + self.service = GatesService(self.gates_repo, self.avconnect_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) + self.gates_repo.get_by_key.return_value = gate + + result = self.service.open_gate(gate_key, credential) + + self.assertFalse(result) + self.gates_repo.get_by_key.assert_called_once_with(gate_key) + self.avconnect_service.open_gate_by_id.assert_not_called() + + 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) + self.gates_repo.get_by_key.return_value = gate + self.avconnect_service.open_gate_by_id.return_value = True + + result = self.service.open_gate(gate_key, credential) + + 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) + + 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) + self.gates_repo.get_by_key.return_value = gate + self.avconnect_service.open_gate_by_id.side_effect = Exception("Test Exception") + + result = self.service.open_gate(gate_key, credential) + + 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) + + def test_open_gate_returns_false_when_gate_does_not_exist(self): + gate_key = "nonexistent_gate" + credential = Credential(username="user", password="pass") + self.gates_repo.get_by_key.return_value = None + + result = self.service.open_gate(gate_key, credential) + + self.assertFalse(result) + self.gates_repo.get_by_key.assert_called_once_with(gate_key) + self.avconnect_service.open_gate_by_id.assert_not_called() + + +if __name__ == "__main__": + unittest.main()