#!/usr/bin/python


import sys
sys.path.insert(0,"../")

import apt
import os
import re
import unittest
import xapian

from softwarecenter.db.application import Application, AppDetails
from softwarecenter.db.database import StoreDatabase
from softwarecenter.db.enquire import AppEnquire
from softwarecenter.db.database import parse_axi_values_file
from softwarecenter.db.pkginfo import get_pkg_info
from softwarecenter.db.update import update_from_app_install_data, update_from_var_lib_apt_lists, update_from_appstream_xml
from softwarecenter.enums import (
    XapianValues,
    PkgStates,
    )

class TestDatabase(unittest.TestCase):
    """ tests the store database """

    def setUp(self):
        apt.apt_pkg.config.set("Dir::State::status",
                               "./data/appdetails/var/lib/dpkg/status")
        self.cache = get_pkg_info()
        self.cache.open()

    def test_update_from_desktop_file(self):
        # ensure we index with german locales to test i18n
        os.environ["LANGUAGE"] = "de"
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        res = update_from_app_install_data(db, self.cache, datadir="./data/desktop")
        self.assertTrue(res)
        self.assertEqual(db.get_doccount(), 5)
        # test if Name[de] was picked up
        i=0
        for it in db.postlist("AAUbuntu Software Zentrum"):
            i+=1
        self.assertEqual(i, 1)

    def test_update_from_appstream_xml(self):
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        res = update_from_appstream_xml(db, self.cache, "./data/app-info/")
        self.assertTrue(res)
        self.assertEqual(db.get_doccount(), 1)
        # FIXME: improve tests
        for p in db.postlist(""):
            doc = db.get_document(p.docid)
            for term in doc.termlist():
                print term, term.term
            for value in doc.values():
                print value, value.num, value.value

    def test_update_from_var_lib_apt_lists(self):
        # ensure we index with german locales to test i18n
        os.environ["LANGUAGE"] = "de"
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        res = update_from_var_lib_apt_lists(db, self.cache, listsdir="./data/app-info/")
        self.assertTrue(res)
        self.assertEqual(db.get_doccount(), 1)
        # test if Name-de was picked up
        i=0
        for it in db.postlist("AAFestplatten Ueberpruefer"):
            i+=1
        self.assertEqual(i, 1)
        # test if gettext worked
        found_gettext_translation = False
        for it in db.postlist("AAFestplatten Ueberpruefer"):
            doc = db.get_document(it.docid)
            for term_iter in doc.termlist():
                # a german term from the app-info file to ensure that
                # it got indexed in german
                if term_iter.term == "platzes":
                    found_gettext_translation = True
                    break
        self.assertTrue(found_gettext_translation)

    def test_update_from_json_string(self):
        from softwarecenter.db.update import update_from_json_string
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        cache = apt.Cache()
        p = os.path.abspath("./data/app-info-json/apps.json")
        res = update_from_json_string(db, cache, open(p).read(), origin=p)
        self.assertTrue(res)
        self.assertEqual(db.get_doccount(), 1)

    def test_build_from_software_center_agent(self):
        from softwarecenter.db.update import update_from_software_center_agent
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        cache = apt.Cache()
        # monkey patch distro to ensure we get data
        import softwarecenter.distro
        distro = softwarecenter.distro.get_distro()
        distro.get_codename = lambda: "natty"
        # we test against the real https://software-center.ubuntu.com here
        # so we need network
        softwarecenter.paths.datadir="../data"
        os.environ["PYTHONPATH"] = ".."
        res = update_from_software_center_agent(db, cache, ignore_cache=True)
        # check results
        self.assertTrue(res)
        self.assertTrue(db.get_doccount() > 1)
        for p in db.postlist(""):
            doc = db.get_document(p.docid)
            ppa = doc.get_value(XapianValues.ARCHIVE_PPA)
            self.assertTrue(ppa.startswith("commercial-ppa") and
                            ppa.count("/") == 1, 
                            "ARCHIVE_PPA value incorrect, got '%s'" % ppa)
            self.assertTrue(
                doc.get_value(XapianValues.ICON).startswith("sc-agent"))

    def test_license_string_data_from_software_center_agent(self):
        from softwarecenter.db.update import update_from_software_center_agent
        from softwarecenter.testutils import get_test_pkg_info
        #os.environ["SOFTWARE_CENTER_DEBUG_HTTP"] = "1"
        os.environ["SOFTWARE_CENTER_BUY_HOST"] = "http://sc.staging.ubuntu.com/"
        # staging does not have a valid cert
        os.environ["PISTON_MINI_CLIENT_DISABLE_SSL_VALIDATION"] = "1"
        cache = get_test_pkg_info()
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        res = update_from_software_center_agent(db, cache, ignore_cache=True)
        self.assertTrue(res)
        for p in db.postlist(""):
            doc = db.get_document(p.docid)
            license = doc.get_value(XapianValues.LICENSE)
            self.assertNotEqual(license, "")
            self.assertNotEqual(license, None)
        del os.environ["SOFTWARE_CENTER_BUY_HOST"]

    def test_application(self):
        db = StoreDatabase("/var/cache/software-center/xapian", self.cache)
        # fail if AppDetails(db) without document= or application=
        # is run
        self.assertRaises(ValueError, AppDetails, db)

    def test_application_details(self):
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        res = update_from_app_install_data(db, self.cache, datadir="./data/desktop")
        self.assertTrue(res)
        db = StoreDatabase("./data/test.db", self.cache)
        db.open(use_axi=False, use_agent=False)
        self.assertEqual(len(db), 5)
        # test details
        app = Application("Ubuntu Software Center Test", "software-center")
        details = app.get_details(db)
        self.assertNotEqual(details, None)
        self.assertEqual(details.component, "main")
        self.assertEqual(details.pkgname, "software-center")
        # get the first document
        for doc in db:
            if doc.get_data() == "Ubuntu Software Center Test":
                appdetails = AppDetails(db, doc=doc)
                break
        # test get_appname and get_pkgname
        self.assertEqual(db.get_appname(doc), "Ubuntu Software Center Test")
        self.assertEqual(db.get_pkgname(doc), "software-center")
        # test appdetails
        self.assertEqual(appdetails.name, "Ubuntu Software Center Test")
        self.assertEqual(appdetails.pkgname, "software-center")
        # FIXME: add a dekstop file with a real channel to test
        #        and monkey-patch/modify the APP_INSTALL_CHANNELS_PATH
        self.assertEqual(appdetails.channelname, None)
        self.assertEqual(appdetails.channelfile, None)
        self.assertEqual(appdetails.component, "main")
        self.assertNotEqual(appdetails.pkg, None)
        # from the fake test/data/appdetails/var/lib/dpkg/status
        self.assertEqual(appdetails.pkg.is_installed, True)
        self.assertEqual(appdetails.pkg_state, PkgStates.INSTALLED)
        # FIXME: test description for unavailable pkg
        self.assertTrue(
            appdetails.description.startswith("Ubuntu Software Center lets you"))
        # FIXME: test appdetails.website
        self.assertEqual(appdetails.icon, "softwarecenter")
        # crude, crude
        self.assertTrue(len(appdetails.version) > 2)
        # FIXME: screenshots will only work on ubuntu
        self.assertTrue(re.match(
                "http://screenshots.ubuntu.com/screenshot-with-version/software-center/[\d.]+", 
                appdetails.screenshot))
        self.assertTrue(re.match(
                "http://screenshots.ubuntu.com/thumbnail-with-version/software-center/[\d.]+",
                appdetails.thumbnail))
        # FIXME: add document that has a price
        self.assertEqual(appdetails.price, '')
        self.assertEqual(appdetails.license, "Open source")
        # test lazy history loading for installation date
        self.ensure_installation_date_and_lazy_history_loading(appdetails)
        # test apturl replacements
        # $kernel
        app = Application("", "linux-headers-$kernel", "channel=$distro-partner")
        self.assertEqual(app.pkgname, 'linux-headers-'+os.uname()[2])
        # $distro
        details = app.get_details(db)
        from softwarecenter.distro import get_distro
        distro = get_distro().get_codename()
        self.assertEqual(app.request, 'channel=' + distro + '-partner')
        
    def ensure_installation_date_and_lazy_history_loading(self, appdetails):
        # we run two tests, the first is to ensure that we get a 
        # result from installation_data immediately (at this point the
        # history is not loaded yet) so we expect "None"
        self.assertEqual(appdetails.installation_date, None)
        # then we need to wait until the history is loaded in the idle
        # handler
        from gi.repository import GObject
        context = GObject.main_context_default()
        while context.pending():
            context.iteration()
        # ... and finally we test that its really there
        # FIXME: this will only work if software-center is installed
        self.assertNotEqual(appdetails.installation_date, None)

    def test_package_states(self):
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        res = update_from_app_install_data(db, self.cache, datadir="./data/desktop")
        self.assertTrue(res)
        db = StoreDatabase("./data/test.db", self.cache)
        db.open(use_axi=False)
        # test PkgStates.INSTALLED
        # FIXME: this will only work if software-center is installed
        app = Application("Ubuntu Software Center Test", "software-center")
        appdetails = app.get_details(db)
        self.assertEqual(appdetails.pkg_state, PkgStates.INSTALLED)
        # test PkgStates.UNINSTALLED
        # test PkgStates.UPGRADABLE
        # test PkgStates.REINSTALLABLE
        # test PkgStates.INSTALLING
        # test PkgStates.REMOVING
        # test PkgStates.UPGRADING
        # test PkgStates.NEEDS_SOURCE
        app = Application("Zynjacku Test", "zynjacku-fake")
        appdetails = app.get_details(db)
        self.assertEqual(appdetails.pkg_state, PkgStates.NEEDS_SOURCE)
        # test PkgStates.NEEDS_PURCHASE
        app = Application("The expensive gem", "expensive-gem")
        appdetails = app.get_details(db)
        self.assertEqual(appdetails.pkg_state, PkgStates.NEEDS_PURCHASE)
        self.assertEqual(appdetails.icon_url, "http://www.google.com/favicon.ico")
        self.assertEqual(appdetails.icon, "favicon")
        # test PkgStates.PURCHASED_BUT_REPO_MUST_BE_ENABLED
        # test PkgStates.UNKNOWN
        app = Application("Scintillant Orange", "scintillant-orange")
        appdetails = app.get_details(db)
        self.assertEqual(appdetails.pkg_state, PkgStates.NOT_FOUND)

    def test_packagename_is_application(self):
        db = StoreDatabase("/var/cache/software-center/xapian", self.cache)
        db.open()
        # apt has no app
        self.assertEqual(db.get_apps_for_pkgname("apt"), set())
        # but software-center has
        self.assertEqual(len(db.get_apps_for_pkgname("software-center")), 1)

    def test_whats_new(self):
        db = StoreDatabase("/var/cache/software-center/xapian", self.cache)
        db.open()
        query = xapian.Query("")
        enquire = xapian.Enquire(db.xapiandb)
        enquire.set_query(query)
        value_time = db._axi_values["catalogedtime"]
        enquire.set_sort_by_value(value_time, reverse=True)
        matches = enquire.get_mset(0, 20)
        last_time = 0
        for m in matches:
            doc = m.document
            doc.get_value(value_time) >= last_time
            last_time = doc.get_value(value_time)
            
    def test_non_axi_apps_cataloged_time(self):
        db = xapian.WritableDatabase("./data/test.db", 
                                     xapian.DB_CREATE_OR_OVERWRITE)
        res = update_from_app_install_data(db, self.cache, datadir="./data/desktop")
        self.assertTrue(res)
        db = StoreDatabase("./data/test.db", self.cache)
        db.open(use_axi=True)

        axi_value_time = db._axi_values["catalogedtime"]
        sc_app = Application("Ubuntu Software Center Test", "software-center")
        sc_doc = db.get_xapian_document(sc_app.appname, sc_app.pkgname)
        sc_cataloged_time = sc_doc.get_value(axi_value_time)
        so_app = Application("Scintillant Orange", "scintillant-orange")
        so_doc = db.get_xapian_document(so_app.appname, so_app.pkgname)
        so_cataloged_time = so_doc.get_value(axi_value_time)
        # the test package Scintillant Orange should be cataloged at a
        # later time than axi package Ubuntu Software Center
        self.assertTrue(so_cataloged_time > sc_cataloged_time)

    def test_parse_axi_values_file(self):
        s = """
# This file contains the mapping between names of numeric values indexed in the
# APT Xapian index and their index
#
# Xapian allows to index numeric values as well as keywords and to use them for
# all sorts of useful querying tricks.  However, every numeric value needs to
# have a unique index, and this configuration file is needed to record which
# indices are allocated and to provide a mnemonic name for them.
#
# The format is exactly like /etc/services with name, number and optional
# aliases, with the difference that the second column does not use the
# "/protocol" part, which would be meaningless here.

version	0	# package version
catalogedtime	1	# Cataloged timestamp
installedsize	2	# installed size
packagesize	3	# package size
app-popcon	4	# app-install .desktop popcon rank
"""
        open("axi-test-values","w").write(s)
        #db = StoreDatabase("/var/cache/software-center/xapian", self.cache)
        axi_values = parse_axi_values_file("axi-test-values")
        self.assertNotEqual(axi_values, {})
        print axi_values

    def test_app_enquire(self):
        db = StoreDatabase("/var/cache/software-center/xapian", self.cache)
        db.open()
        # test the AppEnquire engine
        enquirer = AppEnquire(self.cache, db)
        enquirer.set_query(xapian.Query("a"),
                           nonblocking_load=False)
        self.assertTrue(len(enquirer.get_docids()) > 0)
        # FIXME: test more of the interface

if __name__ == "__main__":
    import logging
    logging.basicConfig(level=logging.DEBUG)
    unittest.main()