#  Copyright (C) 2025 The Android Open Source Project
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

# Lint as: python3
"""Test cases for Telecom."""

import datetime
import logging
import time

from mobly import asserts
from mobly import base_test
from mobly import records
from mobly import test_runner
from mobly import utils as mobly_utils
from mobly.controllers import android_device
from mobly.controllers.android_device_lib import adb as adb_lib

import telecom_utils

_TELECOM_SNIPPET_PKG = 'com.google.snippet.telecom'

_IN_CALL_ACTIVITY = 'com.google.snippet.telecom/.InCallActivity'

_IN_CALL_DURATION = datetime.timedelta(seconds=10)
_WAIT_CALL_RINGING_TIMEOUT = datetime.timedelta(seconds=10)
_WAIT_SCREEN_LOCK_TIMEOUT = datetime.timedelta(seconds=3)


class CtsTelecomTest(base_test.BaseTestClass):

  ad: android_device.AndroidDevice
  ref: android_device.AndroidDevice

  def _setup_android_device(self, ad: android_device.AndroidDevice) -> None:
    ad.load_snippet('tele', _TELECOM_SNIPPET_PKG)
    ad.phone_number = ad.tele.telephonyGetPhoneNumber()
    asserts.abort_all_if(
        not ad.phone_number,
        f'{ad} Failed to get phone number, please insert a valid SIM',
    )

  def setup_class(self) -> None:
    self.ad, self.ref, *_ = self.register_controller(
        android_device, min_number=2
    )
    mobly_utils.concurrent_exec(
        self._setup_android_device,
        [[self.ad], [self.ref]],
        raise_on_exception=True,
    )

  def teardown_test(self) -> None:
    mobly_utils.concurrent_exec(
        lambda d: d.services.create_output_excerpts_all(self.current_test_info),
        [[self.ad], [self.ref]],
        raise_on_exception=True,
    )

  def on_fail(self, record: records.TestResultRecord) -> None:
    logging.info('Taking bug reports, this may take a while...')
    android_device.take_bug_reports(
        [self.ad, self.ref], destination=self.current_test_info.output_path
    )

  def _get_current_activity(self, ad: android_device.AndroidDevice) -> str:
    """Gets the current activity, empty string if screen is off."""
    try:
      return (
          ad.adb.shell([
              'dumpsys',
              'activity',
              'activities',
              '|',
              'grep',
              'topResumedActivity',
          ])
          .decode()
          .strip()
      )
    except adb_lib.AdbError as e:
      if not e.stdout and not e.stderr:
        return ''
      raise

  def test_incall_ui_shows_when_screen_locked(self) -> None:
    """Tests the InCallService visually appears above the lock screen."""
    mobly_utils.concurrent_exec(
        telecom_utils.set_default_dialer,
        [[self.ad, _TELECOM_SNIPPET_PKG], [self.ref, _TELECOM_SNIPPET_PKG]],
        raise_on_exception=True,
    )

    pin_code = '0000'
    self.ad.log.info(f'Set screen lock to PIN: {pin_code}')
    self.ad.adb.shell(['locksettings', 'set-pin', pin_code])

    try:
      self.ad.log.info('Turn off the screen')
      self.ad.adb.shell('input keyevent KEYCODE_SLEEP')
      telecom_utils.wait_until_or_assert(
          lambda: not self.ad.tele.powerIsInteractive()
          and self.ad.tele.keyguardIsDeviceLocked(),
          error_msg=f'{self.ad} Failed to lock the screen',
          timeout=_WAIT_SCREEN_LOCK_TIMEOUT,
      )
      self.ad.log.info('Screen is locked')

      # Reference phone makes a call to the test phone.
      telecom_utils.originate_call(self.ref, self.ad.phone_number)

      telecom_utils.wait_until_or_assert(
          lambda: _IN_CALL_ACTIVITY in self._get_current_activity(self.ad),
          error_msg=(
              f'{self.ad} Failed to show in-call UI when device is locked'
          ),
          timeout=_WAIT_CALL_RINGING_TIMEOUT,
      )
      self.ad.log.info('In-call UI is shown when screen is locked')

      telecom_utils.terminate_call(self.ad, self.ref.phone_number)
      telecom_utils.monitor_call(
          self.ad, self.ref.phone_number, _IN_CALL_DURATION
      )
      telecom_utils.end_call(self.ref, self.ad.phone_number)
    finally:
      self.ad.log.info('Clear screen lock')
      self.ad.adb.shell(['locksettings', 'clear', '--old', pin_code])
      self.ad.log.info('Turn on the screen')
      self.ad.adb.shell('input keyevent KEYCODE_WAKEUP')
      mobly_utils.concurrent_exec(
          lambda d: d.tele.telecomEndAllCalls(),
          [[self.ad], [self.ref]],
          raise_on_exception=True,
      )
      mobly_utils.concurrent_exec(
          telecom_utils.reset_default_dialer,
          [[self.ad], [self.ref]],
          raise_on_exception=True,
      )

  def test_call_audio_for_voice_and_voip_calls(self) -> None:
    """Tests call audio for voice and VoIP calls."""
    try:
      telecom_utils.set_default_dialer(self.ref, _TELECOM_SNIPPET_PKG)

      self.ad.log.info('Registering self-managed phone account')
      self.ad.tele.registerSelfManagedAccount()
      asserts.assert_true(
          self.ad.tele.isSelfManagedTestAccountEnabled(),
          f'{self.ad} Self-managed phone account is not registered',
      )
      self.ad.log.info('Self-managed phone account is registered')

      asserts.assert_true(
          self.ad.tele.isDefaultOutgoingAccountSimBased(),
          f'{self.ad} SIM based phone account is not default outgoing account',
      )
      self.ad.log.info('SIM based phone account is default outgoing account')

      self.ad.log.info('Placing a self-managed call')
      self.ad.tele.addNewIncomingCall()
      self.ad.log.info('Self-managed call is placed')

      self.ad.log.info('Confirming the incoming call')
      self.ad.tele.confirmIncomingCall()
      self.ad.log.info('Incoming call is confirmed')

      time.sleep(3)  # TODO: Verify audio for VoIP call.

      # Use default dialer to make a call.
      self.ad.log.info(f'Originating call to {self.ref.phone_number}')
      self.ad.tele.telecomPlaceCall(
          self.ref.phone_number,
          False,  # startWithSpeakerphone
          0,  # videoState
          False,  # shouldDefaultDialerSet
      )
      self.ad.log.info(f'Call with {self.ref.phone_number} is placed')
      telecom_utils.terminate_call(self.ref, self.ad.phone_number)
      time.sleep(3)  # TODO: Verify audio for voice call.
      telecom_utils.end_call(self.ref, self.ad.phone_number)

      self.ad.log.info('Unholding the self-managed call')
      self.ad.tele.unholdSelfManagedCall()
      self.ad.log.info('Self-managed call is unheld')

      time.sleep(3)  # TODO: Verify audio for VoIP call.

      self.ad.log.info('Disconnecting the self-managed call')
      self.ad.tele.disconnectSelfManagedCall()
      self.ad.log.info('Self-managed call is disconnected')

      time.sleep(3)  # TODO: Verify VoIP call is ended and no audio.
    finally:
      self.ad.tele.unregisterPhoneAccounts()
      telecom_utils.reset_default_dialer(self.ref)


if __name__ == '__main__':
  test_runner.main()

