| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| from pylons import tmpl_context as c |
| from datetime import datetime |
| import urllib2 |
| |
| import mock |
| from ming.orm.ormsession import ThreadLocalORMSession |
| from ming.orm import session |
| from ming import schema |
| from nose.tools import ( |
| raises, |
| assert_equal, |
| assert_in, |
| assert_true, |
| assert_false, |
| ) |
| from forgetracker.model import Ticket, TicketAttachment |
| from forgetracker.tests.unit import TrackerTestWithModel |
| from forgetracker.import_support import ResettableStream |
| from allura.model import Feed, Post, User |
| from allura.lib import helpers as h |
| from allura.tests import decorators as td |
| |
| |
| class TestTicketModel(TrackerTestWithModel): |
| |
| def test_that_label_counts_are_local_to_tool(self): |
| """Test that label queries return only artifacts from the specified |
| tool. |
| """ |
| # create a ticket in two different tools, with the same label |
| from allura.tests import decorators as td |
| |
| @td.with_tool('test', 'Tickets', 'bugs', username='test-user') |
| def _test_ticket(): |
| return Ticket(ticket_num=1, summary="ticket1", labels=["mylabel"]) |
| |
| @td.with_tool('test', 'Tickets', 'bugs2', username='test-user') |
| def _test_ticket2(): |
| return Ticket(ticket_num=2, summary="ticket2", labels=["mylabel"]) |
| |
| # create and save the tickets |
| t1 = _test_ticket() |
| t2 = _test_ticket2() |
| ThreadLocalORMSession.flush_all() |
| |
| # test label query results |
| label_count1 = t1.artifacts_labeled_with( |
| "mylabel", t1.app_config).count() |
| label_count2 = t2.artifacts_labeled_with( |
| "mylabel", t2.app_config).count() |
| assert 1 == label_count1 == label_count2 |
| |
| def test_that_it_has_ordered_custom_fields(self): |
| custom_fields = dict(my_field='my value') |
| Ticket(summary='my ticket', custom_fields=custom_fields, ticket_num=3) |
| ThreadLocalORMSession.flush_all() |
| ticket = Ticket.query.get(summary='my ticket') |
| assert ticket.custom_fields == dict(my_field='my value') |
| |
| @raises(schema.Invalid) |
| def test_ticket_num_required(self): |
| Ticket(summary='my ticket') |
| |
| def test_ticket_num_required2(self): |
| t = Ticket(summary='my ticket', ticket_num=12) |
| try: |
| t.ticket_num = None |
| except schema.Invalid: |
| pass |
| else: |
| raise AssertionError('Expected schema.Invalid to be thrown') |
| |
| def test_activity_extras(self): |
| t = Ticket(summary='my ticket', ticket_num=12) |
| assert_in('allura_id', t.activity_extras) |
| assert_equal(t.activity_extras['summary'], t.summary) |
| |
| def test_has_activity_access(self): |
| t = Ticket(summary='ticket', ticket_num=666) |
| assert_true(t.has_activity_access('read', c.user, 'activity')) |
| t.deleted = True |
| assert_false(t.has_activity_access('read', c.user, 'activity')) |
| |
| def test_comment_has_activity_access(self): |
| t = Ticket(summary='ticket', ticket_num=666, deleted=True) |
| p = t.discussion_thread.add_post(text='test post') |
| assert_equal(p.status, 'ok') |
| assert_true(p.has_activity_access('read', c.user, 'activity')) |
| p.status = 'spam' |
| assert_false(p.has_activity_access('read', c.user, 'activity')) |
| p.status = 'pending' |
| assert_false(p.has_activity_access('read', c.user, 'activity')) |
| p.status = 'ok' |
| p.deleted = True |
| assert_false(p.has_activity_access('read', c.user, 'activity')) |
| |
| def test_private_ticket(self): |
| from allura.model import ProjectRole |
| from allura.model import ACE, DENY_ALL |
| from allura.lib.security import Credentials, has_access |
| from allura.websetup import bootstrap |
| |
| admin = c.user |
| creator = bootstrap.create_user('Not a Project Admin') |
| developer = bootstrap.create_user('Project Developer') |
| observer = bootstrap.create_user('Random Non-Project User') |
| anon = User(_id=None, username='*anonymous', |
| display_name='Anonymous') |
| t = Ticket(summary='my ticket', ticket_num=3, |
| reported_by_id=creator._id) |
| |
| assert creator == t.reported_by |
| role_admin = ProjectRole.by_name('Admin')._id |
| role_developer = ProjectRole.by_name('Developer')._id |
| role_creator = ProjectRole.by_user(t.reported_by, upsert=True)._id |
| ProjectRole.by_user( |
| developer, upsert=True).roles.append(role_developer) |
| ThreadLocalORMSession.flush_all() |
| cred = Credentials.get().clear() |
| |
| t.private = True |
| assert_equal(t.acl, [ |
| ACE.allow(role_developer, 'save_searches'), |
| ACE.allow(role_developer, 'read'), |
| ACE.allow(role_developer, 'create'), |
| ACE.allow(role_developer, 'update'), |
| ACE.allow(role_developer, 'unmoderated_post'), |
| ACE.allow(role_developer, 'post'), |
| ACE.allow(role_developer, 'moderate'), |
| ACE.allow(role_developer, 'delete'), |
| ACE.allow(role_creator, 'read'), |
| ACE.allow(role_creator, 'post'), |
| ACE.allow(role_creator, 'create'), |
| ACE.allow(role_creator, 'unmoderated_post'), |
| DENY_ALL]) |
| assert has_access(t, 'read', user=admin)() |
| assert has_access(t, 'create', user=admin)() |
| assert has_access(t, 'update', user=admin)() |
| assert has_access(t, 'read', user=creator)() |
| assert has_access(t, 'post', user=creator)() |
| assert has_access(t, 'unmoderated_post', user=creator)() |
| assert has_access(t, 'create', user=creator)() |
| assert not has_access(t, 'update', user=creator)() |
| assert has_access(t, 'read', user=developer)() |
| assert has_access(t, 'create', user=developer)() |
| assert has_access(t, 'update', user=developer)() |
| assert not has_access(t, 'read', user=observer)() |
| assert not has_access(t, 'create', user=observer)() |
| assert not has_access(t, 'update', user=observer)() |
| assert not has_access(t, 'read', user=anon)() |
| assert not has_access(t, 'create', user=anon)() |
| assert not has_access(t, 'update', user=anon)() |
| |
| t.private = False |
| assert t.acl == [] |
| assert has_access(t, 'read', user=admin)() |
| assert has_access(t, 'create', user=admin)() |
| assert has_access(t, 'update', user=admin)() |
| assert has_access(t, 'read', user=developer)() |
| assert has_access(t, 'create', user=developer)() |
| assert has_access(t, 'update', user=developer)() |
| assert has_access(t, 'read', user=creator)() |
| assert has_access(t, 'unmoderated_post', user=creator)() |
| assert has_access(t, 'create', user=creator)() |
| assert not has_access(t, 'update', user=creator)() |
| assert has_access(t, 'read', user=observer)() |
| assert has_access(t, 'read', user=anon)() |
| |
| def test_feed(self): |
| t = Ticket( |
| app_config_id=c.app.config._id, |
| ticket_num=1, |
| summary='test ticket', |
| description='test description', |
| created_date=datetime(2012, 10, 29, 9, 57, 21, 465000)) |
| assert_equal(t.created_date, datetime(2012, 10, 29, 9, 57, 21, 465000)) |
| f = Feed.post( |
| t, |
| title=t.summary, |
| description=t.description, |
| pubdate=t.created_date) |
| assert_equal(f.pubdate, datetime(2012, 10, 29, 9, 57, 21, 465000)) |
| assert_equal(f.title, 'test ticket') |
| assert_equal(f.description, |
| '<div class="markdown_content"><p>test description</p></div>') |
| |
| @td.with_tool('test', 'Tickets', 'bugs', username='test-user') |
| @td.with_tool('test', 'Tickets', 'bugs2', username='test-user') |
| def test_ticket_move(self): |
| app1 = c.project.app_instance('bugs') |
| app2 = c.project.app_instance('bugs2') |
| with h.push_context(c.project._id, app_config_id=app1.config._id): |
| ticket = Ticket.new() |
| ticket.summary = 'test ticket' |
| ticket.description = 'test description' |
| ticket.assigned_to_id = User.by_username('test-user')._id |
| ticket.discussion_thread.add_post(text='test comment') |
| |
| assert_equal( |
| Ticket.query.find({'app_config_id': app1.config._id}).count(), 1) |
| assert_equal( |
| Ticket.query.find({'app_config_id': app2.config._id}).count(), 0) |
| assert_equal( |
| Post.query.find(dict(thread_id=ticket.discussion_thread._id)).count(), 1) |
| |
| t = ticket.move(app2.config) |
| assert_equal( |
| Ticket.query.find({'app_config_id': app1.config._id}).count(), 0) |
| assert_equal( |
| Ticket.query.find({'app_config_id': app2.config._id}).count(), 1) |
| assert_equal(t.summary, 'test ticket') |
| assert_equal(t.description, 'test description') |
| assert_equal(t.assigned_to.username, 'test-user') |
| assert_equal(t.url(), '/p/test/bugs2/1/') |
| |
| post = Post.query.find(dict(thread_id=ticket.discussion_thread._id, |
| text={'$ne': 'test comment'})).first() |
| assert post is not None, 'No comment about ticket moving' |
| message = 'Ticket moved from /p/test/bugs/1/' |
| assert_equal(post.text, message) |
| |
| post = Post.query.find(dict(text='test comment')).first() |
| assert_equal(post.thread.discussion_id, app2.config.discussion_id) |
| assert_equal(post.thread.app_config_id, app2.config._id) |
| assert_equal(post.app_config_id, app2.config._id) |
| |
| @td.with_tool('test', 'Tickets', 'bugs', username='test-user') |
| @td.with_tool('test', 'Tickets', 'bugs2', username='test-user') |
| def test_ticket_move_with_different_custom_fields(self): |
| app1 = c.project.app_instance('bugs') |
| app2 = c.project.app_instance('bugs2') |
| app1.globals.custom_fields.extend([ |
| {'name': '_test', 'type': 'string', 'label': 'Test field'}, |
| {'name': '_test2', 'type': 'string', 'label': 'Test field 2'}]) |
| app2.globals.custom_fields.append( |
| {'name': '_test', 'type': 'string', 'label': 'Test field'}) |
| ThreadLocalORMSession.flush_all() |
| ThreadLocalORMSession.close_all() |
| with h.push_context(c.project._id, app_config_id=app1.config._id): |
| ticket = Ticket.new() |
| ticket.summary = 'test ticket' |
| ticket.description = 'test description' |
| ticket.custom_fields['_test'] = 'test val' |
| ticket.custom_fields['_test2'] = 'test val 2' |
| |
| t = ticket.move(app2.config) |
| assert_equal(t.summary, 'test ticket') |
| assert_equal(t.description, 'test description') |
| assert_equal(t.custom_fields['_test'], 'test val') |
| post = Post.query.find( |
| dict(thread_id=ticket.discussion_thread._id)).first() |
| assert post is not None, 'No comment about ticket moving' |
| message = 'Ticket moved from /p/test/bugs/1/' |
| message += '\n\nCan\'t be converted:\n' |
| message += '\n- **_test2**: test val 2' |
| assert_equal(post.text, message) |
| |
| @td.with_tool('test', 'Tickets', 'bugs', username='test-user') |
| @td.with_tool('test', 'Tickets', 'bugs2', username='test-user') |
| def test_ticket_move_with_users_not_in_project(self): |
| app1 = c.project.app_instance('bugs') |
| app2 = c.project.app_instance('bugs2') |
| app1.globals.custom_fields.extend([ |
| {'name': '_user_field', 'type': 'user', 'label': 'User field'}, |
| {'name': '_user_field_2', 'type': 'user', 'label': 'User field 2'}]) |
| app2.globals.custom_fields.extend([ |
| {'name': '_user_field', 'type': 'user', 'label': 'User field'}, |
| {'name': '_user_field_2', 'type': 'user', 'label': 'User field 2'}]) |
| ThreadLocalORMSession.flush_all() |
| ThreadLocalORMSession.close_all() |
| from allura.websetup import bootstrap |
| bootstrap.create_user('test-user-0') |
| with h.push_context(c.project._id, app_config_id=app1.config._id): |
| ticket = Ticket.new() |
| ticket.summary = 'test ticket' |
| ticket.description = 'test description' |
| ticket.custom_fields['_user_field'] = 'test-user' # in project |
| # not in project |
| ticket.custom_fields['_user_field_2'] = 'test-user-0' |
| # not in project |
| ticket.assigned_to_id = User.by_username('test-user-0')._id |
| |
| t = ticket.move(app2.config) |
| assert_equal(t.assigned_to_id, None) |
| assert_equal(t.custom_fields['_user_field'], 'test-user') |
| assert_equal(t.custom_fields['_user_field_2'], '') |
| post = Post.query.find( |
| dict(thread_id=ticket.discussion_thread._id)).first() |
| assert post is not None, 'No comment about ticket moving' |
| message = 'Ticket moved from /p/test/bugs/1/' |
| message += '\n\nCan\'t be converted:\n' |
| message += '\n- **_user_field_2**: test-user-0 (user not in project)' |
| message += '\n- **assigned_to**: test-user-0 (user not in project)' |
| assert_equal(post.text, message) |
| |
| @td.with_tool('test', 'Tickets', 'bugs', username='test-user') |
| def test_attach_with_resettable_stream(self): |
| with h.push_context(c.project._id, app_config_id=c.app.config._id): |
| ticket = Ticket.new() |
| ticket.summary = 'test ticket' |
| ticket.description = 'test description' |
| assert_equal(len(ticket.attachments), 0) |
| f = urllib2.urlopen('file://%s' % __file__) |
| TicketAttachment.save_attachment( |
| 'test_ticket_model.py', ResettableStream(f), |
| artifact_id=ticket._id) |
| ThreadLocalORMSession.flush_all() |
| # need to refetch since attachments are cached |
| session(ticket).expunge(ticket) |
| ticket = Ticket.query.get(_id=ticket._id) |
| assert_equal(len(ticket.attachments), 1) |
| assert_equal(ticket.attachments[0].filename, 'test_ticket_model.py') |
| |
| def test_json_parents(self): |
| ticket = Ticket.new() |
| json_keys = ticket.__json__().keys() |
| assert_in('related_artifacts', json_keys) # from Artifact |
| assert_in('votes_up', json_keys) # VotableArtifact |
| assert_in('ticket_num', json_keys) # Ticket |
| assert ticket.__json__()['assigned_to'] is None |
| |
| @mock.patch('forgetracker.model.ticket.tsearch') |
| @mock.patch.object(Ticket, 'paged_search') |
| @mock.patch.object(Ticket, 'paged_query') |
| def test_paged_query_or_search(self, query, search, tsearch): |
| app_cfg, user = mock.Mock(), mock.Mock() |
| mongo_query = 'mongo query' |
| solr_query = 'solr query' |
| kw = {'kw1': 'test1', 'kw2': 'test2'} |
| filter = None |
| Ticket.paged_query_or_search(app_cfg, user, mongo_query, solr_query, filter, **kw) |
| query.assert_called_once_with(app_cfg, user, mongo_query, sort=None, limit=None, page=0, **kw) |
| assert_equal(tsearch.query_filter_choices.call_count, 1) |
| assert_equal(tsearch.query_filter_choices.call_args[0][0], 'solr query') |
| assert_equal(search.call_count, 0) |
| query.reset_mock(), search.reset_mock(), tsearch.reset_mock() |
| |
| filter = {'status': 'unread'} |
| Ticket.paged_query_or_search(app_cfg, user, mongo_query, solr_query, filter, **kw) |
| search.assert_called_once_with(app_cfg, user, solr_query, filter=filter, sort=None, limit=None, page=0, **kw) |
| assert_equal(query.call_count, 0) |
| assert_equal(tsearch.query_filter_choices.call_count, 0) |