# Copyright (C) 2007, 2008 by Intevation GmbH
# Authors:
# Bernhard Herzog <bh@intevation.de>
#
# This program is free software under the GPL (>=v2)
# Read the file COPYING coming with the software for details.

"""Tests for treepkg.packager"""

import sys
import os
import unittest

from treepkg.run import call
from treepkg.cmdexpand import cmdexpand
from treepkg.util import writefile
from treepkg.packager import PackagerGroup, import_packager_module, \
     CyclicDependencyError
import treepkg.subversion as subversion
import treepkg

from filesupport import FileTestMixin


def create_svn_repository(directory):
    baseurl = "file://" + directory
    call(cmdexpand("svnadmin create --fs-type fsfs $directory",
                   **locals()))
    call(cmdexpand("svn mkdir -q -m 'create directory structure'"
                   " $baseurl/trunk",
                   **locals()))
    return baseurl

def add_svn_files(workingcopy, filedesc, commitmsg):
    for name, contents in filedesc:
        writefile(os.path.join(workingcopy, name), contents)
        call(cmdexpand("svn add -q $name", **locals()),
             cwd=workingcopy)
    call(cmdexpand("svn commit -q -m $commitmsg", **locals()),
         cwd=workingcopy)


class SourcePackager(treepkg.packager.SourcePackager):

    pkg_basename = "testpkg"

    def do_package(self):
        pkgbaseversion, pkgbasedir = self.export_sources()

        pkgbasename = self.pkg_basename + "_" + pkgbaseversion
        origtargz = os.path.join(self.work_dir,
                                 pkgbasename + ".orig.tar.gz")
        self.create_tarball(origtargz, self.work_dir,
                            os.path.basename(pkgbasedir))

        changemsg = ("Update to SVN rev. %d" % (self.revision,))
        self.copy_debian_directory(pkgbasedir, pkgbaseversion,
                                   changemsg)

        self.create_source_package(pkgbasedir, origtargz)
        self.move_source_package(pkgbasename)


class RevisionPackager(treepkg.packager.RevisionPackager):

    source_packager_cls = SourcePackager


class PackageTrack(treepkg.packager.PackageTrack):

    revision_packager_cls = RevisionPackager


class PackagerTest(unittest.TestCase, FileTestMixin):

    revisions = []

    debian_files = [
        ("debian",
         [("control", """\
Source: testpkg
Priority: optional
Maintainer: TreePKG <treepkg@example.com>
Standards-Version: 3.7.2

Package: testpkg
Architecture: all
Description: Test package for treepkg tests
 German (de) internationalized (i18n) files for KDE
 This package contains the German internationalized (i18n) files for
 all KDE core applications.
"""),
          ("changelog", """\
testpkg (0-0) unstable; urgency=low

  * Initial version

 -- TreePKG <treepkg@example.com>  Thu,  8 Mar 2007 18:34:39 +0100
"""),
          ("rules", "binary: echo binary")])]


    def setUp(self):
        self.svndir = self.create_temp_dir("svn")
        self.svnworkdir = self.create_temp_dir("svnwork")
        self.trackdir = self.create_files("track", self.debian_files)
        self.svn_url = create_svn_repository(self.svndir) + "/trunk"
        subversion.checkout(self.svn_url, self.svnworkdir)
        for message, files in self.revisions:
            add_svn_files(self.svnworkdir, files, message)

    def runtest(self, debrevision, group_args=None, **extra_track_args):
        rootcmd = os.path.join(os.path.dirname(__file__), os.pardir, "test",
                               "mocksudopbuilder.py")
        track = PackageTrack(name="testpkg", base_dir=self.trackdir,
                             svn_url=self.svn_url, pbuilderrc="",
                             root_cmd=[sys.executable, rootcmd],
                             deb_email="treepkg@example.com",
                             deb_fullname="treepkg tester",
                             **extra_track_args)

        if group_args is None:
            group_args = {}
        group = PackagerGroup([track], 1, **group_args)
        group.check_package_tracks()

        # determine version that has been packaged.  This assumes that
        # check_package_tracks will leave the checkout in the revision
        # that was actually packaged.
        version = track.last_changed_revision()

        # Now check the source and binary package files
        pkgdir = os.path.join(self.trackdir, "pkg", "%d-1" % version)
        self.assertEquals(sorted(os.listdir(os.path.join(pkgdir, "src"))),
                          [name % locals() for name in
                           ["testpkg_%(version)d-%(debrevision)s.diff.gz",
                            "testpkg_%(version)d-%(debrevision)s.dsc",
                            "testpkg_%(version)d.orig.tar.gz"]])
        self.assertEquals(sorted(os.listdir(os.path.join(pkgdir, "binary"))),
                          [name % locals() for name in
                           ["testpkg_%(version)d-%(debrevision)s_all.deb",
                           "testpkg_%(version)d-%(debrevision)s_i386.changes"]])


class TestPackager(PackagerTest):

    revisions = [
        ("Initial Revision",
         [("README", "and miles to go before I sleep")]),
        ]

    def test_default_debrevision_prefix(self):
        self.runtest("treepkg1")

    def test_custom_debrevision_prefix(self):
        self.runtest("kk1", debrevision_prefix="kk")


class TestPackagerWithMultipleRevisions(PackagerTest):

    revisions = [
        ("Initial Revision",
         [("README", "and miles to go before I sleep")]),
        ("Add some code",
         [("program.c", "int main(void) { return 0; }")]),
        ("Add some more code",
         [("another.c", "int main(void) { return 1; }")]),
        ]

    def test_packaging_specific_revision(self):
        # Package the newest revision and then package an older one.
        self.runtest("treepkg1")
        self.runtest("treepkg1", group_args=dict(revision=3))


class StoppingPackager(treepkg.packager.RevisionPackager):

    def package(self):
        pass

class StoppingTrack(treepkg.packager.PackageTrack):

    def __init__(self, do_package, do_stop, instructions_file, name, trackdir):
        super(StoppingTrack, self).__init__(name, trackdir, "", "", "",
                                            "", "")
        self.do_package = do_package
        self.do_stop = do_stop
        self.instructions_file = instructions_file

    def package_if_updated(self, revision, do_svn_update=True):
        if self.do_stop:
            writefile(self.instructions_file, "stop")
        if self.do_package:
            return StoppingPackager(self, 1)
        else:
            return None


class TestPackageGroupStop(unittest.TestCase, FileTestMixin):

    def setUp(self):
        self.trackdir = self.create_temp_dir("track")
        self.instructions_file = os.path.join(self.trackdir, "instructions")

    def group(self, do_package=True, do_stop=True):
        return PackagerGroup([StoppingTrack(do_package, do_stop,
                                            self.instructions_file,
                                            "test", self.trackdir)],
                             1, instructions_file=self.instructions_file)

    def test_stop(self):
        group = self.group(do_package=True, do_stop=True)
        self.failUnless(group.check_package_tracks())

    def test_no_stop(self):
        group = self.group(do_package=True, do_stop=False)
        self.failIf(group.check_package_tracks())

    def test_instruction_removal(self):
        # run once with stopping
        group = self.group(do_package=True, do_stop=True)
        self.failUnless(group.check_package_tracks())

        # run again without stopping but using the same files.  The
        # instructions file should be removed automatically
        group = self.group(do_package=True, do_stop=False)
        self.failIf(group.check_package_tracks())

    def test_stopping_without_packaging(self):
        group = self.group(do_package=False, do_stop=True)
        self.failUnless(group.check_package_tracks())

    def test_stopping_between_checks(self):
        group = self.group(do_package=False, do_stop=False)
        # run check_package_tracks once
        self.failIf(group.check_package_tracks())

        # tell treepkg to stop
        writefile(self.instructions_file, "stop")

        # check again.  The check_package_tracks() may remove the
        # instructions file but it must do so only the first time
        self.failUnless(group.check_package_tracks())


class TestPackageTrack(unittest.TestCase, FileTestMixin):

    # def setUp(self):
    #     self.trackdir = self.create_temp_dir("track")

    def test_get_revision_numbers(self):
        # Note: The revisions in the pkg dir are not ordered so that we
        # can check whether get_revision_numbers returns a sorted list
        # of revisions
        trackdir = self.create_files("track",
                                     [("pkg",
                                       [("704195-1",
                                         [("status", ""),
                                          ("src", []),
                                          ("binary", [])]),
                                        ("702432-1",
                                         [("status", ""),
                                          ("src", []),
                                          ("binary", [])])])])
        track = PackageTrack("testtrack", trackdir, "", "", "", "", "")
        self.assertEquals(track.get_revision_numbers(), [702432, 704195])

    def test_last_packaged_revision_with_no_revisions(self):
        trackdir = self.create_files("track", [("pkg", [])])
        track = PackageTrack("testtrack", trackdir, "", "", "", "", "")
        self.assertEquals(track.last_packaged_revision(), -1)

    def test_last_packaged_revision_several_revisions(self):
        trackdir = self.create_files("track",
                                     [("pkg",
                                       [("704195-1",
                                         [("status", ""),
                                          ("src", []),
                                          ("binary", [])]),
                                        ("702432-1",
                                         [("status", ""),
                                          ("src", []),
                                          ("binary", [])])])])
        track = PackageTrack("testtrack", trackdir, "", "", "", "", "")
        self.assertEquals(track.last_packaged_revision(), 704195)


class TestRevisionPackager(unittest.TestCase, FileTestMixin):

    # def setUp(self):
    #     self.trackdir = self.create_temp_dir("track")

    def test_list_source_files(self):
        # Note: The revisions in the pkg dir are not ordered so that we
        # can check whether get_revision_numbers returns a sorted list
        # of revisions
        trackdir = self.create_files("track",
                                     [("pkg",
                                       [("704195-1",
                                         [("status",
                                           ("TreePackagerStatus 0.0\n"
                                            "status: binary_package_created\n"
                                            "start: 2007-09-10 17:16:48\n"
                                            "stop: 2007-09-11 00:07:36\n")),
                                          ("src", [("test_1.0.orig.tar.gz", ""),
                                                   ("test_1.0-1.diff.gz", ""),
                                                   ("test_1.0-1.dsc", "")]),
                                          ("binary", [])]),
                                        ("702432-1",
                                         [("status", ""),
                                          ("src", []),
                                          ("binary", [])])])])
        track = PackageTrack("testtrack", trackdir, "", "", "", "", "")
        revpkg = RevisionPackager(track, 704195)
        srcdir = os.path.join(trackdir, "pkg", "704195-1", "src")
        self.assertEquals(revpkg.list_source_files(),
                          [os.path.join(srcdir, filename)
                           for filename in ["test_1.0-1.diff.gz",
                                            "test_1.0-1.dsc",
                                            "test_1.0.orig.tar.gz"]])

    def test_list_binary_files(self):
        # Note: The revisions in the pkg dir are not ordered so that we
        # can check whether get_revision_numbers returns a sorted list
        # of revisions
        trackdir = self.create_files("track",
                                     [("pkg",
                                       [("704195-1",
                                         [("status",
                                           ("TreePackagerStatus 0.0\n"
                                            "status: binary_package_created\n"
                                            "start: 2007-09-10 17:16:48\n"
                                            "stop: 2007-09-11 00:07:36\n")),
                                          ("src", []),
                                          ("binary",
                                           [("test_1.0-1_i386.deb", ""),
                                            ("test_1.0-1_i386.changes", "")])]),
                                        ("702432-1",
                                         [("status", ""),
                                          ("src", []),
                                          ("binary", [])])])])
        track = PackageTrack("testtrack", trackdir, "", "", "", "", "")
        revpkg = RevisionPackager(track, 704195)
        bindir = os.path.join(trackdir, "pkg", "704195-1", "binary")
        self.assertEquals(revpkg.list_binary_files(),
                          [os.path.join(bindir, filename)
                           for filename in ["test_1.0-1_i386.changes",
                                            "test_1.0-1_i386.deb"]])


class TestImportPackagerModule(unittest.TestCase, FileTestMixin):

    files = [("treepkg_importtest",
              [("__init__.py", ""),
               ("withtrack.py", "\n".join(["class PackageTrack:",
                                           "    pass",
                                           ""])),
               ("srconly.py", "\n".join(["class SourcePackager:",
                                         "    pass",
                                         ""])),
               ("srcandbin.py", "\n".join(["class SourcePackager:",
                                           "    pass",
                                           "class BinaryPackager:",
                                           "    pass",
                                           ""])),
               ("revonly.py", "\n".join(["class RevisionPackager:",
                                         "    pass",
                                         ""]))])]

    def setUp(self):
        self.directory = self.create_files("a_module", self.files)
        self.old_path = sys.path
        sys.path = [self.directory] + sys.path

    def tearDown(self):
        sys.path = self.old_path

    def check_class_modules(self, module, classmodules):
        self.assertEquals(classmodules,
                          [(item[0],
                            sys.modules[getattr(module, item[0]).__module__])
                           for item in classmodules])

    def test_import_with_track(self):
        module = import_packager_module("treepkg_importtest.withtrack")
        self.check_class_modules(module, [("PackageTrack", module)])

    def test_import_with_source_packager(self):
        module = import_packager_module("treepkg_importtest.srconly")
        self.check_class_modules(module, [("PackageTrack", treepkg.packager),
                                          ("SourcePackager", module),])

    def test_import_with_source_and_binary_packager(self):
        module = import_packager_module("treepkg_importtest.srcandbin")
        self.check_class_modules(module, [("PackageTrack", treepkg.packager),
                                          ("RevisionPackager",
                                           treepkg.packager),
                                          ("SourcePackager", module),
                                          ("BinaryPackager", module),])
        self.assertEquals(module.PackageTrack.revision_packager_cls,
                          module.RevisionPackager)
        self.assertEquals(module.RevisionPackager.source_packager_cls,
                          module.SourcePackager)
        self.assertEquals(module.RevisionPackager.binary_packager_cls,
                          module.BinaryPackager)

    def test_import_with_revision_packager(self):
        module = import_packager_module("treepkg_importtest.revonly")
        self.check_class_modules(module, [("PackageTrack", treepkg.packager),
                                          ("RevisionPackager", module)])

        self.assertEquals(module.PackageTrack.revision_packager_cls,
                          module.RevisionPackager)


class PackageTrackWithDependencies(treepkg.packager.PackageTrack):

    def __init__(self, name, handle_dependencies, requires, provides):
        defaults = dict(base_dir="/home/builder/tracks/" + name,
                        svn_url="svn://example.com",
                        root_cmd=["false"],
                        pbuilderrc="/home/builder/pbuilderrc",
                        deb_email="treepkg@example.com", deb_fullname="treepkg",
                        handle_dependencies=handle_dependencies)
        super(PackageTrackWithDependencies,
              self).__init__(name, **defaults)
        self.dependencies = (set(requires.split()), set(provides.split()))

    def determine_dependencies(self):
        pass


class TestPackageDependencies(unittest.TestCase):

    def test_track_order(self):
        P = PackageTrackWithDependencies
        tracks = [P("library", True, "base-dev", "library library-dev"),
                  P("other", False, "cdbs base-dev", "other"),
                  P("base", True, "", "base base-dev"),
                  P("program", True, "library-dev libc", "program program-doc"),
                  ]
        group = PackagerGroup(tracks, 3600)
        sorted_tracks = group.get_package_tracks()
        track_indices = dict([(track.name, index) for index, track in
                              enumerate(sorted_tracks)])
        def check_order(track1, track2):
            self.failUnless(track_indices[track1] < track_indices[track2])

        check_order("base", "library")
        check_order("library", "program")
        check_order("base", "program")

        # sanity check whether other is still there.  It doesn't matter
        # where
        self.failUnless("other" in track_indices)

    def test_track_order_cycle(self):
        P = PackageTrackWithDependencies
        tracks = [P("library", True, "base-dev", "library library-dev"),
                  P("cycle", True, "program", "cycle"),
                  P("other", False, "cdbs base-dev", "other"),
                  P("base", True, "cycle", "base base-dev"),
                  P("program", True, "library-dev libc", "program program-doc"),
                  ]
        try:
            group = PackagerGroup(tracks, 3600)
            sorted_tracks = group.get_package_tracks()
        except CyclicDependencyError, exc:
            pass
        else:
            self.fail("PackagerGroup did not detect cyclic dependencies")