import base64 import glob import json import os import uuid import logging import traceback from datetime import datetime import dotenv from behave.model_core import Status from helpers.test_data.users import get_test_user from helpers.integrations.browserstack import BrowserstackHelper from helpers.appium.driver_factory import create_driver from helpers.reporting.screenshot_manager import ScreenshotManager from page_objects.android_pom import ( ArtistScreenAndroid, HomeScreenAndroid, LoginFlowAndroid, MyFavoritesScreenAndroid, SearchScreenAndroid, SoundRecordingScreenAndroid, ProductScreenAndroid, ProfileScreenAndroid, ChartsDigestScreenAndroid) from page_objects.ios_pom import ( ArtistScreenIos, HomeScreenIos, LoginFlowIos, MyFavoritesScreenIos, SearchScreenIos, SoundRecordingScreenIos, ProductScreenIos, ProfileScreenIos) dotenv.load_dotenv() use_browserstack = os.getenv('USE_BROWSERSTACK', 'False').lower() == 'true' TRUE_ENV_VALUES = ("1", "true", "yes", "on") # ----------------------------------------------------------------------------- # Helpers: App lookup and local driver setup # ----------------------------------------------------------------------------- def should_create_behave_logs(): return os.getenv("CI", "True").strip().lower() in TRUE_ENV_VALUES def get_app_extension(mobile_os): if mobile_os == "ios": return ".ipa" if use_browserstack else ".app" return ".apk" def get_app(mobile_os): app_extension = get_app_extension(mobile_os) number_of_apps = len(glob.glob(f"apps/*{app_extension}")) if number_of_apps > 1: raise TypeError(f"More than one {app_extension} found in apps") elif number_of_apps == 0: raise TypeError(f"No {app_extension} found in apps") for application in glob.glob(f"apps/*{app_extension}"): return os.path.abspath(application) def start_driver(app, context, mobile_os): with open(f"{mobile_os}_caps.json") as handle: caps_dict = json.loads(handle.read()) caps_dict["app"] = os.path.abspath(app) if os.environ.get("APPIUM_DEVICE") and os.environ.get("APPIUM_SDK"): caps_dict["deviceName"] = os.environ["APPIUM_DEVICE"] caps_dict["platformVersion"] = os.environ["APPIUM_SDK"] context.driver = create_driver( caps_dict["appium_url"], caps_dict, mobile_os) context.driver.update_settings({ "waitForIdleTimeout": 0, "waitForSelectorTimeout": 1000 }) # ----------------------------------------------------------------------------- # Screenshots & videos # ----------------------------------------------------------------------------- def setup_screenshot_location(context, test_name): filename = (context.driver.capabilities["platformName"] + "_" + test_name + "." + datetime.now().isoformat().replace(':', '.')[:19]) screenshots_dir = "./temp/screenshots" os.makedirs(screenshots_dir, exist_ok=True) screenshot_file_path = "{}/{}.png".format( screenshots_dir, filename) return screenshot_file_path def save_video(encoded_string, scenario_name): os.makedirs(os.path.dirname('temp/'), exist_ok=True) with open(f"temp/{scenario_name}_video.mp4", "wb") as fh: fh.write(base64.b64decode(encoded_string)) # ----------------------------------------------------------------------------- # Behave hooks # ----------------------------------------------------------------------------- def before_all(context, *args, **kwargs): """ Create a unique logger for this Behave execution. BrowserStack SDK restarts Behave internally, so environment variables cannot be used to separate runs. Instead, generate a UUID. """ exec_id = str(uuid.uuid4())[:8] # short unique id logger = logging.getLogger(f"behave_logger_{exec_id}") logger.setLevel(logging.DEBUG) formatter = logging.Formatter( "%(asctime)s - %(levelname)s - %(message)s" ) console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) console_handler.setLevel(logging.DEBUG) logger.addHandler(console_handler) if should_create_behave_logs(): log_filename = f"behave_exec_{exec_id}.log" file_handler = logging.FileHandler(log_filename, mode="w") file_handler.setFormatter(formatter) file_handler.setLevel(logging.DEBUG) logger.addHandler(file_handler) context.logger = logger context.logger.debug(f"=== Behave execution started: {exec_id} ===") def declare_page_objects(mobile_os, context): context.logger.debug(f"Declaring page objects for {mobile_os}") if mobile_os == 'ios': context.home_screen = HomeScreenIos(context.driver) context.login_flow = LoginFlowIos(context.driver) context.sound_recording_screen = SoundRecordingScreenIos( context.driver) context.search_screen = SearchScreenIos(context.driver, use_browserstack) context.artist_screen = ArtistScreenIos(context.driver) context.my_favorites_screen = MyFavoritesScreenIos(context.driver) context.product_screen = ProductScreenIos(context.driver) context.profile_screen = ProfileScreenIos(context.driver) elif mobile_os == 'android': context.home_screen = HomeScreenAndroid(context.driver) context.login_flow = LoginFlowAndroid(context.driver) context.sound_recording_screen = SoundRecordingScreenAndroid( context.driver) context.search_screen = SearchScreenAndroid(context.driver) context.artist_screen = ArtistScreenAndroid(context.driver) context.my_favorites_screen = MyFavoritesScreenAndroid(context.driver) context.product_screen = ProductScreenAndroid(context.driver) context.profile_screen = ProfileScreenAndroid(context.driver) context.charts_digest_screen = ( ChartsDigestScreenAndroid(context.driver)) def before_feature(context, feature, *args, **kwargs): context.logger.debug(f"[before_feature] START {feature.name}") context.feature_failed = False feature.skip_remaining = False mobile_os = os.getenv('OS') brand = os.getenv('BRAND') # Check for brand-specific app tags (e.g., @awal-app, @sme-app) app_tags = [tag for tag in feature.tags if tag.endswith('-app')] if app_tags: # Extract brand names from tags (e.g., @awal-app -> awal) required_brands = [tag.replace('-app', '') for tag in app_tags] # Check if current brand matches ANY of the required brands if brand.lower() not in [b.lower() for b in required_brands]: feature.skip( f"This feature is for {', '.join(required_brands).upper()} " f"brand(s) only. Current brand: {brand.upper()}" ) return # Determine user type based on tags user_type = 'employee' if 'employee' in feature.tags else 'client' try: context.user = get_test_user(mobile_os, brand, user_type) app = get_app(mobile_os) context.app_name = os.path.splitext(os.path.basename(app))[0].lower() context.logger.debug( f"[DEBUG] Active test context → app_name: " f"{getattr(context, 'app_name', 'NOT SET')}") if use_browserstack: context.bs_helper = \ BrowserstackHelper(os.getenv("BROWSERSTACK_USERNAME"), os.getenv("BROWSERSTACK_ACCESS_KEY"), mobile_os, app) context.bs_helper.start_browserstack_driver( context, feature.name) else: start_driver(app, context, mobile_os) declare_page_objects(mobile_os, context) except Exception as e: # Gives better logging of errors in this hook print(traceback.format_exc()) raise e def before_scenario(context, scenario, *args, **kwargs): context.logger.debug(f"[before_scenario] {scenario.name}") context.current_scenario = scenario if getattr(scenario.feature, "skip_remaining", False): scenario.skip( "Previous scenario in this feature failed — " "skipping remaining scenarios") return if "skip" in scenario.tags: scenario.skip("Marked with @skip") if "skip_if_sme" in scenario.tags: current_app_path = get_app(os.getenv("OS")) app_name = os.path.splitext( os.path.basename(current_app_path))[0].lower() if "sme" in app_name: scenario.skip("Skipped for SME app") context.screen_location = "Unknown" def after_step(context, step, *args, **kwargs): if step.status.name in ('failed', 'error'): context.logger.error(f"[after_step] FAILED: {step.name}") context.feature_failed = True scenario = context.current_scenario scenario.feature.skip_remaining = True step_name = step.name.replace(" ", "_") screenshot_file_path = setup_screenshot_location( context, step_name) if use_browserstack: # Screenshot context.driver.save_screenshot(screenshot_file_path) sm = ScreenshotManager() sm.add_screenshot_path(step.name, screenshot_file_path) # Video # video_url = context.browserstack_helper.get_video() # context.embed(video_url, "text/plain") def after_scenario(context, scenario, *args, **kwargs): if scenario.status not in (Status.failed, Status.error): return if "language" not in scenario.tags: return try: context.login_flow.open_profile_screen(context) context.profile_screen.select_default_language(context) except Exception as e: raise e def after_feature(context, feature, *args, **kwargs): context.logger.debug(f"[after_feature] END {feature.name} " f"with status={feature.status}") status = feature.status if use_browserstack: if status == Status.passed: BrowserstackHelper.mark_test(False, context) elif status == Status.failed: BrowserstackHelper.mark_test(True, context) else: BrowserstackHelper.mark_test(True, context) context.driver.quit()