# Path: tests/util.py
# def check_call(cmd, expected_retcode=0, stdin_str="", **kwargs):
# def DataPath(path):
# def GetDataPathContents():
#   def assertMatchesRegex(self, regex, string):
#   def setUp(self):
#   def tearDown(self):
#   def setUp(self):
#   def GetExamplePath(self, name):
#   def GetTestDataPath(self, *path):
#   def GetPath(self, *path):
#   def setUp(self):
#   def tearDown(self):
#   def CheckCallWithPath(self, cmd, expected_retcode=0, stdin_str=""):
#   def ConvertZipToDict(self, zip):
#   def ConvertDictToZip(self, dict):
#   def setUp(self):
#   def tearDown(self):
#   def setUp(self):
#   def MakeLoaderAndLoad(self,
#                         problems=None,
#                         extra_validation=True,
#                         gtfs_factory=None):
#   def AppendToArchiveContents(self, arcname, s):
#   def SetArchiveContents(self, arcname, contents):
#   def GetArchiveContents(self, arcname):
#   def RemoveArchive(self, arcname):
#   def GetArchiveNames(self):
#   def CreateZip(self):
#   def DumpZipFile(self, zf):
#   def setUp(self):
#   def Load(self, feed_name):
#   def ExpectInvalidValue(self, feed_name, column_name):
#   def ExpectMissingFile(self, feed_name, file_name):
#   def setUp(self):
#   def tearDown(self):
#   def ExpectNoProblems(self, object):
#   def ValidateAndExpectMissingValue(self, object, column_name):
#   def ExpectMissingValueInClosure(self, column_name, c):
#   def ValidateAndExpectInvalidValue(self, object, column_name,
#                                     value=INVALID_VALUE):
#   def ExpectInvalidValueInClosure(self, column_name, value=INVALID_VALUE,
#                                   c=None):
#   def ValidateAndExpectInvalidFloatValue(self, object, value):
#   def ValidateAndExpectOtherProblem(self, object):
#   def ExpectOtherProblemInClosure(self, c):
#   def ValidateAndExpectDateOutsideValidRange(self, object, column_name,
#                                              value=INVALID_VALUE):
#   def ExpectException(self, type_name, column_name=None, value=INVALID_VALUE):
#   def SimpleSchedule(self):
#   def __init__(self, test_case, ignore_types=None):
#   def _Report(self, e):
#   def PopException(self, type_name):
#   def FormatException(self, exce, tb):
#   def TearDownAssertNoMoreExceptions(self):
#   def AssertNoMoreExceptions(self):
#   def PopColumnSpecificException(self, type_name, column_name, file_name=None):
#   def PopInvalidValue(self, column_name, file_name=None):
#   def PopMissingValue(self, column_name, file_name=None):
#   def PopDateOutsideValidRange(self, column_name, file_name=None):
#   def PopDuplicateColumn(self, file_name, header, count):
#   def _SortExceptionGroups(self):
#     def ProcessExceptionGroup():
#   def __init__(self, test_case, ignore_types=("ExpirationDate",)):
#   def _Report(self, e):
# def GetTestFailureProblemReporter(test_case,
#                                   ignore_types=("ExpirationDate",)):
#   def __init__(self):
#   def ExpirationDate(self, expiration, context=None):
# class TestCase(unittest.TestCase):
# class RedirectStdOutTestCaseBase(TestCase):
# class GetPathTestCase(TestCase):
# class TempDirTestCaseBase(GetPathTestCase):
# class TempFileTestCaseBase(TestCase):
# class MemoryZipTestCase(TestCase):
# class LoadTestCase(TestCase):
# class ValidationTestCase(TestCase):
# class RecordingProblemAccumulator(transitfeed.ProblemAccumulatorInterface):
# class TestFailureProblemAccumulator(transitfeed.ProblemAccumulatorInterface):
# class ExceptionProblemReporterNoExpiration(transitfeed.ProblemReporter):
#   _IGNORE_TYPES = ["ExpirationDate"]
# INVALID_VALUE = Exception()
# 
# Path: transitfeed/compat.py



import merge
import os.path
import re
import transitfeed
import unittest
import zipfile
from tests import util
from transitfeed.compat import StringIO
    b_schedule = transitfeed.Schedule()
    merged_schedule = transitfeed.Schedule()
    self.feed_merger = merge.FeedMerger(a_schedule, b_schedule,
                                        merged_schedule,
                                        self.problem_reporter)
    self.dataset_merger = merge.TripMerger(None)

  def testGeneratesSomeHTML(self):
    self.problem_reporter.CalendarsNotDisjoint(self.dataset_merger)
    self.problem_reporter.MergeNotImplemented(self.dataset_merger)
    self.problem_reporter.FareRulesBroken(self.dataset_merger)
    self.problem_reporter.SameIdButNotMerged(self.dataset_merger,
                                             'test', 'unknown reason')

    output_file = StringIO()
    old_feed_path = '/path/to/old/feed'
    new_feed_path = '/path/to/new/feed'
    merged_feed_path = '/path/to/merged/feed'
    self.accumulator.WriteOutput(output_file, self.feed_merger,
                                 old_feed_path, new_feed_path,
                                 merged_feed_path)

    html = output_file.getvalue()
    self.assert_(html.startswith('<html>'))
    self.assert_(html.endswith('</html>'))


class MergeInSubprocessTestCase(util.TempDirTestCaseBase):
  def CopyAndModifyTestData(self, zip_path, modify_file, old, new):
    """Return path of zip_path copy with old replaced by new in modify_file."""

# Path: tests/util.py
# def check_call(cmd, expected_retcode=0, stdin_str="", **kwargs):
# def DataPath(path):
# def GetDataPathContents():
#   def assertMatchesRegex(self, regex, string):
#   def setUp(self):
#   def tearDown(self):
#   def setUp(self):
#   def GetExamplePath(self, name):
#   def GetTestDataPath(self, *path):
#   def GetPath(self, *path):
#   def setUp(self):
#   def tearDown(self):
#   def CheckCallWithPath(self, cmd, expected_retcode=0, stdin_str=""):
#   def ConvertZipToDict(self, zip):
#   def ConvertDictToZip(self, dict):
#   def setUp(self):
#   def tearDown(self):
#   def setUp(self):
#   def MakeLoaderAndLoad(self,
#                         problems=None,
#                         extra_validation=True,
#                         gtfs_factory=None):
#   def AppendToArchiveContents(self, arcname, s):
#   def SetArchiveContents(self, arcname, contents):
#   def GetArchiveContents(self, arcname):
#   def RemoveArchive(self, arcname):
#   def GetArchiveNames(self):
#   def CreateZip(self):
#   def DumpZipFile(self, zf):
#   def setUp(self):
#   def Load(self, feed_name):
#   def ExpectInvalidValue(self, feed_name, column_name):
#   def ExpectMissingFile(self, feed_name, file_name):
#   def setUp(self):
#   def tearDown(self):
#   def ExpectNoProblems(self, object):
#   def ValidateAndExpectMissingValue(self, object, column_name):
#   def ExpectMissingValueInClosure(self, column_name, c):
#   def ValidateAndExpectInvalidValue(self, object, column_name,
#                                     value=INVALID_VALUE):
#   def ExpectInvalidValueInClosure(self, column_name, value=INVALID_VALUE,
#                                   c=None):
#   def ValidateAndExpectInvalidFloatValue(self, object, value):
#   def ValidateAndExpectOtherProblem(self, object):
#   def ExpectOtherProblemInClosure(self, c):
#   def ValidateAndExpectDateOutsideValidRange(self, object, column_name,
#                                              value=INVALID_VALUE):
#   def ExpectException(self, type_name, column_name=None, value=INVALID_VALUE):
#   def SimpleSchedule(self):
#   def __init__(self, test_case, ignore_types=None):
#   def _Report(self, e):
#   def PopException(self, type_name):
#   def FormatException(self, exce, tb):
#   def TearDownAssertNoMoreExceptions(self):
#   def AssertNoMoreExceptions(self):
#   def PopColumnSpecificException(self, type_name, column_name, file_name=None):
#   def PopInvalidValue(self, column_name, file_name=None):
#   def PopMissingValue(self, column_name, file_name=None):
#   def PopDateOutsideValidRange(self, column_name, file_name=None):
#   def PopDuplicateColumn(self, file_name, header, count):
#   def _SortExceptionGroups(self):
#     def ProcessExceptionGroup():
#   def __init__(self, test_case, ignore_types=("ExpirationDate",)):
#   def _Report(self, e):
# def GetTestFailureProblemReporter(test_case,
#                                   ignore_types=("ExpirationDate",)):
#   def __init__(self):
#   def ExpirationDate(self, expiration, context=None):
# class TestCase(unittest.TestCase):
# class RedirectStdOutTestCaseBase(TestCase):
# class GetPathTestCase(TestCase):
# class TempDirTestCaseBase(GetPathTestCase):
# class TempFileTestCaseBase(TestCase):
# class MemoryZipTestCase(TestCase):
# class LoadTestCase(TestCase):
# class ValidationTestCase(TestCase):
# class RecordingProblemAccumulator(transitfeed.ProblemAccumulatorInterface):
# class TestFailureProblemAccumulator(transitfeed.ProblemAccumulatorInterface):
# class ExceptionProblemReporterNoExpiration(transitfeed.ProblemReporter):
#   _IGNORE_TYPES = ["ExpirationDate"]
# INVALID_VALUE = Exception()
# 
# Path: transitfeed/compat.py



# Path: tests/testmerge.py
import merge
import os.path
import re
import transitfeed
import unittest
import zipfile
from tests import util
from transitfeed.compat import StringIO

    b_schedule = transitfeed.Schedule()
    merged_schedule = transitfeed.Schedule()
    self.feed_merger = merge.FeedMerger(a_schedule, b_schedule,
                                        merged_schedule,
                                        self.problem_reporter)
    self.dataset_merger = merge.TripMerger(None)

  def testGeneratesSomeHTML(self):
    self.problem_reporter.CalendarsNotDisjoint(self.dataset_merger)
    self.problem_reporter.MergeNotImplemented(self.dataset_merger)
    self.problem_reporter.FareRulesBroken(self.dataset_merger)
    self.problem_reporter.SameIdButNotMerged(self.dataset_merger,
                                             'test', 'unknown reason')

    output_file = StringIO()
    old_feed_path = '/path/to/old/feed'
    new_feed_path = '/path/to/new/feed'
    merged_feed_path = '/path/to/merged/feed'
    self.accumulator.WriteOutput(output_file, self.feed_merger,
                                 old_feed_path, new_feed_path,
                                 merged_feed_path)

    html = output_file.getvalue()
    self.assert_(html.startswith('<html>'))
    self.assert_(html.endswith('</html>'))


class MergeInSubprocessTestCase(util.TempDirTestCaseBase):
  def CopyAndModifyTestData(self, zip_path, modify_file, old, new):
    """Return path of zip_path copy with old replaced by new in modify_file."""

    zipfile_mem = StringIO(open(zip_path, 'rb').read())