#
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.
#
#
import unittest, os, sys, gc
from svn import core, repos, fs
import utils

class RevRange:
  """ Proxy object for a revision range, used for comparison. """

  def __init__(self, start, end):
    self.start = start
    self.end = end

def get_svn_merge_range_t_objects():
  """Returns a list 'svn_merge_range_t' objects being tracked by the
     garbage collector, used for detecting memory leaks."""
  return [
    o for o in gc.get_objects()
      if getattr(o, '__class__', None) is not None and
        o.__class__.__name__ == 'svn_merge_range_t'
  ]

class SubversionMergeinfoTestCase(unittest.TestCase):
  """Test cases for mergeinfo"""

  # Some textual mergeinfo.
  TEXT_MERGEINFO1 = b"/trunk:3-9,27,42*"
  TEXT_MERGEINFO2 = b"/trunk:27-29,41-43*"

  # Meta data used in conjunction with this mergeinfo.
  MERGEINFO_SRC = b"/trunk"
  MERGEINFO_NBR_REV_RANGES = 3

  def setUp(self):
    """Load the mergeinfo-full Subversion repository.  This dumpfile is
       created by dumping the repository generated for command line log
       tests 16.  If it needs to be updated (mergeinfo format changes, for
       example), we can go there to get a new version."""
    self.temper = utils.Temper()
    (self.repos, _, _) = self.temper.alloc_known_repo('data/mergeinfo.dump',
                                                      suffix='-mergeinfo')
    self.fs = repos.fs(self.repos)
    self.rev = fs.youngest_rev(self.fs)

  def tearDown(self):
    del self.fs
    del self.repos
    self.temper.cleanup()

  def test_mergeinfo_parse(self):
    """Test svn_mergeinfo_parse()"""
    mergeinfo = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)
    self.inspect_mergeinfo_dict(mergeinfo, self.MERGEINFO_SRC,
                                self.MERGEINFO_NBR_REV_RANGES)

  def test_rangelist_merge(self):
    mergeinfo1 = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)
    mergeinfo2 = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO2)
    rangelist1 = mergeinfo1.get(self.MERGEINFO_SRC)
    rangelist2 = mergeinfo2.get(self.MERGEINFO_SRC)
    rangelist3 = core.svn_rangelist_merge(rangelist1, rangelist2)
    self.inspect_rangelist_tuple(rangelist3, 3)

  def test_mergeinfo_merge(self):
    """Test svn_mergeinfo_merge()"""
    mergeinfo1 = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)
    mergeinfo2 = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO2)
    mergeinfo3 = core.svn_mergeinfo_merge(mergeinfo1, mergeinfo2)
    self.inspect_mergeinfo_dict(mergeinfo3, self.MERGEINFO_SRC,
                                self.MERGEINFO_NBR_REV_RANGES)

  def test_rangelist_reverse(self):
    mergeinfo = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)
    rangelist = mergeinfo.get(self.MERGEINFO_SRC)
    reversed_rl = core.svn_rangelist_reverse(rangelist)
    expected_ranges = ((42, 41), (27, 26), (9, 2))
    for i in range(0, len(reversed_rl)):
      self.assertEqual(reversed_rl[i].start, expected_ranges[i][0],
                        "Unexpected range start: %d" % reversed_rl[i].start)
      self.assertEqual(reversed_rl[i].end, expected_ranges[i][1],
                        "Unexpected range end: %d" % reversed_rl[i].end)

  def test_mergeinfo_sort(self):
    mergeinfo = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)

    # Swap the order of two revision ranges to misorder the contents
    # of a rangelist.
    rangelist = mergeinfo.get(self.MERGEINFO_SRC)
    rev_range = rangelist[0]
    rangelist[0] = rangelist[1]
    rangelist[1] = rev_range

    mergeinfo = core.svn_mergeinfo_sort(mergeinfo)
    self.inspect_mergeinfo_dict(mergeinfo, self.MERGEINFO_SRC,
                                self.MERGEINFO_NBR_REV_RANGES)

  def test_mergeinfo_get(self):
    mergeinfo = repos.fs_get_mergeinfo(self.repos, [b'/trunk'], self.rev,
                                       core.svn_mergeinfo_inherited,
                                       False, None, None)
    expected_mergeinfo = \
      { b'/trunk' :
          { b'/branches/a' : [RevRange(2, 11)],
            b'/branches/b' : [RevRange(9, 13)],
            b'/branches/c' : [RevRange(2, 16)],
            b'/trunk'      : [RevRange(1, 9)],  },
      }
    self.compare_mergeinfo_catalogs(mergeinfo, expected_mergeinfo)

  def test_mergeinfo_leakage__incorrect_range_t_refcounts(self):
    """Ensure that the ref counts on svn_merge_range_t objects returned by
       svn_mergeinfo_parse() are correct."""
    # When reference counting is working properly, each svn_merge_range_t in
    # the returned mergeinfo will have a ref count of 1...
    mergeinfo = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)
    for (path, rangelist) in core._as_list(mergeinfo.items()):
      # ....and now 2 (incref during iteration of rangelist)

      for (i, r) in enumerate(rangelist):
        # ....and now 3 (incref during iteration of each range object)

        refcount = sys.getrefcount(r)
        # ....and finally, 4 (getrefcount() also increfs)
        expected = 4

        # Note: if path and index are not '/trunk' and 0 respectively, then
        # only some of the range objects are leaking, which is, as far as
        # leaks go, even more impressive.
        self.assertEqual(refcount, expected, (
          "Memory leak!  Expected a ref count of %d for svn_merge_range_t "
          "object, but got %d instead (path: %s, index: %d).  Probable "
          "cause: incorrect Py_INCREF/Py_DECREF usage in libsvn_swig_py/"
          "swigutil_py.c." % (expected, refcount, path, i)))

    del mergeinfo
    gc.collect()

  def test_mergeinfo_leakage__lingering_range_t_objects_after_del(self):
    """Ensure that there are no svn_merge_range_t objects being tracked by
       the garbage collector after we explicitly `del` the results returned
       by svn_mergeinfo_parse().  We call gc.collect() to force an explicit
       garbage collection cycle after the `del`;
       if our reference counts are correct, the allocated svn_merge_range_t
       objects will be garbage collected and thus, not appear in the list of
       objects returned by gc.get_objects()."""
    mergeinfo = core.svn_mergeinfo_parse(self.TEXT_MERGEINFO1)
    del mergeinfo
    gc.collect()
    lingering = get_svn_merge_range_t_objects()
    self.assertEqual(lingering, list(), (
      "Memory leak!  Found lingering svn_merge_range_t objects left over from "
      "our call to svn_mergeinfo_parse(), even though we explicitly deleted "
      "the returned mergeinfo object.  Probable cause: incorrect Py_INCREF/"
      "Py_DECREF usage in libsvn_swig_py/swigutil_py.c.  Lingering objects:\n"
      "%s" % lingering))

  def inspect_mergeinfo_dict(self, mergeinfo, merge_source, nbr_rev_ranges):
    rangelist = mergeinfo.get(merge_source)
    self.inspect_rangelist_tuple(rangelist, nbr_rev_ranges)

  def inspect_rangelist_tuple(self, rangelist, nbr_rev_ranges):
    self.assertTrue(rangelist is not None,
                 "Rangelist for '%s' not parsed" % self.MERGEINFO_SRC)
    self.assertEqual(len(rangelist), nbr_rev_ranges,
                      "Wrong number of revision ranges parsed")
    self.assertEqual(rangelist[0].inheritable, True,
                      "Unexpected revision range 'non-inheritable' flag: %s" %
                      rangelist[0].inheritable)
    self.assertEqual(rangelist[1].start, 26,
                      "Unexpected revision range end: %d" % rangelist[1].start)
    self.assertEqual(rangelist[2].inheritable, False,
                      "Missing revision range 'non-inheritable' flag")

  def compare_mergeinfo_catalogs(self, catalog1, catalog2):
    keys1 = sorted(catalog1.keys())
    keys2 = sorted(catalog2.keys())
    self.assertEqual(keys1, keys2)

    for k in catalog1:
        self.compare_mergeinfos(catalog1[k], catalog2[k])

  def compare_mergeinfos(self, mergeinfo1, mergeinfo2):
    keys1 = sorted(mergeinfo1.keys())
    keys2 = sorted(mergeinfo2.keys())
    self.assertEqual(keys1, keys2)

    for k in mergeinfo1:
        self.compare_rangelists(mergeinfo1[k], mergeinfo2[k])

  def compare_rangelists(self, rangelist1, rangelist2):
    self.assertEqual(len(rangelist1), len(rangelist2))
    for r1, r2 in zip(rangelist1, rangelist2):
        self.assertEqual(r1.start, r2.start)
        self.assertEqual(r1.end, r2.end)


def suite():
    return unittest.defaultTestLoader.loadTestsFromTestCase(
      SubversionMergeinfoTestCase)

if __name__ == '__main__':
    runner = unittest.TextTestRunner()
    runner.run(suite())
