# test_p4python.PY -- Various tests for p4python - python API for Perforce
#
# Robert Cowham, Vaccaperna Systems Ltd
#
# Note that this set of tests demonstrates a variety of ways to use P4Python.
#
# See the example config files config_<hostname>.py
#
# Please download a copy of the Perforce sample depot as available (and described
# in:
# http://www.perforce.com/perforce/technotes/note073.html 

#---Imports
import os
import sys

import copy
import imp
import popen2
import re
import socket
import string
import time
import types
import shutil
import tempfile
import getopt
import fileinput
import win32api
import unzip
import traceback

import unittest

# Add the nearby "code" directory to the module loading path to pick up p4
sys.path.insert(0, os.path.join('..', 'build', 'lib.win32-2.5'))
sys.path.insert(0, '..')

import p4       # Main class to test

#---Code
verbose = 0
slow = 1    

class TestCase(unittest.TestCase):
    def __call__(self, result=None):
        if result is None: result = self.defaultTestResult()
        self.result = result
        unittest.TestCase.__call__(self, result)

    def addFailure(self, msg):
        try:
            raise AssertionError, msg
        except AssertionError:
            self.result.addFailure(self,self._TestCase__exc_info())

hostname = string.lower(string.split(socket.gethostname(), '.')[0])
config_filename = 'config_' + hostname + '.py'
if not os.path.exists(config_filename):
    print "Could not find config file", config_filename
config_file = open(config_filename)
try:
    imp.load_source('config', config_filename, config_file)
finally:
    config_file.close()
original_configuration = copy.copy(sys.modules['config'].__dict__)
import config


# os.environ['PATH'] = os.environ['PATH'] + ";" + config.p4_path

# The default temporary file prefix starts with an '@'.  But
# that would mean that temporary files will look like revision
# specifications to Perforce.  So use a prefix that's acceptable
# to Perforce.
tempfile.gettempprefix = lambda: '%d.' % os.getpid()

# We need to log to a log file.  The P4Python log will get redirected to
# this file, as will the output of various commands.

log_filename = (time.strftime('P4Python.%Y%m%dT%H%M%S.log',
                                 time.gmtime(time.time())))
log_filename = os.path.abspath(log_filename)
log_file = open(log_filename, "a")

def log_exception():
    type, val, tb = sys.exc_info()
    log_message(string.join(traceback.format_exception(type, val, tb), ''))
    del type, val, tb
    
def log_message(msg):
    date = time.strftime('%Y-%m-%d %H:%M:%S UTC',
                         time.gmtime(time.time()))
    log_file.write("%s  %s\n" % (date, msg))
    log_file.flush()
    if verbose:
        print "%s  %s\n" % (date, msg)

sys.stdout.write("P4Python test suite, logging to %s\n" % log_filename)
sys.stdout.flush()

# Perforce
#
# This class supplies the restart_perforce method
# It also provides "system", which is like os.system,
# but captures output, checks for errors, and writes to the log.

class Perforce:

    # Temporary directory for Perforce server and associated files.
    p4dir = None

    # Stop Perforce server
    #
    # We used to be able to stop via "p4 admin stop", but due to password stuff, 
    # just kill the process.
    def stop_perforce(self):
        # Use sysinternals.com utility
        self.system('pskill "%s"' % config.p4_server_executable, ignore_failure = 0)
        
    # Start a new Perforce server
    # Make a completely fresh Perforce server, with a new repository.
    def start_perforce(self, clean=True):
        if clean:
            # Make a new repository directory.
            self.tempdir = tempfile.mktemp()
            os.mkdir(self.tempdir)
            log_message("Perforce repository directory %s." % self.tempdir)
            zipdest = os.path.join(self.tempdir, 'p4root')
            os.mkdir(zipdest)
            self.p4dir = os.path.join(zipdest, 'PerforceSample')
            os.mkdir(self.p4dir)
            
            # Copy the license file - no need with sample depot
            # if config.p4_license_file:
                # shutil.copyfile(config.p4_license_file,
                                # os.path.join(self.p4dir, 'license'))
    
            # Copy depot and checkpoint to appropriate place
            unzipper = unzip.unzip()
            unzipper.extract(config.sample_depot_zip, zipdest)
    
            # Recreate db files from checkpoint
            self.system('""%s"" -r %s -jr %s' %
                        (config.p4_server_path,
                         self.p4dir,
                         os.path.abspath(os.path.join(self.p4dir, 'checkpoint'))),
                         ignore_failure = 0)

            # Now change to use full path name
            self.tempdir = win32api.GetLongPathName(self.tempdir)
            log_message("Reset tempdir to '%s'." % (self.tempdir))
    
        # Work out Perforce's port number and start a Perforce server.
        match = re.match(".*:([0-9]+)$", config.p4_port)
        if match:
            port = int(match.group(1))
        else:
            port = 1999

        # p4d on Windows doesn't detach, to we can't use "system"
        # to start it.
        win32api.WinExec('"%s" -p 0.0.0.0:%d -r %s  -L p4d.log -vserver=3' %
                         (config.p4_server_path, port,
                          self.p4dir))
        time.sleep(2)

    # Restart Perforce
    #
    # By killing the old server and starting a new one.

    def restart_perforce(self, clean=False):
        self.stop_perforce()
        time.sleep(1)
        self.start_perforce(clean)
        time.sleep(2)


    # Run command and check results
    #
    # Calls an external program, raising an exception upon any error
    # and returning standard output and standard error from the program,
    # as well as writing them to the log.

    def system(self, command, ignore_failure = 0, input_lines = None):
        log_file.write('Executing %s\n' % repr(command))
        (child_stdin, child_stdout) = os.popen4(command)
        if input_lines:
            child_stdin.writelines(input_lines)
            child_stdin.close()
        output = child_stdout.read()
        result = child_stdout.close()
        log_file.write(output)
        if not ignore_failure and result:
            message = ('Command "%s" failed with result code %d.' %
                       (command, result))
            log_file.write(message + '\n')
            raise message
        return output
        
# P4Python TEST CASE BASE CLASS
#
# The P4python_base class is a generic test case.  It defines methods
# for setting up the integration.
#
# Other test cases will inherit from this class.
#
# When a class implements several test cases, the methods that implement
# test cases (in the PyUnit sense) should have names starting "test_".
# When a class implements a single test case, the method should be
# called "runTest".

class P4Python_base(TestCase):

    # Set up everything so that test cases can run
    #
    p4d = Perforce()
    p4api_ver = "2005.1"

    def setup_everything(self, config_changes = {}):
        for key, value in config_changes.items():
            setattr(config, key, value)
        self.p4d.restart_perforce(clean=True)

        # Get a permanent Perforce interface using the password.
        self.p4c = p4.P4()
        log_message("p4 api ver %s" % self.p4c.p4_api_ver)
        self.p4c.port = config.p4_port
        self.p4c.user = config.p4_user
        self.p4c.password = config.p4_password
        self.p4c.parse_forms()
        self.p4c.connect()
        
        # Create our client
        # Make a Perforce client.
        client = self.p4c.fetch_client(config.p4_client)
        self.p4_client_root = os.path.join(self.p4d.tempdir, config.p4_client)
        os.mkdir(self.p4_client_root)
        client["Root"] = self.p4_client_root
        self.p4c.save_client(client)
        log_message("Set root of client %s to '%s'." % (config.p4_client, client["Root"]))
        
        self.p4c.disconnect()
        self.p4c.client = config.p4_client
        self.p4c.connect()
        # result = self.p4c.run("sync", "-f")

        # write config file since now client root dir should be there
        p4config = []
        p4config.append("P4PORT=%s\n" % config.p4_port)
        p4config.append("P4USER=%s\n" % config.p4_user)
        p4config.append("P4CLIENT=%s\n" % config.p4_client)
        self.p4_config_file = os.path.join(self.p4_client_root, "p4config.txt")
        self.write_to_file(self.p4_config_file, p4config)
            
    def log_result(self, msg="", output=[]):
        type, val, tb = sys.exc_info()
        log_message(msg + '\n' + string.join(traceback.format_stack(limit=2), ''))
        if len(output) > 0:
            import pprint
            pp = pprint.PrettyPrinter(indent=4)
            log_message(pp.pformat(output))
        if len(self.p4.warnings) > 0:
            log_message("Warnings: " + "\n".join(self.p4.warnings))
        if len(self.p4.errors) > 0:
            log_message("Errors: " + "\n".join(self.p4.errors))
        
    def write_to_file(self, fname, results):
        temp_file = open(fname, "w")
        for l in results:
            temp_file.write(l)
        temp_file.close()

    def edit_file(self, filename):
        output = self.p4.run_where(filename)[0]
        localpath = output['path']
        
        f = open(localpath, 'r')
        lines = f.readlines()
        f.close()
        f = open(localpath, 'w')
        lines.insert(0, "A new line\n")
        f.writelines(lines)
        f.close()

    def setUp(self):
        self.setup_everything()
        self.p4 = p4.P4()
        self.p4.port = config.p4_port
        self.p4.user = config.p4_user
        self.p4.password = config.p4_password
        self.p4.client = config.p4_client

    def run_test(self):
        pass

#--- Test Cases

# Normal operation
class normal(P4Python_base):

    def setUp(self):
        super(normal, self).setUp()

    def runTest(self):
        "Test Normal operations including checkin"
        self.p4.exception_level = 0
        self.p4.connect()
        output = self.p4.run("info")
        for e in self.p4.errors:
            log_message("Error: " + e)
        log_message(output)

        output = self.p4.run("fstat", "//depot/Jam/MAIN/src/jam.c")
        self.assert_(len(self.p4.errors) == 0, "found some errors")
        self.assert_(len(output) == 1, "wrong number of results")
        self.assert_(output[0].has_key('depotFile'), "dict key not found")
        log_message(output)

        output = self.p4.run("fstat", "//depot/Jam/MAIN/src/jam.*")
        self.assertEquals(len(self.p4.errors), 0)
        self.assertEquals(len(output), 6)
        self.assert_(output[0].has_key('depotFile'))
        log_message(output)

        file = "//depot/Jam/MAIN/src/Jambase"
        output = self.p4.run("fstat", file)
        self.assertEquals(len(self.p4.errors), 0)
        self.assertEquals(len(output), 1)
        self.assertEquals(output[0]['depotFile'], file)
        # self.assert_(isinstance(output[0]['otherAction'], list))
        log_message(output)

        # Check for text and binary file contents
        output = self.p4.run("print", "-q", "//depot/Jam/MAIN/src/Build.com")
        self.log_result("print output", output)
        self.assertEquals(len(output), 32)
        self.assertEquals(output[0], "! Bootstrap build script for Jam")
        
        # For a binary file we dump output and compare with result of sync of same file
        bin_file = "//depot/Misc/manuals/triggers.doc"
        output = self.p4.run("sync", bin_file)
        output = self.p4.run("where", bin_file)
        self.log_result("where of %s" % bin_file, output)
        pat = re.compile('\S+ \S+ (.+)')
        m = pat.match(output[0])
        local_file = m.groups(0)[0]
        print_file = local_file + ".b.doc"
        log_message("Local syntax: %s" % local_file)
        output = self.p4.run("print", "-q", bin_file)
        temp_file = open(print_file, "wb")
        for l in output:
            temp_file.write(l)
        temp_file.close()

        result = self.p4d.system("fc /b \"%s\" \"%s\"" % (local_file, print_file))
        if not re.compile("FC: no differences encountered").search(result):
            self.fail("Expected no differences comparing %s and %s." % (local_file, print_file))

        self.p4.disconnect()
        self.p4.tagged()
        self.p4.connect()
        output = self.p4.run("files", "//depot/Jam/MAIN/src/jam.*")
        self.assert_(len(self.p4.errors) == 0, "found unexpected errors")
        self.assert_(len(output) == 6, "wrong number of results")
        self.assertEquals(output[0]['depotFile'], "//depot/Jam/MAIN/src/jam.1", "depotfile key not found")
        for key in ['rev', 'time', 'action', 'type', 'depotFile', 'change']:
            self.assert_(output[0].has_key(key), "expected key not present")
        log_message(output)

        # This is where things get more exciting with nested arrays etc.
        # Commented out until we can find equivalent files in the sampledepot.
        file = "//depot/www/review/index.html"
        output = self.p4.run("filelog", file)
        self.assertEquals(output[0]['rev'], 
                ['3', '2', '1'])
        self.assertEquals(output[0]['file'], 
                [['//depot/www/dev/index.html', '//depot/www/live/index.html'], 
                ['//depot/www/dev/index.html', '//depot/www/live/index.html'], 
                ['//depot/www/dev/index.html', '//depot/www/live/index.html']])
        self.assertEquals(output[0]['user'], 
                ['hera', 'hera', 'hera'])
        self.assertEquals(output[0]['srev'], 
                [['#2', '#2'], ['#1', '#1'], ['#none', '#none']])

        output = self.p4.run("describe", "-s", "712")
        self.assertEquals(output[0]['rev'], ['1', '2', '2', '2', '2', '2', '2'])
        self.assertEquals(output[0]['client'], 'quinn-dev-azalea')
        self.assertEquals(output[0]['depotFile'], 
            ['//depot/Jamgraph/MAIN/src/bin.ntx86/jamgraph.exe', 
            '//depot/Jamgraph/MAIN/src/gparticle.cpp', 
            '//depot/Jamgraph/MAIN/src/gparticle.h', 
            '//depot/Jamgraph/MAIN/src/gworld.cpp', 
            '//depot/Jamgraph/MAIN/src/gworld.h', 
            '//depot/Jamgraph/MAIN/src/jamgraph.vcproj', 
            '//depot/Jamgraph/MAIN/src/main.cpp'])
        self.assertEquals(output[0]['desc'], 
                'Jamgraph functional.  Usage:\n\n' +
                'jam -ndd | jamgraph\n\n' +
                'Still twiddling with the interface, so it\'s undocumented.\n')

        output = self.p4.run("describe", "-s", "709")
        self.assertEquals(output[0]['rev'], 
                ['134', '33'])
        self.assertEquals(output[0]['client'], 
                'dai-beos-locust')
        self.assertEquals(output[0]['depotFile'], 
                ['//depot/Jam/MAIN/src/Jambase', '//depot/Jam/MAIN/src/jambase.c'])
        self.assertEquals(output[0]['desc'], 
                'Upgrade to latest metrowerks on Beos -- the Intel one.\n')

        # Try special form of run
        output2 = self.p4.run_describe("-s", "709")
        self.assertEquals(output, output2)

        # Check for diff output
        output = self.p4.run("describe", "709")
        self.assertEquals(output[0]['rev'], 
                ['134', '33'])
        self.assertEquals(output[0]['client'], 
                'dai-beos-locust')
        self.assertEquals(output[0]['depotFile'], 
                ['//depot/Jam/MAIN/src/Jambase', '//depot/Jam/MAIN/src/jambase.c'])
        self.assertEquals(output[0]['desc'], 
                'Upgrade to latest metrowerks on Beos -- the Intel one.\n')
        self.log_result("describe output", output)
        
        # Check for diff2 working
        file1 = "//depot/Jam/MAIN/src/jam.c"
        file2 = "//depot/Jam/MAIN/src/jam.h"
        output = self.p4.run("diff2", file1, file2)[0]
        self.assertEquals(output["depotFile"], file1)
        self.assertEquals(output["depotFile2"], file2)
        self.assertEquals(output["status"], "content")
        
        file1 = "//depot/Jam/MAIN/src/Jambase"
        self.exception_level = 2
        output = self.p4.run_sync(file1)
        output = self.p4.run_edit(file1)
        output = self.p4.run("diff", file1)

        self.edit_file(file1)
        output = self.p4.run("diff", file1)
        self.log_result("diff output - api ver %s" % self.p4.p4_api_ver, output)
        if self.p4.p4_api_ver == "2005.1":
           self.assert_(re.search(file1, output[0], re.IGNORECASE), output)
        elif self.p4.p4_api_ver == "2005.2":
            self.assertEquals(file1, output[0]['depotFile'])
        self.assert_(len(output[1:]) > 0)
        
# Normal operation
class attributes(P4Python_base):

    def setUp(self):
        super(attributes, self).setUp()

    def runTest(self):
        "Test setting and getting of attributes"
        self.p4.exception_level = 2
        os.chdir(self.p4_client_root)
        self.p4.cwd = self.p4_client_root
        self.assertEquals(self.p4.config, self.p4_config_file)
        self.p4.connect()
        self.p4.prog = "p4python_test"
        self.p4.version = "1.0"
        output = self.p4.run("info")
        self.log_result("info", output)

        # Check you can't read ticket_file but you can set it
        try:
            fred = self.p4.ticket_file
        except AttributeError:
            pass
        else:
            self.fail("Didn't get AttributeError")

        self.p4.ticket_file = os.path.join(self.p4_client_root, "p4ticket.txt")
        
        # Try some unicode stuff
        # self.p4.cwd = unicode(self.p4_client_root)        
        
# Spec output
class specs(P4Python_base):

    def setUp(self):
        super(specs, self).setUp()

    def check_edit(self, file, output):
        self.log_result("edit", output)
        if self.p4.p4_api_ver == "2005.1":
            self.assert_(re.search(r"opened for edit", output[0]), output)
        elif self.p4.p4_api_ver == "2005.2":
            self.assertEquals(file, output[0]['depotFile'])
            
    def check_submit(self, output):
        self.log_result("submit", output)
        if self.p4.p4_api_ver == "2005.1":
            self.assert_(re.search("Change \d+.* submitted\.", output[-1]))
        elif self.p4.p4_api_ver == "2005.2":
            self.assert_(output[-1].has_key('submittedChange'))

    def check_diff(self, file, output):
        self.log_result("diff", output)
        if self.p4.p4_api_ver == "2005.1":
            self.assert_(re.search(file1, output[0], re.IGNORECASE))
        elif self.p4.p4_api_ver == "2005.2":
            self.assertEquals(output[0]['depotFile'], file)

    def runTest(self):
        "Test spec operations"
        self.p4.exception_level = 1
        self.p4.parse_forms()
        self.p4.connect()
        output = self.p4.run("info")

        output = self.p4.run("change", "-o", "709")
        self.assertEquals(output[0]['Change'], '709')
        self.assertEquals(output[0]['Client'], 'dai-beos-locust')
        self.assertEquals(output[0]['Date'], '2001/12/24 01:06:32')
        self.assertEquals(output[0]['Description'], 
                'Upgrade to latest metrowerks on Beos -- the Intel one.\n')

        # Check for diff output
        output = self.p4.run("describe", "709")
        self.log_result("describe output", output)

        file1 = "//depot/Jam/MAIN/src/jam.h"
        output = self.p4.run("edit", file1)
        self.log_result("edit", output)
        self.assert_(re.search(r"file\(s\) not on client", self.p4.warnings[0]))
        output = self.p4.run("sync", file1)
        self.log_result("sync", output)
        output = self.p4.run("edit", file1)
        self.check_edit(file1, output)
        
        submit = self.p4.run("change", "-o")[0]
        submit["Description"] = "Test submit"
        self.assertEquals(len(submit["Files"]), 1)
        self.p4.input = submit
        output = self.p4.run("submit", "-i")
        self.check_submit(output)
        
        # Now try the special options
        file1 = "//depot/Jam/MAIN/src/jam.c"
        file2 = "//depot/Jam/MAIN/src/jam.h"
        output = self.p4.run("sync", file1)
        output = self.p4.run("edit", file1)
        self.check_edit(file1, output)
        output = self.p4.run("sync", file2)
        output = self.p4.run("edit", file2)
        self.check_edit(file2, output)

        submit = self.p4.fetch_change()
        submit["Description"] = "Test submit"
        self.assertEquals(len(submit["Files"]), 2)
        output = self.p4.save_submit(submit)
        self.check_submit(output)
        
        client1 = self.p4.fetch_client()
        client3 = self.p4.fetch_client(config.p4_client)
        client2 = self.p4.run("client", "-o")[0]
        self.assertEquals(client1, client2)
        self.assertEquals(client1, client3)

        job_count = len(self.p4.run("jobs"))
        job = self.p4.fetch_job()
        job["Description"] = "my new job" # Note \n will be tacked on anyway
        self.p4.save_job(job)
        jobs = self.p4.run("jobs")
        job = self.p4.fetch_job("newjob")
        job["Description"] = "another new job\n"
        self.p4.save_job(job)
        new_jobs = self.p4.run("jobs")
        for j in new_jobs:
            if j["Job"] == job["Job"]:
                break
        self.assertEquals(j["Description"], job["Description"])
        self.assertEquals(j["Job"], "newjob")
        self.assertEquals(j["Job"], job["Job"])
        output = self.p4.run("job", "-d", "newjob")
        new_jobs = self.p4.run("jobs")
        self.assertEquals(jobs, new_jobs)
        
        for j in new_jobs:
            if j["Description"] in ["another new job\n", "my new job\n"]:
                output = self.p4.delete_job("-f", j["Job"])
                self.assert_(re.match(r"Job %s deleted." % j["Job"], output))
        jobs = self.p4.run("jobs")
        self.assertEquals(job_count, len(jobs))

        # Check for diff2 working
        file1 = "//depot/Jam/MAIN/src/jam.c"
        file2 = "//depot/Jam/MAIN/src/jam.h"
        output = self.p4.run("diff2", file1, file2)[0]
        self.assertEquals(output["depotFile"], file1)
        self.assertEquals(output["depotFile2"], file2)
        self.assertEquals(output["status"], "content")

        file1 = "//depot/Jam/MAIN/src/jamgram.h"
        self.exception_level = 2
        output = self.p4.run_sync(file1)
        output = self.p4.run_edit(file1)
        output = self.p4.run("diff", file1)

        self.edit_file(file1)
        output = self.p4.run("diff", file1)
        self.check_diff(file1, output)
        self.assert_(len(output[1:]) > 0)

        # Now check that we can set api back to 2005.1 even if using later API.
        self.p4.disconnect()
        self.p4.api("56")
        self.p4.connect()
        output = self.p4.run("diff", file1)
        self.log_result("diff", output)
        self.assert_(re.search(file1, output[0], re.IGNORECASE))
        
        
# Exceptions
class exceptions(P4Python_base):

    def setUp(self):
        super(exceptions, self).setUp()
        
    def runTest(self):
        "Test exceptions being generated for errors and warnings"
        try:
            output = self.p4.run("info")
        except p4.P4Error:
            log_exception()
            for e in self.p4.errors:
                log_message(e)
        else:
            fail("expected an error exception")

        self.p4.connect()
        try:
            output = self.p4.run("fred")
        except p4.P4Error:
            log_exception()
            for e in self.p4.errors:
                log_message(e)
        else:
            fail("expected an error exception")

        self.p4.disconnect()
        self.p4.exception_level = 0
        self.p4.connect()
        output = self.p4.run("fred")
        self.assert_(len(self.p4.errors) > 0, "no errors found")
        for e in self.p4.errors:
            log_message(e)

        self.p4.disconnect()
        self.p4.exception_level = 1 # Default
        self.p4.connect()
        # Repeated sync should generate a warning - but we don't expect exception
        try:
            output = self.p4.run_sync("//depot/Jam/MAIN/src/jam.c#0")
            output = self.p4.run_sync("//depot/Jam/MAIN/src/jam.c")
            output = self.p4.run_sync("//depot/Jam/MAIN/src/jam.c")
        except p4.P4Error:
            log_exception()
            for e in self.p4.errors:
                log_message("Error:" + e)
            for e in self.p4.warnings:
                log_message("Warning:" + e)
                log_message(e)
            self.fail("Didn't expect an exception")

        self.p4.disconnect()
        self.p4.exception_level = 2
        self.p4.connect()

        # Repeated sync should generate a warning - this time we expect exception
        try:
            output = self.p4.run_sync("//depot/Jam/MAIN/src/jam.c#0")
            output = self.p4.run_sync("//depot/Jam/MAIN/src/jam.c")
            output = self.p4.run_sync("//depot/Jam/MAIN/src/jam.c")
        except p4.P4Error:
            log_exception()
            for e in self.p4.errors:
                log_message("Error:" + e)
            for e in self.p4.warnings:
                log_message("Warning:" + e)
                log_message(e)
        else:
            self.fail("Expected an exception")
        
        # Check for correct exception being raised.
        try:
            fred = self.p4.bad_attribute
        except AttributeError:
            pass
        else:
            self.fail("Didn't get AttributeError")

        # Check for correct exception being raised.
        try:
            output = self.p4.run("info", {'key': 'value'})
        except:
            # print "Unexpected error:", sys.exc_info()[0]
            pass
        else:
            self.fail("Didn't get TypeError")

        # Check for correct exception being raised.
        try:
            fred = self.p4.user = ['user']
        except TypeError:
            pass
        else:
            self.fail("Didn't get TypeError")

        # try:
            # self.p4.run("admin", "stop")
            # time.sleep(2)
            # self.p4.disconnect()
        # except p4.P4Error, e:
            # pass
        # else:
            # self.fail("Didn't get P4Error")

        self.p4.exception_level = 2

        # Check for correct exception being raised if we can't connect
        # Do this test last!!
        try:
            self.p4.port = "192.168.255.255:2555"
            self.p4.connect()
        except p4.P4Error, e:
            pass
        else:
            self.fail("Didn't get TypeError")

# Ensure login works OK
class login(P4Python_base):

    def setUp(self):
        super(login, self).setUp()

    def _change_user(self, user):
        self.p4.disconnect()
        self.p4.user = user
        self.p4.connect()
        
    def _delete_user(self, new_user):
        user = self.p4.fetch_user(new_user)
        self.p4.save_user(user)
        self._change_user(config.p4_user)
        self.p4.login(config.p4_password)
        print self.p4.delete_user("-f", new_user)
        
    def runTest(self):
        "Test login"
        self.p4.exception_level = 1
        self.p4.connect()
        self.p4.run("counter", "-f", "security", "2")
        self.p4.disconnect()
        self.p4d.restart_perforce(clean=False)
        self.p4.connect()

        new_user = "bruno"
        
        self.p4.input = self.p4.password
        output = self.p4.run("passwd")
        self.log_result("Set passwd")

        output = self.p4.login(config.p4_password)
        self.log_result("Login")
        self.assert_(re.search("User \w* logged in.", output[0]), output[0])

        self.log_result("users", self.p4.run("users"))
        self.log_result("delete user", self.p4.delete_user("-f", new_user))
        
        output = self.p4.run("login", "-s")
        self.log_result("login -s", output)

        # Alternative forms of login
        output = self.p4.run("logout")
        self.assert_(re.search("User \w* logged out.", output[0]))
        self.p4.input = ""
        output = self.p4.login(config.p4_password)[0]
        self.assert_(re.search("User \w* logged in.", output), output)

        # Check we get an error if no password supplied
        output = self.p4.run("logout")[0]
        self.assert_(re.search("User \w* logged out.", output))
        self.p4.input = ""
        try:
            output = self.p4.login()
            self.log_result("login", output)
        except p4.P4Error:
            pass
        else:
            self.assert_(False, "Expected exception and didn't get one: " + output[0])

        self.p4.input = ""
        output = self.p4.run("login", "-s")
        self.log_result("login -s", output)
        self.assert_(re.search("User \w* ticket expires", output[0]))

        self.log_result("users", self.p4.run("users"))
        
        self.p4.disconnect()
        self.p4.parse_forms()
        self.p4.connect()
        
        # Tests to ensure passwd stuff is OK
        self._change_user(new_user)
        try:
            output = self.p4.run("login", "-s")
            self.log_result("Login status")
        except p4.P4Error, e:
            self.assert_(re.match("Password must be set before access can be granted.", str(e)) or \
                re.match("Perforce password \(P4PASSWD\) invalid or unset.", str(e)))
        else:
            self.assert_(False, "Expected exception and didn't get one: " + output[0])

        # self.p4.input = config.p4_password
        # output = self.p4.run("passwd")
        # self.log_result("passwd", output)
        # self.p4.input = config.p4_password
        # output = self.p4.run("login")
        # self.log_result("login", output)

        # self._delete_user(new_user)
        
        # self._change_user(new_user)
        # try:
            # output = self.p4.run("login", "-s")[0]
        # except p4.P4Error, e:
            # if re.match("Password must be set before access can be granted.", str(e)):
                # output = self.p4.passwd(config.p4_password)
            
# RUNNING THE TESTS

def tests():
    if len(sys.argv) == 1:
        suite = unittest.TestSuite()
        # Decide which tests to run 
        
        tests = [login, normal, specs, exceptions, attributes]
        # tests = [login]
        for t in tests:
            suite.addTest(t())
        return suite
    else:
        # Unfortunately the following doesn't work with pdb, but is OK otherwise
        loader = unittest.defaultTestLoader
        module = __import__('__main__')
        suite = loader.loadTestsFromName(sys.argv[1], module)
        return suite
        
def main():
    # Filter out our arguments before handing unknown ones on
    argv = []
    argv.append(sys.argv[0])
    try:                                
        opts, args = getopt.getopt(sys.argv[1:], "hv", 
                        ["help", "verbose"]) 
    except getopt.GetoptError:           
        print "Usage: -v/verbose"
        sys.exit(2)                     
        
    for opt, arg in opts:                
        if opt in ("-v", "--verbose"):      
            verbose = 1
            argv.append("-v")
        elif opt in ("-h", "--help"):
            argv.append("-h")   # pass it through
        else:   # If unknown pass them on
            argv.append(opt)
            argv.append(arg)
    # unittest.main(defaultTest="tests", argv=argv)
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(tests())

if __name__ == "__main__":
    main()
    
#
# $Id: //depot/projects/p4ofc/main/test/test_p4ofc.py#43 $