#!/usr/bin/python3
# -*- coding: utf-8 -*-

from gi.repository import GLib

import apt_pkg
import aptsources.distro
import aptsources.sourceslist

import dbus
import logging
import glob
import os
import subprocess
import sys
import threading
import time
import unittest
import pathlib

from dbus.mainloop.glib import DBusGMainLoop

sys.path.insert(0, "../")
from softwareproperties.dbus.SoftwarePropertiesDBus import (
    SoftwarePropertiesDBus, DBUS_BUS_NAME, DBUS_PATH, DBUS_INTERFACE_NAME)
from softwareproperties import (
    UPDATE_INST_SEC, UPDATE_DOWNLOAD, UPDATE_NOTIFY)

try:
    DPKG_ARCH = subprocess.check_output(
        ["dpkg", "--print-architecture"]).strip().decode("utf-8")
except subprocess.CalledProcessError:
    sys.stderr.write("WARNING: Failed to read dpkg arch")
    DPKG_ARCH = None

if DPKG_ARCH in ("i386", "amd64"):
    PRIMARY_MIRROR = "http://archive.ubuntu.com/ubuntu"
else:
    PRIMARY_MIRROR = "http://ports.ubuntu.com/ubuntu-ports"


def get_test_source_line():
    distro_release = get_distro_release()
    return "deb %s %s main restricted #"\
           " comment with unicode äöü" % (PRIMARY_MIRROR, distro_release)


def get_dpkg_arch():
    return subprocess.check_output(
        ["dpkg", "--print-architecture"]).strip().decode("utf-8")


def get_distro_release():
    return "bionic"


def clear_apt_config():
    etc_apt = os.path.join(os.path.dirname(__file__), "aptroot", "etc", "apt")
    for dirpath, dirnames, filenames in os.walk(etc_apt):
        for name in filenames:
            path = os.path.join(dirpath, name)
            if os.path.isfile(path):
                os.unlink(path)

    for d in ["apt.conf.d", "sources.list.d", "trusted.gpg.d", "auth.conf.d"]:
        os.makedirs(os.path.join(etc_apt, d), exist_ok=True)

def create_sources_list():
    s = get_test_source_line() + "\n"
    name = os.path.join(os.path.dirname(__file__),
                        "aptroot", "etc", "apt", "sources.list")
    dirname = os.path.dirname(name)
    if not os.path.exists(dirname):
        pathlib.Path(dirname).mkdir(parents=True)
    with open(name, "w") as f:
        f.write(s)
    return name


def session_bus_thread():
    DBusGMainLoop(set_as_default=True)
    loop = GLib.MainLoop()
    threading.current_thread().loop = loop

    bus = dbus.SessionBus(private=True)
    bus.set_exit_on_disconnect(False)

    rootdir = os.path.join(os.path.dirname(__file__), "aptroot")
    spd = SoftwarePropertiesDBus(bus, rootdir=rootdir)
    spd.enforce_polkit = False
    loop.run()
    bus.close()


class TestDBus(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        for k in apt_pkg.config.keys():
            apt_pkg.config.clear(k)
        apt_pkg.init()
        clear_apt_config()
        # create sources.list file
        create_sources_list()
        apt_pkg.config.set("Dir::Etc::sourcelist", "sources.list")
        cls.get_distro = aptsources.distro.get_distro
        aptsources.distro.get_distro = lambda *a, **b: aptsources.distro.UbuntuDistribution(id="Ubuntu", codename="bionic", description="18.04", release="bionic")
        cls.thread = threading.Thread(target=session_bus_thread)
        cls.thread.loop = None
        cls.thread.start()
        while not cls.thread.loop or not cls.thread.loop.is_running():
            time.sleep(1)

    @classmethod
    def tearDownClass(cls):
        cls.thread.loop.quit()
        cls.thread.join()
        aptsources.distro.get_distro = cls.get_distro

    def setUp(self):
        # keep track of signal emissions
        self.sources_list_count = 0
        self.distro_release = get_distro_release()
        create_sources_list()
        self._sourceslist = aptsources.sourceslist.SourcesList()
        # create the client proxy
        self.bus = dbus.SessionBus(private=True, mainloop=DBusGMainLoop())
        self.bus.set_exit_on_disconnect(False)
        self.iface = dbus.Interface(self.bus.get_object(DBUS_BUS_NAME, DBUS_PATH),
                                    DBUS_INTERFACE_NAME)
        self._signal_id = self.iface.connect_to_signal(
            "SourcesListModified", self._on_sources_list_modified)

    def tearDown(self):
        # ensure we remove the "modified" signal again
        self._signal_id.remove()
        self.bus.close()

    def _on_sources_list_modified(self):
        #print("_on_modified_sources_list")
        self.sources_list_count += 1

    @property
    def sourceslist(self):
        self._sourceslist.refresh()
        return ''.join([str(e) for e in self._sourceslist])

    @property
    def enabled_sourceslist(self):
        self._sourceslist.refresh()
        return ''.join([str(e) for e in self._sourceslist
                        if not e.invalid and not e.disabled])

    def _debug_sourceslist(self, text=""):
        logging.debug("sourceslist: %s '%s'" % (text, self.sourceslist))

    # this is an async call - give it a few seconds to catch up with what we expect
    def _assert_eventually(self, prop, n):
        for i in range(9):
            if getattr(self, prop) == n:
                self.assertEqual(getattr(self, prop), n)
            else:
                time.sleep(1)
        # nope, you die now
        self.assertEqual(getattr(self, prop), n)

    def test_enable_disable_component(self):
        # ensure its not there
        self.assertNotIn("universe", self.sourceslist)
        # enable
        self.iface.EnableComponent("universe")
        self._debug_sourceslist("2")
        self.assertIn("universe", self.sourceslist)
        # disable again
        self.iface.DisableComponent("universe")
        self._debug_sourceslist("3")
        self.assertNotIn("universe", self.sourceslist)
        self._assert_eventually("sources_list_count", 2)

    def test_enable_enable_disable_source_code_sources(self):
        # ensure its not there
        self._debug_sourceslist("4")
        self.assertNotIn('deb-src', self.enabled_sourceslist)
        # enable
        self.iface.EnableSourceCodeSources()
        self._debug_sourceslist("5")
        self.assertIn('deb-src', self.enabled_sourceslist)
        # disable again
        self.iface.DisableSourceCodeSources()
        self._debug_sourceslist("6")
        self.assertNotIn('deb-src', self.enabled_sourceslist)
        self._assert_eventually("sources_list_count", 3)

    def test_enable_child_source(self):
        child_source = "%s-updates" % self.distro_release
        # ensure its not there
        self._debug_sourceslist("7")
        self.assertNotIn(child_source, self.sourceslist)
        # enable
        self.iface.EnableChildSource(child_source)
        self._debug_sourceslist("8")
        self.assertIn(child_source, self.sourceslist)
        # disable again
        self.iface.DisableChildSource(child_source)
        self._debug_sourceslist("9")
        self.assertNotIn(child_source, self.sourceslist)
        self._assert_eventually("sources_list_count", 2)

    def test_toggle_source(self):
        # test toggle
        source = get_test_source_line()
        self.iface.ToggleSourceUse(source)
        self._debug_sourceslist("10")
        primary_debline = "# deb %s" % PRIMARY_MIRROR
        self.assertIn(primary_debline, self.sourceslist)
        # to disable the line again, we need to match the new "#"
        source = "# " + source
        self.iface.ToggleSourceUse(source)
        self._debug_sourceslist("11")
        self.assertNotIn(primary_debline, self.sourceslist)

        self._assert_eventually("sources_list_count", 2)

    def test_replace_entry(self):
        # test toggle
        source = get_test_source_line()
        source_new = "deb http://xxx/ %s" % self.distro_release
        self.iface.ReplaceSourceEntry(source, source_new)
        self._debug_sourceslist("11")
        self.assertIn(source_new, self.sourceslist)
        self.assertNotIn(source, self.sourceslist)
        self._assert_eventually("sources_list_count", 1)
        self.iface.ReplaceSourceEntry(source_new, source)
        self._assert_eventually("sources_list_count", 2)

    def test_popcon(self):
        # ensure its set to no
        popcon_p = os.path.join(os.path.dirname(__file__),
                              "aptroot", "etc", "popularity-contest.conf")
        with open(popcon_p) as f:
            popcon = f.read()
            self.assertIn('PARTICIPATE="no"', popcon)
        # toggle
        self.iface.SetPopconPariticipation(True)
        with open(popcon_p) as f:
            popcon = f.read()
            self.assertIn('PARTICIPATE="yes"', popcon)
            self.assertNotIn('PARTICIPATE="no"', popcon)
        # and back
        self.iface.SetPopconPariticipation(False)
        with open(popcon_p) as f:
            popcon = f.read()
            self.assertNotIn('PARTICIPATE="yes"', popcon)
            self.assertIn('PARTICIPATE="no"', popcon)

    def test_updates_automation(self):
        states = [UPDATE_INST_SEC, UPDATE_DOWNLOAD, UPDATE_NOTIFY]
        # security
        self.iface.SetUpdateAutomationLevel(states[0])
        cfg = os.path.join(os.path.dirname(__file__),
                           "aptroot", "etc", "apt", "apt.conf.d",
                           "10periodic")
        with open(cfg) as f:
            config = f.read()
            self.assertIn('APT::Periodic::Unattended-Upgrade "1";', config)
        # download
        self.iface.SetUpdateAutomationLevel(states[1])
        with open(cfg) as f:
            config = f.read()
            self.assertIn('APT::Periodic::Unattended-Upgrade "0";', config)
            self.assertIn('APT::Periodic::Download-Upgradeable-Packages "1";', config)
        # notify
        self.iface.SetUpdateAutomationLevel(states[2])
        with open(cfg) as f:
            config = f.read()
            self.assertIn('APT::Periodic::Unattended-Upgrade "0";', config)
            self.assertIn('APT::Periodic::Download-Upgradeable-Packages "0";', config)

    def test_updates_interval(self):
        # interval
        self.iface.SetUpdateInterval(0)
        cfg = os.path.join(os.path.dirname(__file__),
                           "aptroot", "etc", "apt", "apt.conf.d",
                           "10periodic")
        with open(cfg) as f:
            config = f.read()
            self.assertTrue(
                'APT::Periodic::Update-Package-Lists' not in config or
                'APT::Periodic::Update-Package-Lists "0";' in config)
        self.iface.SetUpdateInterval(1)
        with open(cfg) as f:
            config = f.read()
            self.assertIn('APT::Periodic::Update-Package-Lists "1";', config)
        self.iface.SetUpdateInterval(0)
        with open(cfg) as f:
            config = f.read()
            self.assertIn('APT::Periodic::Update-Package-Lists "0";', config)

    def test_add_remove_source_by_line(self):
        # add invalid
        res = self.iface.AddSourceFromLine("xxx")
        self.assertFalse(res)
        # add real
        s = "deb https://ppa.launchpadcontent.net/ foo bar"
        self.iface.AddSourceFromLine(s)
        self.assertIn(s, self.sourceslist)
        self.assertIn(s.replace("deb", "# deb-src"), self.sourceslist)
        # remove again
        self.iface.RemoveSource(s)
        self.iface.RemoveSource(s.replace("deb", "deb-src"))
        self.assertNotIn(s, self.sourceslist)
        self.assertNotIn(s.replace("deb", "# deb-src"), self.sourceslist)
        self._assert_eventually("sources_list_count", 4)

    def test_add_gpg_key(self):
        # clean
        gpg_glob = os.path.join(os.path.dirname(__file__),
                              "aptroot", "etc", "apt", "trusted.gpg.d", "*.asc")
        trusted_gpg_d = os.path.join(os.path.dirname(__file__),
                                     "aptroot", "etc", "apt", "trusted.gpg.d/")
        testkey = os.path.join(os.path.dirname(__file__),
                               "data", "testkey.asc")
        for f in glob.glob(gpg_glob):
            os.remove(f)
        self.assertTrue(len(glob.glob(gpg_glob)) == 0)
        # add key from file
        res = self.iface.AddKey(os.path.join(os.path.dirname(__file__),
                                             "data", "testkey.asc"))
        self.assertTrue(res)
        self.assertEqual(len(glob.glob(gpg_glob)), 1)
        self.assertNotEqual(os.path.getsize(trusted_gpg_d + "testkey.asc"), 0)
        # remove the key
        res = self.iface.RemoveKey(trusted_gpg_d + "testkey.asc")
        self.assertTrue(res)
        self.assertEqual(len(glob.glob(gpg_glob)), 0)
        # add from data
        with open(testkey) as keyfile:
            data = keyfile.read()
            res = self.iface.AddKeyFromData(data)
            self.assertTrue(res)
            self.assertEqual(len(os.listdir(trusted_gpg_d)), 1)
        # remove the key
        res = self.iface.RemoveKey(trusted_gpg_d + "/46caf96d27ee1eebcf139483275a5c1bb41590474e35f8bb4ae3cceb413f7518.gpg")
        self.assertTrue(res)
        self.assertEqual(len(os.listdir(trusted_gpg_d)), 1)
        # test nonsense
        res = self.iface.AddKeyFromData("nonsens")
        self.assertFalse(res)


if __name__ == "__main__":
    if "-d" in sys.argv:
        logging.basicConfig(level=logging.DEBUG)
    else:
        logging.basicConfig(level=logging.INFO)
    unittest.main()