def test_all(self):
        '''entry point function'''
        self.description = "Checkin unit test"
        self.errors = []

        Batch()

        # FIXME: this is needed for the triggers to be registerd. These
        # triggers have nothing to do with the web
        from pyasm.web import WebInit
        WebInit().execute()


        test_env = UnittestEnvironment()
        test_env.create()
        self.transaction = Transaction.get(create=True)

        Project.set_project("unittest")
       
        try:
            #Command.execute_cmd(self)

            # undo the command
            #undo = UndoCmd()
            #undo.execute()
            self.execute()

        finally:
            self.transaction.rollback()
            test_env.delete()

    def test_all(my):
        
        my.transaction = Transaction.get(create=True)
        try:
            my.create_snapshot()

            my._test_base_alias()

            my._test_file_naming()
            my._test_file_naming_base()
            my._test_dir_naming()
            # this comes after test_dir_naming so the file_object doesn't get polluted
            my._test_file_naming_manual_version()
            my._test_get_naming()
            my._test_checkin_type()
            my._test_naming_util()
        finally:
            my.transaction.rollback()
            Project.set_project('unittest')

            my.test_env.delete()
            my.sample3d_env.delete()

        # reset the unittest project type to whatever it was
        """

    def test_all(my):

        
        Batch()
        from pyasm.web.web_init import WebInit
        WebInit().execute()

        test_env = UnittestEnvironment()
        test_env.create()
        my.transaction = Transaction.get(create=True)
        try:
            
            my.person = Person.create( "Unit", "Test",
                    "ComputerWorld", "Fake Unittest Person")
            my.search_type = my.person.get_search_type()
            my.search_id = my.person.get_id()
            my.context = "test"
            my.full_context = "test/subtest"

            my._test_pipeline()
            my._test_pipeline_connects()

            my._test_schema()
            my._test_snapshot()
            my._test_level()
            my._test_naming_util()
            my._test_sobject_hierarchy()

            my._test_add_tasks()
        finally:
            my.transaction.rollback()
            Project.set_project('unittest')

            test_env.delete()

    def _test_cache(self):
        from pyasm.common import Config

        Config.set_value("security", "mode", "cache", no_exception=True)
        #Config.set_value("security", "authenticate_class", "pyasm.security.authenticate_test.AutocreateAuthenticate", no_exception=True)
        Config.set_value("security", "authenticate_class", "pyasm.security.mms_authenticate.MMSAuthenticate", no_exception=True)
        mode = Config.get_value("security", "authenticate_class", use_cache=False)

        mode = Config.get_value("security", "mode", use_cache=False)
        self.assertEquals(mode, "cache")


        # verify that the user exists in the database
        search = Search("sthpw/login")
        search.add_filter("login", "foofoo")
        login = search.get_sobject()
        self.assertEquals(None, login)


        from pyasm.search import Transaction
        transaction = Transaction.get(create=True)
        transaction.start()

        self.security.login_user("foofoo", "wow")

        # verify that the user exists in the database
        search = Search("sthpw/login")
        search.add_filter("login", "foofoo")
        login = search.get_sobject()
        self.assertNotEquals(None, login)

    def test_all(self):
        
        Batch()
        from pyasm.web.web_init import WebInit
        WebInit().execute()

        test_env = UnittestEnvironment()
        test_env.create()
        self.transaction = Transaction.get(create=True)
        Project.set_project('unittest')
        try:
            self.person = Person.create( "Unit", "Test",
                        "ComputerWorld", "Fake Unittest Person")
            self._test_notification()

            self.transaction = Transaction.get(create=True)
            self._test_result()
        finally:
            self.transaction.rollback()
            test_env.delete()

        return

    def test_all(my):

        my.description = "Checkin unit test"
        my.errors = []

        Batch()
        Project.set_project("unittest")

        
        my.transaction = Transaction.get(create=True)
        try:
            my._test_create()
        finally:
            my.transaction.rollback()

    def test_all(my):
        Batch()
        from pyasm.web.web_init import WebInit
        WebInit().execute()

        test_env = UnittestEnvironment()
        test_env.create()
        my.transaction = Transaction.get(create=True)
        try:
            my._test_multiple_schema()
        finally:
            my.transaction.rollback()
            Project.set_project('unittest')

            test_env.delete()

def import_schema(plugin_code):
    from pyasm.search import Transaction
    transaction = Transaction.get(create=True)

    install_dir = Environment.get_install_dir()
    base_dir = Environment.get_plugin_dir()
    template_dir = "%s/%s" % (base_dir, plugin_code)
    manifest_path = "%s/manifest.xml" % (template_dir)
    print "Reading manifest: ", manifest_path

    xml = Xml()
    xml.read_file(manifest_path)

    # create a new project
    installer = PluginInstaller(base_dir=base_dir, manifest=xml.to_string() )
    installer.execute()

    def test_all(self):

        
        Batch()
        from pyasm.web.web_init import WebInit
        WebInit().execute()
    
        #TODO add the schema entry to the sample3d plugin first
        #sample3d_test_env = Sample3dEnvironment()
        #sample3d_test_env.create()

        test_env = UnittestEnvironment()
        test_env.create()
        self.transaction = Transaction.get(create=True)
        try:
            
            self.person = Person.create( "Unit", "Test",
                    "ComputerWorld", "Fake Unittest Person")
            self.search_type = self.person.get_search_type()
            self.search_id = self.person.get_id()
            self.search_code = self.person.get_value("code")
            self.context = "test"
            self.full_context = "test/subtest"
            self._test_instances()

            self._test_pipeline()
            self._test_pipeline_connects()

            self._test_schema()
            self._test_snapshot()
            self._test_level()
            self._test_naming_util()
            self._test_sobject_hierarchy()

            self._test_add_tasks()
            self._test_time()
        finally:
            self.transaction.rollback()
            Project.set_project('unittest')

            test_env.delete()

    def _test_autocreate(self):
        from pyasm.common import Config

        Config.set_value("security", "mode", "autocreate", no_exception=True)
        Config.set_value("security", "authenticate_class", "pyasm.security.authenticate_test.AutocreateAuthenticate", no_exception=True)

        mode = Config.get_value("security", "mode", use_cache=False)
        self.assertEquals(mode, "autocreate")



        # verify that the user exists in the database
        search = Search("sthpw/login")
        search.add_filter("login", "foofoo")
        login = search.get_sobject()
        self.assertEquals(None, login)


        from pyasm.search import Transaction
        transaction = Transaction.get(create=True)
        transaction.start()

        self.security.login_user("foofoo", "wow")

        # verify that the user exists in the database
        search = Search("sthpw/login")
        search.add_filter("login", "foofoo")
        login = search.get_sobject()
        self.assertNotEquals(None, login)

        email = login.get_value("email")
        self.assertEquals("[email protected]", email)

        transaction.rollback()

        # after rollback, this user should not exist
        search = Search("sthpw/login")
        search.add_filter("login", "foofoo")
        login = search.get_sobject()
        self.assertEquals(None, login)

    def log(cls, level, message, category="default"):
        assert level in ("critical", "error", "warning", "info", "debug")

        # record the exception
        user_name = Environment.get_user_name()
        if not user_name:
            user_name = "UNKNOWN"

        # put the debug in a completely separate transaction from the main
        # transaction
        transaction = Transaction.get(force=True)
        transaction.set_record(False)

        debug_log = SObjectFactory.create("sthpw/debug_log")
        debug_log.set_value("login", user_name)
        debug_log.set_value("level", level)
        debug_log.set_value("category", category)
        debug_log.set_value("message", message )
        debug_log.commit()

        transaction.commit()
        transaction.remove_from_stack()

        return debug_log

    def test_all(self):

        # set up some widget configs in the database

        self.transaction = Transaction.get(create=True)

        widget_config_type = 'config/widget_config'

        def_config = '''
        <config>
        <definition>
          <element name='test'>
            <display class='SelectWdg'/>
          </element>
        </definition>
        </config>
        '''

        config = '''
        <config>
        <test>
          <element name='dan'/>
          <element name='dan2'>
             <display class='SelectWdg'/>
          </element>
          <element name='test'/>
          <element name='test2'>
            <display class='CheckboxWdg'/>
          </element>
          <element name='drop'/>
        </test>
        </config>
        '''

        edit_config = '''
        <config>
        <edit layout='EditWdg'>
          <element name='asset_category'/>
          <element name='code'>
            <display class='EditLevelTextWdg'/>
          </element>
          <element name='drop'/>
        </edit>
        </config>
        '''
        edit_def_config = '''
        <config>
        <edit_definition>
        <element name='asset_category'>
          <display class='ProcessSelectWdg'/>
        </element>
            
        <element name='code'>
            <display class='TextAreaWdg'/>
          </element>
        <element name='drop'>
            <action class='DropElementAction'>
               <path>path_test</path>
            </action>   
        </element>
        </edit_definition>
        </config>
        '''

        ALL_def_config = '''
        <config>
        <definition>
          <element name='dan'>
            <display class='tactic.ui.table.ExpressionElementWdg'/>
          </element>
        </definition>
        </config>
        '''

        try:
            # add the ALL definition view
            widget_config = SearchType.create(widget_config_type)
            widget_config.set_value("search_type", "ALL")
            widget_config.set_value("view", "definition")
            widget_config.set_value("config", ALL_def_config)
            widget_config.commit()

            # add the definition view
            widget_config = SearchType.create(widget_config_type)
            widget_config.set_value("search_type", "unittest/person")
            widget_config.set_value("view", "definition")
            widget_config.set_value("config", def_config)
            widget_config.commit()

            # add the test view for person
            widget_config = SearchType.create(widget_config_type)
            widget_config.set_value("search_type", "unittest/person")
            widget_config.set_value("view", "test")
            widget_config.set_value("config", config)
            widget_config.commit()

            # add the test view for city
            widget_config = SearchType.create(widget_config_type)
            widget_config.set_value("search_type", "unittest/city")
            widget_config.set_value("view", "test")
#.........这里部分代码省略.........

    def postprocess(self):
        web = get_web_container()

        keys = web.get_form_keys()

        from pyasm.search import Transaction
        transaction = Transaction.get()
        assert transaction

        # first get some data based in
        column = self.get_value("%s|column" % self.name)
        if column == "":
            column = self.name
       
        # NOTE: why did this change?
        #prefix = self.get_input_name()
        prefix = self.get_name()
        
        
        context = self.get_value("%s|context" % prefix)
        description = self.get_value("%s|description" % prefix)
        
        field_storage = self.get_value(prefix)
        handoff_path = self.get_value("%s|path" % prefix )
        custom_ticket = self.get_value("%s|ticket" % prefix )

        from pyasm.widget import CheckboxWdg
        cb = CheckboxWdg("%s|is_revision" % prefix)
        is_rev = cb.is_checked()

        if handoff_path:
            handoff_path = handoff_path.replace("\\", "/")

            # This check causes issues.. Just assume it's in the upload location
            #if not os.path.exists(handoff_path):
            security = Environment.get_security()
            ticket = security.get_ticket_key()

            # in case it's supplied by widget like SimpleUploadWdg
            if custom_ticket:
                ticket = custom_ticket

            handoff_path = os.path.basename(handoff_path)
            handoff_path = Common.get_filesystem_name(handoff_path)

            handoff_path = "%s/upload/%s/%s" % (Environment.get_tmp_dir(), ticket, handoff_path)


            print "Uploaded path: ", handoff_path
            if not os.path.exists(handoff_path):
                raise Exception("Uploaded Path [%s] does not exist" % handoff_path)

            self.files = [handoff_path]
            file_types = ['main']

            # create an icon
            icon_creator = IconCreator(handoff_path)
            icon_creator.execute()
            icon_path = icon_creator.get_web_path()
            if icon_path:
                self.files.append(icon_path)
                file_types.append("icon")
            web_path = icon_creator.get_icon_path()
            if web_path:
                self.files.append(web_path)
                file_types.append("web")


        elif field_storage != "":
        #else:
            # process and get the uploaded files
            upload = FileUpload()
            upload.set_field_storage(field_storage)
            upload.execute()

            # get files and file types
            self.files = upload.get_files()
            if not self.files:
                return
            file_types = upload.get_file_types()
        else:
            if self.get_option("file_required") == "true":
                err_msg = _("upload is required")
                raise TacticException("%s %s" % (self.name, err_msg))
            else:
                return


        checkin_class = self.get_option("checkin")
        if checkin_class:
            snapshot_type = self.get_option("snapshot_type")
            self.checkin = Common.create_from_class_path(checkin_class, [self.sobject, self.files, file_types, context, snapshot_type])

        else:
            from pyasm.checkin import FileCheckin
            self.checkin = FileCheckin.get( self.sobject, self.files, file_types,  \
                context=context, column=column, snapshot_type="file" )

        self.sobjects.append(self.sobject)

#.........这里部分代码省略.........

    def execute(my):

        file_path = my.kwargs.get("path")
        project_code = my.kwargs.get("project_code")
        base_dir = my.kwargs.get("base_dir")
        search_type = my.kwargs.get("search_type")
        process = my.kwargs.get("process")
        watch_script_path =  my.kwargs.get("script_path")
        if not process:
            process = "publish"

        basename = os.path.basename(file_path)

        context = my.kwargs.get("context")
        if not context:
            context = '%s/%s'  % (process, basename)


        # find the relative_dir and relative_path
        relative_path = file_path.replace("%s/" % base_dir, "")
        relative_dir = os.path.dirname(relative_path)

        file_name = os.path.basename(file_path)
        log_path = '%s/TACTIC_log.txt' %(base_dir)
        my.create_checkin_log()

        # Define asset type of the file
        asset_type = my.get_asset_type(file_path)
        description = "drop folder check-in of %s" %file_name

        from client.tactic_client_lib import TacticServerStub
        server = TacticServerStub.get(protocol='local')
        server.set_project(project_code)

        transaction = Transaction.get(create=True)
        server.start(title='Check-in of media', description='Check-in of media')

        server_return_value = {}

        try:
            filters = [
                    [ 'name', '=', file_name ],
                    #[ 'relative_dir', '=', relative_dir ]
                ]
            sobj = server.query(search_type, filters=filters, single=True)

            if not sobj:
                # create sobject if it does not yet exist
                sobj = SearchType.create(search_type)
                if SearchType.column_exists(search_type, "name"):
                    sobj.set_value("name", basename)
                if SearchType.column_exists(search_type, "media_type"):
                    sobj.set_value("media_type", asset_type)


                if SearchType.column_exists(search_type, "relative_dir"):
                    sobj.set_value("relative_dir", relative_dir)

                if SearchType.column_exists(search_type, "keywords"):
                    relative_path = relative_path
                    keywords = Common.get_keywords_from_path(relative_path)
                    keywords = " ".join( keywords )
                    sobj.set_value("keywords", keywords)

                sobj.commit()
                search_key = sobj.get_search_key()
            else:
                search_key = sobj.get("__search_key__")


            #task = server.create_task(sobj.get('__search_key__'),process='publish')
            #server.update(task, {'status': 'New'})


            server_return_value = server.simple_checkin(search_key,  context, file_path, description=description, mode='move')

            if watch_script_path:
                cmd = PythonCmd(script_path=watch_script_path,search_type=search_type,drop_path=file_path,search_key=search_key)
                cmd.execute()




            
        except Exception, e:
            print "Error occurred", e
            error_message=str(e)

            import traceback
            tb = sys.exc_info()[2]
            stacktrace = traceback.format_tb(tb)
            stacktrace_str = "".join(stacktrace)
            print "-"*50
            print stacktrace_str


            version_num='Error:'
            system_time=strftime("%Y/%m/%d %H:%M", gmtime())
            pre_log=file_name+(50-len(file_name))*' '+system_time+(33-len(system_time))*' '+version_num+(15-len(version_num))*' ' +error_message+'\n'\
                    + stacktrace_str + '\n' + watch_script_path
#.........这里部分代码省略.........

    def execute(self):

        file_path = self.kwargs.get("path")
        site = self.kwargs.get("site")
        project_code = self.kwargs.get("project_code")
        base_dir = self.kwargs.get("base_dir")
        search_type = self.kwargs.get("search_type")
        process = self.kwargs.get("process")
        watch_script_path = self.kwargs.get("script_path")
        if not process:
            process = "publish"

        basename = os.path.basename(file_path)

        context = self.kwargs.get("context")
        if not context:
            context = '%s/%s'  % (process, basename)


        # find the relative_dir and relative_path
        relative_path = file_path.replace("%s/" % base_dir, "")
        relative_dir = os.path.dirname(relative_path)

        file_name = os.path.basename(file_path)
        log_path = '%s/TACTIC_log.txt' %(base_dir)
        self.create_checkin_log()

        # Define asset type of the file
        asset_type = self.get_asset_type(file_path)
        description = "drop folder check-in of %s" %file_name

        from client.tactic_client_lib import TacticServerStub
        server = TacticServerStub.get(protocol='local')
        server.set_project(project_code)

        transaction = Transaction.get(create=True)
        server.start(title='Check-in of media', description='Check-in of media')

        server_return_value = {}

        try:
            filters = [
                    [ 'name', '=', file_name ],
                    #[ 'relative_dir', '=', relative_dir ]
                ]
            sobj = server.query(search_type, filters=filters, single=True)

            if not sobj:
                # create sobject if it does not yet exist
                sobj = SearchType.create(search_type)
                if SearchType.column_exists(search_type, "name"):
                    sobj.set_value("name", basename)
                if SearchType.column_exists(search_type, "media_type"):
                    sobj.set_value("media_type", asset_type)


                if SearchType.column_exists(search_type, "relative_dir"):
                    sobj.set_value("relative_dir", relative_dir)

                if SearchType.column_exists(search_type, "keywords"):
                    relative_path = relative_path
                    keywords = Common.extract_keywords_from_path(relative_path)
                    keywords = " ".join( keywords )
                    sobj.set_value("keywords", keywords)

                sobj.commit()
                search_key = sobj.get_search_key()
            else:
                search_key = sobj.get("__search_key__")


            #task = server.create_task(sobj.get('__search_key__'),process='publish')
            #server.update(task, {'status': 'New'})
            
            """
            #TEST: simulate different check-in duration
            from random import randint
            sec = randint(1, 5)
            print "checking in for ", sec, "sec"
            server.eval("@SOBJECT(sthpw/login)")
            import shutil
            dir_name,base_name = os.path.split(file_path)
            dest_dir = 'C:/ProgramData/Southpaw/watch_temp'
            if not os.path.exists(dest_dir):
                os.makedirs(dest_dir)
            shutil.move(file_path, '%s/%s'%(dest_dir, base_name))
            time.sleep(sec)
            # move back the file in a few seconds 
            shutil.move('%s/%s'%(dest_dir, base_name), file_path)
            """
            server_return_value = server.simple_checkin(search_key,  context, file_path, description=description, mode='move')

            if watch_script_path:
                cmd = PythonCmd(script_path=watch_script_path,search_type=search_type,drop_path=file_path,search_key=search_key)
                cmd.execute()



            
        except Exception as e:
#.........这里部分代码省略.........