#!/usr/bin/python3
'''GNOME settings daemon tests for power plugin.'''

__author__ = 'Martin Pitt <martin.pitt@ubuntu.com>'
__copyright__ = '(C) 2012 Canonical Ltd.'
__license__ = 'GPL v2 or later'

import unittest
import subprocess
import sys
import time
import os
import os.path
import tempfile
import fcntl

import dbus
import dbusmock

from gi.repository import Gio


class GSDTestCase(dbusmock.DBusTestCase):
    '''Base class for settings daemon tests'''

    @classmethod
    def setUpClass(klass):
        os.environ['GIO_USE_VFS'] = 'local'

        klass.start_system_bus()
        klass.start_session_bus()
        klass.system_bus_con = klass.get_dbus(True)
        klass.session_bus_con = klass.get_dbus(False)

        klass.start_session()

        klass.settings_session = Gio.Settings('org.gnome.desktop.session')

        klass.orig_idle_delay = klass.settings_session['idle-delay']

    @classmethod
    def tearDownClass(klass):
        # restore settings
        klass.settings_session['idle-delay'] = klass.orig_idle_delay
        Gio.Settings.sync()

        klass.stop_session()

        dbusmock.DBusTestCase.tearDownClass()

    @classmethod
    def start_session(klass):
        '''Start minimal GNOME session'''

        klass.session_dir = os.path.expanduser('~/.config/gnome-session/sessions')
        if not os.path.isdir(klass.session_dir):
            os.makedirs(klass.session_dir)
        with open(os.path.join(klass.session_dir, 'null.session'), 'w') as f:
            f.write('''[GNOME Session]
Name=null
RequiredComponents=null
DesktopName=null
''')

        klass.app_dir = os.path.expanduser('~/.local/share/applications')
        if not os.path.isdir(klass.app_dir):
            os.makedirs(klass.app_dir)
        with open(os.path.join(klass.app_dir, 'null.desktop'), 'w') as f:
            f.write('''[Desktop Entry]
Name=null
Exec=sleep 3600
''')

        klass.session_log = tempfile.NamedTemporaryFile()
        klass.session = subprocess.Popen(['gnome-session', '--session=null', '--debug'],
                                         stdout=klass.session_log,
                                         stderr=subprocess.STDOUT)

        # wait until the daemon is on the bus
        try:
            klass.wait_for_bus_object('org.gnome.SessionManager',
                                      '/org/gnome/SessionManager')
        except:
            # on failure, print log
            with open(klass.session_log.name) as f:
                print('----- session log -----\n%s\n------' % f.read())
            raise

        klass.obj_session_mgr = klass.session_bus_con.get_object(
            'org.gnome.SessionManager', '/org/gnome/SessionManager')

        # give g-session some time to lazily initialize idle timers, etc.
        time.sleep(3)

    @classmethod
    def stop_session(klass):
        '''Stop GNOME session'''

        os.unlink(os.path.join(klass.session_dir, 'null.session'))
        os.unlink(os.path.join(klass.app_dir, 'null.desktop'))
        try:
            os.removedirs(klass.session_dir)
        except OSError:
            pass
        try:
            os.removedirs(klass.app_dir)
        except OSError:
            pass

        assert klass.session
        klass.session.terminate()
        klass.session.wait()

        klass.session_log.flush()
        #with open(klass.session_log.name) as f:
        #    print('----- session log -----\n%s\n------' % f.read())
        klass.session_log.close()

    def start_consolekit(self):
        '''start mock ConsoleKit'''

        self.conkit = self.spawn_server('org.freedesktop.ConsoleKit',
                                        '/org/freedesktop/ConsoleKit/Manager',
                                        'org.freedesktop.ConsoleKit.Manager',
                                        system_bus=True,
                                        stdout=subprocess.PIPE)
        self.obj_ck = self.system_bus_con.get_object(
            'org.freedesktop.ConsoleKit', '/org/freedesktop/ConsoleKit/Manager')

        self.obj_ck.AddProperty('org.freedesktop.ConsoleKit.Manager', 'idle',
                                dbus.Boolean(False, variant_level=1))

        self.obj_ck.AddMethods('',
            [
                ('GetSessions', '', 'ao', 'ret = ["/org/freedesktop/ConsoleKit/MockSession"]'),
                ('GetCurrentSession', '', 'o', 'ret = "/org/freedesktop/ConsoleKit/MockSession"'),
                ('GetSessionForUnixProcess', 'i', 'o', 'ret = "/org/freedesktop/ConsoleKit/MockSession"'),
                ('GetSeats', '', 'ao', 'ret = ["/org/freedesktop/ConsoleKit/MockSeat"]'),
                ('GetSystemIdleHint', '', 'b', 'ret = self.Get("org.freedesktop.ConsoleKit.Manager", "idle")'),
            ], dbus_interface='org.freedesktop.DBus.Mock')

        self.obj_ck.AddObject('/org/freedesktop/ConsoleKit/MockSeat',
                              'org.freedesktop.ConsoleKit.Seat',
                              {},
                              [
                                  ('GetSessions', '', 'ao',
                                   'ret = ["/org/freedesktop/ConsoleKit/MockSession"]'),
                              ], dbus_interface='org.freedesktop.DBus.Mock')

        self.obj_ck.AddObject('/org/freedesktop/ConsoleKit/MockSession',
                              'org.freedesktop.ConsoleKit.Session',
                              {},
                              [
                                  ('IsLocal', '', 'b', 'ret = True'),
                                  ('IsActive', '', 'b', 'ret = True'),
                                  ('GetSeatId', '', 'o', 'ret = "/org/freedesktop/ConsoleKit/MockSeat"'),
                                  ('SetIdleHint', 'b', '',
                                   'objects["/org/freedesktop/ConsoleKit/Manager"].Set("org.freedesktop.ConsoleKit.Manager", "idle", args[0])'),
                              ], dbus_interface='org.freedesktop.DBus.Mock')

    def stop_consolekit(self):
        '''stop mock ConsoleKit'''

        self.conkit.terminate()
        self.conkit.wait()


class PowerPluginTest(GSDTestCase):
    '''Test the power plugin'''

    def setUp(self):
        # start mock upowerd
        self.upowerd = self.spawn_server('org.freedesktop.UPower',
                                         '/org/freedesktop/UPower',
                                         'org.freedesktop.UPower',
                                         system_bus=True,
                                         stdout=subprocess.PIPE)
        # set log to nonblocking
        flags = fcntl.fcntl(self.upowerd.stdout, fcntl.F_GETFL)
        fcntl.fcntl(self.upowerd.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK)

        self.obj_upower = self.system_bus_con.get_object(
            'org.freedesktop.UPower', '/org/freedesktop/UPower')

        self.obj_upower.AddMethods('',
                                   [
                                       ('Suspend', '', '', ''),
                                       ('Hibernate', '', '', ''),
                                       ('SuspendAllowed', '', 'b', 'ret = True'),
                                       ('HibernateAllowed', '', 'b', 'ret = True'),
                                       ('EnumerateDevices', '', 'ao', 'ret = objects.keys()'),
                                   ])

        self.obj_upower.AddProperties('org.freedesktop.UPower',
                                      dbus.Dictionary({
                                          'DaemonVersion': '0.8.15',
                                          'CanSuspend': True,
                                          'CanHibernate': True,
                                          'OnBattery': True,
                                          'OnLowBattery': False,
                                          'LidIsPresent': True,
                                          'LidIsClosed': True,
                                          'LidForceSleep': True,
                                          'IsDocked': False,
                                      }, signature='sv'))

        self.start_consolekit()

        self.settings_gsd_power = Gio.Settings('org.gnome.settings-daemon.plugins.power')
        self.orig_gsd_power_settings = {}

        # start power plugin
        self.log_output = ''
        self.set_power_setting('active', False)
        self.daemon_log = tempfile.NamedTemporaryFile()
        self.daemon = subprocess.Popen(
            [os.path.join(os.path.dirname(__file__), 'gsd-test-power')],
            # comment out this line if you want to see the logs in real time
            stdout=self.daemon_log,
            stderr=subprocess.STDOUT)
        # give it some time to settle down
        time.sleep(1)

    def tearDown(self):
        # restore any configuration that tests changed
        for k, v in self.orig_gsd_power_settings.items():
            #print('tearDown(): restoring %s to %s' % (k, v))
            self.settings_gsd_power[k] = v
        Gio.Settings.sync()

        self.daemon.terminate()
        self.daemon.wait()
        self.daemon_log.flush()
        # collect log, run() shows it on failures
        with open(self.daemon_log.name) as f:
            self.log_output = f.read()
        self.daemon_log.close()

        self.upowerd.terminate()
        self.upowerd.wait()
        self.stop_consolekit()

    def run(self, result=None):
        '''Show log output on failed tests'''

        if result:
            orig_err_fail = result.errors + result.failures
        super().run(result)
        if result and result.errors + result.failures > orig_err_fail:
            print('\n----- daemon log -----\n%s\n------\n' % self.log_output)

    def set_power_setting(self, key, value):
        '''Set a power plugin gsetting.

        Changed settings will be automatically reverted in tearDown().
        '''
        if key not in self.orig_gsd_power_settings:
            self.orig_gsd_power_settings[key] = self.settings_gsd_power[key]
        self.settings_gsd_power[key] = value
        Gio.Settings.sync()

    def check_for_suspend(self, timeout):
        '''Check that Suspend() or Hibernate() is requested.

        Fail after the tiven timeout.
        '''
        # check that it request suspend
        while timeout > 0:
            time.sleep(1)
            timeout -= 1
            # check that it requested suspend
            log = self.upowerd.stdout.read()
            if log and (b' Suspend\n' in log or b' Hibernate\n' in log):
                break
        else:
            self.fail('timed out waiting for upower Suspend() call')

    def add_battery(self, name, percent, time_to_empty):
        '''Add a battery to mock upower daemon

        Return DBus proxy object for newly created battery.
        '''
        obj_path = '/org/freedesktop/UPower/devices/' + name
        self.obj_upower.AddObject(obj_path,
                                  'org.freedesktop.UPower.Device',
                                  {
                                      'PowerSupply': dbus.Boolean(True, variant_level=1),
                                      'Model': dbus.String('Mock Bat', variant_level=1),
                                      'Percentage': dbus.Double(30.0, variant_level=1),
                                      'TimeToEmpty': dbus.Int64(1200, variant_level=1),
                                      # UP_DEVICE_STATE_DISCHARGING
                                      'State': dbus.UInt32(2, variant_level=1),
                                      # UP_DEVICE_KIND_BATTERY
                                      'Type': dbus.UInt32(2, variant_level=1),
                                  },
                                  [],
                                  dbus_interface='org.freedesktop.DBus.Mock')
        self.obj_upower.EmitSignal('', 'DeviceAdded', 's', [obj_path],
                                   dbus_interface='org.freedesktop.DBus.Mock')
        return self.system_bus_con.get_object('org.freedesktop.UPower', obj_path)

    def test_sleep_inactive_battery(self):
        '''sleep-inactive-battery-timeout'''

        self.settings_session['idle-delay'] = 5
        self.set_power_setting('sleep-inactive-battery-timeout', 10)
        # would be good to set this, but doesn't work ATM:
        # https://bugzilla.gnome.org/show_bug.cgi?id=685947
        #self.set_power_setting('sleep-inactive-battery-type', 'suspend')

        # wait for idle delay + inactive delay
        time.sleep(15)
        self.check_for_suspend(10)

    def test_action_critical_battery(self):
        '''action on critical battery'''

        # add a fake battery to upower
        obj_bat = self.add_battery('mock_BAT', 30.0, 1200)
        time.sleep(1)

        # now change battery to critical charge
        obj_bat.Set('org.freedesktop.UPower.Device', 'TimeToEmpty',
                    dbus.Int64(30, variant_level=1),
                    dbus_interface=dbus.PROPERTIES_IFACE)
        obj_bat.EmitSignal('', 'Changed', '', [], dbus_interface='org.freedesktop.DBus.Mock')
        self.obj_upower.EmitSignal('', 'DeviceChanged', 's', [obj_bat.object_path],
                                   dbus_interface='org.freedesktop.DBus.Mock')

        # g-s-d waits some 20 seconds before triggering the actions to give the
        # user a chance to panic; so wait longer
        self.check_for_suspend(25)

    def test_action_multiple_batteries(self):
        '''critical actions for multiple batteries'''

        # add two fake batteries to upower
        obj_bat1 = self.add_battery('mock_BAT0', 30.0, 1200)
        obj_bat2 = self.add_battery('mock_BAT1', 40.0, 1600)
        time.sleep(1)

        # now change one battery to critical charge
        obj_bat1.Set('org.freedesktop.UPower.Device', 'TimeToEmpty',
                     dbus.Int64(30, variant_level=1),
                     dbus_interface=dbus.PROPERTIES_IFACE)
        obj_bat1.EmitSignal('', 'Changed', '', [], dbus_interface='org.freedesktop.DBus.Mock')
        self.obj_upower.EmitSignal('', 'DeviceChanged', 's', [obj_bat1.object_path],
                                   dbus_interface='org.freedesktop.DBus.Mock')

        # g-s-d waits some 20 seconds before triggering the actions to give the
        # user a chance to panic; so wait long enough to ensure it didn't do
        # anything (as we still have the second battery)
        time.sleep(25)
        # check that it did not suspend or hibernate
        log = self.upowerd.stdout.read()
        self.assertFalse(b' Suspend\n' in log, 'unexpected Suspend request')
        self.assertFalse(b' Hibernate\n' in log, 'unexpected Hibernate request')

        # now change the other battery to critical charge as well
        obj_bat2.Set('org.freedesktop.UPower.Device', 'TimeToEmpty',
                     dbus.Int64(25, variant_level=1),
                     dbus_interface=dbus.PROPERTIES_IFACE)
        obj_bat2.EmitSignal('', 'Changed', '', [], dbus_interface='org.freedesktop.DBus.Mock')
        self.obj_upower.EmitSignal('', 'DeviceChanged', 's', [obj_bat2.object_path],
                                   dbus_interface='org.freedesktop.DBus.Mock')

        # g-s-d waits some 20 seconds before triggering the actions to give the
        # user a chance to panic; so wait longer
        self.check_for_suspend(25)


# avoid writing to stderr
unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout, verbosity=2))
