class ProductMilestoneTestCase(MilestoneTestCase, MultiproductTestCase):
    def setUp(self):
        self.global_env = self._setup_test_env(create_folder=True)
        self._upgrade_mp(self.global_env)
        self._setup_test_log(self.global_env)
        self._load_product_from_data(self.global_env, self.default_product)

        self.env = ProductEnvironment(self.global_env, self.default_product)
        self._load_default_data(self.env)

    def tearDown(self):
        shutil.rmtree(self.global_env.path)
        self.global_env.reset_db()
        self.env = self.global_env = None

    def test_update_milestone(self):

        self.env.db_transaction("INSERT INTO milestone (name) VALUES ('Test')")

        milestone = Milestone(self.env, 'Test')
        t1 = datetime(2001, 01, 01, tzinfo=utc)
        t2 = datetime(2002, 02, 02, tzinfo=utc)
        milestone.due = t1
        milestone.completed = t2
        milestone.description = 'Foo bar'
        milestone.update()

        self.assertEqual(
            [('Test', to_utimestamp(t1), to_utimestamp(t2), 'Foo bar', 
                    self.default_product)],
            self.env.db_query("SELECT * FROM milestone WHERE name='Test'"))

class ProductAttachmentTestCase(AttachmentTestCase, MultiproductTestCase):

    def setUp(self):
        try:
            AttachmentTestCase.setUp(self)
        except:
            self.global_env = self.env
            self.tearDown()
            raise
        else:
            self.global_env = global_env = self.env
            self._upgrade_mp(global_env)
            self._setup_test_log(global_env)
            self._load_product_from_data(global_env, self.default_product)
            self.env = ProductEnvironment(global_env, self.default_product)

            # Root folder for default product environment
            self.attachments_dir = os.path.join(self.global_env.path,
                    'products', self.default_product, 'files', 'attachments')

    def tearDown(self):
        if os.path.exists(self.global_env.path):
            shutil.rmtree(self.global_env.path)
        self.env.reset_db()

    def test_product_path_isolation(self):
        product_attachment = Attachment(self.env, 'ticket', '42')
        global_attachment = Attachment(self.global_env, 'ticket', '42')
        global_attachment.filename = product_attachment.filename = 'foo.txt'

        self.assertNotEqual(product_attachment.path, global_attachment.path)

    def _get_product_info(self, product, href, resource, max_):
        penv = ProductEnvironment(self.env, product.prefix)
        results = []

        # some queries return a list/tuple, some a generator,
        # hence count() to get the result length
        def count(iter_):
            try:
                return len(iter_)
            except TypeError:
                return sum(1 for _ in iter_)

        query = resource['type'].select(penv)
        for q in itertools.islice(query, max_):
            q.url = href(resource['name'], q.name) \
                if resource.get('hrefurl') \
                else Query.from_string(penv,
                    '%s=%s&%s&col=%s' % (resource['name'], q.name,
                                         self.COMMON_QUERY, resource['name'])
            ).get_href(href)
            q.ticket_count = penv.db_query("""
                SELECT COUNT(*) FROM ticket WHERE ticket.%s='%s'
                AND ticket.status <> 'closed'
                """ % (resource['name'], q.name))[0][0]

            results.append(q)

        # add a '(No <milestone/component/version>)' entry if there are
        # tickets without an assigned resource in the product
        ticket_count = penv.db_query(
            """SELECT COUNT(*) FROM ticket WHERE %s=''
               AND status <> 'closed'""" % (resource['name'],))[0][0]
        if ticket_count != 0:
            q = resource['type'](penv)
            q.name = '(No %s)' % (resource['name'],)
            q.url = Query.from_string(penv,
               'status=!closed&col=id&col=summary&col=owner'
               '&col=status&col=priority&order=priority&%s='
               % (resource['name'],)
            ).get_href(href)
            q.ticket_count = ticket_count
            results.append(q)

        results.sort(key=lambda x: x.ticket_count, reverse=True)

        # add a link to the resource list if there are
        # more than max resources defined
        if count(query) > max_:
            q = resource['type'](penv)
            q.name = _('... more')
            q.ticket_count = None
            q.url = href(resource['name']) if resource.get('hrefurl') \
                else href.dashboard()
            results.append(q)

        return results

    def add_template_data(self, req, data, tickets):
        if isinstance(self.env, ProductEnvironment):
            super(ProductBatchModifyModule, self).add_template_data(req, data,
                                                                    tickets)
            return

        data['batch_modify'] = True
        data['query_href'] = req.session['query_href'] or req.href.query()

        tickets_by_product = {}
        for t in tickets:
            tickets_by_product.setdefault(t['product'], []).append(t)

        data['action_controls'] = []
        global_env = ProductEnvironment.lookup_global_env(self.env)
        cache = {}
        for k, v in tickets_by_product.iteritems():
            batch_module = cache.get(k or '')
            if batch_module is None:
                env = ProductEnvironment(global_env, k) if k else global_env
                cache[k] = batch_module = ProductBatchModifyModule(env)
            data['action_controls'] += batch_module._get_action_controls(req,
                                                                         v)
        batch_list_modes = [
            {'name': _("add"), 'value': "+"},
            {'name': _("remove"), 'value': "-"},
            {'name': _("add / remove"), 'value': "+-"},
            {'name': _("set to"), 'value': "="},
        ]
        add_script_data(req, batch_list_modes=batch_list_modes,
                        batch_list_properties=self._get_list_fields())

 def product_admincmd_mgr(self, prefix):
     try:
         product_env = ProductEnvironment.lookup_env(self.env, prefix)
     except LookupError:
         raise AdminCommandError('Unknown product %s' % (prefix,))
     else:
         return AdminCommandManager(product_env)

    def test_get_tickets(self):
        for pdata in (
            {'prefix': 'p2', 'name':'product, too', 'description': ''},
            {'prefix': 'p3', 'name':'strike three', 'description': ''},
        ):
            num_tickets = 5
            product = Product(self.global_env)
            product._data.update(pdata)
            product.insert()
            self.env = ProductEnvironment(self.global_env, product)
            for i in range(num_tickets):
                ticket = Ticket(self.env)
                ticket['summary'] = 'hello ticket #%s-%d' % (product.prefix, i)
                ticket['reporter'] = 'admin'
                tid = ticket.insert()

            # retrieve tickets using both global and product scope
            tickets_from_global = [(t['product'], t['id']) for t in
                Product.get_tickets(self.global_env, product.prefix)]
            self.assertEqual(len(tickets_from_global), num_tickets)
            tickets_from_product = [(t['product'], t['id']) for t in
                Product.get_tickets(self.env)]
            self.assertEqual(len(tickets_from_product), num_tickets)
            # both lists should contain same elements
            intersection = set(tickets_from_global) & set(tickets_from_product)
            self.assertEqual(len(intersection), num_tickets)

    def setUp(self):
        self.global_env = self._setup_test_env(create_folder=True)
        self._upgrade_mp(self.global_env)
        self._setup_test_log(self.global_env)
        self._load_product_from_data(self.global_env, self.default_product)

        self.env = ProductEnvironment(self.global_env, self.default_product)
        self._load_default_data(self.env)

class ProductResourceTestCase(MultiproductTestCase):
    def setUp(self):
        self._mp_setup()
        self.global_env = self.env
        self._load_product_from_data(self.global_env, u'xü')

        self.env = ProductEnvironment(self.global_env, self.default_product)
        self.env1 = ProductEnvironment(self.global_env, u'xü')

        self._load_default_data(self.global_env)
        self._load_default_data(self.env1)

        # Enable product system component in product context
        self.env.enable_component(MultiProductSystem)

    def tearDown(self):
        self.global_env.reset_db()
        self.global_env = self.env = None

 def setUp(self):
     self._mp_setup()
     self.env.path = '/path/to/env'
     self.env.abs_href = Href('http://globalenv.com/trac.cgi')
     url_pattern = getattr(getattr(self, self._testMethodName).im_func,
                           'product_base_url', '')
     self.env.config.set('multiproduct', 'product_base_url', url_pattern)
     self.env.config.set('trac', 'base_url', 'http://globalenv.com/trac.cgi')
     self.product_env = ProductEnvironment(self.env, self.default_product)

    def post_process_request(self, req, template, data, content_type):
        """Append necessary ticket data
        """
        try:
            tm = self._get_ticket_module()
        except TracError:
            # no ticket module so no create ticket button
            return template, data, content_type

        if (template, data, content_type) != (None,) * 3:  # TODO: Check !
            if data is None:
                data = {}
            dum_req = dummy_request(self.env)
            dum_req.perm = req.perm
            ticket = Ticket(self.env)
            tm._populate(dum_req, ticket, False)
            all_fields = dict([f['name'], f]
                              for f in tm._prepare_fields(dum_req, ticket)
                              if f['type'] == 'select')

            product_field = all_fields.get('product')
            if product_field:
                # When at product scope, set the default selection to the
                # product at current scope. When at global scope the default
                # selection is determined by [ticket] default_product
                if self.env.product and \
                        self.env.product.prefix in product_field['options']:
                    product_field['value'] = self.env.product.prefix
                # Transform the options field to dictionary of product
                # attributes and filter out products for which user doesn't
                #  have TICKET_CREATE permission
                product_field['options'] = [
                    dict(value=p,
                         new_ticket_url=dum_req.href.products(p, 'newticket'),
                         description=ProductEnvironment.lookup_env(self.env, p)
                                                       .product.name
                    )
                for p in product_field['options']
                    if req.perm.has_permission('TICKET_CREATE',
                                               Neighborhood('product', p)
                                               .child(None, None))]
            else:
                msg = _("Missing ticket field '%(field)s'.", field='product')
                if ProductTicketModule is not None and \
                        self.env[ProductTicketModule] is not None:
                    # Display warning alert to users
                    add_warning(req, msg)
                else:
                    # Include message in logs since this might be a failure
                    self.log.warning(msg)
            data['qct'] = {
                'fields': [all_fields[k] for k in self.qct_fields
                           if k in all_fields],
                'hidden_fields': [all_fields[k] for k in all_fields.keys()
                                  if k not in self.qct_fields] }
        return template, data, content_type

    def setUp(self):
        self._mp_setup()
        self.global_env = self.env
        self.env = ProductEnvironment(self.global_env, self.default_product)

        self.global_env.enable_component(TestRequestSpy)
        self.env.enable_component(TestRequestSpy)
        TestRequestSpy(self.global_env).testMatch = self._assert_product_match
        PermissionSystem(self.global_env).grant_permission('testuser', 'PRODUCT_CREATE')
        PermissionSystem(self.global_env).grant_permission('testuser', 'PRODUCT_VIEW')
        PermissionSystem(self.global_env).grant_permission('testuser', 'PRODUCT_MODIFY')

    def post_process_request(self, req, template, data, content_type):
        """Append necessary ticket data
        """
        try:
            tm = self._get_ticket_module()
        except TracError:
            # no ticket module so no create ticket button
            return template, data, content_type

        if (template, data, content_type) != (None,) * 3:  # TODO: Check !
            if data is None:
                data = {}
            req = dummy_request(self.env)
            ticket = Ticket(self.env)
            tm._populate(req, ticket, False)
            all_fields = dict([f['name'], f]
                              for f in tm._prepare_fields(req, ticket)
                              if f['type'] == 'select')

            product_field = all_fields['product']
            if product_field:
                if self.env.product:
                    product_field['value'] = self.env.product.prefix
                else:
                    # Global scope, now check default_product_prefix is valid
                    default_prefix = self.config.get('multiproduct',
                                                     'default_product_prefix')
                    try:
                        ProductEnvironment.lookup_env(self.env, default_prefix)
                    except LookupError:
                        product_field['value'] = product_field['options'][0]
                    else:
                        product_field['value'] = default_prefix

            data['qct'] = {
                'fields': [all_fields[k] for k in self.qct_fields
                           if k in all_fields],
                'hidden_fields': [all_fields[k] for k in all_fields.keys()
                                  if k not in self.qct_fields]
            }
        return template, data, content_type

    def setUp(self):
        self._mp_setup()
        self.global_env = self.env
        self._load_product_from_data(self.global_env, u'xü')

        self.env = ProductEnvironment(self.global_env, self.default_product)
        self.env1 = ProductEnvironment(self.global_env, u'xü')

        self._load_default_data(self.global_env)
        self._load_default_data(self.env1)

        # Enable product system component in product context
        self.env.enable_component(MultiProductSystem)

    def _process_doc(self, doc):
        ui_doc = dict(doc)

        if doc.get('product'):
            env = ProductEnvironment(self.env, doc['product'])
            product_href = ProductEnvironment.resolve_href(env, self.env)
            ui_doc["href"] = product_href(doc['type'], doc['id'])
        else:
            ui_doc["href"] = self.env.href(doc['type'], doc['id'])

        ui_doc['title'] = str(doc['type'] + ": " + doc['_stored_name']).title()

        return ui_doc

    def setUp(self):
        try:
            AttachmentTestCase.setUp(self)
        except:
            self.global_env = self.env
            self.tearDown()
            raise
        else:
            self.global_env = global_env = self.env
            self._upgrade_mp(global_env)
            self._setup_test_log(global_env)
            self._load_product_from_data(global_env, self.default_product)
            self.env = ProductEnvironment(global_env, self.default_product)

            # Root folder for default product environment
            self.attachments_dir = os.path.join(self.global_env.path,
                    'products', self.default_product, 'files', 'attachments')

    def _process_doc(self, doc):
        ui_doc = dict(doc)
        if doc.get('product'):
            env = ProductEnvironment(self.env, doc['product'])
            product_href = ProductEnvironment.resolve_href(env, self.env)
            # pylint: disable=too-many-function-args
            ui_doc["href"] = product_href(doc['type'], doc['id'])
        else:
            ui_doc["href"] = self.req.href(doc['type'], doc['id'])

        if doc['content']:
            ui_doc['content'] = shorten_result(doc['content'])

        if doc['time']:
            ui_doc['date'] = user_time(self.req, format_datetime, doc['time'])

        is_free_text_view = self.view is None
        if is_free_text_view:
            participant = self.allowed_participants[doc['type']]
            ui_doc['title'] = participant.format_search_results(doc)
        return ui_doc

    def render_widget(self, name, context, options):
        """Gather product list and render data in compact view
        """
        data = {}
        req = context.req
        title = ''
        params = ('max', 'cols')
        max_, cols = self.bind_params(name, options, *params)

        if not isinstance(self.env, ProductEnvironment):
            for p in Product.select(self.env):
                if 'PRODUCT_VIEW' in req.perm(Neighborhood('product', p.prefix)):
                    penv = ProductEnvironment(self.env, p.prefix)
                    phref = ProductEnvironment.resolve_href(penv, self.env)
                    for resource in (
                        {'type': Milestone, 'name': 'milestone', 'hrefurl': True},
                        {'type': Component, 'name': 'component'},
                        {'type': Version, 'name': 'version'},
                    ):
                        setattr(p, resource['name'] + 's',
                                self._get_product_info(p, phref, resource, max_))
                    p.owner_link = Query.from_string(self.env,
                        'status!=closed&col=id&col=summary&col=owner'
                        '&col=status&col=priority&order=priority'
                        '&group=product&owner=%s' % (p._data['owner'] or '', )
                    ).get_href(phref)
                    p.href = phref()
                    data.setdefault('product_list', []).append(p)
            title = _('Products')

        data['colseq'] = itertools.cycle(xrange(cols - 1, -1, -1)) if cols \
            else itertools.repeat(1)

        return 'widget_product.html', {
            'title': title,
            'data': data,
            'ctxtnav': [tag.a(_('More'), href=req.href('products'))],
        }, context

#.........这里部分代码省略.........
        product.delete()

        self.assertRaises(TracError, product.delete)

    def test_field_data_get(self):
        """tests that we can use table.field syntax to get to the field data"""
        prefix = self.default_data['prefix']
        name = self.default_data['name']
        description = self.default_data['description']
        product = list(Product.select(self.env, where={'prefix':prefix}))[0]
        self.assertEqual(prefix, product.prefix)
        self.assertEqual(name, product.name)
        self.assertEqual(description, product.description)

    def test_field_set(self):
        """tests that we can use table.field = something to set field data"""
        prefix = self.default_data['prefix']
        product = list(Product.select(self.env, where={'prefix':prefix}))[0]

        new_description = 'test change of description'
        product.description = new_description
        self.assertEqual(new_description, product.description)

    def test_missing_unique_fields(self):
        """ensure that that insert method works when _meta does not specify
        unique fields when inserting more than one ProductResourceMap instances
        """
        class TestModel(ModelBase):
            """A test model with no unique_fields"""
            _meta = {'table_name': 'bloodhound_testmodel',
                     'object_name': 'TestModelObject',
                     'key_fields': ['id',],
                     'non_key_fields': ['value'],
                     'unique_fields': [],}

        from trac.db import DatabaseManager
        schema = [TestModel._get_schema(), ]
        with self.env.db_transaction as db:
            db_connector, dummy = DatabaseManager(self.env)._get_connector()
            for table in schema:
                for statement in db_connector.to_sql(table):
                    db(statement)

        structure =  dict([(table.name, [col.name for col in table.columns])
                           for table in schema])
        tm1 = TestModel(self.env)
        tm1._data.update({'id':1, 'value':'value1'})
        tm1.insert()
        tm2 = TestModel(self.env)
        tm2._data.update({'id':2, 'value':'value2'})
        tm2.insert()

    def test_change_listener_created(self):
        self.assertEqual('created', self.listener.action)
        self.assertIsInstance(self.listener.resource, Product)
        self.assertEqual(self.INITIAL_PREFIX, self.prefix)
        self.assertEqual(self.INITIAL_NAME, self.name)
        self.assertEqual(self.INITIAL_DESCRIPTION, self.description)

    def test_change_listener_changed(self):
        CHANGED_NAME = "changed name"
        self.product.name = CHANGED_NAME
        self.product.update()
        self.assertEqual('changed', self.listener.action)
        self.assertIsInstance(self.listener.resource, Product)
        self.assertEqual(CHANGED_NAME, self.name)
        self.assertEqual({"name":self.INITIAL_NAME}, self.listener.old_values)

    def test_change_listener_deleted(self):
        self.product.delete()
        self.assertEqual('deleted', self.listener.action)
        self.assertIsInstance(self.listener.resource, Product)
        self.assertEqual(self.INITIAL_PREFIX, self.prefix)

    def test_get_tickets(self):
        for pdata in (
            {'prefix': 'p2', 'name':'product, too', 'description': ''},
            {'prefix': 'p3', 'name':'strike three', 'description': ''},
        ):
            num_tickets = 5
            product = Product(self.global_env)
            product._data.update(pdata)
            product.insert()
            self.env = ProductEnvironment(self.global_env, product)
            for i in range(num_tickets):
                ticket = Ticket(self.env)
                ticket['summary'] = 'hello ticket #%s-%d' % (product.prefix, i)
                ticket['reporter'] = 'admin'
                tid = ticket.insert()

            # retrieve tickets using both global and product scope
            tickets_from_global = [(t['product'], t['id']) for t in
                Product.get_tickets(self.global_env, product.prefix)]
            self.assertEqual(len(tickets_from_global), num_tickets)
            tickets_from_product = [(t['product'], t['id']) for t in
                Product.get_tickets(self.env)]
            self.assertEqual(len(tickets_from_product), num_tickets)
            # both lists should contain same elements
            intersection = set(tickets_from_global) & set(tickets_from_product)
            self.assertEqual(len(intersection), num_tickets)

class ProductMilestoneTestCase(MilestoneTestCase, MultiproductTestCase):
    def setUp(self):
        self.global_env = self._setup_test_env(create_folder=True)
        self._upgrade_mp(self.global_env)
        self._setup_test_log(self.global_env)
        self._load_product_from_data(self.global_env, self.default_product)

        self.env = ProductEnvironment(self.global_env, self.default_product)
        self._load_default_data(self.env)

    def tearDown(self):
        shutil.rmtree(self.global_env.path)
        self.global_env.reset_db()
        self.env = self.global_env = None

    @unittest.skipUnless(threading, 'Threading required for test')
    def test_milestone_threads(self):
        """ Ensure that in threaded (e.g. mod_wsgi) situations, we get
        an accurate list of milestones from Milestone.list

        The basic strategy is:
            thread-1 requests a list of milestones
            thread-2 adds a milestone
            thread-1 requests a new list of milestones
        To pass, thread-1 should have a list of milestones that matches
        those that are in the database.
        """
        lock = threading.RLock()
        results = []
        # two events to coordinate the workers and ensure that the threads
        # alternate appropriately
        e1 = threading.Event()
        e2 = threading.Event()

        def task(add):
            """the thread task - either we are discovering or adding events"""
            with lock:
                env = ProductEnvironment(self.global_env,
                                         self.default_product)
                if add:
                    name = 'milestone_from_' + threading.current_thread().name
                    milestone = Milestone(env)
                    milestone.name = name
                    milestone.insert()
                else:
                    # collect the names of milestones reported by Milestone and
                    # directly from the db - as sets to ease comparison later
                    results.append({
                        'from_t': set([m.name for m in Milestone.select(env)]),
                        'from_db': set(
                            [v[0] for v in self.env.db_query(
                                "SELECT name FROM milestone")])})

        def worker1():
            """ check milestones in this thread twice either side of ceding
            control to worker2
            """
            task(False)
            e1.set()
            e2.wait()
            task(False)

        def worker2():
            """ adds a milestone when worker1 allows us to then cede control
            back to worker1
            """
            e1.wait()
            task(True)
            e2.set()

        t1, t2 = [threading.Thread(target=f) for f in (worker1, worker2)]
        t1.start()
        t2.start()
        t1.join()
        t2.join()

        r = results[-1]  # note we only care about the final result
        self.assertEqual(r['from_t'], r['from_db'])

    def test_update_milestone(self):

        self.env.db_transaction("INSERT INTO milestone (name) VALUES ('Test')")

        milestone = Milestone(self.env, 'Test')
        t1 = datetime(2001, 01, 01, tzinfo=utc)
        t2 = datetime(2002, 02, 02, tzinfo=utc)
        milestone.due = t1
        milestone.completed = t2
        milestone.description = 'Foo bar'
        milestone.update()

        self.assertEqual(
            [('Test', to_utimestamp(t1), to_utimestamp(t2), 'Foo bar',
                    self.default_product)],
            self.env.db_query("SELECT * FROM milestone WHERE name='Test'"))

 def admin_url(prefix):
     env = ProductEnvironment.lookup_env(self.env, prefix)
     href = ProductEnvironment.resolve_href(env, self.env)
     return href.admin()