refactor: prepare for local e2e test

app.py

@@ -28,6 +28,9 @@
 )
 
 
+logger = logging.getLogger(__name__)
+
+
 def get_bot():
     """
     All singleton classes must be initialized within this method before bot
@@ -40,25 +43,33 @@ def get_bot():
 
     scheduler = JobScheduler()
 
-    jobs_config_file_key = ConfigManager().get_jobs_config_file_key()
-    if jobs_config_file_key is None:
-        raise Exception("No jobs config file key provided")
-
     args = parser.parse_args()
+    # In local environment, always skip DB update
+    skip_db_update = args.skip_db_update or consts.IS_LOCAL
 
     bot = SysBlokBot(
         config_manager,
         signal_handler=lambda signum, frame: scheduler.stop_running(),
-        skip_db_update=args.skip_db_update,
+        skip_db_update=skip_db_update,
     )
     bot.init_handlers()
 
-    jobs_config_json = bot.app_context.drive_client.download_json(jobs_config_file_key)
-    config_jobs = ConfigManager().set_jobs_config_with_override_from_json(
-        jobs_config_json
-    )
-    if not config_jobs:
-        raise ValueError("Could not load job config, can't go on")
+    # Skip jobs config download in local environment
+    if consts.IS_LOCAL:
+        logger.info("IS_LOCAL=True, skipping jobs config download from Drive")
+        ConfigManager().set_jobs_config_with_override_from_json({})
+    else:
+        jobs_config_file_key = ConfigManager().get_jobs_config_file_key()
+        if jobs_config_file_key is None:
+            raise Exception("No jobs config file key provided")
+        jobs_config_json = bot.app_context.drive_client.download_json(
+            jobs_config_file_key
+        )
+        config_jobs = ConfigManager().set_jobs_config_with_override_from_json(
+            jobs_config_json
+        )
+        if not config_jobs:
+            raise ValueError("Could not load job config, can't go on")
 
     # Setting final logger and sending a message bot is up
     tg_sender = TelegramSender()
@@ -73,17 +84,24 @@ def get_bot():
     scheduler.run()
     scheduler.init_jobs()
 
-    start_msg = f"[{consts.APP_SOURCE}] Bot successfully started"
-    if consts.COMMIT_HASH:
-        start_msg += (
-            f', revision <a href="{consts.COMMIT_URL}">{consts.COMMIT_HASH}</a>.'
-        )
-    tg_sender.send_important_event(start_msg)
+    # Send startup notification (skip in local environment to avoid API calls)
+    if not consts.IS_LOCAL:
+        start_msg = f"[{consts.APP_SOURCE}] Bot successfully started"
+        if consts.COMMIT_HASH:
+            start_msg += (
+                f', revision <a href="{consts.COMMIT_URL}">{consts.COMMIT_HASH}</a>.'
+            )
+        tg_sender.send_important_event(start_msg)
+    else:
+        logger.info("IS_LOCAL=True, skipping startup notification")
 
     return bot
 
 
 def report_critical_error(e: BaseException):
+    if consts.IS_LOCAL:
+        logger.error(f"Critical error (not reporting in local mode): {e}")
+        return
     requests.post(
         url=f"https://api.telegram.org/bot{consts.TELEGRAM_TOKEN}/sendMessage",
         json={

src/consts.py

@@ -30,6 +30,10 @@ class AppSource(Enum):
 TELEGRAM_ERROR_CHAT_ID = os.environ.get("TELEGRAM_ERROR_CHAT_ID", -1)
 TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN", "")
 
+# Environment mode: "local" skips external API calls for testing
+ENVIRONMENT = os.environ.get("ENVIRONMENT", "production")
+IS_LOCAL = ENVIRONMENT == "local"
+
 ROOT_DIR = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
 CONFIG_PATH = os.path.join(ROOT_DIR, "config.json")
 CONFIG_OVERRIDE_PATH = os.path.join(ROOT_DIR, "config_override.json")

src/trello/trello_client.py

@@ -5,7 +5,12 @@
 
 import requests
 
-from ..consts import TrelloCustomFieldTypeAlias, TrelloCustomFieldTypes, BoardListAlias
+from ..consts import (
+    TrelloCustomFieldTypeAlias,
+    TrelloCustomFieldTypes,
+    BoardListAlias,
+    IS_LOCAL,
+)
 from ..strings import load
 from ..utils.singleton import Singleton
 from . import trello_objects as objects
@@ -252,6 +257,15 @@ def _update_from_config(self):
             "key": self.api_key,
             "token": self.token,
         }
+
+        # Skip API calls in local environment
+        if IS_LOCAL:
+            logger.info("IS_LOCAL=True, skipping Trello API calls during init")
+            self.lists_config = {}
+            self.custom_fields_type_config = {}
+            self.custom_fields_config = {}
+            return
+
         # TODO(alexeyqu): move to DB
         lists = self.get_lists()
         self.lists_config = self._fill_alias_id_map(lists, BoardListAlias)