def test_contextual_feeds(self):
        """Test that team/project/user Atom feeds appear as <link> tags."""

        with self.app.app_context():
            user(email='[email protected]', save=True)
            u = user(username='buffy', email="[email protected]",
                     name='Buffy Summers', slug='buffy', save=True)
            team(name='Scooby Gang', slug='scoobies', users=[u], save=True)
            project(name='Kill The Master', slug='master', save=True)

        authenticate(self.client, u)

        site_url = self.app.config.get('SITE_URL')

        rv = self.client.get('/')
        assert ('<link rel="alternate" type="application/atom+xml" '
                'href="%s/statuses.xml"') % site_url in rv.data
        assert ('<link rel="alternate" type="application/atom+xml" '
                'href="%s/project/') % site_url not in rv.data

        rv = self.client.get('/team/scoobies')
        assert ('<link rel="alternate" type="application/atom+xml" '
                'href="%s/team/') % site_url in rv.data

        rv = self.client.get('/project/master')
        assert ('<link rel="alternate" type="application/atom+xml" '
                'href="%s/project/') % site_url in rv.data

        rv = self.client.get('/user/buffy')
        assert ('<link rel="alternate" type="application/atom+xml" '
                'href="%s/user/') % site_url in rv.data

    def test_timeline_filters_project(self):
        """Test the timeline only shows the passed in project."""
        with self.app.app_context():
            u = user(save=True)
            p = project(save=True)
            status(user=u, project=p, save=True)
            p2 = project(name="Test Project 2", slug="test-project-2", save=True)
            status(user=u, project=p2, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(len(data), 1)
        eq_(data[0]["project"], p.dictify())

    def test_timeline_count(self):
        """Test the count parameter of home_timeline"""
        self.app.config['API2_TIMELINE_MAX_RESULTS'] = 50
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            for i in range(60):
                status(project=p, user=u, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(len(data), 20)

        # Test with an acceptable count
        response = self.client.get(self._url(dict(count=50)))
        data = json.loads(response.data)
        eq_(len(data), 50)

        # Test with a count that is too large
        response = self.client.get(self._url(dict(count=60)))
        eq_(response.status_code, 400)

        # Test with a count that is too small
        response = self.client.get(self._url(dict(count=0)))
        eq_(response.status_code, 400)

        # Test with an invalid count
        response = self.client.get(self._url(dict(count='a')))
        eq_(response.status_code, 400)

    def test_timeline_since_id(self):
        """Test the since_id parameter of home_timeline"""
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            for i in range(30):
                status(project=p, user=u, save=True)

        response = self.client.get(self._url(dict(since_id=10, count=20)))
        data = json.loads(response.data)
        eq_(data[19]['id'], 11)

        response = self.client.get(self._url(dict(since_id=10, count=10)))
        data = json.loads(response.data)
        eq_(data[9]['id'], 21)

        response = self.client.get(self._url(dict(since_id=10, count=30)))
        data = json.loads(response.data)
        eq_(len(data), 20)
        eq_(data[19]['id'], 11)

        response = self.client.get(self._url(dict(since_id=0)))
        eq_(response.status_code, 400)

        response = self.client.get(self._url(dict(since_id='a')))
        eq_(response.status_code, 400)

 def test_timeline(self):
     """Test the home_timeline endpoint"""
     with self.app.app_context():
         u = user(save=True, team={})
         p = project(save=True)
         status(user=u, project=p, save=True)
     response = self.client.get(self._url())
     eq_(response.status_code, 200)
     eq_(response.content_type, 'application/json')

    def test_project_view(self):
        """Make sure the project view works like it's supposed to."""
        with self.app.app_context():
            p = project(save=True)

        response = self.client.get('/project/%s' % p.slug)
        eq_(response.status_code, 200)

        response = self.client.get('/project/not-a-real-project')
        eq_(response.status_code, 404)

    def test_paginate(self):
        """Test the paginate helper function."""
        db = get_session(self.app)

        statuses = []
        with self.app.app_context():
            p = project(save=True)
            u = user(save=True)

            # Create 100 statuses
            for i in range(30):
                statuses.append(status(project=p, user=u,
                                       created=datetime(2012, 5, 25),
                                       save=True))

            for i in range(30):
                statuses.append(status(project=p, user=u,
                                       created=datetime(2012, 6, 25),
                                       save=True))

            for i in range(40):
                statuses.append(status(project=p, user=u,
                                       created=datetime(2012, 7, 25),
                                       save=True))

        s = db.query(Status).order_by(Status.id)

        # Test simple pagination
        page = paginate(s, page=1)

        eq_(page.pages, 5)

        eq_(page.has_prev, False)
        eq_(page.has_next, True)

        page = paginate(s, page=3)
        eq_(page.has_prev, True)
        eq_(page.has_next, True)

        page = paginate(s, page=5)
        eq_(page.has_prev, True)
        eq_(page.has_next, False)

        # Test date filtered pagination
        page = paginate(s, page=1, startdate=datetime(2012, 5, 28))
        eq_(page.pages, 4)

        page = paginate(s, page=1, startdate=datetime(2012, 5, 28),
                        enddate=datetime(2012, 6, 28))
        eq_(page.pages, 2)

        page = paginate(s, page=1, enddate=datetime(2012, 6, 28))
        eq_(page.pages, 3)

    def test_status_with_project(self):
        """Test posting a status with a project."""
        with self.app.app_context():
            u = user(email='[email protected]', save=True)
            p = project(name='blackhole', slug='blackhole', save=True)
            data = {'message': 'r1cky rocks!', 'project': p.id}

        authenticate(self.client, u)

        rv = self.client.post('/statusize/', follow_redirects=True,
                              data=data)
        eq_(rv.status_code, 200)

    def test_status_replies(self):
        """Test the loading of replies for a status."""
        with self.app.app_context():
            p = project(save=True)
            u = user(save=True)
            s = status(project=p, user=u, save=True)

            for i in range(30):
                status(project=p, user=u, reply_to=s, save=True)

            page = s.replies()

        eq_(page.pages, 2)

    def test_timeline_trim_project(self):
        """Test the trim_project parameter of home_timeline"""
        with self.app.app_context():
            p = project(save=True)
            status(project=p, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(data[0]["project"], p.dictify())

        response = self.client.get(self._url(dict(trim_project=1)))
        data = json.loads(response.data)
        eq_(data[0]["project"], p.id)

    def test_timeline_trim_user(self):
        """Test the trim_user parameter of home_timeline"""
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            status(user=u, project=p, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(data[0]['user'], u.dictify())

        response = self.client.get(self._url(dict(trim_user=1)))
        data = json.loads(response.data)
        eq_(data[0]['user'], u.id)

    def test_timeline_filters_team(self):
        """Test the timeline only shows the passed in team."""
        with self.app.app_context():
            u = user(save=True, team={})
            u2 = user(
                username="janedoe", email="[email protected]", slug="janedoe", save=True, team={"name": "XXX", "slug": "xxx"}
            )
            p = project(save=True)
            status(user=u, project=p, save=True)
            status(user=u2, project=p, save=True)

        response = self.client.get(self._url(dict(team_id=u.teams[0].id)))
        data = json.loads(response.data)
        eq_(len(data), 1)
        eq_(data[0]["user"], u.dictify())

    def test_timeline_filters_team(self):
        """Test the timeline only shows the passed in team."""
        with self.app.app_context():
            u = user(save=True, team={})
            u2 = user(username='janedoe', email='[email protected]',
                      slug='janedoe', save=True, team={'name': 'XXX',
                                                       'slug': 'xxx'})
            p = project(save=True)
            status(user=u, project=p, save=True)
            status(user=u2, project=p, save=True)

        response = self.client.get(self._url(dict(team_id=u.teams[0].id)))
        data = json.loads(response.data)
        eq_(len(data), 1)
        eq_(data[0]['user'], u.dictify())

    def test_timeline_include_replies(self):
        """Test the include_replies parameter of home_timeline"""
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            for i in range(10):
                s = status(project=p, user=u, save=True)
            for i in range(10):
                status(project=p, user=u, reply_to=s, save=True)

        response = self.client.get(self._url())
        data = json.loads(response.data)
        eq_(len(data), 10)

        response = self.client.get(self._url(dict(include_replies=1)))
        data = json.loads(response.data)
        eq_(len(data), 20)

    def test_project_recent_statuses(self):
        """Test loading of recent statuses for a project."""
        with self.app.app_context():
            p = project(save=True)
            u = user(save=True)

            # Create 70 statuses
            for i in range(70):
                status(project=p, user=u, save=True)

            s = status(project=p, user=u, save=True)

            # Create 30 replies
            for i in range(30):
                status(project=p, user=u, reply_to=s, save=True)

        # Should not include replies
        page = p.recent_statuses()
        eq_(page.pages, 4)

    def test_timeline_week(self):
        """Test the week parameter of home_timeline"""
        year = 2014
        month = 5
        day = 23
        date_str = "%04d-%02d-%02d" % (year, month, day)
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            d = datetime(year, month, day)
            s = status(user=u, project=p, save=True, created=d)

        # A badly formatted date should raise ApiError (returning 400)
        response = self.client.get(self._url(dict(week="May 23, 2014")))
        eq_(response.status_code, 400)

        response = self.client.get(self._url(dict(week=date_str)))
        eq_(response.status_code, 200)
        data = json.loads(response.data)
        ok_(data[0]['created'].startswith(date_str))

 def test_format_update(self):
     content = ("#merge pull #1 and pR 2 to fix bug #3 and BUg 4 by "
                "@jezdez for @r1cky but not [email protected]")
     with self.app.app_context():
         user(username='r1cky', slug='r1cky', save=True)
         user(username='jezdez', slug='jezdez',
              email='[email protected]',
              github_handle='jezdez',
              save=True)
         p = project(name='mdndev', slug='mdndev',
                     repo_url='https://github.com/mozilla/kuma')
         formatted_update = format_update(content, project=p)
     ok_('tag-merge' in formatted_update)
     ok_('pull/1' in formatted_update)
     ok_('pull/2' in formatted_update)
     ok_('show_bug.cgi?id=3' in formatted_update)
     ok_('show_bug.cgi?id=4' in formatted_update)
     ok_('/user/jezdez' in formatted_update)
     ok_('/user/r1cky' in formatted_update)
     ok_('/user/willkg' not in formatted_update)
     ok_('/user/mozilla.com' not in formatted_update)

    def test_feeds(self):
        """Test that the site-wise Atom feed appears and functions properly."""
        with self.app.app_context():
            u = user(email='[email protected]', slug='joe', save=True)
            team(users=[u], slug='a-team', save=True)
            p = project(slug='prjkt', save=True)

            for i in range(20):
                status(user=u, project=p, content='foo', content_html='foo',
                       save=True)

        authenticate(self.client, u)

        site_url = self.app.config.get('SITE_URL')

        # Ensure the Atom link appears in the rendered HTML.
        rv = self.client.get('/')
        assert ('<link rel="alternate" type="application/atom+xml" '
                'href="%s/statuses.xml"') % site_url in rv.data

        # Make sure the Atom feed displays statuses.
        rv = self.client.get('/statuses.xml')
        assert '<entry' in rv.data
        assert ('<content type="html">&lt;h3&gt;Test Project&lt;/h3&gt;&lt;p'
                '&gt;foo&lt;/p&gt;</content>') in rv.data

        rv = self.client.get('/user/joe.xml')
        assert '<entry' in rv.data
        assert ('<content type="html">&lt;h3&gt;Test Project&lt;/h3&gt;&lt;p'
                '&gt;foo&lt;/p&gt;</content>') in rv.data

        rv = self.client.get('/project/prjkt.xml')
        assert '<entry' in rv.data
        assert ('<content type="html">&lt;h3&gt;Test Project&lt;/h3&gt;&lt;p'
                '&gt;foo&lt;/p&gt;</content>') in rv.data

        rv = self.client.get('/team/a-team.xml')
        assert '<entry' in rv.data
        assert ('<content type="html">&lt;h3&gt;Test Project&lt;/h3&gt;&lt;p'
                '&gt;foo&lt;/p&gt;</content>') in rv.data

    def test_timeline_max_id(self):
        """Test the max_id parameter of home_timeline"""
        with self.app.app_context():
            u = user(save=True, team={})
            p = project(save=True)
            for i in range(30):
                status(project=p, user=u, save=True)

        response = self.client.get(self._url(dict(max_id=10, count=20)))
        data = json.loads(response.data)
        eq_(len(data), 10)
        eq_(data[0]["id"], 10)

        response = self.client.get(self._url(dict(max_id=10, since_id=5)))
        data = json.loads(response.data)
        eq_(len(data), 5)

        response = self.client.get(self._url(dict(max_id=0)))
        eq_(response.status_code, 400)

        response = self.client.get(self._url(dict(max_id="a")))
        eq_(response.status_code, 400)

    def test_get_statuses(self):
        """Test getting statuses from API"""
        statuses = []

        with self.app.app_context():
            u = user(save=True)
            p = project(save=True)

            # Create 30 dummy statuses
            for i in range(30):
                if i > 25:
                    p = None
                statuses.append(status(user=u, project=p, save=True))

        # Test with no limit
        response = self.client.get('api/v1/feed/')
        eq_(response.status_code, 200)
        feed = load_json(response.data)

        # Ensure that the default limit works
        eq_(len(feed), 20)

        # Ensure that the correct item are first and last
        eq_(feed.keys()[0], '30', 'First item incorrect')
        eq_(feed.keys()[19], '11', 'Last item incorrect')

        # Test with no limit
        response = self.client.get('api/v1/feed/', query_string={'limit': 5})
        eq_(response.status_code, 200)
        feed = load_json(response.data)

        # Ensure that the default limit works
        eq_(len(feed), 5)

        # Test with no limit
        eq_(feed.keys()[0], '30', 'First item incorrect')
        eq_(feed.keys()[4], '26', 'Last item incorrect')