#!/usr/bin/env python

from MythTV import *
from time import mktime
import traceback
import weakref
import sys
import os

MythLog._setlevel('none')

DBCONN = {'DBHostName': 'sql',
          'DBPort': 3306,
          'DBName': 'mythconverg_trunk',
          'DBUserName': 'mythtv',
          'DBPassword': 'mythtv'}
BEIP = '127.0.0.1'
FRONTEND = ['127.0.0.1',6546]
ERRORLIST = []
OPTS = {}

class testgroup( object ):
    _current = []
    def __repr__(self):
        return '%s(%s)' % (self.name, self._children)
    def __init__(self, name=None):
        self._parent = self._current
        testgroup._current = self
        self._children = []
        self.name = name
        self._parent.append(self)
    def __call__(self):
        if self.name:
            pi('running %s' % self.name, newline=True, color='cyan')
            pi.raiseindent()
        for func in self._children: func()
        if self.name:
            pi.lowerindent()
    def append(self, func):
        self._children.append(func)
    def close(self):
        testgroup._current = self._parent
    @classmethod
    def getcurrent(cls):
        return cls._current

class Critical( Exception ): pass
class Skipped( Exception ): pass

class decor( object ):
    def __init__(self, func):
        self.func = func
        self.__doc__ = self.func.__doc__
        self.__name__ = self.func.__name__
        self.__module__ = self.func.__module__
    def __call__(self, *args, **kwargs):
        self.func(*args, **kwargs)

class required( decor ):
    def __call__(self, *args, **kwargs):
        try: self.func(*args, **kwargs)
        except: raise Critical

class requires( decor ):
    def __call__(self, *args, **kwargs):
        if not OPTS[self.name]:
            raise Skipped
        self.func(*args, **kwargs)

class requires_dbconn( requires ): name = 'dbconn'
class requires_beconn( requires ): name = 'beconn'
class requires_xmlconn( requires ): name = 'xmlconn'
class requires_feconn( requires ): name = 'feconn'
class requires_dbcache( requires ): name = 'dbcache'
class requires_becache( requires ): name = 'becache'
class requires_record_getall( requires ): name = 'record_getall'

class test( decor ):
    def __repr__(self):
        return self.__name__
    def __init__(self, func):
        decor.__init__(self, func)
        testgroup.getcurrent().append(self)
    def __call__(self, *args, **kwargs):
        res = None
        pi('running %s' % self.__name__[5:])
        try:
            res = self.func(*args, **kwargs)
            pi('PASSED', ralign=True, color='green')
            OPTS[self.__name__[5:]] = True
        except Skipped:
            pi('SKIPPED', ralign=True, color='yellow')
            OPTS[self.__name__[5:]] = False
        except Critical:
            pi('CRITICAL', ralign=True, color='magenta')
            ERRORLIST.append((self.__name__, traceback.format_exc()))
            raise
        except:
            pi('FAILED', ralign=True, color='red')
            ERRORLIST.append((self.__name__, traceback.format_exc()))
            OPTS[self.__name__[5:]] = False

class print_indent():
    width = 60
    colors = dict(zip(['black','red','green','yellow','blue','magenta','cyan','white'], range(8)))
    def __init__(self):
        self.col = 0
        self.indent = 0
    def __call__(self, text, newline=False, ralign=False, color=None):
        if self.col == 0:
            sys.stdout.write(' '*self.indent)
            self.col += self.indent
        size = len(text)
        if color is not None:
            text = '\033[3%dm%s\033[37m' % (self.colors[color], text)
        if ralign:
            newline = True
            sys.stdout.write(' '*(self.width-(self.col+size)))
        sys.stdout.write(text)
        self.col += size
        if newline:
            sys.stdout.write('\n')
            self.col = 0
    def raiseindent(self): self.indent += 4
    def lowerindent(self): self.indent = max(self.indent-4, 0)

runtests = testgroup()
#-----------------------------------------
# base classes
test_base = testgroup('base classes')

@test
@required
def test_orddict():
    od = OrdDict((('somedata',1),('otherdata',2)))
    #test attribute access
    od.somedata
    #test attribute set
    od.somedata = 2
    if not od.somedata == 2: raise Exception
    #test item set
    od['somedata'] = 3
    if not od['somedata'] == 3: raise Exception
    #test new attribute set
    od.moredata = 4
    if not od.moredata == 4: raise Exception
    #test iterators
    if ['somedata','otherdata','moredata'] != od.keys(): raise Exception(od.keys())
    if [3,2,4] != od.values(): raise Exception(od.values())
    #test copying
    od2 = od.copy()
    if not od2.moredata == 4: raise Exception

@test
def test_dictinvert():
    di,id = DictInvert.createPair({'somedata':1, 'otherdata':2})
    #test item set
    di['somedata'] = 3
    if not id[3] == 'somedata': raise Exception
    #test new item set
    id[4] = 'moredata'
    if not di['moredata'] == 4: raise Exception
    #test item delete
    del id[4]
    if 'moredata' in di: raise Exception

test_base.close()

#-----------------------------------------
# connections
test_connections = testgroup('connections')
test_base_connections = testgroup('base connections')

@test
def test_dbconn():
    db = DBConnection(DBCONN)
    c = db.cursor()
    c.execute("""SELECT 1""")
    if not c.fetchone()[0] == 1: raise Exception

@test
def test_beconn(): be = BEConnection(BEIP, 6543)

@test
def test_xmlconn():
    xml = XMLConnection(BEIP, port=6544)
    xml._query()

@test
def test_feconn():
    fe = FEConnection(FRONTEND[0], FRONTEND[1])

test_base_connections.close()
test_connection_caches = testgroup('connection caches')

@test
@requires_dbconn
def test_dbcache(): DBCache()

@test
@requires_dbconn
def test_dbcache_manual_args():
    move_dbfile()
    try: DBCache(args=DBCONN)
    except: raise
    finally: move_dbfile(True)

@test
@requires_dbconn
def test_dbcache_manual_kw():
    move_dbfile()
    try:
        DBCache(DBHostName=DBCONN['DBHostName'], DBName=DBCONN['DBName'],
                DBUserName=DBCONN['DBUserName'], DBPassword=DBCONN['DBPassword'])
    except: raise
    finally: move_dbfile(True)

@test
@requires_dbconn
def test_dbcache_upnp():
    move_dbfile()
    try: DBCache()
    except: raise
    finally: move_dbfile(True)

@test
@requires_dbconn
def test_dbcache_recurse():
    move_dbfile()
    try:
        db = DBCache(args=DBCONN)
        db2 = DBCache(db)
    except: raise
    finally: move_dbfile(True)

@test
@requires_dbcache
def test_becache():
    be = BECache()
    host = be.host
    hostname = be.hostname
    BECache(host)
    BECache(hostname)

@test
@requires_feconn
def test_feconn_upnp():
    fe = list(FEConnection.fromUPNP())

test_connection_caches.close()
test_connections.close()

test_collections = testgroup('collections')
test_data_objects = testgroup('database objects')

test_record = testgroup('Record')

@test
def test_record_getall():
    recs = list(Record.getAllEntries())

@test
@requires_record_getall
def test_record_openone():
    rec = Record.getAllEntries().next()
    rec2 = Record(rec.recordid)

test_record.close()

@test
def test_recorded(): pass

@test
def test_recordedprogram(): pass

@test
def test_oldrecorded(): pass

@test
def test_job(): pass

@test
def test_guide(): pass

@test
def test_channel(): pass

test_data_objects.close()

test_method_collections = testgroup('method collections')
test_mythbe = testgroup('MythBE')

@test
@requires_becache
def test_mythbe_getrecordings():
    be = MythBE()
    recs = be.getRecordings()
    rec = be.getRecording(recs[0].chanid, int(mktime(recs[0].starttime.timetuple())))
    be.getPendingRecordings()
    be.getUpcomingRecordings()
    be.getScheduledRecordings()
    be.getConflictedRecordings()

test_mythbe.close()

test_mythdb = testgroup('MythDB')

@test
@requires_dbcache
def test_searchRecorded():
    db = MythDB()
    list(db.searchRecorded())
    list(db.searchRecorded(commflagged=True))

@test
@requires_dbcache
def test_searchOldRecorded():
    db = MythDB()
    list(db.searchRecorded())
    list(db.searchRecorded(commflagged=True))

@test
@requires_dbcache
def test_searchJobs():
    db = MythDB()
    list(db.searchRecorded())
    list(db.searchRecorded(commflagged=True))

@test
@requires_dbcache
def test_searchGuide():
    db = MythDB()
    list(db.searchRecorded())
    list(db.searchRecorded(commflagged=True))

@test
@requires_dbcache
def test_searchRecord():
    db = MythDB()
    list(db.searchRecorded())
    list(db.searchRecorded(commflagged=True))

test_mythdb.close()


def test_mythdb(): db = MythDB()

@test
@requires_dbcache
def test_mythxml(): xml = MythXML()

@test
@requires_dbcache
def test_mythvideo(): mvid = MythVideo()

test_method_collections.close()

#--------------------------------------
# data objects

@test
def test_program(): pass

@test
def test_record(): pass

@test
def test_recorded(): pass

@test
def test_recordedprogram(): pass

@test
def test_oldrecorded(): pass

@test
def test_job(): pass

@test
def test_guide(): pass

@test
def test_channel(): pass

@test
def test_video(): pass

test_collections.close()

#--------------------------------------
#

def finalprint():
    for fail in ERRORLIST:
        print '\n---------------------------------'
        print '%s Failed:' % fail[0]
        print fail[1]
    sys.exit(0)

def move_dbfile(moveback=False):
    cred = os.path.expanduser('~/.mythtv/config.xml')
    back = os.path.expanduser('~/.mythtv/config.xml.bak')
    if moveback:
        os.rename(back,cred)
    else:
        os.rename(cred, back)

pi = print_indent()
if __name__ == '__main__':
    try:
        runtests()
    except: pass
    finalprint()